Fix formatting

python/packages/core/agent_framework/observability.py

@@ -2791,7 +2791,9 @@ def capture_exception(span: trace.Span, exception: Exception, timestamp: int | N
     span.set_status(status=trace.StatusCode.ERROR, description=repr(exception))
 
 
-def _capture_system_instructions_latest_experimental(span: trace.Span, system_instructions: str | list[str] | None) -> None:
+def _capture_system_instructions_latest_experimental(
+    span: trace.Span, system_instructions: str | list[str] | None
+) -> None:
     """Capture system instructions on a span."""
     if not OBSERVABILITY_SETTINGS.use_latest_experimental_gen_ai_semconv or not system_instructions:
         return

python/packages/foundry_hosting/README.md

@@ -46,4 +46,4 @@ locally. Stored checkpoints are scoped under `checkpoints`.
 
 `ResponsesHostServer` persists function approvals durably. By default, it uses the
 `FoundryFunctionApprovalStore`, backed by Foundry storage when hosted and file-based
-storage locally. Stored approvals are scoped under `function_approvals`.
+storage locally. Stored approvals are scoped under `function_approvals`.

python/packages/foundry_hosting/agent_framework_foundry_hosting/_state_store.py

@@ -90,7 +90,6 @@ class FoundryCheckpointStore:
         return await FoundryStateStore.get_or_create(
             f"{self.DEFAULT_ROOT_SCOPE}/{self.context_id}",
             user_isolation=True,
-            user_id=self.platform_context.user_id,
         )
 
     async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID:
@@ -232,7 +231,6 @@ class FoundryFunctionApprovalStore:
         return await FoundryStateStore.get_or_create(
             self.DEFAULT_ROOT_SCOPE,
             user_isolation=True,
-            user_id=self.platform_context.user_id,
         )
 
     async def save_approval_request(self, approval_request_id: str, request: Content) -> None:
@@ -280,7 +278,6 @@ class FoundryAgentSessionStore(SessionStore):
         return await FoundryStateStore.get_or_create(
             f"{self.DEFAULT_ROOT_SCOPE}",
             user_isolation=True,
-            user_id=self.platform_context.user_id,
         )
 
     async def get(self, session_id: str) -> AgentSession | None:

python/packages/foundry_hosting/tests/test_responses.py

@@ -15,6 +15,7 @@ import json
 import uuid
 from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Literal, cast, overload
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -47,8 +48,15 @@ from agent_framework import (
     tool,
 )
 from azure.ai.agentserver.core import get_request_context
-from azure.ai.agentserver.responses import InMemoryResponseProvider, ResponseContext
+from azure.ai.agentserver.responses import (
+    FileResponseStore,
+    InMemoryResponseProvider,
+    ResponseContext,
+    ResponseExitForRecovery,
+    ResponsesServerOptions,
+)
 from azure.ai.agentserver.responses.models import CreateResponse, Item, OutputItem
+from azure.ai.agentserver.responses.streaming._checkpoint import ResponseCheckpointEvent
 from mcp import McpError
 from mcp.types import ErrorData
 from typing_extensions import Any
@@ -59,6 +67,7 @@ from agent_framework_foundry_hosting._responses import (
     ConsentError,
     _item_to_message,  # pyright: ignore[reportPrivateUsage]
     _output_item_to_message,  # pyright: ignore[reportPrivateUsage]
+    _OutputItemTracker,  # pyright: ignore[reportPrivateUsage]
     consent_url_from_error,
 )
 from agent_framework_foundry_hosting._state_store import (
@@ -404,16 +413,38 @@ class TestResponsesHostServerInit:
         with pytest.raises(RuntimeError, match="history provider"):
             ResponsesHostServer(agent)
 
+    def test_init_rejects_resilient_background_for_non_workflow_agent(self, tmp_path: Path) -> None:
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("hi")])])
+        )
+        with pytest.raises(RuntimeError, match="resilient_background"):
+            ResponsesHostServer(
+                agent,
+                store=FileResponseStore(storage_dir=tmp_path),
+                options=ResponsesServerOptions(resilient_background=True),
+            )
+
+    def test_init_rejects_steerable_conversations_for_workflow_agent(self) -> None:
+        workflow_agent = _build_text_workflow_agent("hello from workflow")
+        with pytest.raises(RuntimeError, match="steerable_conversations"):
+            ResponsesHostServer(
+                cast(SupportsAgentRun, workflow_agent),
+                store=InMemoryResponseProvider(),
+                options=ResponsesServerOptions(steerable_conversations=True),
+            )
+
     async def test_previous_response_requires_existing_agent_session(self) -> None:
         agent = _make_agent()
         server = _make_server(agent, session_store=SessionStore())
         request = CreateResponse(model="m", input="hi", previous_response_id="response-missing")
         context = ResponseContext(response_id="response-current", mode_flags=MagicMock())
 
-        handler = await server._handle_response(request, context, asyncio.Event())  # pyright: ignore[reportPrivateUsage]
+        handler = server._handle_response(request, context, asyncio.Event())  # pyright: ignore[reportPrivateUsage]
         events = [event async for event in handler]
 
-        failed_events = [event for event in events if event.get("type") == "response.failed"]
+        failed_events = [
+            event for event in events if isinstance(event, Mapping) and event.get("type") == "response.failed"
+        ]
         assert len(failed_events) == 1
         failed_event = cast(Mapping[str, Any], failed_events[0])
         response = cast(Mapping[str, Any], failed_event["response"])
@@ -464,6 +495,7 @@ class TestAgentSessionPersistence:
 
         provider = server._session_storage_provider  # pyright: ignore[reportPrivateUsage]
         assert provider is not None
+
         session_store = provider.get_store(config=server.config, platform_context=get_request_context())
         assert session_store is not None
         first_session = await session_store.get(first.json()["id"])
@@ -669,7 +701,7 @@ class TestAgentSessionPersistence:
         ):
             handler = cast(
                 AsyncGenerator[Any, None],
-                server._handle_inner_agent(request, context),  # pyright: ignore[reportPrivateUsage]
+                server._handle_response(request, context, asyncio.Event()),  # pyright: ignore[reportPrivateUsage]
             )
             await anext(handler)
             await anext(handler)
@@ -706,7 +738,7 @@ class TestAgentSessionPersistence:
         ):
             handler = cast(
                 AsyncGenerator[Any, None],
-                server._handle_inner_agent(request, context),  # pyright: ignore[reportPrivateUsage]
+                server._handle_response(request, context, asyncio.Event()),  # pyright: ignore[reportPrivateUsage]
             )
             await anext(handler)
             await anext(handler)
@@ -717,6 +749,129 @@ class TestAgentSessionPersistence:
         assert stored is not None
         assert stored.state["started"] is True
 
+    async def test_cancellation_signal_stops_streaming_and_completes(self) -> None:
+        """Steering/explicit-cancel: the loop must break promptly, and the response still completes."""
+        store = SessionStore()
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(contents=[Content.from_text("one")], role="assistant"),
+                AgentResponseUpdate(contents=[Content.from_text("two")], role="assistant"),
+                AgentResponseUpdate(contents=[Content.from_text("three")], role="assistant"),
+            ]
+        )
+        server = _make_server(agent, session_store=store)
+        request = CreateResponse(model="m", input="hi", stream=True)
+        context = ResponseContext(response_id="response-1", mode_flags=MagicMock())
+        cancellation_signal = asyncio.Event()
+
+        with (
+            patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])),
+        ):
+            handler = cast(
+                AsyncGenerator[Any, None],
+                server._handle_response(request, context, cancellation_signal),  # pyright: ignore[reportPrivateUsage]
+            )
+            events: list[Any] = []
+            async for event in handler:
+                events.append(event)
+                if isinstance(event, Mapping) and event.get("type") == "response.output_text.delta":
+                    break
+            # Cancellation arrives after the first delta; the loop must not process "two"/"three".
+            cancellation_signal.set()
+            events.extend([event async for event in handler])
+
+        types = [event.get("type") for event in events if isinstance(event, Mapping)]
+        assert types.count("response.output_text.delta") == 1
+        assert types[-1] == "response.completed"
+
+        stored = await store.get("response-1")
+        assert stored is not None
+
+    async def test_cancellation_signal_preempts_stuck_agent_call(self) -> None:
+        """Steering/explicit-cancel must interrupt an agent call stuck awaiting a slow model/tool
+        response, not merely be checked between already-produced updates."""
+        store = SessionStore()
+        gate = asyncio.Event()  # Never set: simulates a model/tool call that never returns.
+        agent = _make_agent()
+
+        async def _stream_gen() -> AsyncIterator[AgentResponseUpdate]:
+            await gate.wait()
+            yield AgentResponseUpdate(contents=[Content.from_text("too late")], role="assistant")
+
+        def run_streaming(*_args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]:
+            del _args, kwargs
+            return ResponseStream(_stream_gen(), finalizer=AgentResponse.from_updates)
+
+        agent.run = MagicMock(side_effect=run_streaming)
+        server = _make_server(agent, session_store=store)
+        request = CreateResponse(model="m", input="hi", stream=True)
+        context = ResponseContext(response_id="response-1", mode_flags=MagicMock())
+        cancellation_signal = asyncio.Event()
+
+        with (
+            patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])),
+        ):
+            handler = cast(
+                AsyncGenerator[Any, None],
+                server._handle_response(request, context, cancellation_signal),  # pyright: ignore[reportPrivateUsage]
+            )
+            await anext(handler)  # response.created
+            await anext(handler)  # response.in_progress
+            cancellation_signal.set()  # Fires while the agent is stuck awaiting `gate`.
+
+            async def _drain() -> list[Any]:
+                return [event async for event in handler]
+
+            # Bounded well below `gate` never being set: proves cancellation preempted the stuck
+            # call instead of only being observed after it (eventually) produced an update.
+            events = await asyncio.wait_for(_drain(), timeout=1.0)
+
+        types = [event.get("type") for event in events if isinstance(event, Mapping)]
+        assert "response.output_text.delta" not in types
+        assert types[-1] == "response.completed"
+
+    async def test_consumer_failure_cancels_agent_stream_driver_task(self) -> None:
+        """A crash in the consumer (`_OutputItemTracker.handle`) must not leave the background
+        driver task that pumps the agent stream running as an orphaned task."""
+        gate = asyncio.Event()  # Never set: would hang forever if the driver task isn't cancelled.
+        agent = _make_agent()
+
+        async def _stream_gen() -> AsyncIterator[AgentResponseUpdate]:
+            yield AgentResponseUpdate(contents=[Content.from_text("first")], role="assistant")
+            await gate.wait()
+            yield AgentResponseUpdate(contents=[Content.from_text("too late")], role="assistant")
+
+        def run_streaming(*_args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]:
+            del _args, kwargs
+            return ResponseStream(_stream_gen(), finalizer=AgentResponse.from_updates)
+
+        agent.run = MagicMock(side_effect=run_streaming)
+        server = _make_server(agent)
+        request = CreateResponse(model="m", input="hi", stream=True)
+        context = ResponseContext(response_id="response-1", mode_flags=MagicMock())
+
+        tasks_before = asyncio.all_tasks()
+        with (
+            patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])),
+            patch.object(_OutputItemTracker, "handle", side_effect=RuntimeError("tracker exploded")),
+        ):
+            handler = cast(
+                AsyncGenerator[Any, None],
+                server._handle_response(request, context, asyncio.Event()),  # pyright: ignore[reportPrivateUsage]
+            )
+            events = [event async for event in handler]
+
+        types = [event.get("type") for event in events if isinstance(event, Mapping)]
+        assert types[-1] == "response.failed"
+
+        # Give any cancellation triggered during teardown a chance to finish propagating.
+        await asyncio.sleep(0)
+        leaked = asyncio.all_tasks() - tasks_before - {asyncio.current_task()}
+        assert not leaked, f"driver task leaked: {leaked}"
+
 
 # endregion
 
