From 6aae03d6e5d58020f47e409552db0db0b5c776b1 Mon Sep 17 00:00:00 2001 From: Chinedum Echeta <60179183+cecheta@users.noreply.github.com> Date: Wed, 5 Aug 2026 10:27:02 +0000 Subject: [PATCH 1/4] Python: Allow branching from hosted Foundry conversations --- python/packages/foundry_hosting/README.md | 34 ++++--- .../_responses.py | 62 +++++++++--- .../_session_store.py | 6 +- .../foundry_hosting/tests/test_responses.py | 98 +++++++++++++++++++ 4 files changed, 171 insertions(+), 29 deletions(-) diff --git a/python/packages/foundry_hosting/README.md b/python/packages/foundry_hosting/README.md index ea8818d1e8..93c63b7e51 100644 --- a/python/packages/foundry_hosting/README.md +++ b/python/packages/foundry_hosting/README.md @@ -6,14 +6,16 @@ This package provides the integration of Agent Framework agents and workflows wi agents in addition to the Responses provider's message history. By default it uses the experimental `FoundrySessionStore` under `/.sessions` when hosted and an in-memory `SessionStore` locally. Hosted snapshots are partitioned by the -Agent Server request context's platform user ID. Snapshot filenames use the -Responses `conversation_id` or `response_id`, depending on the continuation -mode. +Agent Server request context's platform user ID. Snapshot filenames use Responses `response_id` values, with an additional +`conversation_id` snapshot that points to the latest state of each stored +conversation. Foundry's session file API exposes the hosted `$HOME` directory as `/`, so the API path `/.sessions` is stored on disk at `$HOME/.sessions`. -Workflow agents continue to use their existing checkpoint storage layout. +Workflow agents use the same continuation model for their checkpoints: every +turn is stored under its `response_id`, and stored conversations also maintain +a `conversation_id` checkpoint alias for their latest turn. ## Foundry session isolation @@ -33,11 +35,14 @@ A Foundry session controls hosted compute and filesystem lifetime and may host multiple users and Responses conversations. The Foundry session ID is not used as the MAF session identifier. -When `conversation_id` is used, the host reads and writes the same snapshot -under that ID. When `previous_response_id` is used, the host reads that response -snapshot, runs the loaded MAF session, and writes the updated snapshot under the -current response's `response_id`. Multiple responses can therefore branch from -one prior response without overwriting its snapshot. +When `conversation_id` is used, the host reads the latest snapshot under that +ID, then writes the updated state under both the current `response_id` and the +conversation ID. This preserves every turn while keeping conversation +continuation pointed at the latest state. When `previous_response_id` is used, +the host reads that response snapshot, runs the loaded MAF session, and writes +the updated snapshot under the current response's `response_id`. Responses can +therefore branch from any prior turn without overwriting its snapshot, +including turns originally created through a conversation. Foundry does not infer the hosted `agent_session_id` from `previous_response_id`. Callers using response chains must also reuse the @@ -46,15 +51,18 @@ same sandbox and `$HOME/.sessions` filesystem. Conversation objects bind to a stable hosted session automatically. Workflow checkpoints and function approvals preserve the existing Foundry -Hosting layout. Hosted paths insert the validated raw platform user ID: +Hosting roots. Hosted paths insert the validated raw platform user ID: ```text -/.checkpoints/// +/.checkpoints/// +/.checkpoints/// /.function_approvals//approval_requests.json ``` -Local workflow checkpoints use `{cwd}/.checkpoints//`, and local -function approvals remain in memory. +The conversation directory is a latest-state alias. Each response directory +retains the final checkpoint selected for that turn, allowing a later request +to branch from it. Local workflow checkpoints use the same layout without +``, and local function approvals remain in memory. Hosted requests require container protocol `2.0.0`. The v2-only request `call_id` is checked before session, checkpoint, or approval storage is used, diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index 49882462ce..e2ecb1461b 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -569,7 +569,9 @@ async def _handle_inner_agent( Foundry sessions govern hosted compute and filesystem lifetime and may serve multiple users and Responses conversations. Conversation mode - reads and writes one MAF session snapshot under ``conversation_id``. + reads the latest MAF session snapshot under ``conversation_id`` and + writes each turn under both its immutable ``response_id`` and the + conversation ID. Response chaining reads the snapshot under ``previous_response_id`` and writes the updated session under the current ``response_id``, allowing branches without changing the MAF session's own identifier. The request @@ -688,7 +690,9 @@ async def _handle_inner_agent( session.state.pop(_HOSTED_RESPONSES_HISTORY_SOURCE_ID, None) if session is not None and self._session_store is not None: try: - await self._session_store.set(context.conversation_id or context.response_id, session) + await self._session_store.set(context.response_id, session) + if context.conversation_id is not None: + await self._session_store.set(context.conversation_id, session) except Exception as save_error: if request_interrupted: logger.error( @@ -808,15 +812,10 @@ async def _handle_inner_workflow( if latest_checkpoint is not None: latest_checkpoint_id = latest_checkpoint.checkpoint_id - # Storage that will receive checkpoints written during this turn. - # When the caller chains with previous_response_id, the next turn - # will reference the current response_id as its previous_response_id, - # so new checkpoints must land under the current response_id (or the - # conversation_id when set). When conversation_id is set, this - # matches restore_storage; when only previous_response_id was - # supplied, restore_storage points at the *prior* response's - # directory and write_storage points at the *current* response's. - write_context_id = context.conversation_id or context.response_id + # Each turn writes to response-addressed checkpoint storage. + # Conversation continuation is updated from its latest checkpoint + # after the run. + write_context_id = context.response_id write_storage = _checkpoint_storage_for_context( self._checkpoint_storage_path, write_context_id, @@ -869,7 +868,12 @@ async def _handle_inner_workflow( ): yield item - await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name) + await self._finalize_workflow_checkpoints( + write_storage, + workflow_name=self._agent.workflow.name, + conversation_id=context.conversation_id, + user_id=user_id, + ) yield response_event_stream.emit_completed() return @@ -895,13 +899,45 @@ async def _handle_inner_workflow( for event in tracker.close(): yield event - await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name) + await self._finalize_workflow_checkpoints( + write_storage, + workflow_name=self._agent.workflow.name, + conversation_id=context.conversation_id, + user_id=user_id, + ) yield response_event_stream.emit_completed() except Exception as ex: logger.exception("Failed to produce response for workflow agent") for event in self._emit_failure(response_event_stream, tracker, ex): yield event + async def _finalize_workflow_checkpoints( + self, + response_storage: FileCheckpointStorage, + *, + workflow_name: str, + conversation_id: str | None, + user_id: str | None, + ) -> None: + """Keep one response checkpoint and update the conversation's latest-state alias.""" + await self._delete_not_latest_checkpoints(response_storage, workflow_name) + if conversation_id is None: + return + + latest_checkpoint = await response_storage.get_latest(workflow_name=workflow_name) + if latest_checkpoint is None: + return + if self._checkpoint_storage_path is None: + raise RuntimeError("Checkpoint storage path is not configured for workflow agent.") + + conversation_storage = _checkpoint_storage_for_context( + self._checkpoint_storage_path, + conversation_id, + user_id=user_id, + ) + await conversation_storage.save(latest_checkpoint) + await self._delete_not_latest_checkpoints(conversation_storage, workflow_name) + @staticmethod async def _delete_not_latest_checkpoints(checkpoint_storage: FileCheckpointStorage, workflow_name: str) -> None: """Delete all checkpoints except the latest one. diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_session_store.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_session_store.py index 3fb90154ee..a151faae54 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_session_store.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_session_store.py @@ -19,9 +19,9 @@ class FoundrySessionStore(FileSessionStore): A Foundry hosted session controls platform compute and filesystem lifetime and may host multiple users and Responses conversations. A MAF :class:`AgentSession` contains framework context state. Snapshots are keyed - by ``conversation_id`` for stored conversations or by Responses - ``response_id`` for response chains; these storage keys are independent of - the MAF session's own identifier. + by every Responses ``response_id``. Stored conversations also update a + snapshot keyed by ``conversation_id`` as an alias for the latest turn. + These storage keys are independent of the MAF session's own identifier. This implementation currently persists through :class:`FileSessionStore`, with each validated platform user ID as a child directory. The diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index cc1d863304..a6af9c1c29 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -685,9 +685,46 @@ async def run_with_state(*args: Any, **kwargs: Any) -> AgentResponse: assert seen_session_ids[2] == seen_session_ids[0] assert (tmp_path / "user-1" / "conversation-1.json").is_file() assert (tmp_path / "user-1" / "conversation-2.json").is_file() + assert (tmp_path / "user-1" / "response-1.json").is_file() + assert (tmp_path / "user-1" / "response-2.json").is_file() + assert (tmp_path / "user-1" / "response-3.json").is_file() assert agent.create_session.call_count == 2 assert all(item.kwargs == {} for item in agent.create_session.call_args_list) + async def test_conversation_response_snapshots_support_branching(self) -> None: + seen_counts: list[int] = [] + + async def run_with_state(*args: Any, **kwargs: Any) -> AgentResponse: + session = kwargs["session"] + assert isinstance(session, AgentSession) + count = int(session.state.get("turn_count", 0)) + 1 + session.state["turn_count"] = count + seen_counts.append(count) + return AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text(f"turn {count}")])]) + + agent = _make_agent() + agent.run = AsyncMock(side_effect=run_with_state) + store = SessionStore() + server = _make_server(agent, session_store=store) + + first = await _post(server, input_text="first", conversation_id="conversation-1") + await _post(server, input_text="second", conversation_id="conversation-1") + branch = await _post(server, input_text="branch", previous_response_id=first.json()["id"]) + + assert first.status_code == 200 + assert branch.status_code == 200 + assert seen_counts == [1, 2, 2] + + conversation_snapshot = await store.get("conversation-1") + first_snapshot = await store.get(first.json()["id"]) + branch_snapshot = await store.get(branch.json()["id"]) + assert conversation_snapshot is not None + assert first_snapshot is not None + assert branch_snapshot is not None + assert conversation_snapshot.state["turn_count"] == 2 + assert first_snapshot.state["turn_count"] == 1 + assert branch_snapshot.state["turn_count"] == 2 + async def test_previous_response_chain_restores_session_state(self) -> None: seen_counts: list[int] = [] seen_session_ids: list[str] = [] @@ -4893,6 +4930,67 @@ async def test_basic_text_response_streaming(self) -> None: text_done = [e for e in events if e["event"] == "response.output_text.done"] assert any(e["data"]["text"] == "hello stream" for e in text_done) + @pytest.mark.parametrize("stream", [False, True]) + async def test_conversation_response_checkpoints_support_branching(self, tmp_path: Path, stream: bool) -> None: + @executor + async def count_turns(messages: list[Message], ctx: WorkflowContext[Any, AgentResponse]) -> None: + del messages + turn_count = int(ctx.get_state("turn_count", 0)) + 1 + ctx.set_state("turn_count", turn_count) + await ctx.yield_output( + AgentResponse(messages=[Message("assistant", [Content.from_text(f"turn {turn_count}")])]) + ) + + workflow_agent = WorkflowAgent( + workflow=WorkflowBuilder(start_executor=count_turns).build(), + name="Counting Workflow Agent", + ) + server = _make_server(workflow_agent) + server._checkpoint_storage_path = str(tmp_path) # pyright: ignore[reportPrivateUsage] + + def response_body(response: httpx.Response) -> dict[str, Any]: + if not stream: + return response.json() + return _parse_sse_events(response.text)[-1]["data"]["response"] + + first = await _post( + server, + input_text="first", + conversation_id="conversation-1", + stream=stream, + ) + second = await _post( + server, + input_text="second", + conversation_id="conversation-1", + stream=stream, + ) + first_body = response_body(first) + branch = await _post( + server, + input_text="branch", + previous_response_id=first_body["id"], + stream=stream, + ) + + assert first.status_code == 200 + assert second.status_code == 200 + assert branch.status_code == 200 + assert first_body["status"] == "completed" + assert (tmp_path / first_body["id"]).is_dir() + assert (tmp_path / response_body(second)["id"]).is_dir() + assert (tmp_path / response_body(branch)["id"]).is_dir() + assert (tmp_path / "conversation-1").is_dir() + + branch_text = [ + part["text"] + for item in response_body(branch)["output"] + if item["type"] == "message" + for part in item.get("content", []) + if part["type"] == "output_text" + ] + assert branch_text == ["turn 2"] + async def test_non_streaming_emits_mcp_approval_request_and_persists_to_storage(self) -> None: workflow_agent, mock_agent = _build_approval_workflow_agent(approval_request_id="apr_wf_ns") server = _make_server(workflow_agent) From 02c34afd5d71d3976afb212a1b131fd8cc0d2f68 Mon Sep 17 00:00:00 2001 From: Chinedum Echeta <60179183+cecheta@users.noreply.github.com> Date: Wed, 5 Aug 2026 10:40:39 +0000 Subject: [PATCH 2/4] Fix formatting in README.md for clarity on snapshot filenames --- python/packages/foundry_hosting/README.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/python/packages/foundry_hosting/README.md b/python/packages/foundry_hosting/README.md index 93c63b7e51..219da08f29 100644 --- a/python/packages/foundry_hosting/README.md +++ b/python/packages/foundry_hosting/README.md @@ -6,9 +6,9 @@ This package provides the integration of Agent Framework agents and workflows wi agents in addition to the Responses provider's message history. By default it uses the experimental `FoundrySessionStore` under `/.sessions` when hosted and an in-memory `SessionStore` locally. Hosted snapshots are partitioned by the -Agent Server request context's platform user ID. Snapshot filenames use Responses `response_id` values, with an additional -`conversation_id` snapshot that points to the latest state of each stored -conversation. +Agent Server request context's platform user ID. Snapshot filenames use +Responses `response_id` values, with an additional `conversation_id` snapshot +that points to the latest state of each stored conversation. Foundry's session file API exposes the hosted `$HOME` directory as `/`, so the API path `/.sessions` is stored on disk at `$HOME/.sessions`. From 95eb679bd1d46502055494c1cc3e4ac3b9da41cf Mon Sep 17 00:00:00 2001 From: Chinedum Echeta <60179183+cecheta@users.noreply.github.com> Date: Wed, 5 Aug 2026 11:02:05 +0000 Subject: [PATCH 3/4] Reject previous response ID when a conversation ID is provided --- .../_responses.py | 2 ++ .../foundry_hosting/tests/test_responses.py | 16 ++++++++++++++++ 2 files changed, 18 insertions(+) diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index e2ecb1461b..59561bf34e 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -593,6 +593,8 @@ async def _handle_inner_agent( try: approval_storage = self._approval_storage_for_request() + if request.previous_response_id is not None and context.conversation_id is not None: + raise RuntimeError("Previous response ID cannot be used in conjunction with conversation ID.") read_session_id = context.conversation_id or request.previous_response_id if self._session_store is None: if read_session_id is not None: diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index a6af9c1c29..c6856f0eae 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -501,6 +501,22 @@ async def test_continuation_requires_session_store(self, continuation: str) -> N error = getattr(failed_response, "error", None) assert "Session storage is required" in getattr(error, "message", "") + async def test_previous_response_rejected_with_conversation(self) -> None: + agent = _make_agent() + server = _make_server(agent) + response = await _post( + server, + previous_response_id="caresp_aaaaaaaaaaaaaaaa00" + "1" * 32, + conversation_id="conversation-1", + ) + + assert response.status_code == 200 + body = response.json() + assert body["status"] == "failed" + assert body["error"]["message"] == ("Previous response ID cannot be used in conjunction with conversation ID.") + agent.run.assert_not_called() + agent.create_session.assert_not_called() + async def test_previous_response_requires_existing_snapshot(self, tmp_path: Path) -> None: agent = _make_agent() server = _make_server(agent, session_store=FoundrySessionStore(tmp_path)) From e4d58cb351e0bfd61744b9c4047b56310fa0a4dc Mon Sep 17 00:00:00 2001 From: Chinedum Echeta <60179183+cecheta@users.noreply.github.com> Date: Thu, 6 Aug 2026 13:37:35 +0000 Subject: [PATCH 4/4] Python: Enhance error handling and checkpoint management in ResponsesHostServer --- .../_responses.py | 171 ++++++++++-------- .../foundry_hosting/tests/test_responses.py | 89 +++++++++ 2 files changed, 183 insertions(+), 77 deletions(-) diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py index 59561bf34e..e950377d59 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py @@ -824,89 +824,106 @@ async def _handle_inner_workflow( user_id=user_id, ) - # Multi-turn pattern: when we have a prior checkpoint, restore it - # first (drive the workflow back to idle with prior state intact), - # then make a separate call that delivers the new user input. This - # depends on Workflow.run preserving shared state across calls. The - # restore-only call may yield events from any pending in-flight - # work in the checkpoint; we consume those internally here so they - # don't surface to the response stream as duplicates. - # - # If the restored checkpoint had pending request_info events, the - # restore-only call replays them through - # ``WorkflowAgent._convert_workflow_event_to_agent_response_updates`` - # and populates ``self._agent.pending_requests``. That is the correct - # state: those requests are genuinely outstanding, and the next - # ``run(input_messages, ...)`` call may contain ``function_call_output`` - # items (carried as FunctionResult/FunctionApprovalResponse content) - # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`. - if latest_checkpoint_id is not None: - if is_streaming_request: - async for _ in self._agent.run( - stream=True, - checkpoint_id=latest_checkpoint_id, - checkpoint_storage=restore_storage, - ): - pass - else: - await self._agent.run( + request_failure: Exception | None = None + request_interrupted = False + try: + # Multi-turn pattern: when we have a prior checkpoint, restore it + # first (drive the workflow back to idle with prior state intact), + # then make a separate call that delivers the new user input. This + # depends on Workflow.run preserving shared state across calls. The + # restore-only call may yield events from any pending in-flight + # work in the checkpoint; we consume those internally here so they + # don't surface to the response stream as duplicates. + # + # If the restored checkpoint had pending request_info events, the + # restore-only call replays them through + # ``WorkflowAgent._convert_workflow_event_to_agent_response_updates`` + # and populates ``self._agent.pending_requests``. That is the correct + # state: those requests are genuinely outstanding, and the next + # ``run(input_messages, ...)`` call may contain ``function_call_output`` + # items (carried as FunctionResult/FunctionApprovalResponse content) + # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`. + if latest_checkpoint_id is not None: + if is_streaming_request: + async for _ in self._agent.run( + stream=True, + checkpoint_id=latest_checkpoint_id, + checkpoint_storage=restore_storage, + ): + pass + else: + await self._agent.run( + stream=False, + checkpoint_id=latest_checkpoint_id, + checkpoint_storage=restore_storage, + ) + + if not is_streaming_request: + # Run the agent in non-streaming mode with the new user input. + response = await self._agent.run( + input_messages, stream=False, - checkpoint_id=latest_checkpoint_id, - checkpoint_storage=restore_storage, + checkpoint_storage=write_storage, ) - if not is_streaming_request: - # Run the agent in non-streaming mode with the new user input. - response = await self._agent.run( - input_messages, - stream=False, - checkpoint_storage=write_storage, - ) - - async for item in _to_outputs_for_messages( - response_event_stream, - response.messages, - approval_storage=approval_storage, - ): - yield item - - await self._finalize_workflow_checkpoints( - write_storage, - workflow_name=self._agent.workflow.name, - conversation_id=context.conversation_id, - user_id=user_id, - ) - yield response_event_stream.emit_completed() - return + async for item in _to_outputs_for_messages( + response_event_stream, + response.messages, + approval_storage=approval_storage, + ): + yield item + else: + tracker = _OutputItemTracker(response_event_stream) - tracker = _OutputItemTracker(response_event_stream) + # Run the workflow agent in streaming mode with the new user input. + async for update in self._agent.run( + input_messages, + stream=True, + checkpoint_storage=write_storage, + ): + for content in update.contents: + for event in tracker.handle(content): + yield event + if tracker.needs_async: + async for item in _to_outputs( + response_event_stream, content, approval_storage=approval_storage + ): + yield item + tracker.needs_async = False - # Run the workflow agent in streaming mode with the new user input. - async for update in self._agent.run( - input_messages, - stream=True, - checkpoint_storage=write_storage, - ): - for content in update.contents: - for event in tracker.handle(content): + # Close any remaining active builder + for event in tracker.close(): yield event - if tracker.needs_async: - async for item in _to_outputs( - response_event_stream, content, approval_storage=approval_storage - ): - yield item - tracker.needs_async = False - - # Close any remaining active builder - for event in tracker.close(): - yield event - - await self._finalize_workflow_checkpoints( - write_storage, - workflow_name=self._agent.workflow.name, - conversation_id=context.conversation_id, - user_id=user_id, - ) + except asyncio.CancelledError: + request_interrupted = True + raise + except GeneratorExit: + request_interrupted = True + raise + except Exception as ex: + request_failure = ex + raise + finally: + try: + await self._finalize_workflow_checkpoints( + write_storage, + workflow_name=self._agent.workflow.name, + conversation_id=context.conversation_id, + user_id=user_id, + ) + except Exception as save_error: + if request_interrupted: + logger.error( + "Failed to finalize workflow checkpoints while unwinding an interrupted request", + exc_info=(type(save_error), save_error, save_error.__traceback__), + ) + elif request_failure is not None: + logger.error( + "Failed to finalize workflow checkpoints after a workflow failure", + exc_info=(type(save_error), save_error, save_error.__traceback__), + ) + else: + raise yield response_event_stream.emit_completed() except Exception as ex: logger.exception("Failed to produce response for workflow agent") diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index c6856f0eae..c04f40a531 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -5007,6 +5007,95 @@ def response_body(response: httpx.Response) -> dict[str, Any]: ] assert branch_text == ["turn 2"] + async def test_failed_conversation_workflow_promotes_latest_response_checkpoint(self, tmp_path: Path) -> None: + workflow_agent = _build_text_workflow_agent("ignored") + checkpoint = WorkflowCheckpoint( + workflow_name=workflow_agent.workflow.name, + graph_signature_hash="hash", + ) + + async def failing_run(*args: Any, **kwargs: Any) -> AgentResponse: + await kwargs["checkpoint_storage"].save(checkpoint) + raise RuntimeError("workflow failed") + + server = _make_server(workflow_agent) + server._checkpoint_storage_path = str(tmp_path) # pyright: ignore[reportPrivateUsage] + request = CreateResponse(model="m", input="hi") + context = ResponseContext( + response_id="response-1", + conversation_id="conversation-1", + mode_flags=MagicMock(), + ) + + with ( + patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])), + patch.object(workflow_agent, "run", side_effect=failing_run), + ): + events = [ + event + async for event in server._handle_inner_workflow( # pyright: ignore[reportPrivateUsage] + request, + context, + ) + ] + + conversation_storage = FileCheckpointStorage(tmp_path / "conversation-1") + latest = await conversation_storage.get_latest(workflow_name=workflow_agent.workflow.name) + assert latest is not None + assert latest.checkpoint_id == checkpoint.checkpoint_id + assert getattr(events[-1], "type", None) == "response.failed" + + @pytest.mark.parametrize("interruption", ["cancel", "close"]) + async def test_interrupted_conversation_workflow_promotes_latest_response_checkpoint( + self, + tmp_path: Path, + interruption: str, + ) -> None: + workflow_agent = _build_text_workflow_agent("ignored") + checkpoint = WorkflowCheckpoint( + workflow_name=workflow_agent.workflow.name, + graph_signature_hash="hash", + ) + + async def updates(checkpoint_storage: FileCheckpointStorage) -> AsyncIterator[AgentResponseUpdate]: + await checkpoint_storage.save(checkpoint) + yield AgentResponseUpdate(contents=[Content.from_text("started")], role="assistant") + await asyncio.Event().wait() + + def streaming_run(*args: Any, **kwargs: Any) -> AsyncIterator[AgentResponseUpdate]: + return updates(kwargs["checkpoint_storage"]) + + server = _make_server(workflow_agent) + server._checkpoint_storage_path = str(tmp_path) # pyright: ignore[reportPrivateUsage] + request = CreateResponse(model="m", input="hi", stream=True) + context = ResponseContext( + response_id="response-1", + conversation_id="conversation-1", + mode_flags=MagicMock(), + ) + + with ( + patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])), + patch.object(workflow_agent, "run", side_effect=streaming_run), + ): + handler = cast( + AsyncGenerator[Any, None], + server._handle_inner_workflow(request, context), # pyright: ignore[reportPrivateUsage] + ) + await anext(handler) + await anext(handler) + await anext(handler) + if interruption == "cancel": + with pytest.raises(asyncio.CancelledError): + await handler.athrow(asyncio.CancelledError()) + else: + await handler.aclose() + + conversation_storage = FileCheckpointStorage(tmp_path / "conversation-1") + latest = await conversation_storage.get_latest(workflow_name=workflow_agent.workflow.name) + assert latest is not None + assert latest.checkpoint_id == checkpoint.checkpoint_id + async def test_non_streaming_emits_mcp_approval_request_and_persists_to_storage(self) -> None: workflow_agent, mock_agent = _build_approval_workflow_agent(approval_request_id="apr_wf_ns") server = _make_server(workflow_agent)