diff --git a/python/packages/foundry_hosting/README.md b/python/packages/foundry_hosting/README.md
index 0739b9d212..d73bc714c8 100644
--- a/python/packages/foundry_hosting/README.md
+++ b/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`.
\ No newline at end of file
+storage locally. Stored approvals are scoped under `function_approvals`.
diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
index 17b73b6043..863cf9a06c 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -7,12 +7,13 @@
 import json
 import logging
 import os
-from collections.abc import AsyncIterable, AsyncIterator, Generator, Mapping, Sequence
+from collections.abc import AsyncGenerator, AsyncIterable, AsyncIterator, Generator, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, AsyncExitStack
 from dataclasses import asdict, dataclass, is_dataclass
 from typing import Literal, cast
 
 from agent_framework import (
+    AgentResponseUpdate,
     ChatOptions,
     CheckpointStorage,
     Content,
@@ -59,6 +60,7 @@
     ReasoningSummaryPartBuilder,
     TextContentBuilder,
 )
+from azure.ai.agentserver.responses.streaming._checkpoint import ResponseCheckpointEvent
 from mcp import McpError
 from typing_extensions import Any
 
@@ -103,6 +105,23 @@ def _is_hosted_responses_history_sentinel(provider: ContextProvider) -> bool:
     )
 
 
+def _create_response_event_stream(context: ResponseContext) -> ResponseEventStream:
+    """Create a response stream seeded from recovery state when available."""
+    if context.is_recovery:
+        persisted_response = context.persisted_response
+        if persisted_response is not None:
+            return ResponseEventStream(response=persisted_response, response_id=context.response_id)
+    return ResponseEventStream(response_id=context.response_id)
+
+
+# Reserved response metadata key pinning the workflow checkpoint that was current at the moment of
+# the last successfully persisted response-stream checkpoint. Recovery MUST resume from this specific
+# checkpoint, not simply the latest one in checkpoint_storage: the workflow may have saved further
+# checkpoints after it but before a crash, without their output ever being durably recorded in
+# response.output (see ``_handle_inner_workflow``).
+_LATEST_CHECKPOINT_ID_KEY = "_last_checkpoint_id"
+
+
 # Foundry Toolbox Auth integration
 # Consent-URL error code returned by the Foundry MCP gateway when calling `/list`
 CONSENT_ERROR_CODE = -32006
@@ -215,6 +234,23 @@ def __init__(
             2. The agent must not have any context providers that maintain context
                in memory, because the hosting environment may get deactivated between
                requests, and any in-memory context would be lost.
+            3. Resiliency (resilient_background=True) is ONLY supported for workflows; constructing this
+               server with a non-workflow agent and `resilient_background=True` raises `RuntimeError`.
+               When resiliency is enabled, and the server crashes mid-response:
+               - Background responses are automatically re-invoked on server restart (client won't see the crash).
+               - Stream events are preserved for client reconnection.
+               - State is maintained across crashes.
+            4. Steering (steerable_conversations=True) is ONLY supported for non-workflow agents; constructing
+               this server with a workflow agent and `steerable_conversations=True` raises `RuntimeError`.
+               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. It's also not currently practical to implement: a workflow
+               instance cannot start a new run until its previous (steered-past) run has been garbage
+               collected, and that isn't guaranteed to have happened in time.
+
+        Raises:
+            RuntimeError: If `resilient_background=True` is requested for a non-workflow agent, or if
+                `steerable_conversations=True` is requested for a workflow agent.
         """
         super().__init__(prefix=prefix, options=options, store=store, **kwargs)
 
@@ -226,12 +262,6 @@ def __init__(
                     "There shouldn't be a history provider with `load_messages=True` already present. "
                     "History is managed by the hosting infrastructure."
                 )
-            provider = cast(ContextProvider, provider)
-            logger.warning(
-                "Context provider %s is present. If it maintains context in memory, "
-                "the context may be lost between requests. Use with caution.",
-                provider.source_id,
-            )
 
         self._is_workflow_agent = False
         if isinstance(agent, WorkflowAgent):
@@ -275,6 +305,21 @@ def __init__(
             else function_approval_store_provider
         )
 
+        # Resiliency check: fail loud rather than silently downgrading to a non-recoverable row.
+        self._resilient_background = bool(options and options.resilient_background)
+        if self._resilient_background and not self._is_workflow_agent:
+            raise RuntimeError(
+                "resilient_background=True is only supported for workflow agents. "
+                "Crash recovery cannot be provided for non-workflow agents."
+            )
+
+        # Steering check: steering a workflow is conceptually undefined and also impractical today.
+        if options and options.steerable_conversations and self._is_workflow_agent:
+            raise RuntimeError(
+                "steerable_conversations=True is only supported for non-workflow agents. "
+                "Steering cannot be provided reliably for workflow agents."
+            )
+
         # Lazy agent lifecycle: the agent (and any MCP tools it owns) is entered on
         # the first request rather than at server startup, so that authentication
         # failures during MCP connect can be surfaced to the client as an
@@ -320,27 +365,18 @@ async def _handle_response(
         request: CreateResponse,
         context: ResponseContext,
         cancellation_signal: asyncio.Event,
-    ) -> AsyncIterable[ResponseStreamEvent | dict[str, Any]]:
+    ) -> AsyncIterable[ResponseStreamEvent | ResponseCheckpointEvent]:
         """Handle the creation of a response."""
-        if self._is_workflow_agent:
-            # Workflow agents are handled differently because they require checkpoint restoration
-            return self._handle_inner_workflow(request, context)
-        return self._handle_inner_agent(request, context)
+        # Common per-request setup shared by the workflow and non-workflow paths:
+        # create the response stream and the streaming output-item tracker, emit
+        # the opening lifecycle events, and convert any exception raised while
+        # producing the response into a terminal ``response.failed`` event (which
+        # also drains the tracker so the SSE stream stays well-formed).
+        response_event_stream = _create_response_event_stream(context)
+
+        if context.is_steered_turn:
+            logger.debug("Serving steered turn (pending_input_count=%d)", context.pending_input_count)
 
-    async def _handle_inner_agent(
-        self,
-        request: CreateResponse,
-        context: ResponseContext,
-    ) -> AsyncIterable[ResponseStreamEvent | dict[str, Any]]:
-        """Handle a regular agent with Responses-managed MAF session continuity.
-
-        Conversation mode reads and writes one MAF session snapshot under
-        ``conversation_id``. Response chaining reads the snapshot under
-        ``previous_response_id`` and writes the updated session under the current
-        ``response_id``, allowing branches without changing the MAF session's own
-        identifier. Hosted storage uses the request user as its isolation boundary.
-        """
-        response_event_stream = ResponseEventStream(response_id=context.response_id)
         yield response_event_stream.emit_created()
         yield response_event_stream.emit_in_progress()
 
@@ -355,9 +391,7 @@ async def _handle_inner_agent(
             consent_errors_to_emit = consent_url_from_error(ex)
             if consent_errors_to_emit is None or len(consent_errors_to_emit) == 0:
                 logger.error("Failed to prepare agent: %s", ex, exc_info=(type(ex), ex, ex.__traceback__))
-                for event in self._emit_failure(response_event_stream, None, ex):
-                    yield event
-                return
+                raise
 
             for consent_error in consent_errors_to_emit:
                 logger.warning("Consent URL for tool '%s': %s", consent_error.name, consent_error.consent_url)
@@ -377,6 +411,55 @@ async def _handle_inner_agent(
             )
             return
 
+        tracker = _OutputItemTracker(response_event_stream)
+        try:
+            if self._is_workflow_agent:
+                inner = self._handle_inner_workflow(
+                    request, context, response_event_stream, tracker, cancellation_signal
+                )
+            else:
+                inner = self._handle_inner_agent(request, context, response_event_stream, tracker, cancellation_signal)
+
+            try:
+                async for event in inner:
+                    yield event
+            except BaseException:
+                await inner.aclose()
+                raise
+
+            for event in tracker.close():
+                yield event
+
+            yield response_event_stream.emit_completed()
+        except Exception as ex:
+            logger.error("Failed to produce response for agent", exc_info=(type(ex), ex, ex.__traceback__))
+            for event in tracker.close():
+                yield event
+
+            for event in self._emit_failure(response_event_stream, tracker, ex):
+                yield event
+
+    async def _handle_inner_agent(
+        self,
+        request: CreateResponse,
+        context: ResponseContext,
+        response_event_stream: ResponseEventStream,
+        tracker: _OutputItemTracker,
+        cancellation_signal: asyncio.Event,
+    ) -> AsyncGenerator[ResponseStreamEvent]:
+        """Handle a regular (non-workflow) agent.
+
+        The response stream, tracker, and opening lifecycle events are produced
+        by :meth:`_handle_response`, which also converts any raised exception
+        into a terminal ``response.failed`` event (draining the tracker so the
+        SSE stream stays well-formed).
+        """
+        if context.is_recovery:
+            logger.warning(
+                "Recovery mode is not supported for non-workflow agents. "
+                "The agent will restart from the original input."
+            )
+
         try:
             request_context = get_request_context()
             approval_storage = self._function_approval_storage_provider.get_store(
@@ -398,11 +481,8 @@ async def _handle_inner_agent(
             session_save_id = context.conversation_id or context.response_id
         except Exception as ex:
             logger.error("Failed to prepare state storage: %s", ex, exc_info=(type(ex), ex, ex.__traceback__))
-            for event in self._emit_failure(response_event_stream, None, ex):
-                yield event
-            return
+            raise
 
-        tracker = _OutputItemTracker(response_event_stream)
         request_failure: Exception | None = None
         save_failure: Exception | None = None
         request_interrupted = False
@@ -430,8 +510,12 @@ async def _handle_inner_agent(
                 run_kwargs["options"] = chat_options
 
             async for update in self._agent.run(stream=True, **run_kwargs):  # type: ignore[reportUnknownMemberType]
+                if context.shutdown.is_set() or cancellation_signal.is_set():
+                    # Non-workflow agents can't be resilient, so there is no exit_for_recovery path here:
+                    # both shutdown and steering/cancel just wind the turn down and let it complete normally.
+                    break
                 for content in update.contents:
-                    for event in tracker.handle(content):
+                    for event in tracker.handle(content, message_id=update.message_id):
                         yield event
                     if tracker.needs_async:
                         async for item in _to_outputs(
@@ -441,8 +525,6 @@ async def _handle_inner_agent(
                         ):
                             yield item
                         tracker.needs_async = False
-            for event in tracker.close():
-                yield event
         except (asyncio.CancelledError, GeneratorExit):
             request_interrupted = True
             raise
@@ -468,34 +550,26 @@ async def _handle_inner_agent(
                 logger.error(message, exc_info=(type(save_error), save_error, save_error.__traceback__))
 
         if request_failure is not None and save_failure is not None:
-            failure = RuntimeError(
+            raise RuntimeError(
                 f"Agent request failed: {str(request_failure) or type(request_failure).__name__}; "
                 f"session persistence also failed: {str(save_failure) or type(save_failure).__name__}"
             )
-            for event in self._emit_failure(response_event_stream, tracker, failure):
-                yield event
         elif request_failure is not None:
-            for event in self._emit_failure(response_event_stream, tracker, request_failure):
-                yield event
+            raise request_failure
         elif save_failure is not None:
-            for event in self._emit_failure(response_event_stream, tracker, save_failure):
-                yield event
-        else:
-            yield response_event_stream.emit_completed()
+            raise save_failure
 
     async def _handle_inner_workflow(
         self,
         request: CreateResponse,
         context: ResponseContext,
-    ) -> AsyncIterable[ResponseStreamEvent | dict[str, Any]]:
+        response_event_stream: ResponseEventStream,
+        tracker: _OutputItemTracker,
+        cancellation_signal: asyncio.Event,
+    ) -> AsyncGenerator[ResponseStreamEvent | ResponseCheckpointEvent]:
         """Handle the creation of a response for a workflow agent."""
-        response_event_stream = ResponseEventStream(response_id=context.response_id)
-        yield response_event_stream.emit_created()
-        yield response_event_stream.emit_in_progress()
-
-        # Track the current active output item builder for streaming;
-        # lazily created on matching content, closed when a different type arrives.
-        tracker: _OutputItemTracker | None = None
+        if not isinstance(self._agent, WorkflowAgent):
+            raise RuntimeError("Agent is not a workflow agent.")
 
         try:
             request_context = get_request_context()
@@ -509,14 +583,11 @@ async def _handle_inner_workflow(
             if are_options_set:
                 logger.warning("Workflow agent doesn't support runtime options. They will be ignored.")
 
-            if not isinstance(self._agent, WorkflowAgent):
-                raise RuntimeError("Agent is not a workflow agent.")
-
-            # Workflow agents are not async context managers in any built-in path,
-            # but call _ensure_agent_ready for symmetry with the regular path so
-            # any future async resources owned by the workflow are entered here.
-            await self._ensure_agent_ready()
-
+            # Determine the checkpoint storage for this request. The checkpoint
+            # storage is keyed by the conversation ID (if present) or the response
+            # ID (if no conversation ID is present). On a subsequent turn, the same
+            # conversation ID or a `previous_response_id` can be used to resume the
+            # workflow from the last checkpoint.
             checkpoint_save_id = context.conversation_id or context.response_id
             _validate_checkpoint_context_id(checkpoint_save_id)
             checkpoint_storage = self._checkpoint_storage_provider.get_store(
@@ -525,69 +596,99 @@ async def _handle_inner_workflow(
                 platform_context=request_context,
             )
 
-            # Determine the latest checkpoint (if any) so we can resume the
-            # workflow's prior state for this turn. The directory is keyed by
-            # the platform derived context_id. Multi-turn declarative workflows
-            # need the workflow's internal state (e.g. Conversation.messages,
-            # intermediate Local.* variables) to survive across user turns;
-            # the only place that state lives is the workflow checkpoint, so
-            # on every turn we restore the latest checkpoint and feed the new
-            # input back into the start executor as a continuation rather than
-            # a fresh run.
-            if request.get("previous_response_id") is not None and context.conversation_id is not None:
-                raise RuntimeError("Previous response ID cannot be used in conjunction with conversation ID.")
-            previous_response_id = request.get("previous_response_id")
-            checkpoint_load_id = context.conversation_id or previous_response_id
-            latest_checkpoint = None
-            restore_checkpoint_storage = checkpoint_storage
-            if checkpoint_load_id is not None:
-                _validate_checkpoint_context_id(checkpoint_load_id)
-                if checkpoint_load_id != checkpoint_save_id:
-                    restore_checkpoint_storage = self._checkpoint_storage_provider.get_store(
-                        config=self.config,
-                        context_id=checkpoint_load_id,
-                        platform_context=request_context,
+            if context.is_recovery:
+                if not self._resilient_background:
+                    raise RuntimeError("Recovery mode is only supported when resilient_background=True.")
+                # Resume from the workflow checkpoint durably paired with the last persisted response
+                # snapshot (recorded in that snapshot's own metadata) -- NOT simply the latest workflow
+                # checkpoint in storage, which may be ahead of what response.output actually reflects if
+                # the crash happened between two response-stream checkpoint() calls.
+                checkpoint_id = response_event_stream.internal_metadata.get(_LATEST_CHECKPOINT_ID_KEY)
+                if checkpoint_id is not None:
+                    logger.debug("Serving recovery request from workflow checkpoint %s", checkpoint_id)
+                    run_stream = self._resume_workflow_from_checkpoint(
+                        checkpoint_id, checkpoint_storage, context.response_id
+                    )
+                else:
+                    # No checkpoint was ever paired with a persisted response snapshot (e.g. the crash
+                    # happened before the very first response checkpoint() call); replay the original
+                    # input as a fresh entry, per the recovered-input parity guarantee
+                    # (context.get_input_items() is unchanged from fresh entry).
+                    logger.debug("Serving recovery request with no prior workflow checkpoint; replaying original input")
+                    run_stream = self._agent.run(
+                        input_messages,
+                        stream=True,
+                        checkpoint_storage=checkpoint_storage,
                     )
+            else:
+                # Determine the latest checkpoint (if any) so we can resume the
+                # workflow's prior state for this turn. The directory is keyed by
+                # the conversation id or the previous response id.
+                previous_response_id = request.get("previous_response_id")
+                if previous_response_id is not None and context.conversation_id is not None:
+                    raise RuntimeError("Previous response ID cannot be used in conjunction with conversation ID.")
+                checkpoint_load_id = context.conversation_id or previous_response_id
+                restore_checkpoint_storage = checkpoint_storage
+                if checkpoint_load_id is not None:
+                    _validate_checkpoint_context_id(checkpoint_load_id)
+                    if checkpoint_load_id != checkpoint_save_id:
+                        restore_checkpoint_storage = self._checkpoint_storage_provider.get_store(
+                            config=self.config,
+                            context_id=checkpoint_load_id,
+                            platform_context=request_context,
+                        )
                 latest_checkpoint = await restore_checkpoint_storage.get_latest(workflow_name=self._agent.workflow.name)
+
                 if latest_checkpoint is None and previous_response_id is not None:
+                    # A previous_response_id must have a prior workflow checkpoint to resume from
                     raise RuntimeError(
                         f"Cannot find an existing workflow checkpoint for previous_response_id={previous_response_id}."
                     )
 
-            # Multi-turn pattern: when we have a prior checkpoint, restore it
-            # first (drive the workflow back to idle with prior state intact),
-            # then make a separate call that delivers the new user input. This
-            # depends on Workflow.run preserving shared state across calls. The
-            # restore-only call may yield events from any pending in-flight
-            # work in the checkpoint; we consume those internally here so they
-            # don't surface to the response stream as duplicates.
-            #
-            # If the restored checkpoint had pending request_info events, the
-            # restore-only call replays them through
-            # ``WorkflowAgent._convert_workflow_event_to_agent_response_updates``
-            # and populates ``self._agent.pending_requests``. That is the correct
-            # state: those requests are genuinely outstanding, and the next
-            # ``run(input_messages, ...)`` call may contain ``function_call_output``
-            # items (carried as FunctionResult/FunctionApprovalResponse content)
-            # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`.
-            if latest_checkpoint is not None:
-                async for _ in self._agent.run(
+                if latest_checkpoint is not None:
+                    # If we have a prior checkpoint, restore it first (drive the workflow
+                    # back to idle with prior state intact), then make a separate call that
+                    # delivers the new user input. The restore-only call may yield events
+                    # from any pending in-flight work in the checkpoint; we consume those
+                    # internally here so they don't surface to the response stream as duplicates.
+                    #
+                    # If the restored checkpoint had pending request_info events, the
+                    # restore-only call replays them through
+                    # ``WorkflowAgent._convert_workflow_event_to_agent_response_updates``
+                    # and populates ``self._agent.pending_requests``. That is the correct
+                    # state: those requests are genuinely outstanding, and the next
+                    # ``run(input_messages, ...)`` call may contain ``function_call_output``
+                    # items (carried as FunctionResult/FunctionApprovalResponse content)
+                    # that fulfill them via :meth:`WorkflowAgent._process_pending_requests`.
+                    async for _ in self._agent.run(
+                        stream=True,
+                        checkpoint_id=latest_checkpoint.checkpoint_id,
+                        checkpoint_storage=restore_checkpoint_storage,
+                    ):
+                        if context.shutdown.is_set():
+                            await context.exit_for_recovery()
+                        if cancellation_signal.is_set():
+                            return
+
+                # A cancel signal that fired after the restore-only replay finished (or was never
+                # entered) must still preempt starting a brand new workflow run below.
+                if cancellation_signal.is_set():
+                    return
+
+                run_stream = self._agent.run(
+                    input_messages,
                     stream=True,
-                    checkpoint_id=latest_checkpoint.checkpoint_id,
-                    checkpoint_storage=restore_checkpoint_storage,
-                ):
-                    pass
+                    checkpoint_storage=checkpoint_storage,
+                )
 