@@ -985,6 +1140,57 @@ class TestStreaming:
         assert len(done_events) == 1
         assert done_events[0]["data"]["text"] == "Hello world!"
 
+    async def test_usage_is_aggregated_in_completed_response(self, caplog: pytest.LogCaptureFixture) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(contents=[Content.from_text("Hello ")], role="assistant"),
+                AgentResponseUpdate(
+                    contents=[
+                        Content.from_usage({
+                            "input_token_count": 10,
+                            "output_token_count": 2,
+                            "total_token_count": 12,
+                            "cache_read_input_token_count": 3,
+                            "reasoning_output_token_count": 1,
+                        })
+                    ],
+                    role="assistant",
+                ),
+                AgentResponseUpdate(contents=[Content.from_text("world!")], role="assistant"),
+                AgentResponseUpdate(
+                    contents=[
+                        Content.from_usage({
+                            "input_token_count": 5,
+                            "output_token_count": 4,
+                            "total_token_count": 9,
+                            "cache_read_input_token_count": 2,
+                            "reasoning_output_token_count": 2,
+                        })
+                    ],
+                    role="assistant",
+                ),
+            ]
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+        assert types[-1] == "response.completed"
+        assert types.count("response.output_item.added") == 1
+        assert types.count("response.output_text.delta") == 2
+        completed = events[-1]["data"]["response"]
+        assert completed["usage"] == {
+            "input_tokens": 15,
+            "input_tokens_details": {"cached_tokens": 5},
+            "output_tokens": 6,
+            "output_tokens_details": {"reasoning_tokens": 3},
+            "total_tokens": 21,
+        }
+        assert "Content type 'usage' is not supported yet" not in caplog.text
+
     async def test_function_call_streaming(self) -> None:
         agent = _make_agent(
             stream_updates=[
@@ -3578,13 +3784,16 @@ class TestCheckpointContextValidation:
         with patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])):
             events = [
                 event
-                async for event in server._handle_inner_workflow(  # pyright: ignore[reportPrivateUsage]
+                async for event in server._handle_response(  # pyright: ignore[reportPrivateUsage]
                     request,
                     ResponseContext(**context_kwargs),
+                    asyncio.Event(),
                 )
             ]
 
-        failed_events = [event for event in events if event.get("type") == "response.failed"]
+        failed_events = [
+            event for event in events if isinstance(event, Mapping) and event.get("type") == "response.failed"
+        ]
         assert len(failed_events) == 1
         failed_event = cast(Mapping[str, Any], failed_events[0])
         response = cast(Mapping[str, Any], failed_event["response"])
@@ -3934,6 +4143,50 @@ class TestResponseFailedSurfacing:
         assert types.count("response.output_item.added") == types.count("response.output_item.done")
         assert types[-1] == "response.failed"
 
+    async def test_streaming_run_failure_includes_usage(self) -> None:
+        agent = _make_agent()
+
+        def run_failure(*args: Any, **kwargs: Any) -> ResponseStream[AgentResponseUpdate, AgentResponse]:
+            del args, kwargs
+            return ResponseStream(
+                _raising_updates(
+                    "usage failure",
+                    initial_updates=[
+                        AgentResponseUpdate(
+                            contents=[
+                                Content.from_usage({
+                                    "input_token_count": 8,
+                                    "output_token_count": 3,
+                                    "cache_read_input_token_count": 2,
+                                    "reasoning_output_token_count": 1,
+                                })
+                            ],
+                            role="assistant",
+                        )
+                    ],
+                ),
+                finalizer=AgentResponse.from_updates,
+            )
+
+        agent.run = MagicMock(side_effect=run_failure)
+        server = _make_server(agent)
+
+        resp = await _post(server, input_text="hello", stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+        assert types[-1] == "response.failed"
+        assert "response.completed" not in types
+        failed_response = events[-1]["data"]["response"]
+        assert failed_response["usage"] == {
+            "input_tokens": 8,
+            "input_tokens_details": {"cached_tokens": 2},
+            "output_tokens": 3,
+            "output_tokens_details": {"reasoning_tokens": 1},
+            "total_tokens": 11,
+        }
+
     async def test_workflow_agent_run_failure_emits_response_failed(self) -> None:
         """Exceptions raised by a hosted ``WorkflowAgent`` are converted into a
         terminal ``response.failed`` event in the same way as the regular
@@ -4184,6 +4437,89 @@ def _build_text_workflow_agent(text: str) -> WorkflowAgent:
     return WorkflowAgent(workflow=workflow, name="Text Workflow Agent")
 
 
+class _MultiUpdateWorkflowAgentMock(SupportsAgentRun):
+    """Inner agent that streams one update per text in a single ``run`` call, and tracks ``run_count``."""
+
+    def __init__(self, name: str, texts: Sequence[str], *, gate: asyncio.Event | None = None) -> None:
+        self.id = str(uuid.uuid4())
+        self.name = name
+        self.description: str | None = None
+        self._texts = list(texts)
+        self._gate = gate
+        self.run_count = 0
+        self.started = asyncio.Event()  # Set at the top of run(), before any gate wait.
+
+    def create_session(self, **kwargs: Any) -> AgentSession:
+        del kwargs
+        return AgentSession()
+
+    def get_session(self, service_session_id: str | ServiceSessionId, *, session_id: str | None = None) -> AgentSession:
+        del service_session_id, session_id
+        return AgentSession()
+
+    @overload
+    def run(
+        self,
+        messages: Any = ...,
+        *,
+        stream: Literal[False] = ...,
+        session: AgentSession | None = ...,
+        **kwargs: Any,
+    ) -> Awaitable[AgentResponse[Any]]: ...
+
+    @overload
+    def run(
+        self,
+        messages: Any = ...,
+        *,
+        stream: Literal[True],
+        session: AgentSession | None = ...,
+        **kwargs: Any,
+    ) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
+
+    def run(
+        self,
+        messages: Any = None,
+        *,
+        stream: bool = False,
+        session: AgentSession | None = None,
+        **kwargs: Any,
+    ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
+        del messages, session, kwargs
+        assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents."
+        self.run_count += 1
+        self.started.set()
+        texts = self._texts
+        name = self.name
+        gate = self._gate
+
+        async def _aiter() -> AsyncIterator[AgentResponseUpdate]:
+            if gate is not None:
+                await gate.wait()  # Simulates a stuck model/tool call for preemption tests.
+            for text in texts:
+                yield AgentResponseUpdate(
+                    contents=[Content.from_text(text=text)],
+                    role="assistant",
+                    author_name=name,
+                )
+
+        return ResponseStream(_aiter(), finalizer=AgentResponse.from_updates)
+
+
+def _build_multi_update_workflow_agent(
+    texts: Sequence[str], *, gate: asyncio.Event | None = None
+) -> tuple[WorkflowAgent, _MultiUpdateWorkflowAgentMock]:
+    """Build a ``WorkflowAgent`` whose inner agent streams one update per text in ``texts``."""
+    inner = _MultiUpdateWorkflowAgentMock("multi-update-agent", texts, gate=gate)
+
+    @executor
+    async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
+        await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
+
+    workflow = WorkflowBuilder(start_executor=start).add_edge(start, inner).build()
+    return WorkflowAgent(workflow=workflow, name="Multi Update Workflow Agent"), inner
+
+
 def _build_approval_workflow_agent(
     *,
     approval_request_id: str,
@@ -4249,6 +4585,214 @@ class TestWorkflowAgentHosting:
         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)
 
+    async def test_cancellation_signal_stops_main_loop_and_completes(self) -> None:
+        """Explicit-cancel: the workflow's main loop must break promptly and still complete."""
+        workflow_agent, inner = _build_multi_update_workflow_agent(["one", "two", "three"])
+        server = _make_server(workflow_agent)
+        request = CreateResponse(model="m", input="hi", stream=True)
+        context = ResponseContext(response_id="response-1", mode_flags=MagicMock())
+        cancellation_signal = asyncio.Event()
+
+        with (
+            patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])),
+        ):
+            handler = cast(
+                AsyncGenerator[Any, None],
+                server._handle_response(request, context, cancellation_signal),  # pyright: ignore[reportPrivateUsage]
+            )
+            events: list[Any] = []
+            async for event in handler:
+                events.append(event)
+                if isinstance(event, Mapping) and event.get("type") == "response.output_text.delta":
+                    break
+            # Cancellation arrives after the first delta; the loop must not process "two"/"three".
+            cancellation_signal.set()
+            events.extend([event async for event in handler])
+
+        types = [event.get("type") for event in events if isinstance(event, Mapping)]
+        assert types.count("response.output_text.delta") == 1
+        assert types[-1] == "response.completed"
+        assert inner.run_count == 1
+
+    async def test_cancellation_signal_preempts_stuck_workflow_call(self) -> None:
+        """Explicit-cancel must interrupt the workflow's inner agent call stuck awaiting a slow
+        model/tool response, not merely be checked between already-produced updates."""
+        gate = asyncio.Event()  # Never set: simulates a model/tool call that never returns.
+        workflow_agent, inner = _build_multi_update_workflow_agent(["too late"], gate=gate)
+        server = _make_server(workflow_agent)
+        request = CreateResponse(model="m", input="hi", stream=True)
+        context = ResponseContext(response_id="response-1", mode_flags=MagicMock())
+        cancellation_signal = asyncio.Event()
+
+        with (
+            patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])),
+        ):
+            handler = cast(
+                AsyncGenerator[Any, None],
+                server._handle_response(request, context, cancellation_signal),  # pyright: ignore[reportPrivateUsage]
+            )
+            await anext(handler)  # response.created
+            await anext(handler)  # response.in_progress
+
+            # Pull the first workflow event in the background so we can wait for the inner agent's
+            # run() to actually start (proving it's genuinely stuck on `gate`) before signalling --
+            # otherwise cancellation could preempt the pull before the workflow even reaches it.
+            pending = asyncio.ensure_future(anext(handler))
+            await asyncio.wait_for(inner.started.wait(), timeout=1.0)
+            cancellation_signal.set()  # Fires while the inner agent is stuck awaiting `gate`.
+
+            async def _drain() -> list[Any]:
+                first = await pending
+                return [first, *[event async for event in handler]]
+
+            # Bounded well below `gate` never being set: proves cancellation preempted the stuck
+            # call instead of only being observed after it (eventually) produced an update.
+            events = await asyncio.wait_for(_drain(), timeout=1.0)
+
+        types = [event.get("type") for event in events if isinstance(event, Mapping)]
+        assert "response.output_text.delta" not in types
+        assert types[-1] == "response.completed"
+        assert inner.run_count == 1
+
+    async def test_shutdown_signal_preempts_stuck_workflow_call(self, tmp_path: Path) -> None:
+        """Shutdown must interrupt the workflow's inner agent call stuck awaiting a slow model/tool
+        response, not merely be checked between already-produced updates, and must trigger
+        ``exit_for_recovery()`` because it actually preempted the loop -- not on natural completion."""
+        gate = asyncio.Event()  # Never set: simulates a model/tool call that never returns.
+        workflow_agent, inner = _build_multi_update_workflow_agent(["too late"], gate=gate)
+        server = _make_server(
+            workflow_agent,
+            response_store=FileResponseStore(storage_dir=tmp_path),
+            options=ResponsesServerOptions(resilient_background=True),
+        )
+        request = CreateResponse(model="m", input="hi", stream=True)
+        context = ResponseContext(response_id="response-1", mode_flags=MagicMock())
+        cancellation_signal = asyncio.Event()
+
+        with (
+            patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "exit_for_recovery", new=AsyncMock(side_effect=ResponseExitForRecovery())),
+        ):
+            handler = cast(
+                AsyncGenerator[Any, None],
+                server._handle_response(request, context, cancellation_signal),  # pyright: ignore[reportPrivateUsage]
+            )
+            await anext(handler)  # response.created
+            await anext(handler)  # response.in_progress
+
+            # Pull the first workflow event in the background so we can wait for the inner agent's
+            # run() to actually start (proving it's genuinely stuck on `gate`) before signalling.
+            pending = asyncio.ensure_future(anext(handler))
+            await asyncio.wait_for(inner.started.wait(), timeout=1.0)
+            context.shutdown.set()  # Fires while the inner agent is stuck awaiting `gate`.
+
+            # Bounded well below `gate` never being set: proves shutdown preempted the stuck call
+            # instead of only being observed after it (eventually) produced an update.
+            with pytest.raises(ResponseExitForRecovery):
+                await asyncio.wait_for(pending, timeout=1.0)
+
+        assert inner.run_count == 1
+
+    async def test_cancellation_signal_set_before_turn_skips_new_input(self) -> None:
+        """Explicit-cancel: cancellation set before a continuation turn starts must skip that turn's new
+        input entirely, whether caught by the restore-loop's own check or the standalone check
+        guarding the start of a brand new workflow run."""
+        workflow_agent, inner = _build_multi_update_workflow_agent(["hello"])
+        server = _make_server(workflow_agent)
+
+        first = await _post(server, conversation_id="conv-1", stream=False)
+        assert first.status_code == 200
+        run_count_after_first_turn = inner.run_count
+        assert run_count_after_first_turn == 1
+
+        request = CreateResponse(model="m", input="hi again", stream=True)
+        context = ResponseContext(response_id="response-2", mode_flags=MagicMock(), conversation_id="conv-1")
+        cancellation_signal = asyncio.Event()
+        cancellation_signal.set()  # Steering pressure already present before the turn even starts.
+
+        with (
+            patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])),
+        ):
+            handler = cast(
+                AsyncGenerator[Any, None],
+                server._handle_response(request, context, cancellation_signal),  # pyright: ignore[reportPrivateUsage]
+            )
+            events = [event async for event in handler]
+
+        types = [event.get("type") for event in events if isinstance(event, Mapping)]
+        assert "response.output_text.delta" not in types
+        assert types[-1] == "response.completed"
+        # At most the restore-only replay call happened; the new-turn call (which would deliver
+        # "hi again") must never fire.
+        assert inner.run_count <= run_count_after_first_turn + 1
+
+    async def test_shutdown_signal_set_before_restore_only_triggers_recovery(self, tmp_path: Path) -> None:
+        """Shutdown observed while resuming a checkpoint (whether during the restore-only replay or
+        the standalone check guarding the start of a brand new workflow run) must trigger
+        ``exit_for_recovery()`` -- proving the post-loop ``signalled`` check (not a blind re-check of
+        the flag) correctly gates this action so it doesn't also fire on a replay that merely
+        finished naturally."""
+        workflow_agent, inner = _build_multi_update_workflow_agent(["hello"])
+        server = _make_server(
+            workflow_agent,
+            response_store=FileResponseStore(storage_dir=tmp_path),
+            options=ResponsesServerOptions(resilient_background=True),
+        )
+
+        first = await _post(server, conversation_id="conv-1", stream=False)
+        assert first.status_code == 200
+        run_count_after_first_turn = inner.run_count
+        assert run_count_after_first_turn == 1
+
+        request = CreateResponse(model="m", input="hi again", stream=True)
+        context = ResponseContext(response_id="response-2", mode_flags=MagicMock(), conversation_id="conv-1")
+        cancellation_signal = asyncio.Event()
+        context.shutdown.set()  # Fires before the continuation turn even starts.
+
+        with (
+            patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "get_history", new=AsyncMock(return_value=[])),
+            patch.object(ResponseContext, "exit_for_recovery", new=AsyncMock(side_effect=ResponseExitForRecovery())),
+        ):
+            handler = cast(
+                AsyncGenerator[Any, None],
+                server._handle_response(request, context, cancellation_signal),  # pyright: ignore[reportPrivateUsage]
+            )
+            with pytest.raises(ResponseExitForRecovery):
+                _ = [event async for event in handler]
+
+        # Only the restore-only replay call may have happened; the new-turn call must never fire.
+        assert inner.run_count <= run_count_after_first_turn + 1
+
+    async def test_previous_response_requires_existing_workflow_checkpoint(self) -> None:
+        """A previous_response_id naming a scope with no checkpoint must fail loudly rather than
+        silently starting a fresh workflow run (which could repeat side effects or misinterpret a
+        continuation as a new request)."""
+        workflow_agent, inner = _build_multi_update_workflow_agent(["hello"])
+        server = _make_server(workflow_agent)
+        request = CreateResponse(model="m", input="hi", previous_response_id="response-missing")
+        context = ResponseContext(response_id="response-current", mode_flags=MagicMock())
+
+        with patch.object(ResponseContext, "get_input_items", new=AsyncMock(return_value=[])):
+            handler = server._handle_response(request, context, asyncio.Event())  # pyright: ignore[reportPrivateUsage]
+            events = [event async for event in handler]
+
+        failed_events = [
+            event for event in events if isinstance(event, Mapping) and event.get("type") == "response.failed"
+        ]
+        assert len(failed_events) == 1
+        failed_event = cast(Mapping[str, Any], failed_events[0])
+        response = cast(Mapping[str, Any], failed_event["response"])
+        error = cast(Mapping[str, Any], response["error"])
+        assert (
+            "Cannot find an existing workflow checkpoint for previous_response_id=response-missing." in error["message"]
+        )
+        assert inner.run_count == 0
+
     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)
