diff --git a/helpers.md b/helpers.md index 54902af655..9b4518f49c 100644 --- a/helpers.md +++ b/helpers.md @@ -633,3 +633,24 @@ This only selects the local parser; it does not change the session's schema. `result.parse(Report)` parses an existing raw result. `AgentOutputParseError.result` retains the completed raw answer if validation fails. With `AsyncOpenAI`, await creation and the result getter, and use `async with`. + +### Reattach tool handlers + +Omit `input` to reconnect to an existing session without submitting another prompt. + +```python +with client.beta.agents.sessions.stream( + saved_session_id, tool_handlers={"lookup_order": lookup_order}, +) as stream: + result = stream.get_final_result() +print(result.output_text) +``` + +Pending function calls use the same handlers as a new turn. Result collection +recovers the selected turn's completed output, including messages missed while +disconnected. An already-idle session with no selected turn drains successfully +with `until_done()` but has no final result. Reattachment uses at-least-once +tool-call delivery with application-owned recovery: after process death, restart +the worker, recover its saved session ID, and reattach handlers. An unacknowledged +call may be delivered again; applications own idempotency for handler mutations. +With `AsyncOpenAI`, use `async with` and await the getter. diff --git a/src/openai/lib/beta/agents/_attachment.py b/src/openai/lib/beta/agents/_attachment.py new file mode 100644 index 0000000000..0df29211da --- /dev/null +++ b/src/openai/lib/beta/agents/_attachment.py @@ -0,0 +1,503 @@ +from __future__ import annotations + +from copy import deepcopy +from typing import TYPE_CHECKING, Iterable, Iterator, AsyncIterator + +import anyio + +from ._result import AgentTurnResultError, AgentTurnResultCollector +from ...._types import omit +from ...._streaming import Stream, AsyncStream +from ....types.beta.agent_session import AgentSession +from ....types.beta.agent_session_item import AgentSessionItem +from ....types.beta.agent_session_event import AgentSessionEvent +from ....types.beta.agents.sessions.turn import Turn +from ....types.beta.agent_session_message import AgentSessionMessage + +if TYPE_CHECKING: + from ...streaming.agents._streams import _RequestOptions + from ....resources.beta.agents.sessions.sessions import Sessions, AsyncSessions + +_TERMINAL = ("completed", "failed", "cancelled") + + +class AgentSessionAttachment: + """Turn identity and bounded observation state for a single attachment.""" + + def __init__(self, session_id: str, turn: Turn | None = None, baseline_id: str | None = None) -> None: + self.session_id = session_id + self.turn = turn + self.baseline_id = baseline_id + self.settled = False + self.failed = False + self.reconciled = False + self.observation_interrupted = False + self.messages_read = False + self.manual_diagnostics = False + self.pending_messages: list[AgentSessionEvent] | None = None + + def select(self, turn: Turn) -> None: + if ( + self.turn is None + and turn.session_id == self.session_id + and turn.subagent_id is None + and not (turn.id == self.baseline_id and turn.status in _TERMINAL) + ): + self.turn = deepcopy(turn) + + def refresh(self, turn: Turn) -> None: + if self.turn is None: + self.select(turn) + elif turn.id == self.turn.id and self.turn.status not in _TERMINAL: + self.turn = deepcopy(turn) + + def settle(self, session: AgentSession) -> None: + terminal = self.turn is not None and self.turn.status in _TERMINAL + # Session status can already describe a successor turn. Exact selected + # turn state is authoritative for this attachment's completion boundary. + self.failed = session.status == "failed" and not terminal + self.settled = terminal or self.failed or session.status == "idle" and self.turn is None + + def seed(self, collector: AgentTurnResultCollector) -> None: + if self.turn is not None and ( + collector.turn is None + or collector.turn.id == self.turn.id + and collector.turn.status not in _TERMINAL + and self.turn.status in _TERMINAL + ): + collector.turn = deepcopy(self.turn) + if self.turn.status in _TERMINAL: + collector.required_actions = [] + if self.turn is not None and self.pending_messages is not None: + for event in self.pending_messages: + collector.accept(event) + self.pending_messages = None + if self.settled: + collector.boundary = True + if self.failed: + collector.session_failed = True + + def candidate(self, event: AgentSessionEvent) -> str | None: + if ( + event.type == "agent.session.turn.created" + or event.type == "agent.session.turn.completed" + or event.type == "agent.session.turn.failed" + or event.type == "agent.session.turn.cancelled" + ): + if self.turn is not None and event.turn.subagent_id is None and event.turn.id != self.turn.id: + self.settled = True + else: + self.select(event.turn) + return None + if self.turn is not None: + # Application function calls belong to root turns. A new root's + # replay is a boundary, even if the previous turn's GET still lags. + if ( + event.type == "agent.session.turn.item.added" + and event.item.type == "function_call" + and event.item.turn_id != self.turn.id + ): + self.settled = True + return None + turn_id = getattr(event, "turn_id", None) + if not isinstance(turn_id, str): + turn_id = getattr(getattr(event, "item", None), "turn_id", None) + if isinstance(turn_id, str): + return turn_id + return None + + def collect(self, event: AgentSessionEvent, collector: AgentTurnResultCollector) -> None: + # Only opted-in final output is retained while the turn index catches up. + # Pending function calls are never reconstructed from this buffer. + if self.turn is None and event.type == "agent.session.turn.item.done" and event.item.type == "message": + if event.item.status == "completed" and event.item.phase != "commentary": + if self.pending_messages is None: + self.pending_messages = [] + self.pending_messages.append(deepcopy(event)) + else: + collector.accept(event) + + +def _latest_root(turns: Iterable[Turn]) -> Turn | None: + for turn in turns: + if turn.subagent_id is None: + return turn + return None + + +def select_candidate( + sessions: Sessions, state: AgentSessionAttachment, candidate: str, options: _RequestOptions +) -> None: + latest = _latest_root(sessions.turns.list(state.session_id, order="desc", **options)) + if latest is not None and latest.id == candidate: + state.select(latest) + + +async def async_select_candidate( + sessions: AsyncSessions, state: AgentSessionAttachment, candidate: str, options: _RequestOptions +) -> None: + async for turn in sessions.turns.list(state.session_id, order="desc", **options): + if turn.subagent_id is None: + if turn.id == candidate: + state.select(turn) + return + + +def refresh_idle(sessions: Sessions, state: AgentSessionAttachment, options: _RequestOptions) -> None: + session = sessions.retrieve(state.session_id, **options) + if state.turn is not None: + state.refresh(sessions.turns.retrieve(state.turn.id, session_id=state.session_id, **options)) + else: + latest = _latest_root(sessions.turns.list(state.session_id, order="desc", **options)) + if latest is not None: + state.select(latest) + state.settle(session) + + +async def async_refresh_idle(sessions: AsyncSessions, state: AgentSessionAttachment, options: _RequestOptions) -> None: + session = await sessions.retrieve(state.session_id, **options) + if state.turn is not None: + state.refresh(await sessions.turns.retrieve(state.turn.id, session_id=state.session_id, **options)) + else: + async for turn in sessions.turns.list(state.session_id, order="desc", **options): + if turn.subagent_id is None: + state.select(turn) + break + state.settle(session) + + +def attach( + sessions: Sessions, session_id: str, options: _RequestOptions +) -> tuple[Stream[AgentSessionEvent], AgentSessionAttachment]: + baseline = _latest_root(sessions.turns.list(session_id, order="desc", **options)) + turn = baseline if baseline is not None and baseline.status not in _TERMINAL else None + state = AgentSessionAttachment(session_id, turn, baseline.id if baseline is not None else None) + stream = sessions.events.stream(session_id, **options) + try: + # Subscribe before refreshing state so completion during attachment cannot + # leave us waiting for a terminal event that predates the subscription. + session = sessions.retrieve(session_id, **options) + if state.turn is not None: + state.turn = sessions.turns.retrieve(state.turn.id, session_id=session_id, **options) + else: + latest = _latest_root(sessions.turns.list(session_id, order="desc", **options)) + _select_refreshed(state, latest, baseline) + if session.status == "idle" and state.turn is not None and state.turn.status not in _TERMINAL: + state.refresh(sessions.turns.retrieve(state.turn.id, session_id=session_id, **options)) + state.settle(session) + state.manual_diagnostics = _needs_manual_diagnostics(state, session) + return stream, state + except BaseException: + stream.close() + raise + + +async def async_attach( + sessions: AsyncSessions, session_id: str, options: _RequestOptions +) -> tuple[AsyncStream[AgentSessionEvent], AgentSessionAttachment]: + async def latest_root() -> Turn | None: + async for turn in sessions.turns.list(session_id, order="desc", **options): + if turn.subagent_id is None: + return turn + return None + + baseline = await latest_root() + turn = baseline if baseline is not None and baseline.status not in _TERMINAL else None + state = AgentSessionAttachment(session_id, turn, baseline.id if baseline is not None else None) + stream = await sessions.events.stream(session_id, **options) + try: + session = await sessions.retrieve(session_id, **options) + if state.turn is not None: + state.turn = await sessions.turns.retrieve(state.turn.id, session_id=session_id, **options) + else: + _select_refreshed(state, await latest_root(), baseline) + if session.status == "idle" and state.turn is not None and state.turn.status not in _TERMINAL: + state.refresh(await sessions.turns.retrieve(state.turn.id, session_id=session_id, **options)) + state.settle(session) + state.manual_diagnostics = _needs_manual_diagnostics(state, session) + return stream, state + except BaseException: + # Failed entry leaves the caller without a stream to close. + with anyio.CancelScope(shield=True): + await stream.close() + raise + + +def _select_refreshed(state: AgentSessionAttachment, latest: Turn | None, baseline: Turn | None) -> None: + # A newly visible root can have finished during subscription. An unchanged + # completed root is historical and does not identify the observed work. + if latest is not None and (latest.status not in _TERMINAL or baseline is None or latest.id != baseline.id): + state.select(latest) + + +def _needs_manual_diagnostics(state: AgentSessionAttachment, session: AgentSession) -> bool: + return ( + (state.turn is None or state.turn.status == "waiting") + and session.status == "requires_action" + and any(action.type != "function_call" for action in session.required_actions) + ) + + +def diagnose_manual( + sessions: Sessions, state: AgentSessionAttachment, collector: AgentTurnResultCollector, options: _RequestOptions +) -> None: + if not state.manual_diagnostics: + return + state.manual_diagnostics = False + session = sessions.retrieve(state.session_id, **options) + latest = _latest_root(sessions.turns.list(state.session_id, order="desc", **options)) + _manual_diagnostics(state, session, latest, collector) + + +async def async_diagnose_manual( + sessions: AsyncSessions, + state: AgentSessionAttachment, + collector: AgentTurnResultCollector, + options: _RequestOptions, +) -> None: + if not state.manual_diagnostics: + return + state.manual_diagnostics = False + session = await sessions.retrieve(state.session_id, **options) + latest = None + async for turn in sessions.turns.list(state.session_id, order="desc", **options): + if turn.subagent_id is None: + latest = turn + break + _manual_diagnostics(state, session, latest, collector) + + +def _manual_diagnostics( + state: AgentSessionAttachment, session: AgentSession, latest: Turn | None, collector: AgentTurnResultCollector +) -> None: + if state.turn is None and latest is not None and latest.status not in _TERMINAL: + state.select(latest) + state.seed(collector) + if state.turn is not None and latest is not None and latest.id == state.turn.id: + state.refresh(latest) + state.settle(session) + state.seed(collector) + if ( + _needs_manual_diagnostics(state, session) + and state.turn is None + and (latest is None or latest.status in _TERMINAL) + ): + collector.required_actions = [action for action in collector.required_actions if action.type == "function_call"] + collector.required_actions.extend( + deepcopy([action for action in session.required_actions if action.type == "environment_connection"]) + ) + return + if ( + not _needs_manual_diagnostics(state, session) + or state.turn is None + or latest is None + or latest.id != state.turn.id + or latest.status != "waiting" + ): + return + # Browser authentication replay includes resolved history. Current manual + # required actions are diagnostics only; function dispatch always uses SSE. + collector.required_actions = [ + action for action in collector.required_actions if action.type == "function_call" + ] + deepcopy( + [ + action + for action in session.required_actions + if action.type == "environment_connection" + or action.type == "computer_use_approval_request" + and action.turn_id == state.turn.id + ] + ) + + +def _next_cursor(page: object, data: list[AgentSessionItem], previous: str | None) -> str | None: + if getattr(page, "has_more", None) is False: + return None + if not data and getattr(page, "has_more", None) is not True: + return None + cursor = getattr(page, "last_id", None) or (data[-1].id if data else None) + if not isinstance(cursor, str) or not cursor or cursor == previous: + raise RuntimeError("Cannot recover complete agent output: items page has no advancing cursor") + return cursor + + +def _messages(items: list[AgentSessionItem], turn_id: str) -> list[AgentSessionMessage]: + return [ + item + for item in items + if item.type == "message" + and item.turn_id == turn_id + and item.role == "assistant" + and item.status == "completed" + and item.phase != "commentary" + ] + + +def read_messages( + sessions: Sessions, session_id: str, turn_id: str, options: _RequestOptions +) -> Iterator[AgentSessionMessage]: + after: str | None = None + while True: + page = sessions.items.list( + session_id, order="asc", limit=100, after=after if after is not None else omit, **options + ) + yield from _messages(page.data, turn_id) + after = _next_cursor(page, page.data, after) + if after is None: + return + + +async def async_read_messages( + sessions: AsyncSessions, session_id: str, turn_id: str, options: _RequestOptions +) -> AsyncIterator[AgentSessionMessage]: + after: str | None = None + while True: + page = await sessions.items.list( + session_id, order="asc", limit=100, after=after if after is not None else omit, **options + ) + for message in _messages(page.data, turn_id): + yield message + after = _next_cursor(page, page.data, after) + if after is None: + return + + +def _partial_messages( + recovered: list[AgentSessionMessage], observed: list[AgentSessionMessage] +) -> list[AgentSessionMessage]: + ids = {message.id for message in recovered if message.id is not None} + return recovered + [ + message + for message in observed + if (message.id not in ids if message.id is not None else message not in recovered) + ] + + +def hydrate_error( + sessions: Sessions, state: AgentSessionAttachment, error: AgentTurnResultError, options: _RequestOptions +) -> None: + if error.turn_id is None or state.messages_read: + return + state.messages_read = True + recovered: list[AgentSessionMessage] = [] + try: + recovered.extend(read_messages(sessions, state.session_id, error.turn_id, options)) + except Exception: + error.messages = _partial_messages(recovered, error.messages) + else: + error.messages = _partial_messages(recovered, error.messages) + + +async def async_hydrate_error( + sessions: AsyncSessions, state: AgentSessionAttachment, error: AgentTurnResultError, options: _RequestOptions +) -> None: + if error.turn_id is None or state.messages_read: + return + state.messages_read = True + recovered: list[AgentSessionMessage] = [] + try: + async for message in async_read_messages(sessions, state.session_id, error.turn_id, options): + recovered.append(message) + except Exception: + error.messages = _partial_messages(recovered, error.messages) + else: + error.messages = _partial_messages(recovered, error.messages) + + +def reconcile( + sessions: Sessions, state: AgentSessionAttachment, collector: AgentTurnResultCollector, options: _RequestOptions +) -> bool: + if state.reconciled: + return True + if collector.turn is None: + return False + turn = sessions.turns.retrieve(collector.turn.id, session_id=state.session_id, **options) + # A durable read can lag the terminal event already observed on SSE. + if collector.turn.status in _TERMINAL and turn.status not in _TERMINAL: + turn = collector.turn + collector.turn = turn + state.messages_read = True + messages: list[AgentSessionMessage] = [] + try: + messages.extend(read_messages(sessions, state.session_id, turn.id, options)) + except Exception: + collector.replace_messages(_partial_messages(messages, collector.messages())) + raise + messages = _partial_messages(messages, collector.messages()) + collector.replace_messages(messages) + if turn.status not in _TERMINAL: + return False + _reconcile(state, collector, turn, messages) + return True + + +async def async_reconcile( + sessions: AsyncSessions, + state: AgentSessionAttachment, + collector: AgentTurnResultCollector, + options: _RequestOptions, +) -> bool: + if state.reconciled: + return True + if collector.turn is None: + return False + turn = await sessions.turns.retrieve(collector.turn.id, session_id=state.session_id, **options) + # A durable read can lag the terminal event already observed on SSE. + if collector.turn.status in _TERMINAL and turn.status not in _TERMINAL: + turn = collector.turn + collector.turn = turn + state.messages_read = True + messages: list[AgentSessionMessage] = [] + try: + async for message in async_read_messages(sessions, state.session_id, turn.id, options): + messages.append(message) + except Exception: + collector.replace_messages(_partial_messages(messages, collector.messages())) + raise + messages = _partial_messages(messages, collector.messages()) + collector.replace_messages(messages) + if turn.status not in _TERMINAL: + return False + _reconcile(state, collector, turn, messages) + return True + + +def _reconcile( + state: AgentSessionAttachment, collector: AgentTurnResultCollector, turn: Turn, messages: list[AgentSessionMessage] +) -> None: + collector.turn = turn + collector.cause = None + collector.required_actions = [] + collector.boundary = True + collector.replace_messages(messages) + state.reconciled = True + + +def observe(stream: Stream[AgentSessionEvent], state: AgentSessionAttachment) -> Iterator[AgentSessionEvent]: + iterator = iter(stream) + while True: + try: + event = next(iterator) + except StopIteration: + state.observation_interrupted = True + return + except Exception: + state.observation_interrupted = True + raise + yield event + + +async def async_observe( + stream: AsyncStream[AgentSessionEvent], state: AgentSessionAttachment +) -> AsyncIterator[AgentSessionEvent]: + iterator = stream.__aiter__() + while True: + try: + event = await iterator.__anext__() + except StopAsyncIteration: + state.observation_interrupted = True + return + except Exception: + state.observation_interrupted = True + raise + yield event diff --git a/src/openai/lib/beta/agents/_result.py b/src/openai/lib/beta/agents/_result.py index 9bcb2ba43d..032b9d92f5 100644 --- a/src/openai/lib/beta/agents/_result.py +++ b/src/openai/lib/beta/agents/_result.py @@ -157,6 +157,9 @@ def is_done(self) -> bool: def messages(self) -> list[AgentSessionMessage]: return [message for _, message in sorted(self._messages.values(), key=lambda pair: pair[0])] + def replace_messages(self, messages: list[AgentSessionMessage]) -> None: + self._messages = {str(index): (index, deepcopy(message)) for index, message in enumerate(messages)} + def error(self, reason: ResultErrorReason) -> AgentTurnResultError: if self._error is None: self._error = AgentTurnResultError( diff --git a/src/openai/lib/streaming/agents/_streams.py b/src/openai/lib/streaming/agents/_streams.py index 3be252e9e6..803f0f3ca2 100644 --- a/src/openai/lib/streaming/agents/_streams.py +++ b/src/openai/lib/streaming/agents/_streams.py @@ -8,6 +8,7 @@ from collections import deque from typing_extensions import Self, TypedDict +import anyio import httpx2 from ._tools import arguments, failed_event, result_event, is_pending_call_race @@ -22,6 +23,24 @@ AgentOutputParseError, AgentTurnResultCollection, ) +from ...beta.agents._attachment import ( + AgentSessionAttachment, + attach, + observe, + reconcile, + async_attach, + refresh_idle, + async_observe, + hydrate_error, + async_reconcile, + diagnose_manual, + select_candidate, + async_refresh_idle, + async_hydrate_error, + async_diagnose_manual, + async_select_candidate, +) +from ....types.beta.agent_session import RequiredActionSessionRequiredActionResourceFunctionCall from ....types.beta.agent_session_event import AgentSessionEvent from ....types.beta.agent_function_call_item import AgentFunctionCallItem from ....types.beta.agent_session_input_param import ( @@ -117,14 +136,15 @@ def call(self, event: AgentSessionEvent) -> AgentFunctionCallItem | None: class AgentSessionStream(Generic[OutputT]): - """Submit input to an idle session and stream through the resulting turn's terminal session event. + """Submit input to an idle session, or omit input to attach with the same handlers. Use as a context manager. Only one caller may submit input to this session while the helper runs: the input endpoint does not return a turn ID for correlating - concurrent writers. Initial idle events and subagent turn completions do not - end iteration. Failed/cancelled turns remain visible as events; an unexpected - end of the connection raises RuntimeError. Closing the stream does not cancel - the backend turn. + concurrent writers. Input submission ignores initial idle events and waits + for the resulting root turn to finish. Attachment ends at the identified root + turn's terminal event. Failed/cancelled turns remain visible as events; an + unexpected end of raw iteration raises RuntimeError. Closing the stream does + not cancel the backend turn. Optional tool handlers run sequentially during iteration, after their call event is yielded. Unregistered tools are left for the caller to handle. A @@ -138,7 +158,7 @@ def __init__( sessions: Sessions, session_id: str, *, - input: str | Iterable[AgentSessionInputMessageParam], + input: str | Iterable[AgentSessionInputMessageParam] | Omit = omit, output_type: type[OutputT] | None = None, tool_handlers: Mapping[str, ToolHandler] | None = None, idempotency_key: str | Omit = omit, @@ -147,7 +167,8 @@ def __init__( ) -> None: self._sessions = sessions self._session_id = session_id - self._input = _input_event(input) + self._input = None if isinstance(input, Omit) else _input_event(input) + self._attachment: AgentSessionAttachment | None = None self._handlers = dict(tool_handlers or {}) self._idempotency_key = _input_key(idempotency_key, extra_headers) self._options = _request_options({"extra_headers": extra_headers, "timeout": timeout}) @@ -162,11 +183,16 @@ def __enter__(self) -> Self: if self._entered or self._closed: raise RuntimeError("An AgentSessionStream can only be entered once") self._entered = True + if self._input is None: + self._stream, self._attachment = attach(self._sessions, self._session_id, self._options) + if self._attachment.turn is not None: + self._state.turn_id = self._attachment.turn.id + self._state.turn_ended = self._attachment.turn.status in ("completed", "failed", "cancelled") + self._iterator = self._iterate() + return self session = self._sessions.retrieve(self._session_id, **self._options) if session.status != "idle": - raise ValueError( - "sessions.stream requires an idle session; use sessions.events.stream to follow an active session" - ) + raise ValueError("Submitting input requires an idle session; omit input to attach to existing work") self._stream = self._sessions.events.stream(self._session_id, **self._options) try: self._sessions.events.create( @@ -200,7 +226,9 @@ def until_done(self) -> None: def with_result_collection(self) -> Self: """Beta: retain final messages for a result; call before consuming events.""" - self._collection.enable() + collector = self._collection.enable() + if self._attachment is not None: + self._attachment.seed(collector) return self def get_final_result(self) -> AgentTurnResult[OutputT]: @@ -212,19 +240,50 @@ def get_final_result(self) -> AgentTurnResult[OutputT]: To iterate first, call with_result_collection() before consuming events. """ collector = self._collection.enable() + if self._attachment is not None: + self._attachment.seed(collector) try: + if self._attachment is not None: + diagnose_manual(self._sessions, self._attachment, collector, self._options) + if ( + self._attachment is not None + and self._attachment.observation_interrupted + and collector.cause is not None + ): + raise collector.cause collector.check_outcome(self._handlers) if not collector.is_done(): for _ in self: collector.check_outcome(self._handlers) if collector.is_done(): break + if self._attachment is not None: + reconcile(self._sessions, self._attachment, collector, self._options) return self._collection.result() - except (AgentTurnResultError, AgentOutputParseError): + except AgentTurnResultError as error: + if self._attachment is not None: + hydrate_error(self._sessions, self._attachment, error, self._options) + raise + except AgentOutputParseError: raise except Exception as error: + if self._attachment is not None and self._attachment.observation_interrupted: + self._attachment.observation_interrupted = False + recovered = False + try: + recovered = reconcile(self._sessions, self._attachment, collector, self._options) + except Exception: + pass # Preserve the original observation failure. + if recovered: + return self._collection.result() self._collection.record_error(error) - raise collector.error("observation_failed") from error + try: + collector.check_outcome(self._handlers) + raise collector.error("observation_failed") from error + except AgentTurnResultError as failure: + if self._attachment is not None: + hydrate_error(self._sessions, self._attachment, failure, self._options) + raise finally: self.close() @@ -237,16 +296,82 @@ def close(self) -> None: def _iterate(self) -> Iterator[AgentSessionEvent]: assert self._stream is not None try: - for event in self._stream: + if self._attachment is not None and self._collection.collector is not None: + self._attachment.seed(self._collection.collector) + diagnose_manual(self._sessions, self._attachment, self._collection.collector, self._options) + self._collection.collector.check_outcome(self._handlers) + if self._attachment is not None and self._attachment.settled: + return + events = observe(self._stream, self._attachment) if self._attachment is not None else self._stream + for event in events: if not self._state.accept(event): continue - self._collection.accept(event) + if self._attachment is not None: + discovering = self._attachment.turn is None + candidate = self._attachment.candidate(event) + if candidate is not None: + select_candidate(self._sessions, self._attachment, candidate, self._options) + if self._attachment.settled: + if self._collection.collector is not None: + self._attachment.seed(self._collection.collector) + return + if self._attachment.turn is not None and self._state.turn_id is None: + self._state.turn_id = self._attachment.turn.id + self._state.turn_ended = self._attachment.turn.status in ("completed", "failed", "cancelled") + if self._collection.collector is not None: + self._attachment.seed(self._collection.collector) + if self._attachment.turn is not None and ( + discovering + or event.type == "agent.session.turn.item.added" + and event.item.type == "computer_use_approval_request" + and event.item.turn_id == self._attachment.turn.id + ): + self._attachment.manual_diagnostics = self._attachment.turn.status not in ( + "completed", + "failed", + "cancelled", + ) + diagnose_manual(self._sessions, self._attachment, self._collection.collector, self._options) + self._collection.collector.check_outcome(self._handlers) + if self._attachment is not None and self._collection.collector is not None: + self._attachment.collect(event, self._collection.collector) + else: + self._collection.accept(event) terminal = self._state.terminal(event) + if self._attachment is not None and self._state.turn_ended: + terminal = True + self._attachment.settled = True + if self._collection.collector is not None: + self._collection.collector.boundary = True + if self._attachment is not None and event.type == "agent.session.idle" and not terminal: + refresh_idle(self._sessions, self._attachment, self._options) + terminal = self._attachment.settled + if self._attachment.turn is not None: + self._state.turn_id = self._attachment.turn.id + if self._collection.collector is not None: + self._attachment.seed(self._collection.collector) if terminal: self.close() # Capture routing and arguments before exposing the mutable event. call = self._state.call(event) handler = self._handlers.get(call.name) if call is not None else None + if ( + self._attachment is not None + and call is not None + and handler is None + and self._collection.collector is not None + ): + # On attach, recovered SSE calls are the authoritative pending + # work. A possibly stale session snapshot is not a second queue. + self._collection.collector.required_actions.append( + RequiredActionSessionRequiredActionResourceFunctionCall( + type="function_call", + turn_id=call.turn_id, + call_id=call.call_id, + name=call.name, + arguments=deepcopy(call.arguments), + ) + ) if handler is not None: call = deepcopy(call) yield event @@ -290,7 +415,7 @@ def _submit_result(self, result: SessionInputParamAgentSessionInputToolResult) - class AsyncAgentSessionStream(Generic[OutputT]): """Async counterpart of AgentSessionStream; use with ``async with``. - Requires an idle session with a single input writer. Handlers may return a + Submitting input requires an idle session with a single input writer. Handlers may return a value or an awaitable; synchronous handlers run inline, without a thread pool. Cancellation propagates and closes the event connection. It does not cancel the backend turn. @@ -301,7 +426,7 @@ def __init__( sessions: AsyncSessions, session_id: str, *, - input: str | Iterable[AgentSessionInputMessageParam], + input: str | Iterable[AgentSessionInputMessageParam] | Omit = omit, output_type: type[OutputT] | None = None, tool_handlers: Mapping[str, AsyncToolHandler] | None = None, idempotency_key: str | Omit = omit, @@ -310,7 +435,8 @@ def __init__( ) -> None: self._sessions = sessions self._session_id = session_id - self._input = _input_event(input) + self._input = None if isinstance(input, Omit) else _input_event(input) + self._attachment: AgentSessionAttachment | None = None self._handlers = dict(tool_handlers or {}) self._idempotency_key = _input_key(idempotency_key, extra_headers) self._options = _request_options({"extra_headers": extra_headers, "timeout": timeout}) @@ -325,11 +451,16 @@ async def __aenter__(self) -> Self: if self._entered or self._closed: raise RuntimeError("An AsyncAgentSessionStream can only be entered once") self._entered = True + if self._input is None: + self._stream, self._attachment = await async_attach(self._sessions, self._session_id, self._options) + if self._attachment.turn is not None: + self._state.turn_id = self._attachment.turn.id + self._state.turn_ended = self._attachment.turn.status in ("completed", "failed", "cancelled") + self._iterator = self._iterate() + return self session = await self._sessions.retrieve(self._session_id, **self._options) if session.status != "idle": - raise ValueError( - "sessions.stream requires an idle session; use sessions.events.stream to follow an active session" - ) + raise ValueError("Submitting input requires an idle session; omit input to attach to existing work") self._stream = await self._sessions.events.stream(self._session_id, **self._options) try: await self._sessions.events.create( @@ -363,7 +494,9 @@ async def until_done(self) -> None: def with_result_collection(self) -> Self: """Beta: retain final messages for a result; call before consuming events.""" - self._collection.enable() + collector = self._collection.enable() + if self._attachment is not None: + self._attachment.seed(collector) return self async def get_final_result(self) -> AgentTurnResult[OutputT]: @@ -375,19 +508,50 @@ async def get_final_result(self) -> AgentTurnResult[OutputT]: To iterate first, call with_result_collection() before consuming events. """ collector = self._collection.enable() + if self._attachment is not None: + self._attachment.seed(collector) try: + if self._attachment is not None: + await async_diagnose_manual(self._sessions, self._attachment, collector, self._options) + if ( + self._attachment is not None + and self._attachment.observation_interrupted + and collector.cause is not None + ): + raise collector.cause collector.check_outcome(self._handlers) if not collector.is_done(): async for _ in self: collector.check_outcome(self._handlers) if collector.is_done(): break + if self._attachment is not None: + await async_reconcile(self._sessions, self._attachment, collector, self._options) return self._collection.result() - except (AgentTurnResultError, AgentOutputParseError): + except AgentTurnResultError as error: + if self._attachment is not None: + await async_hydrate_error(self._sessions, self._attachment, error, self._options) + raise + except AgentOutputParseError: raise except Exception as error: + if self._attachment is not None and self._attachment.observation_interrupted: + self._attachment.observation_interrupted = False + recovered = False + try: + recovered = await async_reconcile(self._sessions, self._attachment, collector, self._options) + except Exception: + pass # Preserve the original observation failure. + if recovered: + return self._collection.result() self._collection.record_error(error) - raise collector.error("observation_failed") from error + try: + collector.check_outcome(self._handlers) + raise collector.error("observation_failed") from error + except AgentTurnResultError as failure: + if self._attachment is not None: + await async_hydrate_error(self._sessions, self._attachment, failure, self._options) + raise finally: await self.close() @@ -395,21 +559,90 @@ async def close(self) -> None: """Close the event connection without cancelling the backend turn.""" self._closed = True if self._stream is not None: - await self._stream.close() + with anyio.CancelScope(shield=True): + await self._stream.close() async def _iterate(self) -> AsyncIterator[AgentSessionEvent]: assert self._stream is not None try: - async for event in self._stream: + if self._attachment is not None and self._collection.collector is not None: + self._attachment.seed(self._collection.collector) + await async_diagnose_manual(self._sessions, self._attachment, self._collection.collector, self._options) + self._collection.collector.check_outcome(self._handlers) + if self._attachment is not None and self._attachment.settled: + return + events = async_observe(self._stream, self._attachment) if self._attachment is not None else self._stream + async for event in events: if not self._state.accept(event): continue - self._collection.accept(event) + if self._attachment is not None: + discovering = self._attachment.turn is None + candidate = self._attachment.candidate(event) + if candidate is not None: + await async_select_candidate(self._sessions, self._attachment, candidate, self._options) + if self._attachment.settled: + if self._collection.collector is not None: + self._attachment.seed(self._collection.collector) + return + if self._attachment.turn is not None and self._state.turn_id is None: + self._state.turn_id = self._attachment.turn.id + self._state.turn_ended = self._attachment.turn.status in ("completed", "failed", "cancelled") + if self._collection.collector is not None: + self._attachment.seed(self._collection.collector) + if self._attachment.turn is not None and ( + discovering + or event.type == "agent.session.turn.item.added" + and event.item.type == "computer_use_approval_request" + and event.item.turn_id == self._attachment.turn.id + ): + self._attachment.manual_diagnostics = self._attachment.turn.status not in ( + "completed", + "failed", + "cancelled", + ) + await async_diagnose_manual( + self._sessions, self._attachment, self._collection.collector, self._options + ) + self._collection.collector.check_outcome(self._handlers) + if self._attachment is not None and self._collection.collector is not None: + self._attachment.collect(event, self._collection.collector) + else: + self._collection.accept(event) terminal = self._state.terminal(event) + if self._attachment is not None and self._state.turn_ended: + terminal = True + self._attachment.settled = True + if self._collection.collector is not None: + self._collection.collector.boundary = True + if self._attachment is not None and event.type == "agent.session.idle" and not terminal: + await async_refresh_idle(self._sessions, self._attachment, self._options) + terminal = self._attachment.settled + if self._attachment.turn is not None: + self._state.turn_id = self._attachment.turn.id + if self._collection.collector is not None: + self._attachment.seed(self._collection.collector) if terminal: await self.close() # Capture routing and arguments before exposing the mutable event. call = self._state.call(event) handler = self._handlers.get(call.name) if call is not None else None + if ( + self._attachment is not None + and call is not None + and handler is None + and self._collection.collector is not None + ): + # On attach, recovered SSE calls are the authoritative pending + # work. A possibly stale session snapshot is not a second queue. + self._collection.collector.required_actions.append( + RequiredActionSessionRequiredActionResourceFunctionCall( + type="function_call", + turn_id=call.turn_id, + call_id=call.call_id, + name=call.name, + arguments=deepcopy(call.arguments), + ) + ) if handler is not None: call = deepcopy(call) yield event diff --git a/src/openai/resources/beta/agents/sessions/sessions.py b/src/openai/resources/beta/agents/sessions/sessions.py index 80edcbc259..aa9ff40886 100644 --- a/src/openai/resources/beta/agents/sessions/sessions.py +++ b/src/openai/resources/beta/agents/sessions/sessions.py @@ -81,14 +81,14 @@ def stream( self, session_id: str, *, - input: str | Iterable[AgentSessionInputMessageParam], + input: str | Iterable[AgentSessionInputMessageParam] | Omit = omit, output_type: type[OutputT] | None = None, tool_handlers: Mapping[str, ToolHandler] | None = None, idempotency_key: str | Omit = omit, extra_headers: Headers | None = None, timeout: float | httpx2.Timeout | None | NotGiven = not_given, ) -> AgentSessionStream[OutputT]: - """Stream one turn of an idle session, subscribing before submitting input. + """Stream a new turn, or omit input to attach to existing session work. Use as a context manager. Only one caller may submit input to the session while this helper runs. Optional tool handlers receive an arguments dict; @@ -571,14 +571,14 @@ def stream( self, session_id: str, *, - input: str | Iterable[AgentSessionInputMessageParam], + input: str | Iterable[AgentSessionInputMessageParam] | Omit = omit, output_type: type[OutputT] | None = None, tool_handlers: Mapping[str, AsyncToolHandler] | None = None, idempotency_key: str | Omit = omit, extra_headers: Headers | None = None, timeout: float | httpx2.Timeout | None | NotGiven = not_given, ) -> AsyncAgentSessionStream[OutputT]: - """Stream one turn of an idle session, subscribing before submitting input. + """Stream a new turn, or omit input to attach to existing session work. Use as an async context manager. Only one caller may submit input to the session while this helper runs. Optional tool handlers receive an arguments dict; diff --git a/tests/lib/streaming/agents/test_attachment.py b/tests/lib/streaming/agents/test_attachment.py new file mode 100644 index 0000000000..678b7a448e --- /dev/null +++ b/tests/lib/streaming/agents/test_attachment.py @@ -0,0 +1,1035 @@ +from __future__ import annotations + +from typing import Any, Iterator, AsyncIterator +from typing_extensions import override + +import httpx2 +import pytest + +from openai import OpenAI, AsyncOpenAI, APIConnectionError +from openai.lib.beta.agents import AgentTurnResult, AgentTurnResultError +from openai.lib.streaming.agents._types import ToolHandler +from tests.lib.streaming.agents.test_results import message +from tests.lib.streaming.agents.test_streams import Server, EventBody, sdk as sdk, call, idle, session, turn_event + + +def turn(status: str = "in_progress", turn_id: str = "turn_root", subagent_id: str | None = None) -> dict[str, Any]: + value = turn_event("created", turn_id, subagent_id=subagent_id)["turn"] + assert isinstance(value, dict) + return {**value, "status": status} + + +class AttachmentServer(Server): + def __init__(self) -> None: + super().__init__() + self.status = "in_progress" + self.turns: list[dict[str, Any]] = [turn()] + self.retrieve_status: str | None = None + self.fail_recovery_read = False + self.complete_on_rejected_submission = False + self.after_status: str | None = None + self.after_turns: list[dict[str, Any]] | None = None + self.reads = 0 + self.items: list[dict[str, Any]] = [message("earlier")["item"], message("later", item_id="later")["item"]] + self.body = EventBody([call(), turn_event("completed"), idle()]) + self.stale_required_actions = False + self.manual_actions: list[dict[str, Any]] = [] + self.null_item_cursor = False + self.item_cursor_metadata = False + self.fail_items = False + self.fail_diagnostic = False + self.fail_later_items = False + self.live_turns: list[dict[str, Any]] | None = None + self.live_after_reads = 1 + self.turn_reads = 0 + self.first_turn_status: str | None = None + self.manual_after_reads = 0 + self.turn_list_reads = 0 + self.turns_after_list: int | None = None + + @override + def handle(self, request: httpx2.Request) -> httpx2.Response: + path = request.url.path + if request.method == "POST" and self.complete_on_rejected_submission: + self.retrieve_status = "completed" + if request.method == "GET" and not path.endswith("/events"): + self.requests.append(request) + if path.endswith("/turns"): + self.turn_list_reads += 1 + if self.turns_after_list is not None and self.turn_list_reads >= self.turns_after_list: + self.turns = [turn("waiting")] + if self.body.read_count >= self.live_after_reads and self.live_turns is not None: + self.turns = self.live_turns + after = request.url.params.get("after") + # One item per page exercises existing automatic pagination. + offset = next((i + 1 for i, item in enumerate(self.turns) if item["id"] == after), 0) + data = self.turns[offset : offset + 1] + return httpx2.Response( + 200, json={"data": data, "has_more": offset + 1 < len(self.turns), "object": "list"} + ) + if "/turns/" in path: + self.turn_reads += 1 + turn_id = path.rsplit("/", 1)[-1] + if self.fail_recovery_read and self.body.read_count: + raise httpx2.ConnectError("Synthetic durable read failure", request=request) + status = ( + "completed" + if self.after_status == "idle" + or any( + event.get("type") == "agent.session.turn.completed" and event.get("turn_id") == turn_id + for event in self.body.events[: self.body.read_count] + ) + else "in_progress" + ) + return httpx2.Response( + 200, + json=turn( + "completed" + if turn_id == "old_turn" + else ( + self.first_turn_status + if self.turn_reads == 1 and self.first_turn_status + else self.retrieve_status or status + ), + turn_id, + "child" if turn_id == "child_turn" else None, + ), + ) + if path.endswith("/items"): + if self.fail_items or self.fail_later_items and request.url.params.get("after"): + return httpx2.Response(500, json={"error": {"message": "Synthetic read failure"}}) + after = request.url.params.get("after") + offset = next((i + 1 for i, item in enumerate(self.items) if item["id"] == after), 0) + data = self.items[offset : offset + 1] + payload: dict[str, Any] = {"data": data, "has_more": offset + 1 < len(self.items), "object": "list"} + if self.item_cursor_metadata and data: + payload["last_id"] = data[-1]["id"] + if self.null_item_cursor and data: + payload["data"] = [{**data[0], "id": None}] + return httpx2.Response(200, json=payload) + self.reads += 1 + if self.fail_diagnostic and self.reads > 1: + raise httpx2.ConnectError("Synthetic diagnostic failure", request=request) + if self.after_turns is not None: + self.turns = self.after_turns + pending = self.body.read_count >= self.manual_after_reads + value = session((self.after_status or self.status) if pending else "in_progress") + value["required_actions"] = self.manual_actions if pending else [] + if self.stale_required_actions: + value["required_actions"] = [ + { + "type": "function_call", + "turn_id": "turn_root", + "call_id": "answered", + "name": "missing", + "arguments": {}, + } + ] + return httpx2.Response(200, json=value) + return super().handle(request) + + +@pytest.fixture +def server() -> AttachmentServer: + return AttachmentServer() + + +async def attach_result(sdk: OpenAI | AsyncOpenAI, handlers: dict[str, ToolHandler] | None = None) -> AgentTurnResult: + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test", tool_handlers=handlers) as stream: + result = await stream.get_final_result() + assert await stream.get_final_result() is result + return result + with sdk.beta.agents.sessions.stream("session_test", tool_handlers=handlers) as stream: + result = stream.get_final_result() + assert stream.get_final_result() is result + return result + + +async def test_attach_dispatches_pending_once_and_recovers_paginated_output( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns.insert(0, turn(subagent_id="child", turn_id="child_turn")) + server.body = EventBody([call(), call(), message("later", item_id="later"), turn_event("completed"), idle()]) + server.items.insert(1, {**message("user", item_id="user")["item"], "role": "user"}) + server.items.insert(2, message("child", item_id="child", turn_id="child_turn")["item"]) + server.items.insert(3, message("commentary", item_id="commentary", phase="commentary")["item"]) + calls: list[object] = [] + result = await attach_result(sdk, {"search": lambda args: calls.append(args) or "found"}) + assert calls == [{"query": "test"}] + assert result.output_text == "earlierlater" + assert len(server.inputs()) == 1 + assert server.inputs()[0]["type"] == "agent.session.input.tool_result" + assert len([r for r in server.requests if r.url.path.endswith("/items")]) == len(server.items) + assert server.body.closed + + +async def test_pending_call_can_identify_root_without_turn_created( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [] + server.live_turns = [turn()] + seen: list[object] = [] + result = await attach_result(sdk, {"search": lambda args: seen.append(args) or "found"}) + assert result.turn_id == "turn_root" + assert len(seen) == 1 + + +async def test_completion_between_initial_read_and_subscription( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.after_status = "idle" + server.body = EventBody([]) + result = await attach_result(sdk) + assert result.output_text == "earlierlater" + assert server.body.read_count == 0 + assert not server.inputs() + + +async def test_idle_without_selected_turn_settles_and_has_no_result( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.status = "idle" + server.turns = [turn("completed", "old_turn")] + server.body = EventBody([]) + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test") as stream: + await stream.until_done() + with pytest.raises(AgentTurnResultError, match="incomplete"): + await stream.get_final_result() + else: + with sdk.beta.agents.sessions.stream("session_test") as stream: + stream.until_done() + with pytest.raises(AgentTurnResultError, match="incomplete"): + stream.get_final_result() + assert not server.inputs() + assert not any(r.url.path.endswith("/items") for r in server.requests) + + +async def test_answered_call_not_executed_from_stale_session_snapshot( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.stale_required_actions = True + server.body = EventBody([turn_event("completed"), idle()]) + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + assert not server.inputs() + + +async def test_unhandled_recovered_call_is_reported(sdk: OpenAI | AsyncOpenAI, server: AttachmentServer) -> None: + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk) + assert caught.value.reason == "requires_action" + assert "".join(item.output_text for item in caught.value.messages) == "earlierlater" + action = caught.value.required_actions[0] + assert action.type == "function_call" + assert action.call_id == "call_test" + assert not server.inputs() + assert server.body.closed + + +async def test_raw_attach_does_not_fetch_or_retain_result_messages( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + handlers: dict[str, ToolHandler] = {"search": lambda _args: "found"} + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test", tool_handlers=handlers) as stream: + await stream.until_done() + assert stream._collection.collector is None + else: + with sdk.beta.agents.sessions.stream("session_test", tool_handlers=handlers) as stream: + stream.until_done() + assert stream._collection.collector is None + assert not any(r.url.path.endswith("/items") for r in server.requests) + + +async def test_full_terminal_event_selects_root(sdk: OpenAI | AsyncOpenAI, server: AttachmentServer) -> None: + server.turns = [] + server.body = EventBody( + [turn_event("completed", "child_turn", subagent_id="child"), turn_event("completed"), idle()] + ) + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + + +async def test_attach_handler_failure_submits_generic_result( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + def handler(_args: object) -> str: + raise ValueError("SYNTHETIC_HANDLER_SECRET") + + result = await attach_result(sdk, {"search": handler}) + assert result.turn_id == "turn_root" + assert len(server.inputs()) == 1 + assert "SYNTHETIC_HANDLER_SECRET" not in str(server.inputs()) + + +async def test_attach_tool_submission_failure_preserves_cause( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.reject_input = True + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk, {"search": lambda _args: "found"}) + assert caught.value.reason == "observation_failed" + assert caught.value.__cause__ is not None + assert server.body.closed + + +async def test_reattach_again_after_losing_observation(sdk: OpenAI | AsyncOpenAI, server: AttachmentServer) -> None: + server.body = EventBody([call()]) + seen: list[object] = [] + with pytest.raises(AgentTurnResultError, match="observation_failed"): + await attach_result(sdk, {"search": lambda args: seen.append(args) or "found"}) + # A subsequent attachment receives only calls still awaiting an answer. + server.body = EventBody([turn_event("completed"), idle()]) + result = await attach_result(sdk, {"search": lambda args: seen.append(args) or "found"}) + assert result.output_text == "earlierlater" + assert len(seen) == 1 + assert len(server.inputs()) == 1 + + +@pytest.mark.parametrize("old_root", [False, True]) +async def test_new_root_completes_before_attachment_refresh( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, old_root: bool +) -> None: + server.turns = [turn("completed", "old_turn")] if old_root else [] + server.after_status = "idle" + server.after_turns = [turn("completed")] + server.body = EventBody([turn_event("completed"), idle()]) + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + assert result.output_text == "earlierlater" + assert server.body.read_count == 0 + + +async def test_unchanged_old_completed_root_is_not_selected( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [turn("completed", "old_turn")] + server.after_status = "idle" + server.body = EventBody([]) + with pytest.raises(AgentTurnResultError, match="incomplete"): + await attach_result(sdk) + assert not any(request.url.path.endswith("/items") for request in server.requests) + + +async def test_terminal_selected_turn_settles_while_successor_is_active( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.retrieve_status = "completed" + server.body = EventBody([call(call_id="successor_call")]) + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + assert result.output_text == "earlierlater" + assert server.body.read_count == 0 + assert not server.inputs() + + +async def test_selected_terminal_event_stops_before_successor_calls( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.body = EventBody([turn_event("completed"), call(call_id="successor_call"), idle()]) + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + assert server.body.read_count == 1 + assert not server.inputs() + + +async def test_unhandled_replay_stops_without_waiting_for_batch_end( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + missing: Any = call(call_id="missing") + missing["item"]["name"] = "unregistered" + server.body = EventBody([missing, call(), turn_event("completed"), idle()]) + handled: list[object] = [] + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk, {"search": lambda args: handled.append(args) or "found"}) + assert caught.value.reason == "requires_action" + action = caught.value.required_actions[0] + assert action.type == "function_call" + assert action.call_id == "missing" + assert server.body.read_count == 1 + assert not handled + assert server.body.closed + + +async def test_new_root_finishes_between_idle_baseline_and_refresh( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.status = "idle" + server.turns = [turn("completed", "old_turn")] + server.after_turns = [turn("completed")] + server.after_status = "idle" + server.body = EventBody([]) + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + assert result.output_text == "earlierlater" + + +class InterruptedBody(EventBody): + def __init__(self, server: AttachmentServer, status: str) -> None: + super().__init__([message("partial", item_id="observed_partial")]) + self.server = server + self.status = status + + @override + def __iter__(self) -> Iterator[bytes]: + yield from super().__iter__() + self.server.retrieve_status = self.status + raise httpx2.ReadError("Synthetic observation disconnect") + + @override + async def __aiter__(self) -> AsyncIterator[bytes]: + for chunk in self: + yield chunk + + +@pytest.mark.parametrize("terminal", ["completed", "failed"]) +async def test_interrupted_observation_reconciles_exact_selected_turn( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, terminal: str +) -> None: + server.body = InterruptedBody(server, terminal) + if terminal == "failed": + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk) + assert caught.value.reason == "failed" + assert caught.value.messages + else: + result = await attach_result(sdk) + assert result.output_text == "earlierlaterpartial" + assert len([r for r in server.requests if "/turns/" in r.url.path]) == 2 + assert server.body.closed + + +@pytest.mark.parametrize("read_fails", [False, True]) +async def test_inconclusive_reconciliation_preserves_observation_cause( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, read_fails: bool +) -> None: + server.body = InterruptedBody(server, "in_progress") + server.fail_recovery_read = read_fails + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk) + assert caught.value.reason == "observation_failed" + cause = caught.value.__cause__ + assert isinstance(cause, APIConnectionError) + assert isinstance(cause.__cause__, httpx2.ReadError) + assert str(cause.__cause__) == "Synthetic observation disconnect" + if not read_fails: + assert "".join(item.output_text for item in caught.value.messages) == "earlierlaterpartial" + assert len([r for r in server.requests if "/turns/" in r.url.path]) == 2 + + +async def test_submission_failure_is_not_hidden_by_completed_durable_turn( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.reject_input = True + server.complete_on_rejected_submission = True + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk, {"search": lambda _args: "found"}) + assert caught.value.reason == "observation_failed" + assert len([r for r in server.requests if "/turns/" in r.url.path]) == 1 + assert "".join(item.output_text for item in caught.value.messages) == "earlierlater" + assert caught.value.__cause__ is not None + + +async def test_collected_progress_can_recover_after_observation_error( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.body = InterruptedBody(server, "completed") + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + with pytest.raises(APIConnectionError): + async for _ in stream: + pass + result = await stream.get_final_result() + else: + with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + with pytest.raises(APIConnectionError): + list(stream) + result = stream.get_final_result() + assert result.output_text == "earlierlaterpartial" + + +@pytest.mark.parametrize("metadata", [False, True]) +async def test_legacy_null_item_ids_do_not_truncate_recovered_output( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, metadata: bool +) -> None: + server.body = EventBody([turn_event("completed")]) + server.null_item_cursor = True + server.item_cursor_metadata = metadata + if metadata: + result = await attach_result(sdk) + assert result.output_text == "earlierlater" + else: + with pytest.raises(AgentTurnResultError, match="observation_failed") as caught: + await attach_result(sdk) + assert "advancing cursor" in str(caught.value.__cause__) + + +@pytest.mark.parametrize("status", ["failed", "cancelled"]) +@pytest.mark.parametrize("read_fails", [False, True]) +async def test_bootstrap_terminal_error_recovers_available_partial_output( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, status: str, read_fails: bool +) -> None: + server.retrieve_status = status + server.fail_items = read_fails + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk) + assert caught.value.reason == status + if not read_fails: + assert "".join(item.output_text for item in caught.value.messages) == "earlierlater" + assert server.body.read_count == 0 + + +def manual_action(kind: str, turn_id: str = "turn_root") -> dict[str, Any]: + if kind == "environment_connection": + return {"type": kind, "environment_id": "env_test"} + request: dict[str, Any] = {"type": kind, "reason": "Synthetic approval"} + if kind == "browser_authentication": + request.update(fields=[], options=[], credential_origin="https://example.com") + else: + request["origin"] = "https://example.com" + return { + "type": "computer_use_approval_request", + "request_id": "request_test", + "turn_id": turn_id, + "request": request, + } + + +@pytest.mark.parametrize("kind", ["environment_connection", "browser_authentication", "browser_origin_access"]) +async def test_current_manual_action_is_diagnostic_without_waiting_for_replay( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, kind: str +) -> None: + server.turns = [turn("waiting")] + server.retrieve_status = "waiting" + server.status = "requires_action" + server.manual_actions = [manual_action(kind)] + server.body = EventBody([]) + with pytest.raises(AgentTurnResultError, match="requires_action") as caught: + await attach_result(sdk) + assert caught.value.required_actions[0].type == server.manual_actions[0]["type"] + assert server.body.read_count == 0 + assert not server.inputs() + + +@pytest.mark.parametrize("kind", ["environment_connection", "browser_authentication", "browser_origin_access"]) +async def test_stale_manual_action_does_not_override_selected_completion( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, kind: str +) -> None: + server.status = "requires_action" + server.retrieve_status = "completed" + server.manual_actions = [manual_action(kind, "successor")] + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + assert server.body.read_count == 0 + + +async def test_manual_action_from_successor_is_not_selected( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [turn("waiting")] + server.after_turns = [turn("waiting", "successor")] + server.retrieve_status = "waiting" + server.status = "requires_action" + server.manual_actions = [manual_action("environment_connection")] + server.body = EventBody([turn_event("completed")]) + # The selected terminal SSE event remains authoritative even if a subsequent + # GET fixture still returns the older waiting state. + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + assert result.output_text == "earlierlater" + + +async def test_terminal_failure_wins_over_partial_recovery_read_failure( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.body = InterruptedBody(server, "failed") + server.fail_items = True + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk) + assert caught.value.reason == "failed" + assert caught.value.messages[0].output_text == "partial" + + +async def test_raw_attachment_does_not_retain_manual_request_payload( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [turn("waiting")] + server.retrieve_status = "waiting" + server.status = "requires_action" + server.manual_actions = [manual_action("browser_authentication")] + server.body = EventBody([turn_event("completed")]) + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test") as stream: + await stream.until_done() + state = stream._attachment + assert stream._collection.collector is None + else: + with sdk.beta.agents.sessions.stream("session_test") as stream: + stream.until_done() + state = stream._attachment + assert stream._collection.collector is None + assert state is not None + assert not any(isinstance(value, (dict, list)) for value in vars(state).values()) + assert server.reads == 1 # No diagnostic refetch for raw iteration. + assert not any(r.url.path.endswith("/items") for r in server.requests) + + +@pytest.mark.parametrize("baseline", ["old_turn", "newer_old_turn"]) +async def test_historical_browser_request_cannot_select_prior_completed_root( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, baseline: str +) -> None: + server.turns = [turn("completed", baseline)] + server.live_turns = [turn()] + old_request: Any = call() + old_request["turn_id"] = "old_turn" + old_request["event_id"] = "old_request" + old_request["item"] = { + "type": "computer_use_approval_request", + "id": "old_request", + "turn_id": "old_turn", + "request_id": "old_request", + "request": manual_action("browser_authentication", "old_turn")["request"], + } + server.body = EventBody([old_request, call(), turn_event("completed")]) + handled: list[object] = [] + result = await attach_result(sdk, {"search": lambda args: handled.append(args) or "found"}) + assert result.turn_id == "turn_root" + assert result.output_text == "earlierlater" + assert len(handled) == 1 + + +async def test_preentered_collection_seeds_manual_error_identity( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [turn("waiting")] + server.retrieve_status = "waiting" + server.status = "requires_action" + server.manual_actions = [manual_action("browser_authentication")] + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + with pytest.raises(AgentTurnResultError) as raw: + await stream.until_done() + with pytest.raises(AgentTurnResultError) as final: + await stream.get_final_result() + else: + with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + with pytest.raises(AgentTurnResultError) as raw: + stream.until_done() + with pytest.raises(AgentTurnResultError) as final: + stream.get_final_result() + assert raw.value is final.value + assert final.value.turn_id == "turn_root" + assert "".join(item.output_text for item in final.value.messages) == "earlierlater" + + +@pytest.mark.parametrize("status", ["failed", "cancelled", "in_progress"]) +async def test_later_page_failure_preserves_already_recovered_messages( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, status: str +) -> None: + server.fail_later_items = True + server.retrieve_status = status + if status == "in_progress": + server.body = InterruptedBody(server, "in_progress") + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk) + assert caught.value.reason == ("observation_failed" if status == "in_progress" else status) + assert any(item.output_text == "earlier" for item in caught.value.messages) + if status == "in_progress": + assert any(item.output_text == "partial" for item in caught.value.messages) + + +async def test_buffered_idle_does_not_settle_current_active_attachment( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [turn("completed", "old_turn")] + server.live_turns = [turn()] + server.live_after_reads = 2 # The active root is not yet visible at the stale idle frame. + server.body = EventBody([idle(), call(), turn_event("completed")]) + calls: list[object] = [] + result = await attach_result(sdk, {"search": lambda args: calls.append(args) or "found"}) + assert result.turn_id == "turn_root" + assert len(calls) == 1 + + +@pytest.mark.parametrize("historical", [False, True]) +async def test_environment_connection_without_active_turn_is_reported( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, historical: bool +) -> None: + server.turns = [turn("completed", "old_turn")] if historical else [] + server.status = "requires_action" + server.manual_actions = [manual_action("environment_connection")] + server.body = EventBody([]) + with pytest.raises(AgentTurnResultError, match="requires_action") as caught: + await attach_result(sdk) + assert caught.value.session_id == "session_test" + assert caught.value.turn_id is None + assert caught.value.required_actions[0].type == "environment_connection" + assert server.body.read_count == 0 + assert not any(r.url.path.endswith("/items") for r in server.requests) + + +@pytest.mark.parametrize("stale_first_lookup", [False, True]) +@pytest.mark.parametrize("already_selected", [False, True]) +async def test_first_waiting_root_discovered_from_browser_item_checks_manual_action( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, stale_first_lookup: bool, already_selected: bool +) -> None: + server.turns = [turn("waiting")] if already_selected else [] + server.retrieve_status = "waiting" + server.live_turns = [turn("waiting")] + server.manual_after_reads = 1 if already_selected else 0 + server.status = "requires_action" + server.manual_actions = [manual_action("browser_authentication")] + request: Any = call() + request["turn_id"] = None + request["item"] = { + "type": "computer_use_approval_request", + "id": "request_test", + "turn_id": "turn_root", + "request_id": "request_test", + "request": server.manual_actions[0]["request"], + } + server.live_after_reads = 2 if stale_first_lookup and not already_selected else 1 + repeated: dict[str, object] = {**request, "event_id": "second_request_observation"} + server.body = EventBody([request, repeated] if stale_first_lookup else [request]) + with pytest.raises(AgentTurnResultError, match="requires_action") as caught: + await attach_result(sdk) + assert caught.value.turn_id == "turn_root" + assert caught.value.required_actions[0].type == "computer_use_approval_request" + assert server.body.read_count == (2 if stale_first_lookup and not already_selected else 1) + + +async def test_diagnostic_read_failure_retains_durable_partial_messages( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [turn("waiting")] + server.retrieve_status = "waiting" + server.status = "requires_action" + server.manual_actions = [manual_action("browser_authentication")] + server.fail_diagnostic = True + with pytest.raises(AgentTurnResultError, match="observation_failed") as caught: + await attach_result(sdk) + assert caught.value.turn_id == "turn_root" + assert "".join(item.output_text for item in caught.value.messages) == "earlierlater" + assert isinstance(caught.value.__cause__, APIConnectionError) + assert server.body.read_count == 0 + + +@pytest.mark.parametrize("status", ["completed", "failed", "cancelled"]) +async def test_terminal_event_survives_stale_reconciliation( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, status: str +) -> None: + server.retrieve_status = "in_progress" + server.body = EventBody([turn_event(status)]) + if status == "completed": + result = await attach_result(sdk) + assert result.turn.status == "completed" + assert result.output_text == "earlierlater" + else: + with pytest.raises(AgentTurnResultError) as caught: + await attach_result(sdk) + assert caught.value.reason == status + assert caught.value.turn is not None + assert caught.value.turn.status == status + + +async def test_idle_handshake_rechecks_selected_turn_after_stale_read( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.after_status = "idle" + server.first_turn_status = "in_progress" + server.retrieve_status = "completed" + server.body = EventBody([]) + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + assert result.output_text == "earlierlater" + assert server.body.read_count == 0 + + +async def test_idle_event_refreshes_exact_selected_turn(sdk: OpenAI | AsyncOpenAI, server: AttachmentServer) -> None: + server.first_turn_status = "in_progress" + server.retrieve_status = "completed" + server.body = EventBody([idle()]) + result = await attach_result(sdk) + assert result.turn_id == "turn_root" + assert result.output_text == "earlierlater" + assert server.body.read_count == 1 + + +async def test_reconciliation_preserves_completed_sse_tail_when_items_read_lags( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.items = [message("earlier")["item"]] + server.body = EventBody([message("later", item_id="later"), turn_event("completed")]) + result = await attach_result(sdk) + assert result.output_text == "earlierlater" + assert len(result.messages) == 2 + + +async def test_unhandled_call_diagnostic_does_not_alias_yielded_arguments( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + event: Any = call() + event["item"]["arguments"] = {"nested": {"value": "original"}} + server.body = EventBody([event]) + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + observed = await stream.__anext__() + assert observed.type == "agent.session.turn.item.added" and observed.item.type == "function_call" + observed.item.arguments["nested"]["value"] = "changed" # type: ignore[index] + with pytest.raises(AgentTurnResultError) as caught: + await stream.get_final_result() + else: + with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + observed = next(stream) + assert observed.type == "agent.session.turn.item.added" and observed.item.type == "function_call" + observed.item.arguments["nested"]["value"] = "changed" # type: ignore[index] + with pytest.raises(AgentTurnResultError) as caught: + stream.get_final_result() + action = caught.value.required_actions[0] + assert action.type == "function_call" + assert action.arguments == {"nested": {"value": "original"}} + + +async def test_progress_collection_retains_distinct_unhandled_calls( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.body = EventBody([call(call_id="first"), call(call_id="second")]) + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + await stream.__anext__() + await stream.__anext__() + with pytest.raises(AgentTurnResultError) as caught: + await stream.get_final_result() + else: + with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + next(stream) + next(stream) + with pytest.raises(AgentTurnResultError) as caught: + stream.get_final_result() + assert [action.call_id for action in caught.value.required_actions if action.type == "function_call"] == [ + "first", + "second", + ] + + +async def test_manual_diagnostics_selects_newly_visible_waiting_root( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [] + server.turns_after_list = 3 + server.status = "requires_action" + server.manual_actions = [manual_action("environment_connection")] + server.body = EventBody([]) + with pytest.raises(AgentTurnResultError, match="requires_action") as caught: + await attach_result(sdk) + assert caught.value.turn_id == "turn_root" + assert caught.value.required_actions[0].type == "environment_connection" + assert server.body.read_count == 0 + + +async def test_error_hydration_preserves_observed_completed_tail( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.items = [message("earlier")["item"]] + server.body = EventBody([message("later", item_id="later"), call()]) + with pytest.raises(AgentTurnResultError, match="requires_action") as caught: + await attach_result(sdk) + assert "".join(item.output_text for item in caught.value.messages) == "earlierlater" + + +@pytest.mark.parametrize("backend", ["asyncio", "trio"]) +@pytest.mark.parametrize("during_refresh", [False, True]) +def test_cancelled_attachment_closes_open_sse(backend: str, during_refresh: bool) -> None: + import anyio + + class CheckpointCloseBody(EventBody): + @override + async def aclose(self) -> None: + await anyio.sleep(0) + self.closed = True + + server = AttachmentServer() + server.body = CheckpointCloseBody([idle()]) + server.turns = [] + cancelled: list[BaseException] = [] + + async def run() -> None: + with anyio.CancelScope() as scope: + session_reads = 0 + + async def handle(request: httpx2.Request) -> httpx2.Response: + nonlocal session_reads + if request.url.path.endswith("/sessions/session_test"): + session_reads += 1 + if session_reads == (2 if during_refresh else 1): + scope.cancel() + await anyio.sleep(0) + return server.handle(request) + + async with AsyncOpenAI( + api_key="synthetic", + base_url="https://sdk-test.example/v1", + max_retries=0, + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handle), trust_env=False), + ) as client: + try: + async with client.beta.agents.sessions.stream("session_test") as stream: + assert during_refresh + await stream.until_done() + pytest.fail("Cancelled attachment must not finish successfully") + except anyio.get_cancelled_exc_class() as error: + cancelled.append(error) + raise + + anyio.run(run, backend=backend) + assert len(cancelled) == 1 + assert server.body.closed + + +async def test_manual_refresh_preserves_observed_function_diagnostics( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [turn("waiting")] + server.retrieve_status = "waiting" + server.status = "requires_action" + server.manual_after_reads = 2 + server.manual_actions = [manual_action("browser_authentication")] + approval: Any = call(call_id="approval") + approval["item"] = { + "type": "computer_use_approval_request", + "id": "request_test", + "turn_id": "turn_root", + "request_id": "request_test", + "request": server.manual_actions[0]["request"], + } + server.body = EventBody([call(), approval]) + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + await stream.__anext__() + with pytest.raises(AgentTurnResultError) as caught: + await stream.__anext__() + else: + with sdk.beta.agents.sessions.stream("session_test").with_result_collection() as stream: + next(stream) + with pytest.raises(AgentTurnResultError) as caught: + next(stream) + assert [action.type for action in caught.value.required_actions] == [ + "function_call", + "computer_use_approval_request", + ] + assert not server.inputs() + + +@pytest.mark.parametrize("event_kind", ["created", "function_call"]) +@pytest.mark.parametrize("collect", [False, True]) +async def test_successor_root_stops_attachment_before_dispatch_even_when_selected_read_lags( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, event_kind: str, collect: bool +) -> None: + server.retrieve_status = "in_progress" + # Both the selected turn GET and latest-turn index still describe the old root. + successor_call: Any = call(call_id="successor_call") + successor_call["turn_id"] = "successor" + successor_call["item"]["turn_id"] = "successor" + server.body = EventBody( + [turn_event("created", "successor"), successor_call] if event_kind == "created" else [successor_call] + ) + handled: list[object] = [] + handlers: dict[str, ToolHandler] = {"search": lambda args: handled.append(args) or "found"} + if collect: + with pytest.raises(AgentTurnResultError, match="incomplete"): + await attach_result(sdk, handlers) + elif isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test", tool_handlers=handlers) as stream: + await stream.until_done() + else: + with sdk.beta.agents.sessions.stream("session_test", tool_handlers=handlers) as stream: + stream.until_done() + assert server.body.read_count == 1 + assert not handled + assert not server.inputs() + assert server.body.closed + + +@pytest.mark.parametrize("status", ["completed", "failed", "cancelled"]) +async def test_output_before_root_discovery_survives_lagging_history( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, status: str +) -> None: + server.turns = [] + server.items = [] + server.body = EventBody([message("observed"), turn_event("created"), turn_event(status)]) + if status == "completed": + result = await attach_result(sdk) + assert result.output_text == "observed" + else: + with pytest.raises(AgentTurnResultError, match=status) as caught: + await attach_result(sdk) + assert "".join(item.output_text for item in caught.value.messages) == "observed" + + +async def test_manual_refresh_terminal_turn_finishes_without_reading_sse( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.turns = [turn("waiting")] + server.retrieve_status = "waiting" + server.status = "requires_action" + server.manual_actions = [manual_action("browser_authentication")] + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test") as stream: + server.turns = [turn("completed")] + result = await stream.get_final_result() + else: + with sdk.beta.agents.sessions.stream("session_test") as stream: + server.turns = [turn("completed")] + result = stream.get_final_result() + assert result.output_text == "earlierlater" + assert server.body.read_count == 0 + assert not server.inputs() + + +async def test_recovery_scans_interleaved_turns_with_maximum_page_size( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer +) -> None: + server.retrieve_status = "completed" + server.items = [ + message("first", item_id="first")["item"], + message("other", item_id="other", turn_id="other_turn")["item"], + message("last", item_id="last")["item"], + ] + result = await attach_result(sdk) + assert result.output_text == "firstlast" + requests = [request for request in server.requests if request.url.path.endswith("/items")] + assert len(requests) == 3 + assert all(request.url.params["limit"] == "100" for request in requests) + + +@pytest.mark.parametrize("collect", [False, True]) +async def test_pre_discovery_output_retention_requires_opt_in_and_copies_events( + sdk: OpenAI | AsyncOpenAI, server: AttachmentServer, collect: bool +) -> None: + server.turns = [] + server.items = [] + server.body = EventBody([message("original"), turn_event("created"), turn_event("completed")]) + if isinstance(sdk, AsyncOpenAI): + async with sdk.beta.agents.sessions.stream("session_test") as stream: + if collect: + stream.with_result_collection() + observed = await stream.__anext__() + assert observed.type == "agent.session.turn.item.done" and observed.item.type == "message" + observed.item.content.clear() + if collect: + assert (await stream.get_final_result()).output_text == "original" + else: + assert stream._attachment is not None and stream._attachment.pending_messages is None + else: + with sdk.beta.agents.sessions.stream("session_test") as stream: + if collect: + stream.with_result_collection() + observed = next(stream) + assert observed.type == "agent.session.turn.item.done" and observed.item.type == "message" + observed.item.content.clear() + if collect: + assert stream.get_final_result().output_text == "original" + else: + assert stream._attachment is not None and stream._attachment.pending_messages is None