Python: Foundry Hosted Agent Resiliency Support (#7670)

* Migrate FHA to responses==2.0.0b1 and add Foundry state store * Fix session id error * Fix tests * Improve tests * Fix copilot comments * Address comments * Revert sample changes * Address comments * Add ContextScopedStoreProvider * Fix type check * Fix type check * LRA on top of state store * Temp disable state store user isolation * Simulate shutdown * Remove sim shutdown * Add sample * refine resiliency sample * Add steerable conversation support * Revert uv.lock * Add last_checkpoint_id and checkpoint existence check * Tighted resilient-recovery states * Tests for tightened resilient-recovery states * Make cancellation effective even when the iterator is stuck * Add more tests and fix sample * Small adjustment after review * Fix typing * Fix typing * Close driver background task in case of exceptions raised in the consumer * Handle usage content * xfail an integration test due to a known gap * Fix formatting --------- Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>
2026-08-21 17:37:47 +00:00
parent eb8548e582
commit 7b7b9a128c
26 changed files with 2700 additions and 345 deletions
@@ -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

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