@@ -4457,3 +5001,36 @@ class TestWorkflowAgentHosting:
 
 
 # endregion
+
+
+# region Resilient background checkpointing
+
+
+class TestResilientBackgroundCheckpointing:
+    """``ResponseEventStream.checkpoint()`` only persists when its returned event is ``yield``-ed, so
+    ``_handle_inner_workflow`` fetches the latest saved workflow checkpoint and yields it after every update
+    when resilient_background is enabled.
+    """
+
+    async def test_workflow_yields_checkpoint_event_when_resilient_background(self, tmp_path: Path) -> None:
+        workflow_agent = _build_text_workflow_agent("hello from workflow")
+        server = _make_server(
+            workflow_agent,
+            response_store=FileResponseStore(storage_dir=tmp_path),
+            options=ResponsesServerOptions(resilient_background=True),
+        )
+        request = CreateResponse(model="m", input="hi", background=True, stream=True, store=True)
+        context = ResponseContext(response_id="response-current", mode_flags=MagicMock())
+
+        events = [
+            event
+            async for event in server._handle_response(  # pyright: ignore[reportPrivateUsage]
+                request, context, asyncio.Event()
+            )
+        ]
+
+        checkpoint_events = [e for e in events if isinstance(e, ResponseCheckpointEvent)]
+        assert checkpoint_events, "expected at least one checkpoint event yielded for a resilient background run"
+
+
+# endregion

python/packages/foundry_hosting/tests/test_responses_int.py

@@ -14,20 +14,39 @@ Required environment variables:
 
 from __future__ import annotations
 
+import asyncio
 import base64
 import json
+import multiprocessing
+import multiprocessing.process
 import os
+import re
+import socket
+import time
+from collections.abc import Callable
 from pathlib import Path
 from typing import Annotated, Any
 from unittest.mock import MagicMock
 
 import httpx
 import pytest
-from agent_framework import Agent, SlidingWindowStrategy, tool
+from agent_framework import (
+    Agent,
+    Content,
+    Executor,
+    Message,
+    SlidingWindowStrategy,
+    WorkflowBuilder,
+    WorkflowContext,
+    executor,
+    handler,
+    tool,
+)
 from agent_framework.foundry import FoundryChatClient
-from azure.ai.agentserver.responses import InMemoryResponseProvider
+from azure.ai.agentserver.responses import InMemoryResponseProvider, ResponsesServerOptions
 from azure.identity import AzureCliCredential
 from openai import AsyncOpenAI
+from typing_extensions import Never
 
 from agent_framework_foundry_hosting import ResponsesHostServer
 
@@ -774,3 +793,399 @@ class TestOptions:
         body = resp.json()
         assert body["status"] == "completed"
         assert len(body["output"]) > 0