-            tracker = _OutputItemTracker(response_event_stream)
+            async for update in run_stream:
+                if context.shutdown.is_set():
+                    await context.exit_for_recovery()
+                if cancellation_signal.is_set():
+                    break
 
-            # Run the workflow agent in streaming mode with the new user input.
-            async for update in self._agent.run(
-                input_messages,
-                stream=True,
-                checkpoint_storage=checkpoint_storage,
-            ):
                 for content in update.contents:
-                    for event in tracker.handle(content):
+                    for event in tracker.handle(content, message_id=update.message_id):
                         yield event
                     if tracker.needs_async:
                         async for item in _to_outputs(
@@ -596,14 +697,50 @@ async def _handle_inner_workflow(
                             yield item
                         tracker.needs_async = False
 
-            # Close any remaining active builder
-            for event in tracker.close():
-                yield event
-            yield response_event_stream.emit_completed()
-        except Exception as ex:
+                # Pin the workflow checkpoint that is current right now -- before persisting the response
+                # snapshot -- so a future crash recovery resumes exactly here, not from a later workflow
+                # checkpoint whose output was never recorded in this response snapshot. The generator is
+                # paused while we hold control, so no further workflow checkpoints can appear underneath us.
+                if self._resilient_background:
+                    latest_checkpoint = await checkpoint_storage.get_latest(workflow_name=self._agent.workflow.name)
+                    if latest_checkpoint is not None:
+                        response_event_stream.internal_metadata[_LATEST_CHECKPOINT_ID_KEY] = (
+                            latest_checkpoint.checkpoint_id
+                        )
+                    yield response_event_stream.checkpoint()
+        except Exception:
             logger.exception("Failed to produce response for workflow agent")
-            for event in self._emit_failure(response_event_stream, tracker, ex):
-                yield event
+            raise
+
+    async def _resume_workflow_from_checkpoint(
+        self,
+        checkpoint_id: str,
+        checkpoint_storage: CheckpointStorage,
+        response_id: str,
+    ) -> AsyncGenerator[AgentResponseUpdate]:
+        """Resume a crashed background workflow run, forwarding every event it produces.
+
+        ``WorkflowAgent.run(checkpoint_id=..., messages=None)`` treats a message-less resume as
+        "restore only": it drives the workflow with the checkpoint's own already-queued internal
+        messages, but silently discards every event produced while doing so, on the assumption
+        that the workflow merely settles back to idle awaiting the next turn's input. That
+        assumption doesn't hold for crash recovery: the countdown (and any other self-driving
+        workflow) genuinely continues -- and may run to completion -- from its own queued
+        messages, and that output must not be lost. Drive the underlying ``Workflow`` directly so
+        none of it is discarded, converting each event the same way ``WorkflowAgent.run`` does.
+        """
+        if not isinstance(self._agent, WorkflowAgent):
+            raise RuntimeError("Agent is not a workflow agent.")
+        agent = self._agent
+        async for event in agent.workflow.run(
+            stream=True,
+            checkpoint_id=checkpoint_id,
+            checkpoint_storage=checkpoint_storage,
+        ):
+            for update in agent._convert_workflow_event_to_agent_response_updates(  # pyright: ignore[reportPrivateUsage]
+                response_id, event
+            ):
+                yield update
 
     @staticmethod
     def _emit_failure(
@@ -647,6 +784,10 @@ def __init__(self, stream: ResponseEventStream) -> None:
         self._stream = stream
         self._active_type: str | None = None
         self._active_id: str | None = None
+        # message_id of the update that opened the active text item, used to detect a new
+        # logical message (e.g. a fresh workflow yield_output call) even when the content
+        # type doesn't change, so it isn't silently merged into the still-open item.
+        self._active_message_id: str | None = None
         # Accumulated delta text for the current active builder
         self._accumulated: list[str] = []
         # Builder state — only one is active at a time
@@ -659,16 +800,26 @@ def __init__(self, stream: ResponseEventStream) -> None:
         self._mcp_builder: OutputItemMcpCallBuilder | None = None
         self.needs_async = False
 
-    def handle(self, content: Content) -> Generator[ResponseStreamEvent]:
+    def handle(self, content: Content, message_id: str | None = None) -> Generator[ResponseStreamEvent]:
         """Process a content item, yielding sync events.
 
+        Args:
+            content: The content item to process.
+            message_id: The ``message_id`` of the update ``content`` came from, if any. A
+                change in ``message_id`` across otherwise same-typed text content marks a new
+                logical message and forces the previous output item closed, rather than being
+                merged into it.
+
         Sets ``needs_async = True`` if the caller must also drain an
         async ``_to_outputs`` call for this content.
         """
         if content.type == "text" and content.text is not None:
-            if self._active_type != "text":
+            if self._active_type != "text" or (
+                message_id is not None and self._active_message_id is not None and message_id != self._active_message_id
+            ):
                 yield from self._close()
                 yield from self._open_message()
+            self._active_message_id = message_id
             self._accumulated.append(content.text)
             if self._text_content is not None:
                 yield self._text_content.emit_delta(content.text)
@@ -820,6 +971,7 @@ def _close(self) -> Generator[ResponseStreamEvent]:
 
         self._active_type = None
         self._active_id = None
+        self._active_message_id = None
         self._accumulated.clear()
 
 
diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_state_store.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_state_store.py
index 9df38a4332..d73b582f9b 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_state_store.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_state_store.py
@@ -90,7 +90,6 @@ async def _get_store(self) -> FoundryStateStore:
         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 @@ async def _get_store(self) -> FoundryStateStore:
         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 @@ async def _get_store(self) -> FoundryStateStore:
         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:
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index b12497e530..e3c1aa357f 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -15,6 +15,7 @@
 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,14 @@
     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,
+    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
@@ -404,16 +411,38 @@ async def save_messages(
         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 +493,7 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]:
 
         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 +699,7 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]:
         ):
             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 +736,7 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]:
         ):
             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 +747,45 @@ async def updates() -> AsyncIterator[AgentResponseUpdate]:
         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
