Use distinct private AgentServer wheel versions

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 1674d491-af7d-4f9e-99aa-5a829db81591

python/pyrightconfig.samples.json

@@ -9,7 +9,14 @@
         "**/05-end-to-end/**",
         "**/harness/**",
         "**/agent_with_foundry_tracing.py",
-        "**/azure_responses_client_with_foundry.py"
+        "**/azure_responses_client_with_foundry.py",
+        "**/monty_code_act.py",
+        "**/monty_code_interpreter.py",
+        "**/monty_code_interpreter_manual_wiring.py",
+        "**/11_monty_codeact/**",
+        "**/local_shell_with_allowlist.py",
+        "**/local_shell_with_environment_provider.py",
+        "**/client_with_local_shell.py"
     ],
     "typeCheckingMode": "basic",
     "reportMissingImports": "error",

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

@@ -38,6 +38,11 @@ To have a multi-turn conversation with the agent, include the previous response
 curl -X POST http://localhost:8088/responses -H "Content-Type: application/json" -d '{"input": "How are you?", "previous_response_id": "REPLACE_WITH_PREVIOUS_RESPONSE_ID"}'
 ```
 
+## Next steps
+
+- [13_steering](../13_steering) — steerable conversations: queue a new turn while the previous one is still running
+- [15_workflow_resilience](../15_workflow_resilience) — durable workflow recovery across host restarts
+
 ## Deploying the Agent to Foundry
 
 To host the agent on Foundry, follow the instructions in the [Deploying the Agent to Foundry](../../README.md#deploying-the-agent-to-foundry) section of the README in the parent directory.

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

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

python/samples/04-hosting/foundry-hosted-agents/responses/13_steering/Dockerfile

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

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

@@ -0,0 +1,65 @@
+# What this sample demonstrates
+
+How to enable **steerable conversations** with `ResponsesHostServer`.
+
+When `steerable_conversations=True`, a client can send a new turn to the same
+conversation while the previous turn is still running.  Instead of receiving
+HTTP 409 `conversation_locked`, the new turn is queued and the running handler
+receives a cooperative cancellation signal.  Once the current turn terminates,
+the queued turn runs.
+
+## How It Works
+
+1. The server starts with `steerable_conversations=True`.
+2. Turn 1 is sent as a foreground streaming request on a shared conversation.
+3. Two seconds later, while turn 1 is still streaming, turn 2 arrives on the
+   same conversation.  The server immediately returns `status=queued`.
+4. The running handler observes `cancellation_signal.is_set()` (with
+   `context.client_cancelled=False`, distinguishing steering from a real cancel)
+   and emits `response.completed` with partial output.
+5. Once turn 1 finishes the framework drains the queue and invokes the handler
+   again for turn 2 (with `context.is_steered_turn=True`).
+
+See [main.py](main.py) for the server and [client.py](client.py) for the
+interactive demo.
+
+## Running the Sample
+
+**Terminal 1 — start the server:**
+
+```bash
+uv run python main.py
+```
+
+**Terminal 2 — run the demo client:**
+
+```bash
+uv run python client.py
+```
+
+You will see turn 1 streaming, turn 2 arriving as `status=queued`, turn 1
+cutting off, and then turn 2's answer appearing.  The server terminal shows
+the steering signal and handler cancellation in real time.
+
+You can also point the client at a deployed instance:
+
+```bash
+uv run python client.py https://your-deployed-agent-url
+```
+
+## Environment Variables
+
+Copy `.env.example` to `.env` and fill in your values:
+
+```
+FOUNDRY_PROJECT_ENDPOINT=https://...
+AZURE_AI_MODEL_DEPLOYMENT_NAME=gpt-4o
+```
+
+Run `az login` before starting the server.
+
+## Deploying the Agent to Foundry
+
+To host the agent on Foundry, follow the instructions in the
+[Deploying the Agent to Foundry](../../README.md#deploying-the-agent-to-foundry)
+section of the README in the parent directory.

python/samples/04-hosting/foundry-hosted-agents/responses/13_steering/client.py

@@ -0,0 +1,175 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Steerable conversations demo client.
+
+Demonstrates steerable_conversations by sending two concurrent turns to a
+running ResponsesHostServer.  Run this while main.py is running in a second
+terminal to see steering in action.
+
+Usage
+-----
+Terminal 1 — start the server:
+    uv run python main.py
+
+Terminal 2 — run this client:
+    uv run python client.py [SERVER_URL]
+
+SERVER_URL defaults to http://localhost:8088.
+Set it to your deployed agent URL to test against a hosted instance.
+
+What you will see
+-----------------
+Turn 1 starts streaming a long response (counting to 50).  Two seconds later,
+turn 2 arrives on the same conversation with a different question.  Because the
+server has steerable_conversations=True, turn 2 is accepted immediately with
+status=queued rather than rejected with 409.  The server then cancels turn 1's
+handler, which emits a partial response.completed event.  Once turn 1 finishes,
+turn 2 runs and streams its answer.
+
+Watch the server terminal to see the steering signal, cancellation, and drain
+logged in real time.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import sys
+from typing import Any
+
+import httpx
+from dotenv import load_dotenv
+
+load_dotenv()
+
+_DEFAULT_BASE = "http://localhost:8088"
+_MODEL = os.environ.get("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4o")
+
+
+def _parse_sse(chunk: str) -> dict[str, Any] | None:
+    for line in chunk.splitlines():
+        if line.startswith("data: "):
+            try:
+                return json.loads(line[6:])  # type: ignore[no-any-return]
+            except json.JSONDecodeError:
+                pass
+    return None
+
+
+async def demo(base: str) -> None:
+    """Run the steering demonstration against *base* URL."""
+    import uuid
+
+    conv_id = f"demo-{uuid.uuid4().hex[:8]}"
+    turn1_text: list[str] = []
+    turn1_terminal: str | None = None
+
+    limits = httpx.Limits(max_keepalive_connections=5, max_connections=5)
+    async with httpx.AsyncClient(base_url=base, timeout=120, limits=limits) as client:
+        print(f"\nConnected to {base}")
+        print(f"Conversation ID: {conv_id}\n")
+
+        # ------------------------------------------------------------------
+        # Turn 1: ask the agent to count slowly.  Stream the response in
+        # real-time so we can see the partial output before steering.
+        # ------------------------------------------------------------------
+        async def stream_turn1() -> None:
+            nonlocal turn1_terminal
+            print(">>> Turn 1: 'Count from 1 to 50, one number per line.'")
+            async with client.stream(
+                "POST",
+                "/responses",
+                json={
+                    "model": _MODEL,
+                    "input": "Count from 1 to 50, one number per line.",
+                    "stream": True,
+                    "store": True,
+                    # "conversation" is the b8 field for a shared conversation ID.
+                    # Both turns must carry the same value to share the same
+                    # multi-turn steerable task.
+                    "conversation": conv_id,
+                },
+            ) as resp:
+                assert resp.status_code == 200, f"Turn 1 failed: {resp.status_code}"
+                async for chunk in resp.aiter_text():
+                    event = _parse_sse(chunk)
+                    if event is None:
+                        continue
+                    et = event.get("type", "")
+                    if et == "response.output_item.delta":
+                        delta = (
+                            event.get("delta", {}).get("text", "")
+                            or event.get("delta", {}).get("content", "")
+                            or ""
+                        )
+                        if delta:
+                            print(delta, end="", flush=True)
+                            turn1_text.append(delta)
+                    elif et in ("response.completed", "response.failed", "response.cancelled"):
+                        turn1_terminal = et.split(".")[-1]
+                        break
+
+        # ------------------------------------------------------------------
+        # Turn 2: sent 2 s later while turn 1 is still streaming.
+        # background=True returns status=queued immediately so we don't block.
+        # ------------------------------------------------------------------
+        async def send_turn2() -> dict[str, Any]:
+            await asyncio.sleep(2)
+            print("\n\n>>> Turn 2 (sent while turn 1 is still running): 'What is the capital of France?'")
+            r2 = await client.post(
+                "/responses",
+                json={
+                    "model": _MODEL,
+                    "input": "What is the capital of France?",
+                    "background": True,
+                    "store": True,
+                    "conversation": conv_id,
+                },
+            )
+            assert r2.status_code == 200, f"Turn 2 failed: {r2.status_code} {r2.text}"
+            return r2.json()  # type: ignore[no-any-return]
+
+        t1 = asyncio.create_task(stream_turn1())
+        r2 = await send_turn2()
+
+        r2_id = r2["id"]
+        r2_status = r2["status"]
+        print(f"\n    Turn 2 response ID : {r2_id}")
+        print(f"    Turn 2 status      : {r2_status}")
+
+        if r2_status != "queued":
+            print(
+                f"\nUnexpected status {r2_status!r} — expected 'queued'.\n"
+                "  If 'conflict': the server may not have steerable_conversations=True.\n"
+                "  If 'in_progress': turn 1 completed before turn 2 arrived; "
+                "the model may be responding too quickly for this prompt."
+            )
+
+        # Wait for turn 1 to drain
+        await t1
+        print(f"\n    Turn 1 terminated  : response.{turn1_terminal}")
+        print(f"    Turn 1 text so far : {len(''.join(turn1_text))} chars")
+
+        # Poll turn 2 until it completes, then print the agent's reply
+        print("\n>>> Waiting for turn 2 to complete...")
+        for _ in range(120):
+            await asyncio.sleep(0.5)
+            s2_resp = (await client.get(f"/responses/{r2_id}")).json()
+            s2 = s2_resp["status"]
+            print(f"    status: {s2}")
+            if s2 in ("completed", "failed", "cancelled"):
+                if s2 == "completed":
+                    output_text = ""
+                    for item in s2_resp.get("output", []):
+                        for part in item.get("content", []):
+                            output_text += part.get("text", "") or part.get("content", "")
+                    print(f"\n>>> Turn 2 answer: {output_text.strip()}")
+                break
+        else:
+            print("Turn 2 did not complete within the timeout.")
+
+
+if __name__ == "__main__":
+    base_url = sys.argv[1] if len(sys.argv) > 1 else _DEFAULT_BASE
+    asyncio.run(demo(base_url))

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

@@ -0,0 +1,42 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+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 environment variables from .env file
+load_dotenv()
+
+
+def main():
+    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 friendly assistant. Keep your answers brief.",
+        default_options={"store": False},
+    )
+
+    # steerable_conversations=True allows a client to send a new turn while
+    # the current one is still in progress. The new turn is queued and the
+    # running handler is cooperatively cancelled (via cancellation_signal)
+    # rather than the client receiving HTTP 409 conversation_locked.
+    # Once the current turn reaches a terminal event the queued turn runs.
+    server = ResponsesHostServer(
+        agent,
+        options=ResponsesServerOptions(steerable_conversations=True),
+    )
+    server.run()
+
+
+if __name__ == "__main__":
+    main()

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

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

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

@@ -0,0 +1,18 @@
+# This sample makes no external model calls and requires no credentials.
+# The variables below are optional tuning knobs for main.py (both have
+# sensible defaults); demo.py sets WORKFLOW_STATE_DIR and
+# WORKFLOW_STAGE_DELAY_SECONDS itself for its isolated crash-recovery run,
+# overriding whatever is set here.
+
+# Directory for audit.jsonl and stage marker files.
+# Default: a .workflow_state folder next to main.py.
+# WORKFLOW_STATE_DIR="./.workflow_state"
+
+# Seconds each pipeline stage sleeps to simulate work.
+# Default: 4
+# WORKFLOW_STAGE_DELAY_SECONDS="4"
+
+# The sample disables Azure Monitor's internal Statsbeat metrics when no
+# Application Insights connection is configured. This avoids a local probe of
+# the Azure instance metadata endpoint without disabling normal telemetry.
+# APPLICATIONINSIGHTS_STATSBEAT_DISABLED_ALL="true"

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

@@ -0,0 +1,199 @@
+# Durable workflow recovery
+
+This sample shows a long-running background request continuing after its host
+process is interrupted and replaced.
+
+The request runs through a deterministic local pipeline:
+
+```text
+ingest -> transform -> validate -> finalize
+```
+
+No model deployment, external service, credentials, or environment variables
+are required.
+
+## Private preview wheel setup
+
+The private AgentServer preview versions are newer than the packages currently
+available from PyPI. Running the repository workspace sync can still replace
+them with the public versions recorded in the workspace lock.
+
+Create an isolated environment and install the local wheel files explicitly.
+The Agent Framework wheel directory must contain
+`agent_framework_foundry_hosting-*.whl`. The AgentServer wheel directory must
+contain the `core`, `invocations`, and `responses` wheels. The released
+`agent-framework-core` package and any other Agent Framework packages are
+resolved normally from the configured package index.
+
+PowerShell:
+
+```powershell
+$afWheels = "C:\path\to\agent-framework-wheels"
+$agentServerWheels = "C:\path\to\agent-server-wheels"
+
+uv venv .venv-preview
+uv pip install --python .venv-preview\Scripts\python.exe `
+    (Get-ChildItem "$afWheels\agent_framework_foundry_hosting-*.whl").FullName `
+    (Get-ChildItem "$agentServerWheels\azure_ai_agentserver_core-*.whl").FullName `
+    (Get-ChildItem "$agentServerWheels\azure_ai_agentserver_invocations-*.whl").FullName `
+    (Get-ChildItem "$agentServerWheels\azure_ai_agentserver_responses-*.whl").FullName `
+    httpx
+```
+
+macOS/Linux:
+
+```bash
+export AF_WHEEL_DIR=/path/to/agent-framework-wheels
+export AGENTSERVER_WHEEL_DIR=/path/to/agent-server-wheels
+
+uv venv .venv-preview
+uv pip install --python .venv-preview/bin/python \
+    "$AF_WHEEL_DIR"/agent_framework_foundry_hosting-*.whl \
+    "$AGENTSERVER_WHEEL_DIR"/azure_ai_agentserver_core-*.whl \
+    "$AGENTSERVER_WHEEL_DIR"/azure_ai_agentserver_invocations-*.whl \
+    "$AGENTSERVER_WHEEL_DIR"/azure_ai_agentserver_responses-*.whl \
+    httpx
+```
+
+Verify that the installed AgentServer build exposes the preview API:
+
+```powershell
+.venv-preview\Scripts\python.exe -c "from azure.ai.agentserver.responses import ResponsesServerOptions; assert ResponsesServerOptions(resilient_background=True).resilient_background"
+```
+
+```bash
+.venv-preview/bin/python -c "from azure.ai.agentserver.responses import ResponsesServerOptions; assert ResponsesServerOptions(resilient_background=True).resilient_background"
+```
+
+Do not run `uv sync` in this environment. When using the repository workspace
+environment instead, reinstall the three local AgentServer wheels after every
+sync and run commands with `uv run --no-sync`.
+
+## Run the demo
+
+From the repository's `python` directory, using the isolated environment above:
+
+```powershell
+.venv-preview\Scripts\python.exe .\samples\04-hosting\foundry-hosted-agents\responses\15_workflow_resilience\demo.py
+```
+
+```bash
+.venv-preview/bin/python ./samples/04-hosting/foundry-hosted-agents/responses/15_workflow_resilience/demo.py
+```
+
+Run `demo.py`, not `main.py`. `main.py` is the hosted agent server and waits for
+API requests; `demo.py` starts that server and drives the complete scenario.
+
+## What the demo proves
+
+The demo:
+
+1. Starts one background workflow request.
+2. Stops the host while `transform` is running.
+3. Starts a replacement host, which recovers the existing request.
+4. Stops the replacement host while `finalize` is running.
+5. Starts another replacement host and waits for the original request to
+   complete.
+6. Verifies that completed stages, intermediate progress output, and the final
+   response output survived both interruptions.
+
+The console shows only the customer-visible recovery story. Detailed server and
+telemetry output is captured in an isolated diagnostic log and is printed only
+if the demo fails.
+
+Expected output:
+
+```text
+Durable workflow recovery demo
+==============================
+Pipeline: ingest -> transform -> validate -> finalize
+The host will be interrupted twice while one request is running.
+
+[START] Starting the first host instance...
+[OK] Host is ready.
+
+[REQUEST] Starting one background workflow request...
+[OK] Request accepted: caresp_...
+[OK] Initial status: in_progress
+
+[WORKFLOW] Waiting for the first stage to complete...
+[OK] 'ingest' completed and its progress was saved.
+[INTERRUPT] Stopping the host while 'transform' is running...
+[INTERRUPT] Host stopped unexpectedly.
+
+[RECOVER] Starting a replacement host...
+[OK] The existing request was recovered automatically.
+
+[WORKFLOW] Waiting for two more stages to complete...
+[OK] 'transform' and 'validate' completed; saved progress was preserved.
+[INTERRUPT] Stopping the host while 'finalize' is running...
+[INTERRUPT] Host stopped unexpectedly.
+
+[RECOVER] Starting another replacement host...
+[OK] The same request was recovered again.
+
+[REQUEST] Waiting for the original request to finish...
+
+Result
+------
+[OK] The original request completed after two host interruptions.
+[OK] Progress output from every completed stage was preserved.
+[OK] Final output: Pipeline complete for 'run the resilient pipeline'. Stages executed: ingest, transform, validate, finalize.
+[OK] Every completed stage ran exactly once:
+     ingest: 1
+     transform: 1
+     validate: 1
+     finalize: 1
+
+PASS: Durable progress and response output survived both interruptions.
+```
+
+## Application setup
+
+The durability configuration in `main.py` is intentionally small:
+
+```python
+server = ResponsesHostServer(
+    agent,
+    options=ResponsesServerOptions(resilient_background=True),
+)
+```
+
+The hosting layer manages durable workflow progress and recovers stored
+background requests when a replacement host starts. The application does not
+configure workflow checkpoint storage itself.
+
+## Idempotency
+
+An interrupted stage may restart from the beginning if its latest work had not
+yet been saved. External side effects such as charging a payment, sending an
+email, or writing to another system must use an idempotency key, upsert, or
+equivalent duplicate-safe design.
+
+The demo uses an append-only audit ledger to detect repeated completed work and
+asserts that each completed stage appears exactly once.
+
+## Manual server runs
+
+To start only the hosted agent server:
+
+```powershell
+.venv-preview\Scripts\python.exe .\samples\04-hosting\foundry-hosted-agents\responses\15_workflow_resilience\main.py
+```
+
+```bash
+.venv-preview/bin/python ./samples/04-hosting/foundry-hosted-agents/responses/15_workflow_resilience/main.py
+```
+
+The server then waits for Responses API requests on `http://localhost:8088`.
+Use `Ctrl+C` to stop it.
+
+`.env.example` documents optional settings for the workflow state directory,
+stage delay, and local Azure Monitor Statsbeat behavior.
+
+## Deploying to Foundry
+
+Follow the parent sample's
+[Foundry deployment instructions](../../README.md#deploying-the-agent-to-foundry).
+In Foundry, host replacements exercise the same durable request recovery
+behavior demonstrated locally.

python/samples/04-hosting/foundry-hosted-agents/responses/15_workflow_resilience/demo.py

@@ -0,0 +1,273 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Demonstrate that a long-running workflow survives host interruptions.
+
+The demo submits one background request to a four-stage pipeline, stops the
+host twice while work is in progress, and starts a replacement host each time.
+It then verifies that:
+
+- the original request still completes,
+- completed stages are not repeated,
+- interrupted work continues automatically, and
+- intermediate progress and final response output are preserved.
+
+Server and telemetry output is written to an isolated diagnostic log instead
+of the console. The temporary state is removed after a successful run.
+
+Usage:
+    .venv-preview/Scripts/python.exe demo.py  # Windows
+    .venv-preview/bin/python demo.py          # macOS/Linux
+
+No external services, model deployments, or credentials are required. The
+private preview Agent Framework and AgentServer wheels must be installed as
+described in README.md.
+"""
+
+import asyncio
+import inspect
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+_BASE = "http://localhost:8088"
+_MAIN = os.path.join(os.path.dirname(os.path.abspath(__file__)), "main.py")
+_SERVER_LOG = "server.log"
+
+_STAGES = ("ingest", "transform", "validate", "finalize")
+
+# Allow the host to durably record a completed stage before interrupting the
+# following stage.
+_KILL_GRACE_SECONDS = 2.0
+
+
+def _require_private_preview() -> None:
+    """Fail before starting a child process when the public AgentServer build is installed."""
+    from azure.ai.agentserver.responses import ResponsesServerOptions
+
+    if "resilient_background" not in inspect.signature(ResponsesServerOptions).parameters:
+        raise RuntimeError(
+            "This interpreter does not contain the private AgentServer durability preview: "
+            f"{sys.executable}. Follow the README.md private preview wheel setup, then run "
+            "the demo with .venv-preview's Python directly. Do not use `uv run` without `--no-sync`."
+        )
+
+
+async def _wait_for_server(base_url: str, timeout: float = 30) -> bool:
+    """Poll /readiness until the server responds or the timeout expires."""
+    deadline = time.monotonic() + timeout
+    async with httpx.AsyncClient() as probe:
+        while time.monotonic() < deadline:
+            try:
+                r = await probe.get(f"{base_url}/readiness", timeout=1.0)
+                if r.status_code == 200:
+                    return True
+            except Exception:  # noqa: BLE001
+                pass
+            await asyncio.sleep(0.3)
+    return False
+
+
+async def _wait_for_marker(markers_dir: Path, stage: str, timeout: float = 60) -> None:
+    """Poll the filesystem until <stage>.json appears under markers_dir."""
+    marker = markers_dir / f"{stage}.json"
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if marker.exists():
+            return
+        await asyncio.sleep(0.2)
+    raise TimeoutError(f"Timed out waiting for stage marker: {marker}")
+
+
+def _start_server(state_dir: Path, stage_delay_seconds: float) -> subprocess.Popen[bytes]:
+    """Start main.py with isolated durable state and captured diagnostics."""
+    env = dict(os.environ)
+    env["HOME"] = str(state_dir)
+    env["USERPROFILE"] = str(state_dir)
+    env["WORKFLOW_STATE_DIR"] = str(state_dir)
+    env["WORKFLOW_STAGE_DELAY_SECONDS"] = str(stage_delay_seconds)
+    with (state_dir / _SERVER_LOG).open("ab") as server_log:
+        server_log.write(b"\n--- starting host instance ---\n")
+        server_log.flush()
+        return subprocess.Popen(
+            [sys.executable, _MAIN],
+            cwd=str(state_dir),
+            env=env,
+            stdout=server_log,
+            stderr=subprocess.STDOUT,
+        )
+
+
+def _interrupt(server: subprocess.Popen[bytes], running_stage: str) -> None:
+    print(f"[INTERRUPT] Stopping the host while '{running_stage}' is running...")
+    server.terminate()
+    try:
+        server.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        server.kill()
+        server.wait(timeout=10)
+    print("[INTERRUPT] Host stopped unexpectedly.\n")
+
+
+def _print_log_tail(state_dir: Path, line_count: int = 40) -> None:
+    log_path = state_dir / _SERVER_LOG
+    if not log_path.exists():
+        return
+    lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines()
+    print("\nLast server diagnostic lines:")
+    for line in lines[-line_count:]:
+        print(f"  {line}")
+
+
+def _read_audit(state_dir: Path) -> list[dict[str, Any]]:
+    audit_path = state_dir / "audit.jsonl"
+    if not audit_path.exists():
+        return []
+    with audit_path.open(encoding="utf-8") as f:
+        return [json.loads(line) for line in f if line.strip()]
+
+
+async def demo() -> None:
+    """Run the customer-facing durability demonstration."""
+    _require_private_preview()
+    state_dir = Path(tempfile.mkdtemp(prefix="workflow_resilience_demo_"))
+    markers_dir = state_dir / "markers"
+    stage_delay_seconds = 4.0
+    succeeded = False
+    print("Durable workflow recovery demo")
+    print("==============================")
+    print("Pipeline: ingest -> transform -> validate -> finalize")
+    print("The host will be interrupted twice while one request is running.\n")
+
+    server: subprocess.Popen[bytes] | None = None
+    try:
+        print("[START] Starting the first host instance...")
+        server = _start_server(state_dir, stage_delay_seconds)
+        if not await _wait_for_server(_BASE):
+            _interrupt(server, "startup")
+            raise RuntimeError("Server did not start within 30 seconds.")
+        print("[OK] Host is ready.\n")
+
+        async with httpx.AsyncClient(base_url=_BASE, timeout=60) as client:
+            print("[REQUEST] Starting one background workflow request...")
+            r = await client.post(
+                "/responses",
+                json={
+                    "model": "n/a",  # ignored: WorkflowAgent doesn't use runtime model options
+                    "input": "run the resilient pipeline",
+                    "background": True,
+                    "store": True,
+                },
+            )
+            assert r.status_code == 200, f"POST failed: {r.status_code} {r.text}"
+            body: dict[str, Any] = r.json()
+            resp_id: str = body["id"]
+            print(f"[OK] Request accepted: {resp_id}")
+            print(f"[OK] Initial status: {body['status']}\n")
+
+        print("[WORKFLOW] Waiting for the first stage to complete...")
+        await _wait_for_marker(markers_dir, "ingest")
+        print("[OK] 'ingest' completed and its progress was saved.")
+        await asyncio.sleep(_KILL_GRACE_SECONDS)
+        _interrupt(server, "transform")
+
+        print("[RECOVER] Starting a replacement host...")
+        server = _start_server(state_dir, stage_delay_seconds)
+        if not await _wait_for_server(_BASE):
+            _interrupt(server, "recovery startup")
+            raise RuntimeError("Replacement host did not start within 30 seconds.")
+        print("[OK] The existing request was recovered automatically.\n")
+
+        print("[WORKFLOW] Waiting for two more stages to complete...")
+        await _wait_for_marker(markers_dir, "validate")
+        print("[OK] 'transform' and 'validate' completed; saved progress was preserved.")
+        await asyncio.sleep(_KILL_GRACE_SECONDS)
+        _interrupt(server, "finalize")
+
+        print("[RECOVER] Starting another replacement host...")
+        server = _start_server(state_dir, stage_delay_seconds)
+        if not await _wait_for_server(_BASE):
+            _interrupt(server, "recovery startup")
+            raise RuntimeError("Replacement host did not start within 30 seconds.")
+        print("[OK] The same request was recovered again.\n")
+
+        print("[REQUEST] Waiting for the original request to finish...")
+        async with httpx.AsyncClient(base_url=_BASE, timeout=60) as client:
+            status = "unknown"
+            for _ in range(150):
+                await asyncio.sleep(0.5)
+                poll = (await client.get(f"/responses/{resp_id}")).json()
+                status = poll.get("status", "unknown")
+                if status in ("completed", "failed", "cancelled"):
+                    break
+            assert status == "completed", f"Response did not complete: status={status!r}"
+
+            output_texts = [
+                "".join(part.get("text", "") or part.get("content", "") for part in item.get("content", []))
+                for item in poll.get("output", [])
+                if item.get("type") == "message"
+            ]
+            expected_outputs = [
+                "[ingest] received request: 'run the resilient pipeline'",
+                "[transform] normalized request: 'run the resilient pipeline'",
+                "[validate] validated request: 'run the resilient pipeline'",
+                (
+                    "Pipeline complete for 'run the resilient pipeline'. "
+                    "Stages executed: ingest, transform, validate, finalize."
+                ),
+            ]
+            assert output_texts == expected_outputs, (
+                f"Recovered response did not preserve the expected workflow outputs: {output_texts!r}"
+            )
+            final_output = output_texts[-1]
+
+        for stage in _STAGES:
+            marker = markers_dir / f"{stage}.json"
+            assert marker.exists(), f"Missing stage marker: {marker}"
+
+        audit = _read_audit(state_dir)
+        counts = {stage: sum(1 for entry in audit if entry["stage"] == stage) for stage in _STAGES}
+        for stage, count in counts.items():
+            assert count == 1, (
+                f"Stage {stage!r} appears {count} times in audit.jsonl; expected exactly once. "
+                "A count > 1 means a completed stage replayed; a count of 0 means it never ran."
+            )
+
+        print("\nResult")
+        print("------")
+        print("[OK] The original request completed after two host interruptions.")
+        print("[OK] Progress output from every completed stage was preserved.")
+        print(f"[OK] Final output: {final_output}")
+        print("[OK] Every completed stage ran exactly once:")
+        for stage in _STAGES:
+            print(f"     {stage}: {counts[stage]}")
+        print("\nPASS: Durable progress and response output survived both interruptions.")
+        succeeded = True
+
+    except Exception:
+        print("\nFAIL: The durability demonstration did not complete.")
+        _print_log_tail(state_dir)
+        raise
+
+    finally:
+        if server is not None and server.poll() is None:
+            server.terminate()
+            try:
+                server.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                server.kill()
+        if succeeded:
+            shutil.rmtree(state_dir, ignore_errors=True)
+        else:
+            print(f"\nDiagnostic state retained at: {state_dir}")
+
+
+if __name__ == "__main__":
+    asyncio.run(demo())

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

@@ -0,0 +1,161 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""A deterministic workflow used to demonstrate durable background requests.
+
+The local pipeline has four stages:
+
+``ingest -> transform -> validate -> finalize``
+
+With ``resilient_background=True``, the hosting layer durably records workflow
+progress. If the host stops while a request is running, a replacement host
+continues the same request from its latest saved progress rather than starting
+the entire workflow again.
+
+``demo.py`` interrupts the host twice and verifies that completed stages are
+preserved, interrupted work continues, and the original response completes
+with the expected output. No model deployment, external service, or credential
+is required.
+
+Production note: a stage interrupted before its progress is saved may run
+again. External side effects should therefore use idempotency keys or upsert
+semantics.
+"""
+
+import asyncio
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from dotenv import load_dotenv
+
+# Load configuration before importing the hosting stack, whose telemetry distro
+# installs its default resource-detector list during import.
+load_dotenv()
+
+# Azure Monitor Statsbeat probes the Azure instance metadata endpoint after a
+# short warmup. Disable only those internal SDK metrics when no Application
+# Insights connection is configured, avoiding a noisy local-only connection
+# error without disabling the sample's normal traces and metrics.
+if not os.environ.get("APPLICATIONINSIGHTS_CONNECTION_STRING"):
+    os.environ.setdefault("APPLICATIONINSIGHTS_STATSBEAT_DISABLED_ALL", "true")
+
+from agent_framework import Message, WorkflowBuilder, WorkflowContext, executor  # noqa: E402
+from agent_framework_foundry_hosting import ResponsesHostServer  # noqa: E402
+from azure.ai.agentserver.responses import ResponsesServerOptions  # noqa: E402
+from typing_extensions import Never  # noqa: E402
+
+# Keep durable workflow state in built-in, serializable types.
+PipelineState = dict[str, Any]
+
+
+def _state_dir() -> Path:
+    """Directory where audit.jsonl and stage markers are written.
+
+    Defaults to a folder next to this file for ad-hoc manual runs, but
+    ``demo.py`` always overrides ``WORKFLOW_STATE_DIR`` to an isolated
+    temporary directory so demo runs never touch real user state.
+    """
+    raw = os.environ.get("WORKFLOW_STATE_DIR")
+    path = Path(raw) if raw else Path(__file__).parent / ".workflow_state"
+    (path / "markers").mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def _stage_delay_seconds() -> float:
+    """How long each stage sleeps to simulate work. See .env.example."""
+    return float(os.environ.get("WORKFLOW_STAGE_DELAY_SECONDS", "4"))
+
+
+async def _run_stage(stage_id: str) -> None:
+    """Simulate work for *stage_id*, then durably record that it ran.
+
+    The delay happens before the audit entry so the demo can interrupt a stage
+    while it is still in progress. Only completed stage attempts are recorded.
+    """
+    await asyncio.sleep(_stage_delay_seconds())
+
+    state_dir = _state_dir()
+    pid = os.getpid()
+    timestamp = datetime.now(timezone.utc).isoformat()
+    record = {"stage": stage_id, "pid": pid, "timestamp": timestamp}
+
+    # Append-only so the demo can detect if completed work was repeated.
+    with (state_dir / "audit.jsonl").open("a", encoding="utf-8") as f:
+        f.write(json.dumps(record) + "\n")
+
+    # The marker lets demo.py reliably interrupt the following stage.
+    (state_dir / "markers" / f"{stage_id}.json").write_text(json.dumps(record), encoding="utf-8")
+
+    # Visible progress marker for anyone watching the server's stdout.
+    print(f"[{stage_id}] complete (pid={pid})", flush=True)
+
+
+@executor(id="ingest")
+async def ingest(messages: list[Message], ctx: WorkflowContext[PipelineState, str]) -> None:
+    """Stage 1: accept the incoming request and start the pipeline state."""
+    await _run_stage("ingest")
+    prompt = messages[0].text if messages else ""
+    state: PipelineState = {"request_summary": prompt[:200], "stages_completed": ["ingest"]}
+    await ctx.yield_output(f"[ingest] received request: {state['request_summary']!r}")
+    await ctx.send_message(state)
+
+
+@executor(id="transform")
+async def transform(state: PipelineState, ctx: WorkflowContext[PipelineState, str]) -> None:
+    """Stage 2: pretend to normalize/transform the request."""
+    await _run_stage("transform")
+    state["stages_completed"].append("transform")
+    await ctx.yield_output(f"[transform] normalized request: {state['request_summary']!r}")
+    await ctx.send_message(state)
+
+
+@executor(id="validate")
+async def validate(state: PipelineState, ctx: WorkflowContext[PipelineState, str]) -> None:
+    """Stage 3: pretend to validate the transformed request."""
+    await _run_stage("validate")
+    state["stages_completed"].append("validate")
+    await ctx.yield_output(f"[validate] validated request: {state['request_summary']!r}")
+    await ctx.send_message(state)
+
+
+@executor(id="finalize")
+async def finalize(state: PipelineState, ctx: WorkflowContext[Never, str]) -> None:
+    """Stage 4 (terminal): emit the single Workflow Output summary."""
+    await _run_stage("finalize")
+    state["stages_completed"].append("finalize")
+    await ctx.yield_output(
+        f"Pipeline complete for {state['request_summary']!r}. Stages executed: {', '.join(state['stages_completed'])}."
+    )
+
+
+def build_workflow():
+    """Build the 4-stage deterministic pipeline with explicit output selection."""
+    return (
+        WorkflowBuilder(
+            start_executor=ingest,
+            # Only finalize's yield is the terminal Workflow Output.
+            output_from=[finalize],
+            # ingest/transform/validate yields are visible progress notes.
+            intermediate_output_from=[ingest, transform, validate],
+        )
+        .add_chain([ingest, transform, validate, finalize])
+        .build()
+    )
+
+
+def main() -> None:
+    agent = build_workflow().as_agent(name="resilient-pipeline")
+
+    # Durable background execution is host-managed; no workflow checkpoint
+    # storage needs to be configured by the application.
+    server = ResponsesHostServer(
+        agent,
+        options=ResponsesServerOptions(resilient_background=True),
+    )
+    server.run()
+
+
+if __name__ == "__main__":
+    main()

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

@@ -0,0 +1,11 @@
+# This sample makes no external model calls, so agent-framework-foundry
+# (FoundryChatClient) is not needed -- only the hosting layer is.
+# For the private durability preview, follow README.md and install the local
+# Agent Framework and AgentServer wheel files explicitly instead of using this
+# package-name-only requirements file.
+agent-framework-foundry-hosting
+
+# demo.py's HTTP client. (Previously pulled in transitively via
+# agent-framework-foundry -> agent-framework-openai -> openai; declared
+# explicitly here since that transitive path no longer exists.)
+httpx

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/.gitignore

@@ -0,0 +1 @@
+.durability-response.json

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

@@ -0,0 +1,145 @@
+# Resilient translation workflow
+
+This sample adapts the Foundry three-agent translation workflow into a durable
+background request:
+
+```text
+English -> French -> Spanish -> English
+```
+
+Each stage intentionally terminates its host once, then calls a real model
+through `FoundryChatClient` after recovery. Before exiting, the stage atomically
+writes and flushes a marker in the session's persistent home directory. The
+replacement host sees that marker and continues instead of crashing again.
+After the recovered stage completes, it removes its marker so a later workflow
+request demonstrates the same three crashes again.
+`ResponsesServerOptions(resilient_background=True)` keeps the original
+background response and workflow checkpoints alive across all three restarts.
+
+`azure.yaml` defines the `azd` project and provisioned services. The source
+directory's `agent.yaml` is an Agent Manifest with the required `template`
+field. It is intentionally not a ContainerAgent-schema document; current `azd`
+validates `agent.yaml` against the Agent Manifest schema.
+
+## Prerequisites
+
+1. Install `azd` and the Foundry agents extension declared by `azure.yaml`:
+
+   ```powershell
+   azd extension install azure.ai.agents
+   azd extension upgrade azure.ai.agents
+   azd auth login
+   az login
+   ```
+
+2. Obtain these private preview wheels:
+
+   - `agent_framework_foundry_hosting-*.whl`
+   - `azure_ai_agentserver_core-*.whl`
+   - `azure_ai_agentserver_invocations-*.whl`
+   - `azure_ai_agentserver_responses-*.whl`
+
+The AgentServer preview versions are newer than the packages currently
+available from PyPI, so the wheels must be installed by file.
+
+## Prepare the deployment
+
+From this directory:
+
+```powershell
+python .\prepare_wheels.py `
+  --agent-framework-wheels C:\path\to\agent-framework-wheels `
+  --agent-server-wheels C:\path\to\agent-server-wheels
+```
+
+The script requires exactly one matching wheel for each package, copies the four
+files into the container build context, and generates
+`wheelhouse/private-wheels.txt` with their actual filenames. The generated files
+are intentionally gitignored, but `.agentignore` explicitly includes them in
+the Foundry code deployment ZIP. `requirements.txt` includes the generated
+requirements fragment, so code deployment installs the unpublished preview
+versions without depending on one build's wheel filenames.
+
+Initialize or select an `azd` environment, then provision:
+
+```powershell
+azd env new
+azd provision
+```
+
+If you already have a Foundry project and model deployment, use the normal
+`azd ai agent init` or environment configuration flow to target them.
+
+## Run locally
+
+```powershell
+azd ai agent run
+```
+
+In another terminal:
+
+```powershell
+uv run .\durability_client.py --endpoint http://localhost:8088/responses
+```
+
+With `WORKFLOW_CRASH_ONCE_PER_STAGE=true`, the local host exits at the start of
+each stage. Restart `azd ai agent run` after each exit while the client keeps
+polling. Set the variable to `false` to run without injected crashes.
+
+## Deploy and prove recovery
+
+Deploy the container:
+
+```powershell
+azd deploy
+azd ai agent show --output json
+```
+
+Create a hosted session with persistent filesystem state:
+
+```powershell
+azd ai agent sessions create --output json
+```
+
+Copy `agent_session_id` from the result and the Responses endpoint from
+`azd ai agent show`, then start one stored background response bound to that
+session:
+
+```powershell
+uv run .\durability_client.py `
+  --endpoint "https://<account>.services.ai.azure.com/api/projects/<project>/agents/resilient-translation-workflow/endpoint/protocols/openai/responses?api-version=v1" `
+  --session-id "<agent-session-id>"
+```
+
+The first attempt at each translation stage persists a marker and exits with
+code 70. Foundry automatically starts a replacement container with the same
+session ID and persistent home directory. The client continues polling the
+original response through all three replacements. The container logs show one
+intentional termination and one recovered continuation for each stage. Each
+marker is removed after its recovered stage succeeds, so another request in the
+same session repeats the demonstration.
+
+A successful run ends with the persisted French, Spanish, and round-trip
+English output followed by:
+
+```text
+PASS: The original response completed.
+```
+
+The response ID and endpoint are also saved in `.durability-response.json`. If
+the client itself is interrupted, resume polling with:
+
+```powershell
+uv run .\durability_client.py `
+  --endpoint "<endpoint>" `
+  --session-id "<agent-session-id>" `
+  --response-id "<response-id>"
+```
+
+## Durability boundary
+
+Workflow progress is saved between executor supersteps. If a host stops during
+a model call, that current stage may run again; completed prior stages are
+restored. Model translation is side-effect free. Production executors that
+write to external systems must use idempotency keys, upserts, or an equivalent
+duplicate-safe design.

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/azure.yaml

@@ -0,0 +1,52 @@
+# yaml-language-server: $schema=https://raw.githubusercontent.com/Azure/azure-dev/main/schemas/v1.0/azure.yaml.json
+
+requiredVersions:
+  extensions:
+    azure.ai.agents: '>=1.0.0-beta.4'
+name: resilient-translation-workflow
+services:
+  ai-project:
+    host: azure.ai.project
+    deployments:
+      - name: gpt-5.4-mini
+        model:
+          format: OpenAI
+          name: gpt-5.4-mini
+          version: '2026-03-17'
+        sku:
+          name: GlobalStandard
+          capacity: 10
+  resilient-translation-workflow:
+    host: azure.ai.agent
+    metadata:
+      tags:
+        - Agent Framework
+        - AI Agent Hosting
+        - Azure AI AgentServer
+        - Responses Protocol
+        - Durable Workflows
+    project: src/resilient-translation-workflow
+    language: python
+    codeConfiguration:
+      runtime: python_3_13
+      entryPoint: main.py
+    uses:
+      - ai-project
+    kind: hosted
+    name: resilient-translation-workflow
+    displayName: Resilient Translation Workflow
+    description: A durable English to French to Spanish to English workflow.
+    protocols:
+      - protocol: responses
+        version: 2.0.0
+    environmentVariables:
+      - name: AZURE_AI_MODEL_DEPLOYMENT_NAME
+        value: ${AZURE_AI_MODEL_DEPLOYMENT_NAME}
+      - name: WORKFLOW_CRASH_ONCE_PER_STAGE
+        value: "true"
+    container:
+      resources:
+        cpu: '0.5'
+        memory: 1Gi
+infra:
+  provider: microsoft.foundry

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/durability_client.py

@@ -0,0 +1,161 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Submit and monitor one durable background response across host replacements."""
+
+# /// script
+# requires-python = ">=3.10"
+# dependencies = [
+#   "azure-identity>=1.25.2",
+#   "httpx>=0.28.1",
+# ]
+# ///
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+from pathlib import Path
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+import httpx
+from azure.identity.aio import AzureCliCredential
+
+TERMINAL_STATUSES = {"completed", "failed", "cancelled", "incomplete"}
+
+
+def _is_local_endpoint(endpoint: str) -> bool:
+    return urlsplit(endpoint).hostname in {"localhost", "127.0.0.1"}
+
+
+def _response_url(endpoint: str, response_id: str | None = None, session_id: str | None = None) -> str:
+    parsed = urlsplit(endpoint)
+    path = parsed.path.rstrip("/")
+    if response_id:
+        path = f"{path}/{response_id}"
+    query = dict(parse_qsl(parsed.query, keep_blank_values=True))
+    if session_id:
+        query["agent_session_id"] = session_id
+    return urlunsplit((parsed.scheme, parsed.netloc, path, urlencode(query), parsed.fragment))
+
+
+def _output_text(response: dict[str, object]) -> str:
+    parts: list[str] = []
+    output = response.get("output")
+    if not isinstance(output, list):
+        return ""
+    for item in output:
+        if not isinstance(item, dict):
+            continue
+        content = item.get("content")
+        if not isinstance(content, list):
+            continue
+        for part in content:
+            if isinstance(part, dict) and isinstance(part.get("text"), str):
+                parts.append(part["text"])
+    return "\n".join(parts)
+
+
+async def _authorization_headers(endpoint: str) -> dict[str, str]:
+    if _is_local_endpoint(endpoint):
+        return {}
+    credential = AzureCliCredential()
+    try:
+        token = await credential.get_token("https://ai.azure.com/.default")
+        return {"Authorization": f"Bearer {token.token}"}
+    finally:
+        await credential.close()
+
+
+async def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--endpoint", required=True, help="Responses endpoint printed by `azd ai agent show`.")
+    parser.add_argument(
+        "--session-id",
+        help="Hosted session created by `azd ai agent sessions create`; required for remote endpoints.",
+    )
+    parser.add_argument("--response-id", help="Resume monitoring an existing response instead of creating one.")
+    parser.add_argument(
+        "--input",
+        default="The quick brown fox jumps over the lazy dog.",
+        help="English text to send when creating a response.",
+    )
+    parser.add_argument("--poll-seconds", type=float, default=5)
+    parser.add_argument("--state-file", type=Path, default=Path(".durability-response.json"))
+    args = parser.parse_args()
+
+    if not _is_local_endpoint(args.endpoint) and not args.session_id:
+        parser.error("--session-id is required for remote endpoints")
+
+    headers = await _authorization_headers(args.endpoint)
+    headers.update(
+        {
+            "Content-Type": "application/json",
+            "x-agent-user-isolation-key": "durability-demo-user",
+            "x-agent-chat-isolation-key": "durability-demo-chat",
+        }
+    )
+    async with httpx.AsyncClient(headers=headers, timeout=60) as client:
+        response_id = args.response_id
+        if not response_id:
+            result = await client.post(
+                _response_url(args.endpoint, session_id=args.session_id),
+                json={
+                    "model": "workflow",
+                    "input": args.input,
+                    "background": True,
+                    "store": True,
+                    "stream": False,
+                },
+            )
+            result.raise_for_status()
+            body = result.json()
+            response_id = body["id"]
+            args.state_file.write_text(
+                json.dumps(
+                    {
+                        "endpoint": args.endpoint,
+                        "session_id": args.session_id,
+                        "response_id": response_id,
+                    },
+                    indent=2,
+                ),
+                encoding="utf-8",
+            )
+            if args.session_id:
+                print(f"Hosted session: {args.session_id}")
+            print(f"Created durable background response: {response_id}")
+            print(f"Recovery state saved to: {args.state_file.resolve()}")
+            print("Redeploy or replace the host while this client continues polling.\n")
+
+        last_status: str | None = None
+        while True:
+            try:
+                result = await client.get(_response_url(args.endpoint, response_id, args.session_id))
+                result.raise_for_status()
+            except (httpx.HTTPError, httpx.TimeoutException) as exc:
+                print(f"Host temporarily unavailable; retrying: {exc}")
+                await asyncio.sleep(args.poll_seconds)
+                continue
+
+            body = result.json()
+            status = body.get("status")
+            if status != last_status:
+                print(f"Response status: {status}")
+                last_status = status if isinstance(status, str) else None
+
+            if status in TERMINAL_STATUSES:
+                text = _output_text(body)
+                if text:
+                    print("\nPersisted response output\n-------------------------")
+                    print(text)
+                if status != "completed":
+                    raise RuntimeError(json.dumps(body, indent=2))
+                print("\nPASS: The original response completed.")
+                return
+
+            await asyncio.sleep(args.poll_seconds)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/prepare_wheels.py

@@ -0,0 +1,75 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Copy the private preview wheels used by the deployment."""
+
+from __future__ import annotations
+
+import argparse
+import shutil
+from pathlib import Path
+
+EXPECTED_WHEELS = (
+    "agent_framework_foundry_hosting-*.whl",
+    "azure_ai_agentserver_core-*.whl",
+    "azure_ai_agentserver_invocations-*.whl",
+    "azure_ai_agentserver_responses-*.whl",
+)
+GENERATED_REQUIREMENTS = "private-wheels.txt"
+
+
+def _find_one(pattern: str, directories: list[Path]) -> Path:
+    matches = [path for directory in directories for path in directory.glob(pattern)]
+    if len(matches) != 1:
+        locations = ", ".join(str(path) for path in directories)
+        raise RuntimeError(f"Expected exactly one {pattern!r} wheel in {locations}; found {len(matches)}.")
+    return matches[0]
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--agent-framework-wheels",
+        type=Path,
+        required=True,
+        help="Directory containing agent_framework_foundry_hosting-*.whl.",
+    )
+    parser.add_argument(
+        "--agent-server-wheels",
+        type=Path,
+        required=True,
+        help="Directory containing the private core, invocations, and responses AgentServer wheels.",
+    )
+    args = parser.parse_args()
+
+    agent_framework_directory = args.agent_framework_wheels.resolve()
+    agent_server_directory = args.agent_server_wheels.resolve()
+    source_directories = [agent_framework_directory, agent_server_directory]
+    for directory in source_directories:
+        if not directory.is_dir():
+            raise RuntimeError(f"Wheel directory does not exist: {directory}")
+
+    sources: list[Path] = []
+    for index, pattern in enumerate(EXPECTED_WHEELS):
+        directories = [agent_framework_directory] if index == 0 else [agent_server_directory]
+        sources.append(_find_one(pattern, directories))
+
+    destination = Path(__file__).parent / "src" / "resilient-translation-workflow" / "wheelhouse"
+    destination.mkdir(parents=True, exist_ok=True)
+    for old_wheel in destination.glob("*.whl"):
+        old_wheel.unlink()
+
+    for source in sources:
+        shutil.copy2(source, destination / source.name)
+        print(f"Copied {source.name}")
+
+    requirements = destination / GENERATED_REQUIREMENTS
+    requirements.write_text(
+        "".join(f"./wheelhouse/{source.name}\n" for source in sources),
+        encoding="utf-8",
+    )
+    print(f"Generated {requirements.name}")
+    print(f"Prepared private deployment wheels in {destination}")
+
+
+if __name__ == "__main__":
+    main()

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/src/resilient-translation-workflow/.agentignore

@@ -0,0 +1,9 @@
+.env
+.venv/
+__pycache__/
+*.py[cod]
+.durability-response.json
+
+!wheelhouse/
+!wheelhouse/*.whl
+!wheelhouse/private-wheels.txt

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/src/resilient-translation-workflow/.azdignore

@@ -0,0 +1,8 @@
+.env
+.venv/
+__pycache__/
+*.py[cod]
+.durability-response.json
+
+!wheelhouse/
+!wheelhouse/*.whl

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/src/resilient-translation-workflow/.dockerignore

@@ -0,0 +1,5 @@
+.env
+.venv/
+__pycache__/
+*.py[cod]
+.durability-response.json

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/src/resilient-translation-workflow/.env.example

@@ -0,0 +1,3 @@
+FOUNDRY_PROJECT_ENDPOINT=https://<account>.services.ai.azure.com/api/projects/<project>
+AZURE_AI_MODEL_DEPLOYMENT_NAME=gpt-5.4-mini
+WORKFLOW_CRASH_ONCE_PER_STAGE=true

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/src/resilient-translation-workflow/Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.13-slim
+
+WORKDIR /app
+
+COPY . user_agent/
+WORKDIR /app/user_agent
+
+# requirements.txt references the private wheels by file so both container and
+# Foundry code deployments install the same artifacts.
+RUN python -m pip install --no-cache-dir -r requirements.txt \
+    && python -c "from azure.ai.agentserver.responses import ResponsesServerOptions; assert ResponsesServerOptions(resilient_background=True).resilient_background"
+
+EXPOSE 8088
+
+CMD ["python", "main.py"]

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/src/resilient-translation-workflow/agent.yaml

@@ -0,0 +1,19 @@
+name: resilient-translation-workflow
+description: >
+  A durable Agent Framework workflow that translates English to French to
+  Spanish and back to English.
+template:
+  name: resilient-translation-workflow
+  kind: hosted
+  protocols:
+    - protocol: responses
+      version: 2.0.0
+  environment_variables:
+    - name: AZURE_AI_MODEL_DEPLOYMENT_NAME
+      value: "{{chat}}"
+    - name: WORKFLOW_CRASH_ONCE_PER_STAGE
+      value: "true"
+resources:
+  - name: chat
+    kind: model
+    id: gpt-5.4

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/src/resilient-translation-workflow/main.py

@@ -0,0 +1,163 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""A model-backed translation workflow with durable background execution."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+from agent_framework import Agent, Message, WorkflowBuilder, WorkflowContext, executor
+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 typing_extensions import Never
+
+TranslationState = dict[str, Any]
+_agents: dict[str, Agent] = {}
+
+
+def _crash_marker(stage: str) -> Path:
+    return Path.home() / ".workflow-resilience-crashes" / f"{stage}.crashed"
+
+
+def _crash_once(stage: str) -> None:
+    """Terminate the host once for this stage within the persistent session home."""
+    if os.getenv("WORKFLOW_CRASH_ONCE_PER_STAGE", "true").lower() not in {"1", "true", "yes"}:
+        return
+
+    marker = _crash_marker(stage)
+    marker.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        descriptor = os.open(marker, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+    except FileExistsError:
+        print(f"[{stage}] crash marker found; continuing recovered stage", flush=True)
+        return
+
+    with os.fdopen(descriptor, "w", encoding="utf-8") as marker_file:
+        marker_file.write(f"Intentional crash for {stage}\n")
+        marker_file.flush()
+        os.fsync(marker_file.fileno())
+
+    print(f"[{stage}] crash marker persisted; terminating host intentionally", flush=True)
+    os._exit(70)
+
+
+def _clear_crash_marker(stage: str) -> None:
+    _crash_marker(stage).unlink(missing_ok=True)
+
+
+async def _translate(stage: str, text: str) -> str:
+    response = await _agents[stage].run(text)
+    return response.text.strip()
+
+
+@executor(id="english-to-french")
+async def english_to_french(
+    messages: list[Message],
+    ctx: WorkflowContext[TranslationState, str],
+) -> None:
+    source = messages[0].text if messages else ""
+    _crash_once("english-to-french")
+    french = await _translate("english-to-french", source)
+    state: TranslationState = {"source": source, "french": french}
+    await ctx.yield_output(f"[French]\n{french}")
+    await ctx.send_message(state)
+    _clear_crash_marker("english-to-french")
+
+
+@executor(id="french-to-spanish")
+async def french_to_spanish(
+    state: TranslationState,
+    ctx: WorkflowContext[TranslationState, str],
+) -> None:
+    _crash_once("french-to-spanish")
+    spanish = await _translate("french-to-spanish", state["french"])
+    state["spanish"] = spanish
+    await ctx.yield_output(f"[Spanish]\n{spanish}")
+    await ctx.send_message(state)
+    _clear_crash_marker("french-to-spanish")
+
+
+@executor(id="spanish-to-english")
+async def spanish_to_english(
+    state: TranslationState,
+    ctx: WorkflowContext[Never, str],
+) -> None:
+    _crash_once("spanish-to-english")
+    english = await _translate("spanish-to-english", state["spanish"])
+    await ctx.yield_output(
+        "\n".join((
+            f"[Original English]\n{state['source']}",
+            f"[French]\n{state['french']}",
+            f"[Spanish]\n{state['spanish']}",
+            f"[Round-trip English]\n{english}",
+        ))
+    )
+    _clear_crash_marker("spanish-to-english")
+
+
+def _create_agent(client: FoundryChatClient, *, name: str, instructions: str) -> Agent:
+    return Agent(client=client, name=name, instructions=instructions)
+
+
+def build_workflow() -> Any:
+    client = FoundryChatClient(
+        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
+        model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
+        credential=DefaultAzureCredential(),
+    )
+    _agents.update({
+        "english-to-french": _create_agent(
+            client,
+            name="english-to-french",
+            instructions=(
+                "Translate the user's English text into French. "
+                "Return only the translation, without explanations or labels."
+            ),
+        ),
+        "french-to-spanish": _create_agent(
+            client,
+            name="french-to-spanish",
+            instructions=(
+                "Translate the user's French text into Spanish. "
+                "Return only the translation, without explanations or labels."
+            ),
+        ),
+        "spanish-to-english": _create_agent(
+            client,
+            name="spanish-to-english",
+            instructions=(
+                "Translate the user's Spanish text into English. "
+                "Return only the translation, without explanations or labels."
+            ),
+        ),
+    })
+
+    return (
+        WorkflowBuilder(
+            start_executor=english_to_french,
+            output_from=[spanish_to_english],
+            intermediate_output_from=[english_to_french, french_to_spanish],
+        )
+        .add_chain([english_to_french, french_to_spanish, spanish_to_english])
+        .build()
+    )
+
+
+def main() -> None:
+    workflow_agent = build_workflow().as_agent(
+        name="resilient-translation-workflow",
+        description="Durable English to French to Spanish to English translation workflow.",
+    )
+    server = ResponsesHostServer(
+        workflow_agent,
+        options=ResponsesServerOptions(resilient_background=True),
+    )
+    server.run()
+
+
+if __name__ == "__main__":
+    main()

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/src/resilient-translation-workflow/requirements.txt

@@ -0,0 +1,2 @@
+-r ./wheelhouse/private-wheels.txt
+agent-framework-foundry>=1.10.1,<2

python/samples/04-hosting/foundry-hosted-agents/responses/16_foundry_workflow_resilience/src/resilient-translation-workflow/wheelhouse/.gitignore

@@ -0,0 +1,3 @@
+*.whl
+private-wheels.txt
+!.gitignore