+
+
+# ---------------------------------------------------------------------------
+# Tests — real crash/recovery for resilient-background workflows
+#
+# A real ResponsesHostServer is force-killed mid-workflow, then a freshly
+# started process pointed at the same on-disk state recovers and completes the same response.
+# The workflow is deterministic and model-free so the test needs no credentials.
+# ---------------------------------------------------------------------------
+
+
+class _CountdownStartExecutor(Executor):
+    """Extract the countdown target from the input text without calling a model."""
+
+    def __init__(self, id: str = "start") -> None:
+        super().__init__(id=id)
+
+    @handler
+    async def extract_target(self, messages: list[Message], ctx: WorkflowContext[int, str]) -> None:
+        match = re.search(r"\d+", " ".join(m.text for m in messages))
+        if not match:
+            await ctx.yield_output("The message must contain a positive integer counter target.")
+            return
+        await ctx.send_message(int(match.group()))
+
+
+class _CountdownExecutor(Executor):
+    """Decrement the target through a self-loop, then signal completion."""
+
+    def __init__(self, sleep_seconds: float, id: str = "countdown") -> None:
+        super().__init__(id=id)
+        self._sleep_seconds = sleep_seconds
+
+    @handler
+    async def countdown(self, target: int, ctx: WorkflowContext[int | str, str]) -> None:
+        if target <= 0:
+            await ctx.send_message("Countdown complete.", target_id="complete")
+            return
+
+        await asyncio.sleep(self._sleep_seconds)  # Simulate a long-running operation
+        await ctx.yield_output(str(target))
+        await ctx.send_message(target - 1, target_id=self.id)
+
+
+class _PairedYieldExecutor(Executor):
+    """Yield two separate, already-complete output items in a single superstep, then self-loop.
+
+    Exercises the case the single-item-per-superstep countdown workflow can't: a superstep whose
+    checkpoint only becomes visible after *both* items have been pulled from the stream, so the
+    second item is still the tracker's dangling "active" item at the moment the checkpoint for
+    this superstep is (or isn't yet) safe to pin and persist.
+    """
+
+    def __init__(self, sleep_seconds: float, id: str = "paired") -> None:
+        super().__init__(id=id)
+        self._sleep_seconds = sleep_seconds
+
+    @handler
+    async def step(self, target: int, ctx: WorkflowContext[int | str, str]) -> None:
+        if target <= 0:
+            await ctx.send_message("Countdown complete.", target_id="complete")
+            return
+
+        await asyncio.sleep(self._sleep_seconds)  # Simulate a long-running operation
+        await ctx.yield_output(f"first-{target}")
+        await ctx.yield_output(f"second-{target}")
+        await ctx.send_message(target - 1, target_id=self.id)
+
+
+class _ToolCallExecutor(Executor):
+    """Emit a deterministic function-call/result pair, then a final message, without a real model.
+
+    Exercises the function-call accumulation path in ``_OutputItemTracker`` (a distinct code path
+    from plain text) under a real crash/recovery cycle: the result must not be re-invoked or
+    duplicated after recovery.
+    """
+
+    def __init__(self, sleep_seconds: float, id: str = "tool_call") -> None:
+        super().__init__(id=id)
+        self._sleep_seconds = sleep_seconds
+
+    @handler
+    async def call_tool(self, target: int, ctx: WorkflowContext[str, Content | str]) -> None:
+        call_id = f"call_{target}"
+        await asyncio.sleep(self._sleep_seconds)  # Simulate a long-running operation
+        await ctx.yield_output(
+            Content.from_function_call(call_id, "get_number_fact", arguments=json.dumps({"number": target}))
+        )
+        await ctx.yield_output(Content.from_function_result(call_id, result=f"{target} is a deterministic number."))
+        await ctx.yield_output(f"The number is {target}.")
+        await ctx.send_message("Countdown complete.", target_id="complete")
+
+
+@executor(id="complete")
+async def _countdown_complete(message: str, ctx: WorkflowContext[Never, str]) -> None:  # zuban: ignore
+    """Yield the workflow's completion output."""
+    await ctx.yield_output(message)
+
+
+def _build_countdown_workflow(sleep_seconds: float):
+    """Build the target extraction, countdown, and completion workflow."""
+    start = _CountdownStartExecutor()
+    countdown = _CountdownExecutor(sleep_seconds)
+
+    return (
+        WorkflowBuilder(start_executor=start, output_from="all")
+        .add_edge(start, countdown)
+        .add_edge(countdown, countdown)
+        .add_edge(countdown, _countdown_complete)
+        .build()
+    )
+
+
+def _build_paired_yield_workflow(sleep_seconds: float):
+    """Build a workflow whose self-looping executor yields two output items per superstep."""
+    start = _CountdownStartExecutor()
+    paired = _PairedYieldExecutor(sleep_seconds)
+
+    return (
+        WorkflowBuilder(start_executor=start, output_from="all")
+        .add_edge(start, paired)
+        .add_edge(paired, paired)
+        .add_edge(paired, _countdown_complete)
+        .build()
+    )
+
+
+def _build_tool_call_workflow(sleep_seconds: float):
+    """Build a workflow that emits a function-call/result pair, then text, then a second superstep."""
+    start = _CountdownStartExecutor()
+    tool_call = _ToolCallExecutor(sleep_seconds)
+
+    return (
+        WorkflowBuilder(start_executor=start, output_from="all")
+        .add_edge(start, tool_call)
+        .add_edge(tool_call, _countdown_complete)
+        .build()
+    )
+
+
+def _run_resilient_server(
+    *,
+    port: int,
+    state_root: str,
+    sleep_seconds: float,
+    log_path: str,
+    build_workflow: Callable[[float], Any],
+) -> None:
+    """Multiprocessing target: hosts the given workflow as a real, killable server process."""
+    log_file = open(log_path, "a", buffering=1)  # noqa: SIM115
+    os.dup2(log_file.fileno(), 1)
+    os.dup2(log_file.fileno(), 2)
+    os.environ["AGENTSERVER_STATE_ROOT"] = state_root
+
+    workflow_agent = build_workflow(sleep_seconds).as_agent(name="resilient-workflow")
+    server = ResponsesHostServer(workflow_agent, options=ResponsesServerOptions(resilient_background=True))
+    server.run(host="127.0.0.1", port=port)
+
+
+def _free_port() -> int:
+    """Find an available TCP port on localhost."""
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+def _start_resilient_server(
+    *, port: int, state_root: Path, log_path: Path, build_workflow: Callable[[float], Any] = _build_countdown_workflow
+) -> multiprocessing.process.BaseProcess:
+    ctx = multiprocessing.get_context("spawn")
+    proc = ctx.Process(
+        target=_run_resilient_server,
+        kwargs={
+            "port": port,
+            "state_root": str(state_root),
+            "sleep_seconds": 0.05,
+            "log_path": str(log_path),
+            "build_workflow": build_workflow,
+        },
+    )
+    proc.start()
+    return proc
+
+
+async def _wait_for_ready(base_url: str, *, timeout: float = 30.0) -> None:
+    deadline = asyncio.get_event_loop().time() + timeout
+    async with httpx.AsyncClient() as client:
+        while asyncio.get_event_loop().time() < deadline:
+            try:
+                resp = await client.get(f"{base_url}/readiness", timeout=2.0)
+                if resp.status_code == 200:
+                    return
+            except httpx.HTTPError:
+                pass
+            await asyncio.sleep(0.2)
+    raise RuntimeError("Server did not become ready in time.")
+
+
+def _kill(proc: multiprocessing.process.BaseProcess) -> None:
+    # kill() sends SIGKILL on POSIX and calls TerminateProcess on Windows -- an ungraceful hard
+    # kill, just like a real crash.
+    if proc.is_alive():
+        proc.kill()
+        proc.join(timeout=10)
+
+
+def _clear_stale_stream_lock(state_root: Path, response_id: str) -> None:
+    # On Windows, the local stream store falls back to a plain lock *file* (no fcntl), which isn't
+    # cleaned up when the process is force-killed. Retry briefly since the killed process's file
+    # handle may not be released immediately.
+    lock_path = state_root / "streams" / f"{response_id}.jsonl.lock"
+    if not lock_path.exists():
+        return
+    for attempt in range(10):
+        try:
+            lock_path.unlink()
+            return
+        except PermissionError:
+            if attempt == 9:
+                raise
+            time.sleep(0.5)
+
+
+def _output_texts(output_items: list[dict[str, Any]]) -> list[str]:
+    texts: list[str] = []
+    for item in output_items:
+        if item.get("type") != "message":
+            continue
+        for part in item.get("content", []):
+            if part.get("type") == "output_text":
+                texts.append(part["text"])
+    return texts
+
+
+async def _run_until_nth_output_item_then_crash(
+    *, base_url: str, server: multiprocessing.process.BaseProcess, input_text: str, crash_after_count: int
+) -> str:
+    """POST a real streaming background response, force-kill the server after ``crash_after_count``
+    ``response.output_item.done`` events, and return the response id.
+    """
+    response_id: str | None = None
+    count = 0
+    async with (
+        httpx.AsyncClient(timeout=30) as client,
+        client.stream(
+            "POST",
+            f"{base_url}/responses",
+            json={"input": input_text, "store": True, "background": True, "stream": True},
+        ) as resp,
+    ):
+        assert resp.status_code == 200
+        current_event: str | None = None
+        async for line in resp.aiter_lines():
+            if line.startswith("event:"):
+                current_event = line[len("event:") :].strip()
+            elif line.startswith("data:"):
+                data = json.loads(line[len("data:") :].strip())
+                if current_event == "response.created" and response_id is None:
+                    response_id = data["response"]["id"]
+                elif current_event == "response.output_item.done":
+                    count += 1
+                    if count >= crash_after_count:
+                        break
+    assert response_id is not None
+    assert count >= crash_after_count
+    return response_id
+
+
+async def _wait_for_recovery_completion(*, base_url: str, response_id: str) -> dict[str, Any]:
+    """Replay the response's SSE stream to completion after a restart, then return the final body."""
+    async with httpx.AsyncClient(timeout=60) as client:
+        async with client.stream("GET", f"{base_url}/responses/{response_id}", params={"stream": "true"}) as resp:
+            assert resp.status_code == 200
+            current_event: str | None = None
+            async for line in resp.aiter_lines():
+                if line.startswith("event:"):
+                    current_event = line[len("event:") :].strip()
+                elif line.startswith("data:") and current_event in (
+                    "response.completed",
+                    "response.failed",
+                    "response.incomplete",
+                ):
+                    break
+
+        final = await client.get(f"{base_url}/responses/{response_id}")
+    return final.json()
+
+
+async def _crash_and_recover(
+    *,
+    build_workflow: Callable[[float], Any],
+    input_text: str,
+    crash_after_count: int,
+    tmp_path: Path,
+) -> dict[str, Any]:
+    """Force-kill a real server after ``crash_after_count`` output items, restart it against the
+    same on-disk state, and return the recovered response's final body.
+    """
+    state_root = tmp_path / "state"
+    log_path = tmp_path / "server.log"
+    port = _free_port()
+    base_url = f"http://127.0.0.1:{port}"
+
+    server = _start_resilient_server(port=port, state_root=state_root, log_path=log_path, build_workflow=build_workflow)
+    try:
+        await _wait_for_ready(base_url)
+        response_id = await _run_until_nth_output_item_then_crash(
+            base_url=base_url, server=server, input_text=input_text, crash_after_count=crash_after_count
+        )
+    finally:
+        _kill(server)
+
+    _clear_stale_stream_lock(state_root, response_id)
+
+    server = _start_resilient_server(port=port, state_root=state_root, log_path=log_path, build_workflow=build_workflow)
+    try:
+        await _wait_for_ready(base_url)
+        body = await _wait_for_recovery_completion(base_url=base_url, response_id=response_id)
+    finally:
+        _kill(server)
+
+    assert body["status"] == "completed", log_path.read_text(errors="replace")
+    return body
+
+
+@pytest.mark.xfail(
+    reason=("Known gap: 1. 'RuntimeError: Server did not become ready in time.' consistenly in CI. 2. #7809"),
+    strict=False,
+)
+class TestWorkflowResilientRecoveryRealCrash:
+    """Force-kill a real ResponsesHostServer process mid-workflow and verify a freshly started
+    process, pointed at the same on-disk state, recovers and completes the response with no
+    lost or duplicated output.
+
+    Crash points are parametrized across each workflow's item boundaries rather than a single
+    fixed point, since the checkpoint pin/persist ordering being tested depends on exactly where,
+    relative to a superstep boundary, the crash lands.
+    """
+
+    @pytest.mark.integration
+    @pytest.mark.parametrize("crash_after_count", [1, 3, 6])
+    async def test_countdown_workflow_crash_and_recover(self, tmp_path: Path, crash_after_count: int) -> None:
+        """One output item per superstep: a baseline where the tracker's active item always
+        auto-closes via a fresh message_id before the next checkpoint check runs.
+        """
+        target = 6
+        expected_texts = [str(n) for n in range(target, 0, -1)] + ["Countdown complete."]
+
+        body = await _crash_and_recover(
+            build_workflow=_build_countdown_workflow,
+            input_text=f"Count down from {target}",
+            crash_after_count=crash_after_count,
+            tmp_path=tmp_path,
+        )
+        assert _output_texts(body["output"]) == expected_texts
+
+    @pytest.mark.integration
+    @pytest.mark.parametrize("crash_after_count", [1, 2, 6])
+    async def test_paired_yield_workflow_crash_and_recover(self, tmp_path: Path, crash_after_count: int) -> None:
+        """Two output items per superstep: the second item is still the tracker's dangling
+        "active" item at the exact moment the checkpoint for that superstep would be pinned and
+        persisted.
+        """
+        target = 6
+        expected_texts = [text for n in range(target, 0, -1) for text in (f"first-{n}", f"second-{n}")] + [
+            "Countdown complete."
+        ]
+
+        body = await _crash_and_recover(
+            build_workflow=_build_paired_yield_workflow,
+            input_text=f"Count down from {target}",
+            crash_after_count=crash_after_count,
+            tmp_path=tmp_path,
+        )
+        assert _output_texts(body["output"]) == expected_texts
+
+    @pytest.mark.integration
+    @pytest.mark.parametrize("crash_after_count", [1, 3])
+    async def test_tool_call_workflow_crash_and_recover(self, tmp_path: Path, crash_after_count: int) -> None:
+        """Function-call/result accumulation is a distinct ``_OutputItemTracker`` code path from
+        plain text. Crashing right at the boundary into the next superstep (after the call/result/
+        text superstep completes) must resume without re-emitting the call.
+        """
+        body = await _crash_and_recover(
+            build_workflow=_build_tool_call_workflow,
+            input_text="Look up a fact about 7",
+            crash_after_count=crash_after_count,
+            tmp_path=tmp_path,
+        )
+        function_calls = [item for item in body["output"] if item.get("type") == "function_call"]
+        function_call_outputs = [item for item in body["output"] if item.get("type") == "function_call_output"]
+        assert len(function_calls) == 1
+        assert function_calls[0]["call_id"] == "call_7"
+        assert len(function_call_outputs) == 1
+        assert function_call_outputs[0]["call_id"] == "call_7"
+        assert _output_texts(body["output"]) == ["The number is 7.", "Countdown complete."]

python/packages/foundry_hosting/tests/test_state_store.py

@@ -81,7 +81,7 @@ async def test_save_uses_context_scoped_store() -> None:
         result = await FoundryCheckpointStore("context-1", _platform_context()).save(checkpoint)
 
     assert result == "checkpoint-1"
-    get_or_create.assert_awaited_once_with("checkpoints/context-1", user_isolation=True, user_id="user-1")
+    get_or_create.assert_awaited_once_with("checkpoints/context-1", user_isolation=True)
     store.set_item.assert_awaited_once_with("checkpoint-1", checkpoint.to_dict(), call_id="call-1")
 
 