+
 
 # endregion
 
@@ -3518,13 +3587,16 @@ async def test_workflow_rejects_invalid_checkpoint_scope(
         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"])
@@ -4113,6 +4185,81 @@ async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorReque
     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]) -> None:
+        self.id = str(uuid.uuid4())
+        self.name = name
+        self.description: str | None = None
+        self._texts = list(texts)
+        self.run_count = 0
+
+    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
+        texts = self._texts
+        name = self.name
+
+        async def _aiter() -> AsyncIterator[AgentResponseUpdate]:
+            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]) -> tuple[WorkflowAgent, _MultiUpdateWorkflowAgentMock]:
+    """Build a ``WorkflowAgent`` whose inner agent streams one update per text in ``texts``."""
+    inner = _MultiUpdateWorkflowAgentMock("multi-update-agent", texts)
+
+    @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,
@@ -4178,6 +4325,95 @@ async def test_basic_text_response_streaming(self) -> None:
         text_done = [e for e in events if e["event"] == "response.output_text.done"]
         assert any(e["data"]["text"] == "hello stream" for e in text_done)
 
+    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_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_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)
@@ -4386,3 +4622,36 @@ async def test_round_trip_approval_response_rejected(self) -> None:
 
 
 # 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
diff --git a/python/packages/foundry_hosting/tests/test_state_store.py b/python/packages/foundry_hosting/tests/test_state_store.py
index e08d4dfabd..23a488c03d 100644
--- a/python/packages/foundry_hosting/tests/test_state_store.py
+++ b/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")
 
 
diff --git a/python/samples/04-hosting/foundry-hosted-agents/README.md b/python/samples/04-hosting/foundry-hosted-agents/README.md
index 4c7f1f34d6..f89da96e1b 100644
--- a/python/samples/04-hosting/foundry-hosted-agents/README.md
+++ b/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
 
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/.dockerignore b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/.dockerignore
new file mode 100644
index 0000000000..0d6619bae0
--- /dev/null
+++ b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/.dockerignore
@@ -0,0 +1,7 @@
+.venv
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.Python
+.env
\ No newline at end of file
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/.env.example b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/.env.example
new file mode 100644
index 0000000000..4d268b931b
--- /dev/null
+++ b/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="..."
\ No newline at end of file
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/Dockerfile b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/Dockerfile
new file mode 100644
index 0000000000..eaffb94f19
--- /dev/null
+++ b/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"]
\ No newline at end of file
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/README.md b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/README.md
new file mode 100644
index 0000000000..cdf9df3eac
--- /dev/null
+++ b/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.
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/agent.manifest.yaml b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/agent.manifest.yaml
new file mode 100644
index 0000000000..c36ecba231
--- /dev/null
+++ b/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
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/agent.yaml b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/agent.yaml
new file mode 100644
index 0000000000..423c31bdea
--- /dev/null
+++ b/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}
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/main.py b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/main.py
new file mode 100644
index 0000000000..e941f23cd3
--- /dev/null
+++ b/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()
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/requirements.txt b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/requirements.txt
new file mode 100644
index 0000000000..0b85d57fae
--- /dev/null
+++ b/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
\ No newline at end of file
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/verify_resiliency.py b/python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/verify_resiliency.py
new file mode 100644
index 0000000000..9a5966456d
--- /dev/null
+++ b/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()
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/.dockerignore b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/.dockerignore
new file mode 100644
index 0000000000..31ed562a7e
--- /dev/null
+++ b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/.dockerignore
@@ -0,0 +1,7 @@
+.venv
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.Python
+.env
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/.env.example b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/.env.example
new file mode 100644
index 0000000000..4d268b931b
--- /dev/null
+++ b/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="..."
\ No newline at end of file
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/Dockerfile b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/Dockerfile
new file mode 100644
index 0000000000..0cc939d9b3
--- /dev/null
+++ b/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"]
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/README.md b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/README.md
new file mode 100644
index 0000000000..32720cee1c
--- /dev/null
+++ b/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.
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/agent.manifest.yaml b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/agent.manifest.yaml
new file mode 100644
index 0000000000..90c39b67d4
--- /dev/null
+++ b/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
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/agent.yaml b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/agent.yaml
new file mode 100644
index 0000000000..eb5b534040
--- /dev/null
+++ b/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}
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/main.py b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/main.py
new file mode 100644
index 0000000000..8ed3a5a6ac
--- /dev/null
+++ b/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()
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/requirements.txt b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/requirements.txt
new file mode 100644
index 0000000000..0b85d57fae
--- /dev/null
+++ b/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
\ No newline at end of file
diff --git a/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/verify_steering.py b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/verify_steering.py
new file mode 100644
index 0000000000..01c6139923
--- /dev/null
+++ b/python/samples/04-hosting/foundry-hosted-agents/responses/steerable_long_running_agent/verify_steering.py
@@ -0,0 +1,284 @@
+# 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 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,
+        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()