@@ -235,7 +235,7 @@ async def test_save_and_load_function_approval_request() -> None:
         loaded = await storage.load_approval_request("approval-1")
 
     assert get_or_create.await_count == 2
-    get_or_create.assert_awaited_with("function_approvals", user_isolation=True, user_id="user-1")
+    get_or_create.assert_awaited_with("function_approvals", user_isolation=True)
     store.create_item.assert_awaited_once_with("approval-1", request.to_dict(), call_id="call-1")
     store.get_item.assert_awaited_once_with("approval-1", call_id="call-1")
     assert loaded == request
@@ -311,7 +311,7 @@ async def test_set_agent_session_uses_scoped_store() -> None:
     ) as get_or_create:
         await FoundryAgentSessionStore(_platform_context()).set("storage-session-1", session)
 
-    get_or_create.assert_awaited_once_with("agent_sessions", user_isolation=True, user_id="user-1")
+    get_or_create.assert_awaited_once_with("agent_sessions", user_isolation=True)
     store.set_item.assert_awaited_once_with("storage-session-1", session.to_dict(), call_id="call-1")
 
 
@@ -330,7 +330,7 @@ async def test_get_agent_session_returns_deserialized_session() -> None:
     assert result is not None
     assert result.to_dict() == session.to_dict()
     assert result is not session
-    get_or_create.assert_awaited_once_with("agent_sessions", user_isolation=True, user_id="user-1")
+    get_or_create.assert_awaited_once_with("agent_sessions", user_isolation=True)
     store.get_item.assert_awaited_once_with("storage-session-1", call_id="call-1")
 
 

python/samples/04-hosting/foundry-hosted-agents/README.md

@@ -22,7 +22,10 @@ This directory contains samples that demonstrate how to use hosted [Agent Framew
 | 9 | [Foundry Memory](responses/foundry_memory/) | An agent with persistent semantic memory backed by a Microsoft Foundry Memory Store, using `FoundryMemoryProvider` to remember user facts across sessions. |
 | 10 | [Monty CodeAct](responses/monty_codeact/) | An agent with a Monty-backed CodeAct context provider, exposing a single `execute_code` tool that runs Python in a [pydantic-monty](https://github.com/pydantic/monty) interpreter and invokes typed host tools (`compute`, `fetch_data`) from inside the sandbox. Uses the beta `agent-framework-monty` package. |
 | 11 | [Foundry Toolbox MCP Skills](responses/foundry_toolbox_mcp_skills/) | An agent that discovers MCP-based skills attached to a Foundry Toolbox and serves them via `SkillsProvider(MCPSkillsSource(...))`, fetching `SKILL.md` bodies and supplementary resources on demand. |
-| 12 | [Using deployed agent](responses/using_deployed_agent.py) | Invoke an agent already deployed to Foundry using either a service-created or user-created hosted session, then delete the session after use. |
+| 13 | [Custom Storage](responses/custom_storage/) | An agent demonstrating how to implement a custom storage provider for agent sessions (in-memory and Cosmos DB). |
+| 14 | [Resilient Long-Running Workflow](responses/resilient_long_running_workflow/) | A long-running, crash-resilient workflow demonstrating how `resilient_background=True` lets a background response survive a hard crash of the server process and resume from its last checkpoint instead of restarting from scratch. |
+| 15 | [Steerable Long-Running Agent](responses/steerable_long_running_agent/) | A long-running, non-workflow agent demonstrating how `steerable_conversations=True` lets a new turn on the same conversation cancel and replace a still-running turn instead of waiting for it to finish. Steering is only supported for non-workflow agents. |
+| 16 | [Using deployed agent](responses/using_deployed_agent.py) | Invoke an agent already deployed to Foundry using either a service-created or user-created hosted session, then delete the session after use. |
 
 ## Session Identifiers
 

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/.dockerignore

@@ -0,0 +1,7 @@
+.venv
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.Python
+.env

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/.env.example

@@ -0,0 +1,2 @@
+FOUNDRY_PROJECT_ENDPOINT="..."
+AZURE_AI_MODEL_DEPLOYMENT_NAME="..."

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY . user_agent/
+WORKDIR /app/user_agent
+
+RUN if [ -f requirements.txt ]; then \
+        pip install -r requirements.txt; \
+    else \
+        echo "No requirements.txt found"; \
+    fi
+
+EXPOSE 8088
+
+CMD ["python", "main.py"]

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/README.md

@@ -0,0 +1,112 @@
+# What this sample demonstrates
+
+A long-running, crash-resilient [Agent Framework](https://github.com/microsoft/agent-framework) workflow
+hosted using the **Responses protocol**. The workflow extracts a target number from the user's message and
+then counts down from it one step per second, demonstrating how `resilient_background=True` lets a
+background response survive a hard crash of the server process and resume from its last checkpoint instead
+of restarting from scratch.
+
+## How It Works
+
+### Workflow
+
+The workflow has three executors (see [main.py](main.py)):
+
+- **`StartExecutor`** uses a `FoundryChatClient`-backed agent to extract a positive integer target from the
+  user's message. If no valid target is found, the workflow yields an error message instead of counting down.
+- **`CountdownExecutor`** decrements the target through a self-loop, sleeping for a second and yielding an
+  output on each tick, to simulate a long-running operation.
+- **`complete`** yields the workflow's final `"Countdown complete."` output once the countdown reaches zero.
+
+### Agent Hosting
+
+The workflow is hosted as an agent using the [Agent Framework](https://github.com/microsoft/agent-framework)
+`ResponsesHostServer`, which provisions a REST API endpoint compatible with the OpenAI Responses protocol.
+Setting `resilient_background=True` in `ResponsesServerOptions` enables the framework to checkpoint the
+workflow's progress and durably persist streamed output, so a background response can be recovered and
+resumed after a crash (see "Testing resiliency" below).
+
+## Running the Agent Host
+
+Follow the instructions in the [Running the Agent Host Locally](../../README.md#running-the-agent-host-locally) section of the README in the parent directory to run the agent host.
+
+## Interacting with the agent
+
+> Depending on how you run the agent host, you can invoke the agent using `curl` (`Invoke-WebRequest` in PowerShell) or `azd`. Please refer to the [parent README](../../README.md) for more details. Use this README for sample queries you can send to the agent.
+
+Send a POST request to the server with a JSON body containing an `"input"` field with a positive integer target to count down from. For example:
+
+```bash
+curl -X POST http://localhost:8088/responses -H "Content-Type: application/json" -d '{"input": "Count down from 5"}'
+```
+
+The server will respond with a JSON object containing the response output (one item per countdown step) and a response ID. You can use this response ID to continue the conversation in subsequent requests.
+
+
+## Testing resiliency (crash recovery)
+
+This sample enables `resilient_background=True`, so a long-running countdown survives a hard crash of the
+server process and resumes from its last checkpoint instead of restarting from scratch. Locally, the
+server persists responses, streams, and checkpoints under `${AGENTSERVER_STATE_ROOT:-~/.agentserver}/`, so
+this state survives a process restart as long as you run from the same working directory.
+
+On startup, the server's task manager scans that persisted state for any tasks that were still in flight
+when the process died and automatically reclaims and resumes each one from its last checkpoint -- this
+happens for every incomplete resilient background response, not just the one a client happens to reconnect
+to. This is why a stale response from an earlier run can still be found "recovering" in the server logs
+long after you've moved on to a new test: every restart re-triggers the same scan, so the stale task keeps
+getting resumed until it either reaches a terminal state or the persisted state directory is cleared (as
+`verify_resiliency.py` does).
+
+### Automated
+
+[verify_resiliency.py](verify_resiliency.py) runs the whole scenario end to end: it clears any leftover
+`${AGENTSERVER_STATE_ROOT:-~/.agentserver}` state from a previous run, starts the server, kicks off a
+background+streaming countdown, force-kills the server once half the countdown has completed, restarts the
+server, and asserts the recovered response completes with the exact expected output (no lost or duplicated
+steps). Progress is printed as each countdown item completes, both before and after the crash, by reading
+the response's own `stream=true` SSE feed -- a plain (non-streaming) `GET` only ever reflects the response's
+initial or terminal snapshot, never anything in between.
+
+```bash
+python verify_resiliency.py --target 20
+```
+
+### Manual
+
+To exercise crash recovery by hand:
+
+1. Start the server, then kick off a long background+streaming countdown and note the response `id` from the
+   first `response.created` event:
+
+   ```bash
+   curl -X POST http://localhost:8088/responses -H "Content-Type: application/json" \
+     -d '{"input": "Count down from 200", "stream": true, "store": true, "background": true}'
+   ```
+
+2. While the countdown is still running, kill the server process abruptly — use `kill -9 <pid>`
+   (`Stop-Process -Id <pid> -Force` on Windows), not `Ctrl+C`. A `Ctrl+C` triggers a graceful shutdown, which
+   is handled differently than a crash; a hard kill is required to exercise crash recovery. The sample server
+   prints its PID (`PID: <pid>`) on startup so you don't need to look it up separately.
+
+3. Restart the server (`python main.py`) from the same working directory.
+
+4. Reconnect to the response to observe recovery:
+
+   ```bash
+   curl "http://localhost:8088/responses/REPLACE_WITH_RESPONSE_ID?stream=true"
+   ```
+
+   The recovered stream emits a fresh `response.in_progress` event first, then resumes the countdown from
+   where it left off — the output already produced before the crash is neither lost nor duplicated.
+   Alternatively, poll `GET /responses/REPLACE_WITH_RESPONSE_ID` (without `stream`) until `status` is
+   `completed` and inspect the `output` array for a contiguous, non-duplicated sequence.
+
+> **Windows note:** the local stream store falls back to a plain lock *file* (no `fcntl`), which isn't
+> cleaned up when the process is force-killed. If restart fails with `another process holds the lock-file
+> on ...jsonl`, delete the stale `<response-id>.jsonl.lock` file under
+> `%USERPROFILE%\.agentserver\streams\` before restarting the server.
+
+## Deploying the Agent to Foundry
+
+To host the agent on Foundry, follow the instructions in the [Deploying the Agent to Foundry](../../README.md#deploying-the-agent-to-foundry) section of the README in the parent directory.

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/agent.manifest.yaml

@@ -0,0 +1,25 @@
+name: agent-framework-resilient-long-running-workflow
+description: >
+  A hosted agent that demonstrates a resilient long-running workflow using the Responses protocol.
+metadata:
+  tags:
+    - Agent Framework
+    - AI Agent Hosting
+    - Azure AI AgentServer
+    - Responses Protocol
+    - Streaming
+    - Workflows
+    - Resilience
+template:
+  name: agent-framework-resilient-long-running-workflow
+  kind: hosted
+  protocols:
+    - protocol: responses
+      version: 2.0.0
+  environment_variables:
+    - name: AZURE_AI_MODEL_DEPLOYMENT_NAME
+      value: "{{AZURE_AI_MODEL_DEPLOYMENT_NAME}}"
+resources:
+  - kind: model
+    id: gpt-4.1-mini
+    name: AZURE_AI_MODEL_DEPLOYMENT_NAME

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/agent.yaml

@@ -0,0 +1,12 @@
+# yaml-language-server: $schema=https://raw.githubusercontent.com/microsoft/AgentSchema/refs/heads/main/schemas/v1.0/ContainerAgent.yaml
+kind: hosted
+name: agent-framework-resilient-long-running-workflow
+protocols:
+  - protocol: responses
+    version: 2.0.0
+resources:
+  cpu: "0.25"
+  memory: "0.5Gi"
+environment_variables:
+  - name: AZURE_AI_MODEL_DEPLOYMENT_NAME
+    value: ${AZURE_AI_MODEL_DEPLOYMENT_NAME}

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/main.py

@@ -0,0 +1,119 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Host a three-executor workflow that extracts and counts down a target.
+
+The start executor asks a Foundry-backed agent to extract a positive integer
+from the incoming message. The countdown executor repeatedly decrements that
+integer and sends it back to itself. At zero, it sends a completion message to
+the terminal executor, which yields the workflow output.
+
+Environment variables:
+    FOUNDRY_PROJECT_ENDPOINT: Microsoft Foundry project endpoint.
+    AZURE_AI_MODEL_DEPLOYMENT_NAME: Model deployment name.
+"""
+
+import asyncio
+import os
+
+from agent_framework import Agent, Executor, Message, WorkflowBuilder, WorkflowContext, executor, handler
+from agent_framework.foundry import FoundryChatClient
+from agent_framework_foundry_hosting import ResponsesHostServer
+from azure.ai.agentserver.responses import ResponsesServerOptions
+from azure.identity import DefaultAzureCredential
+from dotenv import load_dotenv
+from pydantic import BaseModel, Field
+from typing_extensions import Never
+
+load_dotenv()
+
+
+class CounterTarget(BaseModel):
+    """The counter target extracted from the user's message."""
+
+    target: int | None = Field(
+        description="The positive integer to count down from, or null when no valid target was provided."
+    )
+
+
+class StartExecutor(Executor):
+    """Extract a valid counter target and start the countdown."""
+
+    def __init__(self, agent: Agent, id: str = "start") -> None:
+        super().__init__(id=id)
+        self._agent = agent
+
+    @handler
+    async def extract_target(self, messages: list[Message], ctx: WorkflowContext[int, str]) -> None:
+        """Ask the model for a target and forward valid positive integers."""
+        response = await self._agent.run(messages, options={"response_format": CounterTarget})
+        extraction = response.value
+        if not isinstance(extraction, CounterTarget) or extraction.target is None or extraction.target <= 0:
+            await ctx.yield_output("The message must contain a positive integer counter target.")
+            return
+
+        await ctx.send_message(extraction.target)
+
+
+class CountdownExecutor(Executor):
+    def __init__(self, id: str = "countdown") -> None:
+        super().__init__(id=id)
+
+    @handler
+    async def countdown(self, target: int, ctx: WorkflowContext[int | str, str]) -> None:
+        """Decrement the target through a self-loop, then signal completion."""
+        if target <= 0:
+            await ctx.send_message("Countdown complete.", target_id="complete")
+            return
+
+        await asyncio.sleep(1)  # Simulate a long-running operation
+        await ctx.yield_output(str(target))
+        await ctx.send_message(target - 1, target_id=self.id)
+
+
+@executor(id="complete")
+async def complete(message: str, ctx: WorkflowContext[Never, str]) -> None:
+    """Yield the workflow's completion output."""
+    await ctx.yield_output(message)
+
+
+def build_workflow():
+    """Build the target extraction, countdown, and completion workflow."""
+    client = FoundryChatClient(
+        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
+        model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
+        credential=DefaultAzureCredential(),
+    )
+    target_agent = Agent(
+        client=client,
+        name="counter_target_extractor",
+        instructions=(
+            "Extract the counter target requested by the user. Return the target only when it is a positive integer. "
+            "Return null for zero, negative numbers, fractions, or messages without a clear counter target."
+        ),
+    )
+    start = StartExecutor(target_agent)
+    countdown = CountdownExecutor()
+
+    return (
+        WorkflowBuilder(start_executor=start, output_from="all")
+        .add_edge(start, countdown)
+        .add_edge(countdown, countdown)
+        .add_edge(countdown, complete)
+        .build()
+    )
+
+
+def main() -> None:
+    """Run the workflow as a durable Responses API host."""
+    print(f"PID: {os.getpid()}")  # lets crash-recovery testing find and kill this process
+    workflow_agent = build_workflow().as_agent(name="countdown-workflow")
+    server = ResponsesHostServer(
+        workflow_agent,
+        options=ResponsesServerOptions(resilient_background=True),
+        log_level="DEBUG",
+    )
+    server.run()
+
+
+if __name__ == "__main__":
+    main()

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/requirements.txt

@@ -0,0 +1,2 @@
+agent-framework-foundry
+agent-framework-foundry-hosting

python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/verify_resiliency.py

@@ -0,0 +1,253 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""End-to-end crash-recovery test for the resilient countdown workflow sample.
+
+Starts the server, kicks off a background countdown, force-kills the server mid-countdown to
+simulate a real crash, clears the stale Windows stream lock file, restarts the server, and
+verifies the countdown resumes and completes with the exact expected output (no loss, no
+duplication). Requires the same environment (.env) as running main.py directly.
+
+Usage:
+    python verify_resiliency.py [--target N] [--crash-after-count N]
+"""
+
+import argparse
+import json
+import shutil
+import subprocess
+import sys
+import threading
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import IO, Any
+
+HOST = "127.0.0.1"
+PORT = 8088
+BASE_URL = f"http://{HOST}:{PORT}"
+SAMPLE_DIR = Path(__file__).parent
+LOG_PATH = SAMPLE_DIR / "verify_resiliency.log"
+
+
+def _http_get(path: str, timeout: float = 5.0) -> tuple[int, dict[str, Any]]:
+    with urllib.request.urlopen(urllib.request.Request(f"{BASE_URL}{path}"), timeout=timeout) as resp:
+        return resp.status, json.loads(resp.read())
+
+
+def _start_server(log_file: IO[str]) -> subprocess.Popen:  # type: ignore
+    return subprocess.Popen([sys.executable, "main.py"], cwd=SAMPLE_DIR, stdout=log_file, stderr=subprocess.STDOUT)
+
+
+def _wait_for_ready(timeout: float = 30.0) -> None:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            status, _ = _http_get("/readiness", timeout=2.0)
+            if status == 200:
+                return
+        except (urllib.error.URLError, ConnectionError, TimeoutError):
+            pass
+        time.sleep(0.5)
+    raise RuntimeError("Server did not become ready in time.")
+
+
+def _kill(server: subprocess.Popen) -> None:  # type: ignore
+    # Popen.kill() maps to TerminateProcess on Windows -- an ungraceful hard kill, just like a real crash.
+    if server.poll() is None:
+        server.kill()
+        server.wait(timeout=10)
+
+
+def _clear_stale_stream_lock(response_id: str) -> None:
+    # On Windows, the local stream store falls back to a plain lock *file* (no `fcntl`), which isn't
+    # cleaned up when the process is force-killed. If restart fails with `another process holds the
+    # lock-file on ...jsonl`, delete the stale `<response-id>.jsonl.lock` file under
+    lock_path = Path.home() / ".agentserver" / "streams" / f"{response_id}.jsonl.lock"
+    if not lock_path.exists():
+        return
+    # Windows may not release the killed process's file handle immediately; retry briefly.
+    for attempt in range(10):
+        try:
+            lock_path.unlink()
+            print(f"      removed stale lock file: {lock_path}")
+            return
+        except PermissionError:
+            if attempt == 9:
+                raise
+            time.sleep(0.5)
+
+
+def _create_streaming_background_response(payload: dict[str, Any], progress: dict[str, Any]) -> None:
+    """POST a streaming background create-response request and track its progress.
+
+    Background responses only expose incremental output over SSE when the *creation*
+    request itself sets ``stream=true`` (``ResponseExecution.replay_enabled`` requires it);
+    a plain (non-streaming) GET only ever reflects the initial (empty output) or terminal
+    (full output) snapshot, never anything in between. So the create call itself must be
+    the streaming one, and its own response body is read here as the progress feed.
+    """
+    data = json.dumps({**payload, "stream": True}).encode("utf-8")
+    request = urllib.request.Request(
+        f"{BASE_URL}/responses", data=data, headers={"Content-Type": "application/json"}, method="POST"
+    )
+    try:
+        with urllib.request.urlopen(request) as resp:
+            current_event: str | None = None
+            for raw_line in resp:
+                line = raw_line.decode("utf-8").rstrip("\n")
+                if line.startswith("event:"):
+                    current_event = line[len("event:") :].strip()
+                    continue
+                if not line.startswith("data:"):
+                    continue
+                data_obj = json.loads(line[len("data:") :].strip())
+                if current_event == "response.created" and "id" not in progress:
+                    progress["id"] = data_obj["response"]["id"]
+                    progress["status"] = data_obj["response"]["status"]
+                    progress["ready"].set()
+                elif current_event == "response.output_item.done":
+                    progress["count"] += 1
+                    for part in data_obj.get("item", {}).get("content", []):
+                        if part.get("type") == "output_text":
+                            print(f"      output item {progress['count']}: {part['text']!r}")
+    except urllib.error.HTTPError as exc:
+        progress["error"] = f"HTTP {exc.code}: {exc.read().decode('utf-8', errors='replace')}"
+    except (urllib.error.URLError, ConnectionError, TimeoutError, OSError) as exc:
+        progress["error"] = f"{type(exc).__name__}: {exc}"
+    finally:
+        progress["ready"].set()  # Unblock a waiter even if response.created never arrived.
+
+
+def _watch_recovery_progress(response_id: str, progress: dict[str, Any]) -> None:
+    """Replay the response's SSE stream after recovery and print each item as it arrives.
+
+    ``starting_after`` is omitted, so the replay starts from the beginning of the retained
+    history -- pre-crash items are reprinted, then live items follow as the recovered
+    workflow produces them.
+    """
+    url = f"{BASE_URL}/responses/{response_id}?stream=true"
+    try:
+        with urllib.request.urlopen(url) as resp:
+            current_event: str | None = None
+            for raw_line in resp:
+                line = raw_line.decode("utf-8").rstrip("\n")
+                if line.startswith("event:"):
+                    current_event = line[len("event:") :].strip()
+                    continue
+                if not line.startswith("data:"):
+                    continue
+                data_obj = json.loads(line[len("data:") :].strip())
+                if current_event == "response.output_item.done":
+                    progress["count"] += 1
+                    for part in data_obj.get("item", {}).get("content", []):
+                        if part.get("type") == "output_text":
+                            print(f"      output item {progress['count']}: {part['text']!r}")
+                elif current_event in ("response.completed", "response.failed", "response.incomplete"):
+                    progress["done"].set()
+    except (urllib.error.URLError, ConnectionError, TimeoutError, OSError):
+        pass  # Connection drops when the server crashes or exits; the caller already knows.
+    finally:
+        progress["done"].set()
+
+
+def _extract_message_texts(output_items: list[dict[str, Any]]) -> list[str]:
+    texts: list[str] = []
+    for item in output_items:
+        if item.get("type") != "message":
+            continue
+        for part in item.get("content", []):
+            if part.get("type") == "output_text":
+                texts.append(part["text"])
+    return texts
+
+
+def _clear_stale_state() -> None:
+    """Wipe ~/.agentserver so a prior run's incomplete response is never auto-recovered
+    on startup and left competing with this run's request for the event loop.
+    """
+    state_root = Path.home() / ".agentserver"
+    if state_root.exists():
+        shutil.rmtree(state_root, ignore_errors=True)
+        print(f"      cleared stale state: {state_root}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--target", type=int, default=20, help="Countdown starting value.")
+    args = parser.parse_args()
+
+    crash_after_count = args.target // 2
+    expected_texts = [str(n) for n in range(args.target, 0, -1)] + ["Countdown complete."]
+
+    _clear_stale_state()
+
+    log_file = LOG_PATH.open("w", encoding="utf-8")
+    print(f"Server logs (DEBUG level) are redirected to {LOG_PATH}.")
+
+    print(f"[1/6] Starting server (target={args.target})...")
+    server = _start_server(log_file)  # type: ignore
+    print(f"      PID: {server.pid}")
+    try:
+        _wait_for_ready()
+
+        print("[2/6] Starting background countdown...")
+        progress: dict[str, Any] = {"count": 0, "ready": threading.Event()}
+        watcher = threading.Thread(
+            target=_create_streaming_background_response,
+            args=({"input": f"Count down from {args.target}", "store": True, "background": True}, progress),
+            daemon=True,
+        )
+        watcher.start()
+        if not progress["ready"].wait(timeout=60):
+            raise SystemExit("FAIL: did not receive response.created in time.")
+        if "id" not in progress:
+            raise SystemExit(f"FAIL: streaming create request failed: {progress.get('error', 'unknown error')}")
+        response_id = progress["id"]
+        print(f"      response id: {response_id}, status: {progress['status']}")
+
+        print(f"[3/6] Waiting for the countdown to reach {crash_after_count} completed item(s)...")
+        while progress["count"] < crash_after_count:
+            time.sleep(0.5)
+        print(f"      output items observed via SSE before crash: {progress['count']}")
+
+        print("[4/6] Force-killing the server (simulated crash)...")
+    finally:
+        _kill(server)
+
+    _clear_stale_stream_lock(response_id)
+
+    print("[5/6] Restarting the server...")
+    server = _start_server(log_file)  # type: ignore
+    print(f"      PID: {server.pid}")
+    try:
+        _wait_for_ready()
+
+        print("[6/6] Waiting for the recovered countdown to complete...")
+        recovery: dict[str, Any] = {"count": 0, "done": threading.Event()}
+        recovery_watcher = threading.Thread(target=_watch_recovery_progress, args=(response_id, recovery), daemon=True)
+        recovery_watcher.start()
+        recovery["done"].wait(timeout=args.target * 2 + 30)
+        final = _http_get(f"/responses/{response_id}")[1]
+    finally:
+        _kill(server)
+        log_file.close()
+
+    if final["status"] != "completed":
+        raise SystemExit(
+            f"FAIL: response did not complete in time; last status: {final['status']}. See {LOG_PATH} for server logs."
+        )
+
+    texts = _extract_message_texts(final.get("output", []))
+    print(f"      final status: {final['status']}, output items: {len(final['output'])}")
+    if texts != expected_texts:
+        raise SystemExit(
+            f"FAIL: recovered output mismatch.\n  expected: {expected_texts}\n  got:      {texts}\n"
+            f"See {LOG_PATH} for server logs."
+        )
+
+    print("PASS: countdown crashed mid-flight and recovered with no lost or duplicated output.")
+
+
+if __name__ == "__main__":
+    main()

python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/.dockerignore

@@ -0,0 +1,7 @@
+.venv
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.Python
+.env

python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/.env.example

@@ -0,0 +1,2 @@
+FOUNDRY_PROJECT_ENDPOINT="..."
+AZURE_AI_MODEL_DEPLOYMENT_NAME="..."

python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY . user_agent/
+WORKDIR /app/user_agent
+
+RUN if [ -f requirements.txt ]; then \
+        pip install -r requirements.txt; \
+    else \
+        echo "No requirements.txt found"; \
+    fi
+
+EXPOSE 8088
+
+CMD ["python", "main.py"]

python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/README.md

@@ -0,0 +1,106 @@
+# What this sample demonstrates
+
+A steerable multi-turn [Agent Framework](https://github.com/microsoft/agent-framework) agent hosted using the
+**Responses protocol**. The agent is asked to count down from a target number, pacing its own output with a short
+remark before each number so a real response takes a while to fully generate. With `steerable_conversations=True`,
+sending a new turn on the same conversation while the countdown is still streaming **cancels the in-progress turn**
+and drains the new turn next.
+
+Steering is only supported for non-workflow agents. Steering a workflow is conceptually undefined: a workflow's
+graph may have loops or parallel branches with no single well-defined "current point" to cancel and resume from,
+unlike an agent's strictly linear execution. `ResponsesHostServer` rejects `steerable_conversations=True` for a
+workflow agent with `RuntimeError`.
+
+## How It Works
+
+### Agent
+
+The agent (see [main.py](main.py)) is a single `Agent` backed by `FoundryChatClient`, with no workflow and no custom
+agent class. Its instructions ask it to count down one integer per line, prefacing each with a brief remark, so a
+real streamed generation takes long enough for a second turn to arrive mid-stream. Compared to the
+[Basic](../basic/) sample, the only differences are the instructions and passing
+`ResponsesServerOptions(steerable_conversations=True)` -- steering needs no special agent-side code.
+
+### Agent Hosting
+
+The agent is hosted using the [Agent Framework](https://github.com/microsoft/agent-framework) `ResponsesHostServer`.
+Setting `steerable_conversations=True` in `ResponsesServerOptions` lets a new turn on the same conversation chain
+preempt a still-running one:
+
+- The framework signals the in-progress turn's handler via the 3rd positional `cancellation_signal` argument.
+- The handler (here, the Agent Framework agent-hosting layer) checks that signal between streamed model updates and
+  winds the turn down promptly, letting it complete with whatever partial output it had already produced.
+- The new turn is then drained with `context.is_steered_turn == True` and `context.pending_input_count` reflecting
+  how many further turns are still queued behind it.
+- **A single, linear chain**: every turn after the first must reference the immediately preceding turn's `id` via
+  `previous_response_id`, and a `previous_response_id` that doesn't point at the latest turn is rejected with HTTP
+  409 (`conversation_fork_not_supported`). Chain identity in turn also depends on session continuity: without an
+  explicit `conversation`, the server derives a session id per request, and it's only deterministic across turns
+  when a client forwards the `x-agent-session-id` header from a prior response back as `agent_session_id` on the
+  next one (see the `curl` walkthrough below) -- otherwise a later turn resolves to a different session and starts
+  a brand new response instead of steering the earlier one. Sending the same explicit `conversation` value on every
+  turn sidesteps this entirely: it makes the derived session id (and the chain itself) a deterministic function of
+  that id, so no header needs to be echoed back, and `previous_response_id` becomes unnecessary for continuity.
+
+## Running the Agent Host
+
+Follow the instructions in the [Running the Agent Host Locally](../../README.md#running-the-agent-host-locally) section
+of the README in the parent directory to run the agent host.
+
+## Interacting with the agent
+
+> Depending on how you run the agent host, you can invoke the agent using `curl` (`Invoke-WebRequest` in PowerShell) or
+> `azd`. Please refer to the [parent README](../../README.md) for more details. Use this README for sample queries you
+> can send to the agent.
+
+Start a long background countdown and note the response `id` from the JSON body and the `x-agent-session-id`
+response header:
+
+```bash
+curl -i -X POST http://localhost:8088/responses -H "Content-Type: application/json" \
+  -d '{"input": "Count down from 30, slowly and with commentary.", "store": true, "background": true}'
+```
+
+While it is still generating, send a second turn on the same conversation with `previous_response_id` set to steer
+it to a new target. Without an explicit `conversation_id`, also forward the `x-agent-session-id` value from the
+first response as `agent_session_id` -- otherwise this turn resolves to a different session and starts a brand new
+response instead of steering the first one:
+
+```bash
+curl -X POST http://localhost:8088/responses -H "Content-Type: application/json" \
+  -d '{"input": "Actually, count down from 3 instead.", "store": true, "background": true, "previous_response_id": "REPLACE_WITH_FIRST_RESPONSE_ID", "agent_session_id": "REPLACE_WITH_X-AGENT-SESSION-ID_HEADER"}'
+```
+
+This second request returns immediately with `"status": "queued"`. Polling the *first* response's id will show it
+completed early, with fewer tokens than a full 30-count run. Polling the *second* response's id will show a fresh
+countdown from 3.
+
+Alternatively, send an explicit `conversation` id on every turn instead of forwarding `x-agent-session-id`. This is
+simpler and also works without `previous_response_id` at all, since the `conversation` id alone identifies the chain:
+
+```bash
+curl -X POST http://localhost:8088/responses -H "Content-Type: application/json" \
+  -d '{"input": "Count down from 30, slowly and with commentary.", "store": true, "background": true, "conversation": "my-conversation-id"}'
+
+curl -X POST http://localhost:8088/responses -H "Content-Type: application/json" \
+  -d '{"input": "Actually, count down from 3 instead.", "store": true, "background": true, "conversation": "my-conversation-id"}'
+```
+
+## Testing steering
+
+[verify_steering.py](verify_steering.py) runs the whole scenario end to end: it starts the server, kicks off a
+background streaming countdown, waits for it to stream a minimum number of tokens, sends a second turn with a new
+target via `previous_response_id`, and asserts that the second turn is accepted immediately as `"queued"`, that the
+first turn completes early, and that the second (steered) turn's output contains the new target's countdown in
+order. Because this sample calls a real model, the assertions here are intentionally loose rather than an exact
+output match.
+
+```bash
+python verify_steering.py --first-target 30 --second-target 3
+```
+
+## Deploying the Agent to Foundry
+
+To host the agent on Foundry, follow the instructions in the
+[Deploying the Agent to Foundry](../../README.md#deploying-the-agent-to-foundry) section of the README in the parent
+directory.

python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/agent.manifest.yaml

@@ -0,0 +1,26 @@
+name: agent-framework-steerable-long-running-agent
+description: >
+  A hosted agent that demonstrates steerable multi-turn conversations for a plain (non-workflow)
+  agent using the Responses protocol.
+metadata:
+  tags:
+    - Agent Framework
+    - AI Agent Hosting
+    - Azure AI AgentServer
+    - Responses Protocol
+    - Streaming
+    - Steering
+    - Multi-turn
+template:
+  name: agent-framework-steerable-long-running-agent
+  kind: hosted
+  protocols:
+    - protocol: responses
+      version: 2.0.0
+  environment_variables:
+    - name: AZURE_AI_MODEL_DEPLOYMENT_NAME
+      value: "{{AZURE_AI_MODEL_DEPLOYMENT_NAME}}"
+resources:
+  - kind: model
+    id: gpt-4.1-mini
+    name: AZURE_AI_MODEL_DEPLOYMENT_NAME

python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/agent.yaml

@@ -0,0 +1,12 @@
+# yaml-language-server: $schema=https://raw.githubusercontent.com/microsoft/AgentSchema/refs/heads/main/schemas/v1.0/ContainerAgent.yaml
+kind: hosted
+name: agent-framework-steerable-long-running-agent
+protocols:
+  - protocol: responses
+    version: 2.0.0
+resources:
+  cpu: "0.25"
+  memory: "0.5Gi"
+environment_variables:
+  - name: AZURE_AI_MODEL_DEPLOYMENT_NAME
+    value: ${AZURE_AI_MODEL_DEPLOYMENT_NAME}

python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/main.py

@@ -0,0 +1,54 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Host a single (non-workflow) agent that counts down slowly, steerably.
+
+The agent is asked to count down from a target number, pacing its own output with a short remark
+before each number so a real response takes a while to fully generate. With
+`steerable_conversations=True`, sending a new turn on the same conversation while the countdown is
+still streaming cancels the in-progress turn and drains the new turn next. Steering is only
+supported for non-workflow agents like this one.
+
+Environment variables:
+    FOUNDRY_PROJECT_ENDPOINT: Microsoft Foundry project endpoint.
+    AZURE_AI_MODEL_DEPLOYMENT_NAME: Model deployment name.
+"""
+
+import os
+
+from agent_framework import Agent
+from agent_framework.foundry import FoundryChatClient
+from agent_framework_foundry_hosting import ResponsesHostServer
+from azure.ai.agentserver.responses import ResponsesServerOptions
+from azure.identity import DefaultAzureCredential
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+def main() -> None:
+    client = FoundryChatClient(
+        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
+        model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
+        credential=DefaultAzureCredential(),
+    )
+
+    agent = Agent(
+        client=client,
+        instructions=(
+            "You are a counting assistant. When asked to count down from a positive integer, count down one "
+            "integer per line, and before each number add a brief, unique one-sentence remark, so your full "
+            "response takes some time to generate. If no valid positive integer target is given, reply with "
+            "'Please provide a positive integer to count down from.' and nothing else."
+        ),
+    )
+
+    server = ResponsesHostServer(
+        agent,
+        options=ResponsesServerOptions(steerable_conversations=True),
+        log_level="DEBUG",
+    )
+    server.run()
+
+
+if __name__ == "__main__":
+    main()

python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/requirements.txt

@@ -0,0 +1,2 @@
+agent-framework-foundry
+agent-framework-foundry-hosting

python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/verify_steering.py

@@ -0,0 +1,286 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""End-to-end steering test for the steerable single-agent countdown sample.
+
+Starts the server, kicks off a background streaming countdown, then -- while the model is still
+generating -- sends a second turn on the same conversation with a new target. Verifies the second
+turn is accepted immediately as "queued", that the first turn is cancelled and completes early
+(fewer tokens than a full run), and that the second (steered) turn completes with a countdown for
+its own target. Because this sample uses a real model (no deterministic per-tick pacing),
+assertions here are necessarily looser than an exact output match. Requires the
+same environment (.env) as running main.py directly.
+
+Usage:
+    python verify_steering.py [--first-target N] [--second-target N] [--min-deltas-before-steering N]
+"""
+
+import argparse
+import json
+import os
+import shutil
+import subprocess
+import sys
+import threading
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import IO, Any
+
+HOST = "127.0.0.1"
+PORT = 8088
+BASE_URL = f"http://{HOST}:{PORT}"
+SAMPLE_DIR = Path(__file__).parent
+LOG_PATH = SAMPLE_DIR / "verify_steering.log"
+
+
+def _http_get(path: str, timeout: float = 5.0) -> tuple[int, dict[str, Any]]:
+    with urllib.request.urlopen(urllib.request.Request(f"{BASE_URL}{path}"), timeout=timeout) as resp:
+        return resp.status, json.loads(resp.read())
+
+
+def _http_post(path: str, payload: dict[str, Any], timeout: float = 30.0) -> tuple[int, dict[str, Any]]:
+    data = json.dumps(payload).encode("utf-8")
+    request = urllib.request.Request(
+        f"{BASE_URL}{path}", data=data, headers={"Content-Type": "application/json"}, method="POST"
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as resp:
+            return resp.status, json.loads(resp.read())
+    except urllib.error.HTTPError as exc:
+        return exc.code, json.loads(exc.read())
+
+
+def _poll_until_terminal(response_id: str, timeout: float) -> dict[str, Any]:
+    """Poll ``GET /responses/{id}`` until the response reaches a terminal status.
+
+    A ``stream=false`` create only guarantees a fast initial ack (e.g. ``"queued"``); this polls
+    for the actual outcome instead of trying to replay it live (a ``?stream=true`` GET replay is
+    only valid for a response that was itself created with ``stream=true``).
+    """
+    deadline = time.monotonic() + timeout
+    snapshot: dict[str, Any] = {}
+    while time.monotonic() < deadline:
+        snapshot = _http_get(f"/responses/{response_id}")[1]
+        if snapshot.get("status") in ("completed", "failed", "incomplete", "cancelled"):
+            return snapshot
+        time.sleep(0.5)
+    return snapshot
+
+
+def _start_server(log_file: IO[str]) -> subprocess.Popen:  # type: ignore
+    return subprocess.Popen(
+        [sys.executable, "main.py"],
+        cwd=SAMPLE_DIR,
+        env={**os.environ, "PYTHONIOENCODING": "utf-8"},
+        stdout=log_file,
+        stderr=subprocess.STDOUT,
+    )
+
+
+def _wait_for_ready(timeout: float = 30.0) -> None:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            status, _ = _http_get("/readiness", timeout=2.0)
+            if status == 200:
+                return
+        except (urllib.error.URLError, ConnectionError, TimeoutError):
+            pass
+        time.sleep(0.5)
+    raise RuntimeError("Server did not become ready in time.")
+
+
+def _kill(server: subprocess.Popen) -> None:  # type: ignore
+    if server.poll() is None:
+        server.kill()
+        server.wait(timeout=10)
+
+
+def _watch_sse(request: "urllib.request.Request | str", progress: dict[str, Any]) -> None:
+    """Read an SSE stream from a streaming create POST and track its progress.
+
+    Tracks the response id (on ``response.created``), a running count of text delta events (a
+    single-agent response streams as one message, not discrete output items), and signals
+    ``progress["done"]`` on any terminal event.
+    """
+    try:
+        with urllib.request.urlopen(request) as resp:
+            # Without an explicit conversation_id, the session id (which scopes the conversation
+            # chain id used to attach a steered turn to the same task) must be forwarded by the
+            # caller on later turns -- otherwise each turn derives a different session id locally.
+            session_id = resp.headers.get("x-agent-session-id")
+            if session_id:
+                progress["session_id"] = session_id
+            current_event: str | None = None
+            for raw_line in resp:
+                line = raw_line.decode("utf-8").rstrip("\n")
+                if line.startswith("event:"):
+                    current_event = line[len("event:") :].strip()
+                    continue
+                if not line.startswith("data:"):
+                    continue
+                data_obj = json.loads(line[len("data:") :].strip())
+                if current_event == "response.created" and "id" not in progress:
+                    progress["id"] = data_obj["response"]["id"]
+                    progress["status"] = data_obj["response"]["status"]
+                    progress["ready"].set()
+                elif current_event == "response.output_text.delta":
+                    progress["delta_count"] += 1
+                elif current_event in ("response.completed", "response.failed", "response.incomplete"):
+                    progress["done"].set()
+    except urllib.error.HTTPError as exc:
+        progress["error"] = f"HTTP {exc.code}: {exc.read().decode('utf-8', errors='replace')}"
+    except (urllib.error.URLError, ConnectionError, TimeoutError, OSError) as exc:
+        progress["error"] = f"{type(exc).__name__}: {exc}"
+    finally:
+        progress["ready"].set()
+        progress["done"].set()
+
+
+def _extract_output_text(output_items: list[dict[str, Any]]) -> str:
+    parts: list[str] = []
+    for item in output_items:
+        if item.get("type") != "message":
+            continue
+        for part in item.get("content", []):
+            if part.get("type") == "output_text":
+                parts.append(part["text"])
+    return "".join(parts)
+
+
+def _clear_stale_state() -> None:
+    """Wipe ~/.agentserver so a prior run's task/queue state never leaks into this run."""
+    state_root = Path.home() / ".agentserver"
+    if state_root.exists():
+        shutil.rmtree(state_root, ignore_errors=True)
+        print(f"      cleared stale state: {state_root}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--first-target", type=int, default=30, help="First turn's countdown starting value.")
+    parser.add_argument("--second-target", type=int, default=3, help="Steered turn's countdown starting value.")
+    parser.add_argument(
+        "--min-deltas-before-steering",
+        type=int,
+        default=15,
+        help="Minimum text delta events to observe on turn 1 before sending the steering turn.",
+    )
+    args = parser.parse_args()
+
+    _clear_stale_state()
+
+    log_file = LOG_PATH.open("w", encoding="utf-8")
+    print(f"Server logs (DEBUG level) are redirected to {LOG_PATH}.")
+
+    print(f"[1/5] Starting server (first target={args.first_target}, second target={args.second_target})...")
+    server = _start_server(log_file)  # type: ignore
+    print(f"      PID: {server.pid}")
+    try:
+        _wait_for_ready()
+
+        print("[2/5] Starting the first turn's background streaming countdown...")
+        first_progress: dict[str, Any] = {
+            "delta_count": 0,
+            "ready": threading.Event(),
+            "done": threading.Event(),
+        }
+        first_payload = {
+            "input": f"Count down from {args.first_target}, slowly and with commentary.",
+            "store": True,
+            "background": True,
+            "stream": True,
+        }
+        first_data = json.dumps(first_payload).encode("utf-8")
+        first_request = urllib.request.Request(
+            f"{BASE_URL}/responses", data=first_data, headers={"Content-Type": "application/json"}, method="POST"
+        )
+        first_watcher = threading.Thread(target=_watch_sse, args=(first_request, first_progress), daemon=True)
+        first_watcher.start()
+        if not first_progress["ready"].wait(timeout=60):
+            raise SystemExit("FAIL: did not receive response.created for turn 1 in time.")
+        if "id" not in first_progress:
+            raise SystemExit(f"FAIL: turn 1 create request failed: {first_progress.get('error', 'unknown error')}")
+        first_id = first_progress["id"]
+        print(f"      turn 1 response id: {first_id}, status: {first_progress['status']}")
+
+        print(f"[3/5] Waiting for turn 1 to stream at least {args.min_deltas_before_steering} tokens...")
+        deadline = time.monotonic() + 60
+        while first_progress["delta_count"] < args.min_deltas_before_steering:
+            if first_progress["done"].is_set() or time.monotonic() > deadline:
+                raise SystemExit(
+                    "FAIL: turn 1 finished or timed out before enough tokens streamed to steer reliably; "
+                    f"observed {first_progress['delta_count']} delta(s). See {LOG_PATH} for server logs."
+                )
+            time.sleep(0.1)
+        count_at_steer_time = first_progress["delta_count"]
+        print(f"      turn 1 text deltas observed before steering: {count_at_steer_time}")
+
+        print(f"[4/5] Sending the steering turn (new target={args.second_target})...")
+        second_payload = {
+            "input": f"Actually, count down from {args.second_target} instead.",
+            "store": True,
+            "background": True,
+            "stream": False,
+            "previous_response_id": first_id,
+        }
+        # Forward the session id turn 1 was assigned so this turn resolves to the same
+        # conversation chain and is queued as a steer instead of starting a fresh task.
+        if "session_id" in first_progress:
+            second_payload["agent_session_id"] = first_progress["session_id"]
+        status, body = _http_post("/responses", second_payload)
+        if status != 200 or body.get("status") != "queued":
+            raise SystemExit(f"FAIL: expected an immediate queued response for the steering turn, got: {body}")
+        second_id = body["id"]
+        print(f"      steering turn accepted immediately as queued; response id: {second_id}")
+
+        print("[5/5] Watching turn 1 end early and the steered turn complete...")
+        first_progress["done"].wait(timeout=120)
+        first_final = _http_get(f"/responses/{first_id}")[1]
+        second_final = _poll_until_terminal(second_id, timeout=60)
+    finally:
+        _kill(server)
+        log_file.close()
+
+    first_text = _extract_output_text(first_final.get("output", []))
+    second_text = _extract_output_text(second_final.get("output", []))
+
+    print(f"      turn 1 final status: {first_final['status']}, {len(first_text)} character(s): {first_text}")
+    print(f"      turn 2 final status: {second_final['status']}, {len(second_text)} character(s): {second_text}")
+
+    if "Serving steered turn" in LOG_PATH.read_text(encoding="utf-8"):
+        print("      confirmed 'Serving steered turn' in the server log.")
+
+    if first_final["status"] != "completed":
+        raise SystemExit(
+            f"FAIL: turn 1 did not complete; last status: {first_final['status']}. See {LOG_PATH} for server logs."
+        )
+    # Loose bound: a steered turn 1 should have generated only a bit more than what we observed
+    # right before steering, not a whole additional full run's worth of tokens.
+    if first_progress["delta_count"] > count_at_steer_time * 3 + 20:
+        raise SystemExit(
+            "FAIL: turn 1 kept streaming long after the steering turn was sent -- steering did not "
+            f"cancel it in time. See {LOG_PATH} for server logs."
+        )
+
+    if second_final["status"] != "completed":
+        raise SystemExit(
+            f"FAIL: turn 2 did not complete; last status: {second_final['status']}. See {LOG_PATH} for server logs."
+        )
+    # Weak ordering check: each number from the new target down to 1 must appear, in order.
+    search_from = 0
+    for n in range(args.second_target, 0, -1):
+        idx = second_text.find(str(n), search_from)
+        if idx == -1:
+            raise SystemExit(
+                f"FAIL: steered turn output is missing '{n}' in order.\n  got: {second_text!r}\n"
+                f"See {LOG_PATH} for server logs."
+            )
+        search_from = idx + 1
+
+    print("PASS: the steering turn cancelled the in-progress countdown early and completed its own countdown.")
+
+
+if __name__ == "__main__":
+    main()