From 0efc9c136cf768dc4a0152b959c3bab856e49a32 Mon Sep 17 00:00:00 2001 From: Naman Tyagi Date: Fri, 14 Aug 2026 13:28:24 +0530 Subject: [PATCH 1/9] [agentserver] Gate resilient TaskManager on the explicit enablement flag Make the durable-response/task subsystem strictly opt-in via `set_resilient_tasks_enabled` (auto-set by `resilient_background` / `steerable_conversations`). Previously every responses host wrapped all `store=true` responses in internal resilient tasks and therefore forced the boot-time recovery scan, even for a plain host that opted into nothing. Core (_base.py): - Construct the TaskManager (and run startup recovery) ONLY when the switch is enabled. When off, no manager is installed; `get_task_manager()` raises `TaskManagerNotInitialized` and callers degrade to non-durable in-process execution. A plain host pays nothing: no manager, no task-store call. Responses: - ResilientResponseOrchestrator auto-enables the switch when `resilient_background` or `steerable_conversations` is set, so resilient deployments keep full Row 1/2/3 recovery. - _start_resilient_background swallows `TaskManagerNotInitialized` at the single outer catch and runs the handler in-process (non-durable). Removed the hosted fail-loud branch and the now-unused `_is_hosted_environment` helper. Tests/samples: - Updated the opt-in gate tests for gated construction. - Rewrote the resilient-start-failure contract test: no-manager now swallows and runs in-process (real task-store start failures with a manager present still fail loud as platform errors). - Conformance `_test_handler` enables the switch so Row 2 (resilient_background=False) still validates mark-failed-on-crash. - Invocations resilient samples (research, multiturn, langgraph) enable the switch explicitly. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../azure/ai/agentserver/core/_base.py | 91 +++++++-------- .../tests/tasks/test_task_manager_optin.py | 104 +++++++++--------- .../samples/resilient_langgraph/app.py | 8 ++ .../samples/resilient_multiturn/app.py | 9 +- .../samples/resilient_research/app.py | 8 ++ .../responses/hosting/_orchestrator.py | 54 +++------ .../hosting/_resilient_orchestrator.py | 17 +++ .../contract/test_resilient_start_failure.py | 63 +++++------ .../e2e/resilience_contract/_test_handler.py | 7 ++ 9 files changed, 187 insertions(+), 174 deletions(-) diff --git a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py index b2f578865a8e..99267cc26556 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py +++ b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py @@ -311,63 +311,50 @@ async def _lifespan(_app: Starlette) -> AsyncGenerator[None, None]: # noqa: RUF # --- Resilient task manager initialization --- # - # The TaskManager is CONSTRUCTED unconditionally (whenever the - # resilient tasks module is importable). Construction is cheap and - # makes NO task-store calls — it only builds in-memory state (an - # idle provider client, empty routing tables, a lease-owner string). - # This keeps ``get_task_manager()`` working so a ``@task``-based app - # can run tasks without hitting ``TaskManagerNotInitialized``, - # while paying zero network cost until a task is actually used. + # The resilient ``TaskManager`` is CONSTRUCTED ONLY when resilient + # tasks were explicitly enabled via + # ``set_resilient_tasks_enabled(True)``. Recovery (and the durable + # task subsystem as a whole) is strictly opt-in: a host that does + # not set the switch pays nothing — no manager, no provider client, + # no task-store call, no recovery loop. # - # The network-backed startup RECOVERY SCAN (a blocking hosted - # task-store ``list()`` plus ``DefaultAzureCredential`` token - # acquisition, which would otherwise gate server readiness) — and - # the periodic recovery loop it spawns — run when EITHER holds: - # (1) at least one durable task was declared (``@task`` / - # ``@multi_turn_task``, tracked in ``_REGISTERED_DESCRIPTORS``) - # — an app that uses tasks gets recovery automatically, OR - # (2) resilient tasks were explicitly enabled via - # ``set_resilient_tasks_enabled(True)`` — a force-enable that - # starts the recovery loop even before any task is declared, - # so a task declared later is picked up by the loop. - # A plain server that neither declares a task nor sets the switch - # (e.g. an invocations-only host) makes no task-store call at - # startup. - # - # NOTE (deferred): if the switch is OFF and no task is declared at - # startup, the recovery loop is not started, so a task declared - # LATER in that lifetime will run but its prior-crash orphans are - # not scanned until the next restart. Fully closing that requires a - # lazy manager-start on first late registration; tracked as future - # work (the manager cannot be made fully async, only lazily - # started). + # When the switch is OFF the manager is NOT installed, so + # ``get_task_manager()`` raises ``TaskManagerNotInitialized``. + # Callers that route through the manager (e.g. the responses + # protocol's ``store=true`` path) are expected to SWALLOW that + # signal and degrade to non-durable, in-process execution — the + # response still runs and persists, it simply is not crash- + # recoverable. Declaring a durable task (``@task`` / + # ``@multi_turn_task``) does NOT implicitly turn the subsystem on; + # an app that wants durable tasks / recovery must set the switch. task_manager = None - try: - from .tasks._manager import ( # pylint: disable=import-outside-toplevel - TaskManager, - set_task_manager, - ) - - task_manager = TaskManager( - config=cfg, - shutdown_event=asyncio.Event(), - shutdown_grace_seconds=_read_task_manager_shutdown_grace(), - ) - set_task_manager(task_manager) + if _resilient_tasks_enabled(): + try: + from .tasks._manager import ( # pylint: disable=import-outside-toplevel + TaskManager, + set_task_manager, + ) - if _resilient_tasks_enabled() or _has_registered_tasks(): + task_manager = TaskManager( + config=cfg, + shutdown_event=asyncio.Event(), + shutdown_grace_seconds=_read_task_manager_shutdown_grace(), + ) + set_task_manager(task_manager) await task_manager.startup() logger.info("TaskManager initialized with startup recovery") - else: - logger.info( - "TaskManager initialized (recovery deferred; enabled=%s, tasks_declared=%s)", - _resilient_tasks_enabled(), - _has_registered_tasks(), - ) - except ImportError: - pass # resilient module not available - except Exception: # pylint: disable=broad-exception-caught - logger.warning("Failed to initialize TaskManager", exc_info=True) + except ImportError: + pass # resilient module not available + except Exception: # pylint: disable=broad-exception-caught + logger.warning("Failed to initialize TaskManager", exc_info=True) + else: + logger.info( + "TaskManager NOT initialized (resilient tasks disabled; enable via " + "set_resilient_tasks_enabled(True)). Durable tasks and crash recovery " + "are inactive; store=true work degrades to non-durable execution. " + "tasks_declared=%s", + _has_registered_tasks(), + ) yield diff --git a/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py b/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py index 6cff55e7d56f..d5695a32becb 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py +++ b/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py @@ -1,21 +1,20 @@ # --------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # --------------------------------------------------------- -"""Tests for the gated resilient ``TaskManager`` startup recovery scan. - -``AgentServerHost`` always constructs the resilient ``TaskManager`` (a cheap, -in-memory object that makes no task-store calls), so ``get_task_manager()`` and -``.run()`` / ``.start()`` work regardless. Its network-backed **startup -recovery scan** (and the periodic recovery loop it spawns) runs when EITHER: - -1. at least one durable task has been declared (``@task`` / - ``@multi_turn_task``, tracked in the ``_REGISTERED_DESCRIPTORS`` list), OR -2. the switch was set via ``set_resilient_tasks_enabled(True)`` (default - ``False``) — a force-enable. - -Both signals are read directly at lifespan startup. When neither is true, no -task-store call is made — plain servers (e.g. invocations-only hosts) pay -nothing. +"""Tests for the gated resilient ``TaskManager`` construction + recovery. + +The resilient ``TaskManager`` is CONSTRUCTED ONLY when resilient tasks were +explicitly enabled via ``set_resilient_tasks_enabled(True)`` (default +``False``). Recovery — and the durable task subsystem as a whole — is strictly +opt-in. + +When the switch is off the manager is NOT installed, so ``get_task_manager()`` +raises ``TaskManagerNotInitialized``; callers that route through the manager +(e.g. the responses ``store=true`` path) swallow that at their outer catch and +degrade to non-durable in-process execution. Merely declaring a durable task +(``@task`` / ``@multi_turn_task``) — including internal protocol primitives — +does NOT turn the subsystem on. A plain server pays nothing: no manager, no +task-store call. """ import logging @@ -27,6 +26,7 @@ ) from azure.ai.agentserver.core.tasks import ( TaskContext, + TaskManagerNotInitialized, multi_turn_task, resilient_tasks_enabled, set_resilient_tasks_enabled, @@ -136,63 +136,64 @@ async def _probe(ctx: "TaskContext[dict]") -> None: class TestLifespanManagerAndRecovery: - """The TaskManager is ALWAYS constructed (cheap, no task-store calls); - the network-backed startup recovery scan runs when EITHER the switch is - enabled OR at least one durable task is declared.""" + """The resilient TaskManager is CONSTRUCTED ONLY when the switch is + explicitly enabled via ``set_resilient_tasks_enabled(True)``. With the + switch off no manager is installed and ``get_task_manager()`` raises — + callers swallow that and degrade to non-durable execution.""" @pytest.mark.asyncio - async def test_neither_enabled_nor_task_no_recovery( + async def test_neither_enabled_nor_task_no_manager( self, _clean_state, _fake_task_manager ) -> None: - """No switch, no task: the manager is constructed and installed so - ``get_task_manager()`` works (no ``TaskManagerNotInitialized``) — but - the startup recovery scan does NOT run (plain invocations host).""" + """No switch, no task: the manager is NOT constructed and + ``get_task_manager()`` raises ``TaskManagerNotInitialized``.""" from azure.ai.agentserver.core import AgentServerHost from azure.ai.agentserver.core.tasks._manager import get_task_manager app = AgentServerHost() async with app.router.lifespan_context(app): - # A manager exists and is retrievable during the active lifespan. - assert len(_fake_task_manager.instances) == 1 - assert get_task_manager() is _fake_task_manager.instances[0] - # No recovery scan happened (neither gate true). - assert _fake_task_manager.instances[0].startup_called is False - - # Torn down + cleared on shutdown. - assert _fake_task_manager.instances[0].shutdown_called is True + # No manager constructed (switch off). + assert len(_fake_task_manager.instances) == 0 + with pytest.raises(TaskManagerNotInitialized): + get_task_manager() @pytest.mark.asyncio - async def test_task_declared_runs_recovery_without_switch(self, _clean_state, _fake_task_manager) -> None: - """A declared task alone runs recovery (backward compatible — an - existing ``@task`` app gets recovery without calling the switch).""" + async def test_task_declared_without_switch_no_manager(self, _clean_state, _fake_task_manager) -> None: + """A declared task alone does NOT construct the manager: the durable + task subsystem is opt-in and gated solely on the switch.""" from azure.ai.agentserver.core import AgentServerHost + from azure.ai.agentserver.core.tasks._manager import get_task_manager _declare_task() # switch left at default False app = AgentServerHost() async with app.router.lifespan_context(app): - pass - assert len(_fake_task_manager.instances) == 1 - assert _fake_task_manager.instances[0].startup_called is True + assert len(_fake_task_manager.instances) == 0 + with pytest.raises(TaskManagerNotInitialized): + get_task_manager() @pytest.mark.asyncio - async def test_switch_alone_runs_recovery_without_task(self, _clean_state, _fake_task_manager) -> None: - """The switch alone runs recovery (force-enable) — starting the - periodic recovery loop so a task declared later is picked up.""" + async def test_switch_alone_builds_manager_and_runs_recovery(self, _clean_state, _fake_task_manager) -> None: + """The switch constructs the manager and runs recovery (force-enable) + — starting the periodic recovery loop so a task declared later is + picked up.""" from azure.ai.agentserver.core import AgentServerHost + from azure.ai.agentserver.core.tasks._manager import get_task_manager set_resilient_tasks_enabled(True) app = AgentServerHost() async with app.router.lifespan_context(app): - pass - assert len(_fake_task_manager.instances) == 1 - assert _fake_task_manager.instances[0].startup_called is True + assert len(_fake_task_manager.instances) == 1 + assert get_task_manager() is _fake_task_manager.instances[0] + assert _fake_task_manager.instances[0].startup_called is True + assert _fake_task_manager.instances[0].shutdown_called is True @pytest.mark.asyncio - async def test_switch_and_task_runs_recovery( + async def test_switch_and_task_builds_manager_and_runs_recovery( self, _clean_state, _fake_task_manager, caplog: pytest.LogCaptureFixture ) -> None: - """Both signals true -> manager built and startup recovery runs.""" + """Switch on and a task declared -> manager built and startup recovery + runs.""" from azure.ai.agentserver.core import AgentServerHost set_resilient_tasks_enabled(True) @@ -210,9 +211,13 @@ async def test_switch_and_task_runs_recovery( assert any("TaskManager initialized with startup recovery" in r.message for r in caplog.records) @pytest.mark.asyncio - async def test_multi_turn_task_runs_recovery_without_switch(self, _clean_state, _fake_task_manager) -> None: - """A declared ``@multi_turn_task`` alone also runs recovery.""" + async def test_multi_turn_task_declared_without_switch_no_manager( + self, _clean_state, _fake_task_manager + ) -> None: + """A declared ``@multi_turn_task`` alone also does NOT construct the + manager (opt-in).""" from azure.ai.agentserver.core import AgentServerHost + from azure.ai.agentserver.core.tasks._manager import get_task_manager @multi_turn_task(name="gate_lifespan_mt") async def _probe(ctx: "TaskContext[dict]") -> None: @@ -220,7 +225,6 @@ async def _probe(ctx: "TaskContext[dict]") -> None: app = AgentServerHost() async with app.router.lifespan_context(app): - pass - - assert len(_fake_task_manager.instances) == 1 - assert _fake_task_manager.instances[0].startup_called is True + assert len(_fake_task_manager.instances) == 0 + with pytest.raises(TaskManagerNotInitialized): + get_task_manager() diff --git a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_langgraph/app.py b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_langgraph/app.py index ba58ee46bdf9..9b33070c3033 100644 --- a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_langgraph/app.py +++ b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_langgraph/app.py @@ -67,6 +67,7 @@ EventStreamNotFoundError, streams, ) +from azure.ai.agentserver.core.tasks import set_resilient_tasks_enabled from azure.ai.agentserver.invocations import InvocationAgentServerHost try: @@ -83,6 +84,13 @@ app = InvocationAgentServerHost() +# Recovery is opt-in and gated solely on this switch. Declaring a +# ``@multi_turn_task`` makes the framework recovery-capable, but the startup +# recovery scan that reclaims tasks orphaned by a prior crash runs only when it +# is explicitly enabled. Enable it here so a fresh process reclaims in-flight +# sessions at startup. +set_resilient_tasks_enabled(True) + async def _sse_from_stream( stream: EventStream, invocation_id: str, *, initial_status: str = "queued" diff --git a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_multiturn/app.py b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_multiturn/app.py index b61a88ea70c9..e7d764629f98 100644 --- a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_multiturn/app.py +++ b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_multiturn/app.py @@ -37,7 +37,7 @@ from starlette.responses import JSONResponse, Response from azure.ai.agentserver.core.storage import FoundryStateStore -from azure.ai.agentserver.core.tasks import TaskConflictError +from azure.ai.agentserver.core.tasks import TaskConflictError, set_resilient_tasks_enabled from azure.ai.agentserver.invocations import InvocationAgentServerHost try: @@ -47,6 +47,13 @@ app = InvocationAgentServerHost() +# Recovery is opt-in and gated solely on this switch. Declaring a +# ``@multi_turn_task`` makes the framework recovery-capable, but the durable +# task subsystem (and its crash recovery) is only active when explicitly +# enabled. Enable it here so a fresh process reclaims in-flight sessions at +# startup. +set_resilient_tasks_enabled(True) + @app.invoke_handler async def handle_invoke(request: Request) -> Response: diff --git a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_research/app.py b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_research/app.py index 00b64c2813ff..e31057cb44c7 100644 --- a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_research/app.py +++ b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_research/app.py @@ -77,6 +77,7 @@ EventStreamNotFoundError, streams, ) +from azure.ai.agentserver.core.tasks import set_resilient_tasks_enabled from azure.ai.agentserver.invocations import InvocationAgentServerHost try: @@ -101,6 +102,13 @@ app = InvocationAgentServerHost() +# Recovery is opt-in and gated solely on this switch. Declaring a +# ``@multi_turn_task`` makes the framework recovery-capable, but the startup +# recovery scan that reclaims tasks orphaned by a prior crash runs only when it +# is explicitly enabled. Enable it here so a fresh process reclaims in-flight +# sessions at startup. +set_resilient_tasks_enabled(True) + # --- SSE rendering --------------------------------------------------------- diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py index 9e779c8111a6..e8856a4b1fe0 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py @@ -73,23 +73,6 @@ logger = logging.getLogger("azure.ai.agentserver") -def _is_hosted_environment() -> bool: - """Return whether the agent is running in a Foundry-hosted container. - - Uses the canonical :class:`~azure.ai.agentserver.core.AgentConfig` derivation - (the same public API ``_routing`` already uses for Foundry auto-activation). - In a hosted deployment the resilient-task subsystem is auto-initialized with - no opt-out, so a missing TaskManager is a platform-infrastructure failure - rather than a reason to silently run a response non-durably. - - :return: ``True`` if running in a Foundry-hosted environment. - :rtype: bool - """ - from azure.ai.agentserver.core import AgentConfig # pylint: disable=import-outside-toplevel - - return AgentConfig.from_env().is_hosted - - _STORAGE_ERROR_MESSAGE = ( "An internal error occurred while storing the response. " "Subsequent retrieval is not guaranteed. Please retry the request." @@ -4004,27 +3987,22 @@ async def _start_resilient_background( # `previous_response_id`. Propagate so the endpoint layer # surfaces HTTP 409 `conversation_fork_not_supported`. raise - except TaskManagerNotInitialized as exc: - # No resilient-task subsystem is installed in this process. - if _is_hosted_environment(): - # Hosted deployments auto-initialize the subsystem with no - # opt-out, so its absence means initialization failed at boot - # (e.g. a misconfigured backend or a missing dependency). - # Durability is mandatory in production — fail loudly as a - # platform error rather than silently degrading a store=true - # response to a non-durable, connection-scoped task. - logger.error( - "Resilient task subsystem missing in hosted environment for response %s; failing the request", - ctx.response_id, - ) - setattr(exc, PLATFORM_ERROR_TAG, True) - await self._runtime_state.delete(ctx.response_id) - raise - # Non-hosted (local dev, or unit/contract tests whose ASGI lifespan - # never ran). Nothing to recover — run the handler in-process. This - # is the legitimate non-durable path, NOT a failure. (When a manager - # IS present — hosted, or a local file-provider deployment — the - # start above succeeds and we use the resilient task.) + except TaskManagerNotInitialized: + # No resilient-task subsystem is installed in this process — the + # host did not enable resilient tasks + # (``set_resilient_tasks_enabled``), so recovery/durability is + # opt-out here. SWALLOW the signal and run the handler in-process: + # the response still executes and persists to the store (so GET + # works), it simply is not crash-recoverable. This is the deliberate + # non-durable path for a host that opted out of resilient tasks — + # NOT a failure. (When a manager IS present the start above succeeds + # and we use the resilient task with full recovery.) + logger.info( + "Resilient task subsystem not enabled for response %s; running handler " + "in-process (non-durable). Enable via set_resilient_tasks_enabled(True) " + "for crash recovery.", + ctx.response_id, + ) record.execution_task = asyncio.create_task(fallback_runner()) except Exception as exc: # pylint: disable=broad-exception-caught # The resilient-task subsystem IS present but starting the task diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py index 72b1e2dc9061..d92d8b2ba881 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py @@ -427,6 +427,23 @@ def __init__( # function and does not need this reference. self._parent_orchestrator = parent_orchestrator + # Opting into resilience implies the durable task subsystem must be + # constructed (and recovery enabled) for this deployment. The subsystem + # is gated SOLELY on the process-global switch (``AgentServerHost`` + # constructs the ``TaskManager`` only when it is set), so translate the + # explicit resilience opt-in — ``resilient_background`` / + # ``steerable_conversations`` — into that switch here, at host + # construction time (before the ASGI lifespan runs). A plain host that + # sets neither leaves the switch off: no TaskManager is constructed and + # ``store=true`` work degrades to non-durable in-process execution + # (``_start_resilient_background`` swallows ``TaskManagerNotInitialized``). + if options.resilient_background or options.steerable_conversations: + from azure.ai.agentserver.core.tasks import ( # pylint: disable=import-outside-toplevel + set_resilient_tasks_enabled, + ) + + set_resilient_tasks_enabled(True) + # Spec 023 — per-request primitive dispatch (SOT §6.6). # Two task primitives are registered per deployment; ``_pick_primitive`` # selects per request based on (conversation_id, previous_response_id, diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py index 12bcf8f93728..9a4227b3fb85 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py @@ -1,16 +1,20 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. -"""Contract tests: a resilient task-start failure must FAIL the request. +"""Contract tests: resilient task-start behavior. -When the resilient-task subsystem IS installed (hosted) but starting the -task-backed background execution fails, the server must NOT silently degrade -to a non-durable, connection-scoped ``asyncio.create_task`` (which loses crash -recovery while looking healthy). Instead it must fail immediately and surface -the failure as a *platform* error source — exactly like a Foundry storage -failure. +Two distinct cases with DIFFERENT contracts: -The legitimate "no task subsystem at all" case (e.g. a test client without a -TaskManager) must STILL run the handler in-process — that is not a failure. +1. The resilient-task subsystem IS installed but starting the task-backed + execution *fails* (the start call raises): the server must NOT silently + degrade to a non-durable ``asyncio.create_task`` — it must fail immediately + and surface a *platform* error source (like a Foundry storage failure). A + real durability failure must not hide behind a healthy-looking response. + +2. No task subsystem is installed at all (the host did not enable resilient + tasks via ``set_resilient_tasks_enabled``): this is the deliberate opt-out + path. ``TaskManagerNotInitialized`` is SWALLOWED and the handler runs + in-process — the response still executes and persists (GET works), it is + simply not crash-recoverable. This applies regardless of hosted vs local. """ from __future__ import annotations @@ -24,7 +28,6 @@ from azure.ai.agentserver.core._platform_headers import ERROR_DETAIL, ERROR_SOURCE from azure.ai.agentserver.responses import ResponsesAgentServerHost -from azure.ai.agentserver.responses.hosting import _orchestrator as _orch from azure.ai.agentserver.responses.hosting import _resilient_orchestrator as _ro from azure.ai.agentserver.responses.streaming._event_stream import ResponseEventStream @@ -114,12 +117,12 @@ def test_streaming_start_failure_emits_standalone_error_event(self, _start_fails class TestNoTaskManagerStillRunsHandler: - """Regression: no task subsystem (non-hosted) → handler runs in-process.""" + """No task subsystem (opt-out) → handler runs in-process (non-durable).""" def test_no_manager_background_runs_handler_ok(self) -> None: # Plain TestClient (no `with` → no lifespan → no TaskManager installed) - # and non-hosted (FOUNDRY_HOSTING_ENVIRONMENT unset) → legitimate - # in-process fallback, NOT a failure. + # → the outer catch swallows TaskManagerNotInitialized and runs the + # handler in-process. Legitimate opt-out path, NOT a failure. client = _build_client() resp = client.post( "/responses", @@ -129,36 +132,30 @@ def test_no_manager_background_runs_handler_ok(self) -> None: assert ERROR_SOURCE not in resp.headers -class TestHostedNoTaskManagerFailsLoudly: - """Hosted + no manager → durability is mandatory → fail as platform error. - - In a hosted deployment the resilient-task subsystem is auto-initialized - with no opt-out, so its absence is a platform-infrastructure failure — the - server must NOT silently degrade a ``store=true`` response to a non-durable - in-process run. - """ +class TestNoTaskManagerSwallowsAndRunsInProcess: + """Recovery is opt-in: with no TaskManager installed, ``store=true`` work is + NOT failed as a platform error — the outer catch swallows + ``TaskManagerNotInitialized`` and runs the handler in-process (non-durable). + This holds regardless of hosted vs local; enabling durability is the + operator's explicit choice via ``set_resilient_tasks_enabled(True)``.""" - def test_hosted_no_manager_background_fails_platform_500(self, monkeypatch: pytest.MonkeyPatch) -> None: - # Simulate hosted at the gate (no real env change → no Foundry store - # auto-activation) while the bare TestClient has no manager installed. - monkeypatch.setattr(_orch, "_is_hosted_environment", lambda: True) + def test_no_manager_background_runs_in_process(self) -> None: client = _build_client() resp = client.post( "/responses", json={"model": "test", "input": "hi", "stream": False, "store": True, "background": True}, ) - assert resp.status_code == 500, resp.text - assert resp.headers.get(ERROR_SOURCE) == "platform" - assert ERROR_DETAIL in resp.headers - assert "in_progress" not in resp.text + assert resp.status_code == 200, resp.text + # Swallowed, not surfaced as a platform error. + assert ERROR_SOURCE not in resp.headers - def test_hosted_no_manager_streaming_emits_error_event(self, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(_orch, "_is_hosted_environment", lambda: True) + def test_no_manager_streaming_runs_in_process(self) -> None: client = _build_client() resp = client.post( "/responses", json={"model": "test", "input": "hi", "stream": True, "store": True, "background": True}, ) types = [e["type"] for e in _collect_sse_events(resp.text)] - assert "error" in types, f"expected a standalone error event, got: {types}" - assert "response.completed" not in types, f"must not complete, got: {types}" + # In-process fallback runs the handler to completion; no error surface. + assert "error" not in types, f"must not surface an error on opt-out, got: {types}" + assert "response.completed" in types, f"expected completion, got: {types}" diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/e2e/resilience_contract/_test_handler.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/e2e/resilience_contract/_test_handler.py index b528eca60fb8..e7d796589285 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/tests/e2e/resilience_contract/_test_handler.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/e2e/resilience_contract/_test_handler.py @@ -53,6 +53,7 @@ import asyncio import os +from azure.ai.agentserver.core.tasks import set_resilient_tasks_enabled from azure.ai.agentserver.responses import ( CreateResponse, ResponseContext, @@ -101,6 +102,12 @@ def _env_int(name: str, default: int) -> int: resilient_background=_RESILIENT_BG, shutdown_grace_period_seconds=_SHUTDOWN_GRACE_S, ) +# Conformance exercises the durable-response subsystem (Rows 1/2/3), which is +# gated on the resilient-tasks switch. Row 2 runs with resilient_background=False +# (which does NOT auto-enable the switch), so enable it explicitly here — the +# server must construct the TaskManager and run recovery to honour the Row 2 +# mark-failed-on-crash contract. +set_resilient_tasks_enabled(True) app = ResponsesAgentServerHost(options=options) From 3e95cd961bdde32b8e840c15f7c5b55f355287d7 Mon Sep 17 00:00:00 2001 From: Naman Tyagi Date: Fri, 14 Aug 2026 13:47:19 +0530 Subject: [PATCH 2/9] [agentserver] Log a startup notice when responses resilience is disabled Emit a one-time WARNING at responses host startup when the durable-response subsystem is off (store=true responses are non-durable across an ungraceful crash), and an INFO when it is enabled, so operators are not surprised by the plain non-durable default. Reflects the final resolved state (after the resilient orchestrator's resilient_background/steerable auto-enable). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../agentserver/responses/hosting/_routing.py | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_routing.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_routing.py index d0fb1a76e257..436bd208bb3c 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_routing.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_routing.py @@ -496,6 +496,33 @@ def __init__( runtime_options.shutdown_grace_period_seconds, ) + # Announce whether the durable-response subsystem is active. The + # resilient orchestrator (constructed above) auto-enables the switch + # when ``resilient_background`` / ``steerable_conversations`` is set, so + # this reflects the final resolved state. When disabled we emit a + # one-time startup WARNING so operators are not surprised that a + # ``store=true`` response killed mid-flight by an ungraceful crash stays + # ``in_progress`` (there is no crash recovery) — matching a plain + # stateless server. + from azure.ai.agentserver.core.tasks import ( # pylint: disable=import-outside-toplevel + resilient_tasks_enabled, + ) + + if resilient_tasks_enabled(): + logger.info( + "Responses resilience: ENABLED - store=true responses run inside durable " + "tasks with crash recovery (in-flight responses are recovered/marked failed " + "on restart)." + ) + else: + logger.warning( + "Responses resilience: DISABLED - store=true responses run in-process and are " + "NOT durable across an ungraceful crash: a response in-flight when the process " + "is hard-killed stays in_progress on a later GET (no mark-failed/recovery). " + "Enable durability via resilient_background / steerable_conversations, or " + "set_resilient_tasks_enabled(True)." + ) + # ------------------------------------------------------------------ # Shutdown notification # ------------------------------------------------------------------ From 769776c82f0b818e0caea2c534cd7dbd725e1026 Mon Sep 17 00:00:00 2001 From: Naman Tyagi Date: Fri, 14 Aug 2026 14:12:28 +0530 Subject: [PATCH 3/9] [agentserver] Auto-enable resilient tasks only for resilient_background Narrow the responses auto-enable of the resilient-tasks switch to `resilient_background` only; `steerable_conversations` no longer implicitly enables the durable subsystem. Recovery is tied to resilient_background alone; a steerable host that wants durability sets resilient_background=True (or the switch explicitly). The two-switch UX is a known rough edge to smooth over post-Public-Preview. Steering conformance handler now enables the switch explicitly, since steering (multi-turn input queuing) needs the TaskManager regardless of resilient_background. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../hosting/_resilient_orchestrator.py | 31 ++++++++++++------- .../resilience_contract/_steering_handler.py | 6 ++++ 2 files changed, 26 insertions(+), 11 deletions(-) diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py index d92d8b2ba881..c170e27997a5 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py @@ -427,17 +427,26 @@ def __init__( # function and does not need this reference. self._parent_orchestrator = parent_orchestrator - # Opting into resilience implies the durable task subsystem must be - # constructed (and recovery enabled) for this deployment. The subsystem - # is gated SOLELY on the process-global switch (``AgentServerHost`` - # constructs the ``TaskManager`` only when it is set), so translate the - # explicit resilience opt-in — ``resilient_background`` / - # ``steerable_conversations`` — into that switch here, at host - # construction time (before the ASGI lifespan runs). A plain host that - # sets neither leaves the switch off: no TaskManager is constructed and - # ``store=true`` work degrades to non-durable in-process execution - # (``_start_resilient_background`` swallows ``TaskManagerNotInitialized``). - if options.resilient_background or options.steerable_conversations: + # Opting into resilient background responses implies the durable task + # subsystem must be constructed (and recovery enabled) for this + # deployment. The subsystem is gated SOLELY on the process-global switch + # (``AgentServerHost`` constructs the ``TaskManager`` only when it is + # set), so translate the explicit ``resilient_background`` opt-in into + # that switch here, at host construction time (before the ASGI lifespan + # runs). + # + # NOTE: only ``resilient_background`` auto-enables the subsystem — + # ``steerable_conversations`` intentionally does NOT. Recovery is tied to + # ``resilient_background`` alone; a steerable host that wants durability + # must set ``resilient_background=True`` (or call + # ``set_resilient_tasks_enabled(True)`` explicitly). The two-switch UX is + # a known rough edge to smooth over post-Public-Preview. + # + # A host that leaves ``resilient_background`` off (and does not set the + # switch) constructs no ``TaskManager``: ``store=true`` work degrades to + # non-durable in-process execution (``_start_resilient_background`` + # swallows ``TaskManagerNotInitialized``). + if options.resilient_background: from azure.ai.agentserver.core.tasks import ( # pylint: disable=import-outside-toplevel set_resilient_tasks_enabled, ) diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/e2e/resilience_contract/_steering_handler.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/e2e/resilience_contract/_steering_handler.py index b8ae32eb4051..d5f68eda85e3 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/tests/e2e/resilience_contract/_steering_handler.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/e2e/resilience_contract/_steering_handler.py @@ -41,6 +41,7 @@ import asyncio import os +from azure.ai.agentserver.core.tasks import set_resilient_tasks_enabled from azure.ai.agentserver.responses import ( CreateResponse, ResponseContext, @@ -77,6 +78,11 @@ def _env_int(name: str, default: int) -> int: steerable_conversations=True, shutdown_grace_period_seconds=_SHUTDOWN_GRACE_S, ) +# Steering (mid-turn input queuing) is implemented by the multi-turn task +# primitive and needs the TaskManager. Since only ``resilient_background`` now +# auto-enables the subsystem (steerable does not), enable it explicitly so the +# steering conformance is valid even when CONFORMANCE_RESILIENT_BACKGROUND=false. +set_resilient_tasks_enabled(True) app = ResponsesAgentServerHost(options=options) _turn_counts: dict[str, int] = {} From 588c885dbebf44b6edaf188642291cb942c3a4ce Mon Sep 17 00:00:00 2001 From: Naman Tyagi Date: Fri, 14 Aug 2026 14:47:32 +0530 Subject: [PATCH 4/9] [agentserver] Address PR feedback: docs, docstring, test, revert invocations samples - Rewrite set_resilient_tasks_enabled/_enablement docs to describe the switch as the opt-in gate for TaskManager construction (not just the recovery scan), and that a declared @task no longer implicitly enables it. - Update core README + tasks-guide to show the required opt-in, and core/responses CHANGELOGs with the breaking-change note. - Fix _start_resilient_background docstring: TaskManagerNotInitialized is now swallowed (in-process fallback) regardless of hosting; only real task-start failures raise/tag platform errors. - Rewrite the swallow contract test to enter the ASGI lifespan with the switch off, assert no TaskManager is installed, and poll GET to completion. - Revert the 3 invocations resilient sample edits (unrelated api.md whitespace consistency drift); sample opt-in to be handled as a follow-up. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../azure-ai-agentserver-core/CHANGELOG.md | 11 +++ .../azure-ai-agentserver-core/README.md | 11 ++- .../ai/agentserver/core/tasks/_enablement.py | 54 ++++++----- .../docs/tasks-guide.md | 34 +++---- .../samples/resilient_langgraph/app.py | 8 -- .../samples/resilient_multiturn/app.py | 9 +- .../samples/resilient_research/app.py | 8 -- .../CHANGELOG.md | 10 ++ .../responses/hosting/_orchestrator.py | 24 +++-- .../contract/test_resilient_start_failure.py | 95 ++++++++++++++----- 10 files changed, 163 insertions(+), 101 deletions(-) diff --git a/sdk/agentserver/azure-ai-agentserver-core/CHANGELOG.md b/sdk/agentserver/azure-ai-agentserver-core/CHANGELOG.md index 9b594533e897..335a81f85327 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/CHANGELOG.md +++ b/sdk/agentserver/azure-ai-agentserver-core/CHANGELOG.md @@ -8,6 +8,17 @@ ### Breaking Changes +- The resilient task subsystem is now **strictly opt-in**. `AgentServerHost` + constructs the `TaskManager` only when resilient tasks are enabled via + `set_resilient_tasks_enabled(True)` (or a protocol option that maps to it, + e.g. the responses `resilient_background`). Previously the manager was always + constructed and a declared `@task` / `@multi_turn_task` implicitly enabled the + startup recovery scan. Now, declaring a task does **not** turn the subsystem + on: with the switch off, `get_task_manager()` raises `TaskManagerNotInitialized` + and `.run()` / `.start()` cannot run a task. Existing apps that rely on `@task` + must call `set_resilient_tasks_enabled(True)` (before host startup) to keep + durable tasks and crash recovery. Plain servers that use no tasks are + unaffected and continue to pay nothing. - Removed `TaskMetadata`, `TaskContext.metadata`, and `TaskRun.metadata`. Durable application state now belongs in an explicit `FoundryStateStore` and no longer shares task lifecycle PATCHes or lease renewal. Typed task diff --git a/sdk/agentserver/azure-ai-agentserver-core/README.md b/sdk/agentserver/azure-ai-agentserver-core/README.md index 5bb689abac11..12b170a2042b 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/README.md +++ b/sdk/agentserver/azure-ai-agentserver-core/README.md @@ -191,8 +191,17 @@ python my_agent.py The `@task` decorator builds crash-resilient agents that survive container restarts, OOM kills, and redeployments. Task state is persisted to a task store, enabling automatic recovery and multi-turn suspend/resume patterns. +The resilient task subsystem is **opt-in**: call `set_resilient_tasks_enabled(True)` (before host startup, typically at import time) so the framework constructs the `TaskManager` and runs crash recovery. Without it, `.run()` / `.start()` raise `TaskManagerNotInitialized`. + ```python -from azure.ai.agentserver.core.tasks import task, TaskContext +from azure.ai.agentserver.core.tasks import ( + set_resilient_tasks_enabled, + task, + TaskContext, +) + +# Opt in to the durable task subsystem (required for @task to run). +set_resilient_tasks_enabled(True) @task(name="process_document") async def process_document(ctx: TaskContext[dict]) -> dict: diff --git a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_enablement.py b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_enablement.py index 84dae9c48963..c90e84360b75 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_enablement.py +++ b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_enablement.py @@ -1,24 +1,24 @@ # --------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # --------------------------------------------------------- -"""Explicit force-enable switch for the resilient task recovery scan. +"""Explicit opt-in switch for the resilient task subsystem. -The resilient ``TaskManager`` is always constructed by ``AgentServerHost`` (a -cheap, in-memory object that makes no task-store calls), so ``get_task_manager`` -and ``.run()`` / ``.start()`` work regardless of this switch. What this switch -affects is only the network-backed **startup recovery scan** (a hosted -task-store ``list()`` plus credential-token acquisition) and the periodic -recovery loop it spawns. +The resilient ``TaskManager`` is constructed by ``AgentServerHost`` **only when +this switch is on**. It is the single source of truth for the durable task +subsystem: when the switch is off, no ``TaskManager`` is installed, so +``get_task_manager()`` raises +:class:`~azure.ai.agentserver.core.tasks.TaskManagerNotInitialized` and +``.run()`` / ``.start()`` cannot run a task. When on, the manager is +constructed and the startup crash-recovery scan (plus the periodic recovery +loop) runs. -That recovery runs at startup when EITHER a durable task has been declared -(``@task`` / ``@multi_turn_task``) OR this switch is on. So an app that uses -tasks gets recovery automatically; this switch is a **force-enable** that -starts the recovery loop even before any task is declared (useful when tasks -are registered lazily after startup — the running loop then picks them up). +Recovery — and durable tasks as a whole — is therefore strictly opt-in. Merely +declaring a durable task (``@task`` / ``@multi_turn_task``) does **not** turn +the subsystem on; you must set this switch (directly, or via a protocol option +that maps to it, e.g. the responses ``resilient_background`` server option). -The switch is process-global and defaults to ``False``. It is intentionally -decoupled from any ``AgentServerHost`` instance so it can be flipped -independently at import time, e.g.:: +The switch is process-global and defaults to ``False``. Set it before +``AgentServerHost`` lifespan startup (typically at import time):: from azure.ai.agentserver.core.tasks import set_resilient_tasks_enabled @@ -32,18 +32,22 @@ @experimental def set_resilient_tasks_enabled(value: bool = True) -> None: - """Force-enable (or clear) the resilient task recovery scan process-wide. + """Opt in to (or clear) the resilient task subsystem process-wide. - Setting this to ``True`` starts the startup recovery scan + periodic - recovery loop even when no durable task is declared at startup. It does - NOT gate the ``TaskManager``'s existence: ``.run()`` / ``.start()`` work - whether or not this is set — this only controls automatic crash recovery. + This gates whether ``AgentServerHost`` constructs the ``TaskManager`` at + lifespan startup. Setting it to ``True`` constructs the manager, runs the + startup crash-recovery scan, and starts the periodic recovery loop. When it + is ``False`` no manager is installed: ``get_task_manager()`` raises + :class:`~azure.ai.agentserver.core.tasks.TaskManagerNotInitialized` and a + durable task cannot run (callers such as the responses ``store=true`` path + swallow that and degrade to non-durable in-process execution). Must be called before ``AgentServerHost`` lifespan startup (typically at import time) to take effect. Defaults to enabling when called with no argument. - :param value: ``True`` to force-enable recovery, ``False`` to clear. + :param value: ``True`` to enable the resilient task subsystem, ``False`` to + clear. :type value: bool """ global _RESILIENT_TASKS_ENABLED # pylint: disable=global-statement @@ -52,11 +56,11 @@ def set_resilient_tasks_enabled(value: bool = True) -> None: @experimental def resilient_tasks_enabled() -> bool: - """Return whether the recovery scan was explicitly force-enabled. + """Return whether the resilient task subsystem is enabled. - Note this reflects only the switch — recovery also runs automatically when - a durable task is declared, so a ``False`` return does not mean recovery is - off, nor that tasks are unavailable. + This is the authoritative gate: a ``False`` return means no ``TaskManager`` + is constructed and durable tasks / crash recovery are inactive (declaring a + task does NOT change this — the subsystem is opt-in). :return: ``True`` if :func:`set_resilient_tasks_enabled` turned it on. :rtype: bool diff --git a/sdk/agentserver/azure-ai-agentserver-core/docs/tasks-guide.md b/sdk/agentserver/azure-ai-agentserver-core/docs/tasks-guide.md index 5de9ab44010f..686482afc2fc 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/docs/tasks-guide.md +++ b/sdk/agentserver/azure-ai-agentserver-core/docs/tasks-guide.md @@ -115,31 +115,31 @@ What this primitive deliberately does **not** do: ### Enabling resilient tasks -The resilient `TaskManager`'s **startup recovery scan** — a network round-trip -to the hosted task store that reclaims tasks left in-flight by a crashed prior -instance — runs at startup when **either** of the following holds: - -1. at least one durable task has been declared (`@task` / `@multi_turn_task`) — - an app that uses tasks gets recovery automatically, **or** -2. it was explicitly force-enabled via `set_resilient_tasks_enabled(True)`. - -The force-enable is useful when tasks are registered *lazily* (declared after -startup): it starts the periodic recovery loop up front so a task declared -later is still recovered. +The resilient task subsystem is **opt-in**. The `TaskManager` is constructed — +and its **startup recovery scan** (a network round-trip to the hosted task store +that reclaims tasks left in-flight by a crashed prior instance) plus the +periodic recovery loop run — **only when the switch is on**: ```python from azure.ai.agentserver.core.tasks import set_resilient_tasks_enabled -set_resilient_tasks_enabled(True) # force-enable recovery before any task +set_resilient_tasks_enabled(True) # opt in BEFORE host startup (e.g. at import) ``` Use `resilient_tasks_enabled()` to read the current switch state. -The `TaskManager` itself is always constructed (a cheap, in-memory object that -makes no task-store calls until a task is used), so `get_task_manager()` and -`.run()` / `.start()` work regardless of the switch. A server that neither -declares a task nor sets the switch (e.g. an invocations-only host) simply -skips the startup recovery scan and pays none of its latency. +Merely declaring a durable task (`@task` / `@multi_turn_task`) does **not** turn +the subsystem on. When the switch is off, no `TaskManager` is installed: +`get_task_manager()` raises `TaskManagerNotInitialized` and `.run()` / `.start()` +cannot run a task. A server that does not set the switch (e.g. an +invocations-only host) constructs no manager and pays none of the recovery-scan +latency. + +> **Protocol note:** the responses protocol exposes a `resilient_background` +> server option that maps to this switch — constructing a +> `ResponsesAgentServerHost(options=ResponsesServerOptions(resilient_background=True))` +> enables the subsystem automatically, so responses apps typically don't call +> `set_resilient_tasks_enabled` directly. ### One-shot diff --git a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_langgraph/app.py b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_langgraph/app.py index 9b33070c3033..ba58ee46bdf9 100644 --- a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_langgraph/app.py +++ b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_langgraph/app.py @@ -67,7 +67,6 @@ EventStreamNotFoundError, streams, ) -from azure.ai.agentserver.core.tasks import set_resilient_tasks_enabled from azure.ai.agentserver.invocations import InvocationAgentServerHost try: @@ -84,13 +83,6 @@ app = InvocationAgentServerHost() -# Recovery is opt-in and gated solely on this switch. Declaring a -# ``@multi_turn_task`` makes the framework recovery-capable, but the startup -# recovery scan that reclaims tasks orphaned by a prior crash runs only when it -# is explicitly enabled. Enable it here so a fresh process reclaims in-flight -# sessions at startup. -set_resilient_tasks_enabled(True) - async def _sse_from_stream( stream: EventStream, invocation_id: str, *, initial_status: str = "queued" diff --git a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_multiturn/app.py b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_multiturn/app.py index e7d764629f98..b61a88ea70c9 100644 --- a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_multiturn/app.py +++ b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_multiturn/app.py @@ -37,7 +37,7 @@ from starlette.responses import JSONResponse, Response from azure.ai.agentserver.core.storage import FoundryStateStore -from azure.ai.agentserver.core.tasks import TaskConflictError, set_resilient_tasks_enabled +from azure.ai.agentserver.core.tasks import TaskConflictError from azure.ai.agentserver.invocations import InvocationAgentServerHost try: @@ -47,13 +47,6 @@ app = InvocationAgentServerHost() -# Recovery is opt-in and gated solely on this switch. Declaring a -# ``@multi_turn_task`` makes the framework recovery-capable, but the durable -# task subsystem (and its crash recovery) is only active when explicitly -# enabled. Enable it here so a fresh process reclaims in-flight sessions at -# startup. -set_resilient_tasks_enabled(True) - @app.invoke_handler async def handle_invoke(request: Request) -> Response: diff --git a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_research/app.py b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_research/app.py index e31057cb44c7..00b64c2813ff 100644 --- a/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_research/app.py +++ b/sdk/agentserver/azure-ai-agentserver-invocations/samples/resilient_research/app.py @@ -77,7 +77,6 @@ EventStreamNotFoundError, streams, ) -from azure.ai.agentserver.core.tasks import set_resilient_tasks_enabled from azure.ai.agentserver.invocations import InvocationAgentServerHost try: @@ -102,13 +101,6 @@ app = InvocationAgentServerHost() -# Recovery is opt-in and gated solely on this switch. Declaring a -# ``@multi_turn_task`` makes the framework recovery-capable, but the startup -# recovery scan that reclaims tasks orphaned by a prior crash runs only when it -# is explicitly enabled. Enable it here so a fresh process reclaims in-flight -# sessions at startup. -set_resilient_tasks_enabled(True) - # --- SSE rendering --------------------------------------------------------- diff --git a/sdk/agentserver/azure-ai-agentserver-responses/CHANGELOG.md b/sdk/agentserver/azure-ai-agentserver-responses/CHANGELOG.md index 312ac5bb7990..c06d63abf27e 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/CHANGELOG.md +++ b/sdk/agentserver/azure-ai-agentserver-responses/CHANGELOG.md @@ -4,6 +4,16 @@ ### Breaking Changes +- The durable-response subsystem is now **opt-in**. A `store=true` response is + wrapped in a resilient task (with crash recovery) only when the resilient task + subsystem is enabled — which `resilient_background=True` (or + `set_resilient_tasks_enabled(True)`) now does automatically. On a host that + enables neither, `store=true` responses run **non-durably in-process**: they + execute and persist (GET works), but a response in-flight when the process is + ungracefully killed stays `in_progress` on a later GET (no mark-failed/recovery) + — matching a plain stateless server. A one-time startup log announces which + mode is active. Previously every responses host implicitly used the task + subsystem (and paid the boot recovery scan) regardless of these options. - Removed `ResponseContext.conversation_chain_metadata` and the `ConversationChainMetadataNamespace` protocol. Resilient response applications now persist cross-turn state explicitly with diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py index e8856a4b1fe0..c00fed495660 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py @@ -3909,12 +3909,15 @@ async def _start_resilient_background( Two outcomes when the resilient start cannot proceed: - **No task subsystem installed** (the start raises - :class:`~azure.ai.agentserver.core.tasks.TaskManagerNotInitialized` — - e.g. an in-process test client whose lifespan never ran). When NOT - hosted, run the handler in-process via ``fallback_runner`` — there is - nothing to recover, so this is the legitimate non-durable path, NOT a - failure. When hosted, the subsystem is auto-initialized with no - opt-out, so its absence is a platform failure and is re-raised tagged. + :class:`~azure.ai.agentserver.core.tasks.TaskManagerNotInitialized`). + This happens whenever resilient tasks were not enabled for the host + (the durable subsystem is opt-in via ``set_resilient_tasks_enabled`` / + ``resilient_background``), or in an in-process test client whose + lifespan never ran. The signal is **swallowed** and the handler runs + in-process via ``fallback_runner`` — there is no manager to recover + through, so this is the legitimate non-durable path, NOT a failure. + The response still executes and persists (GET works); it is simply not + crash-recoverable. This applies regardless of hosted vs local. - **Subsystem present but the start fails**: fail immediately. The exception is tagged as a platform infrastructure error and re-raised (no silent degradation to a non-durable task — that would hide a real @@ -3934,11 +3937,12 @@ async def _start_resilient_background( recovery). Stamped into task framework metadata so recovery dispatch can route without re-deriving the gate from request params. :paramtype disposition: str - :raises Exception: If durability is required but unavailable — either the - task subsystem is present and the resilient start fails, or the - deployment is hosted and the subsystem is missing. The exception is + :raises Exception: If the task subsystem is present and the resilient + start fails (e.g. the task-store write is rejected). The exception is tagged with ``PLATFORM_ERROR_TAG`` so the endpoint surfaces - ``x-platform-error-source: platform``. + ``x-platform-error-source: platform``. A *missing* subsystem + (``TaskManagerNotInitialized``) is NOT raised — it is swallowed and + handled via the in-process fallback (see above). """ from ._resilient_orchestrator import ( ResilientResponseOrchestrator, diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py index 9a4227b3fb85..379419a2a082 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py @@ -27,11 +27,33 @@ from starlette.testclient import TestClient from azure.ai.agentserver.core._platform_headers import ERROR_DETAIL, ERROR_SOURCE +from azure.ai.agentserver.core.tasks import ( + TaskManagerNotInitialized, + resilient_tasks_enabled, + set_resilient_tasks_enabled, +) +from azure.ai.agentserver.core.tasks._manager import get_task_manager from azure.ai.agentserver.responses import ResponsesAgentServerHost from azure.ai.agentserver.responses.hosting import _resilient_orchestrator as _ro from azure.ai.agentserver.responses.streaming._event_stream import ResponseEventStream +@pytest.fixture() +def _switch_off() -> Any: + """Ensure the process-global resilient-tasks switch is OFF for the test. + + The switch is process-global and other tests may have flipped it on (e.g. + by constructing a ``resilient_background=True`` host). Reset it to False so + the switch-off host path is genuinely exercised, and restore afterwards. + """ + saved = resilient_tasks_enabled() + set_resilient_tasks_enabled(False) + try: + yield + finally: + set_resilient_tasks_enabled(saved) + + async def _noop_handler(request: Any, context: Any, cancellation_signal: asyncio.Event) -> AsyncIterator[Any]: async def _events() -> AsyncIterator[Any]: stream = ResponseEventStream(response_id=context.response_id, model=getattr(request, "model", None) or "") @@ -133,29 +155,54 @@ def test_no_manager_background_runs_handler_ok(self) -> None: class TestNoTaskManagerSwallowsAndRunsInProcess: - """Recovery is opt-in: with no TaskManager installed, ``store=true`` work is - NOT failed as a platform error — the outer catch swallows - ``TaskManagerNotInitialized`` and runs the handler in-process (non-durable). - This holds regardless of hosted vs local; enabling durability is the - operator's explicit choice via ``set_resilient_tasks_enabled(True)``.""" + """Recovery is opt-in: with the switch OFF, the ASGI lifespan installs NO + TaskManager, so ``store=true`` work is NOT failed as a platform error — the + outer catch swallows ``TaskManagerNotInitialized`` and runs the handler + in-process (non-durable). The response still executes AND persists (GET + works). Enabling durability is the operator's explicit choice via + ``set_resilient_tasks_enabled(True)`` / ``resilient_background``.""" + + def test_switch_off_no_manager_installed_and_response_completes(self, _switch_off: Any) -> None: + # Enter the lifespan with the switch explicitly OFF so this exercises the + # real production opt-out path (not the bare no-lifespan test client). + with _build_client() as client: + # Lifespan ran but installed NO manager (switch off). + with pytest.raises(TaskManagerNotInitialized): + get_task_manager() - def test_no_manager_background_runs_in_process(self) -> None: - client = _build_client() - resp = client.post( - "/responses", - json={"model": "test", "input": "hi", "stream": False, "store": True, "background": True}, - ) - assert resp.status_code == 200, resp.text - # Swallowed, not surfaced as a platform error. - assert ERROR_SOURCE not in resp.headers + resp = client.post( + "/responses", + json={"model": "test", "input": "hi", "stream": False, "store": True, "background": True}, + ) + assert resp.status_code == 200, resp.text + # Swallowed → not a platform error. + assert ERROR_SOURCE not in resp.headers + response_id = resp.json()["id"] + + # The in-process fallback runs AND persists: GET reaches a terminal. + import time + + deadline = time.monotonic() + 10.0 + status = None + while time.monotonic() < deadline: + got = client.get(f"/responses/{response_id}") + if got.status_code == 200: + status = got.json().get("status") + if status in ("completed", "failed", "cancelled"): + break + time.sleep(0.05) + assert status == "completed", f"expected completed via in-process fallback, got {status}" + + def test_switch_off_no_manager_streaming_runs_in_process(self, _switch_off: Any) -> None: + with _build_client() as client: + with pytest.raises(TaskManagerNotInitialized): + get_task_manager() - def test_no_manager_streaming_runs_in_process(self) -> None: - client = _build_client() - resp = client.post( - "/responses", - json={"model": "test", "input": "hi", "stream": True, "store": True, "background": True}, - ) - types = [e["type"] for e in _collect_sse_events(resp.text)] - # In-process fallback runs the handler to completion; no error surface. - assert "error" not in types, f"must not surface an error on opt-out, got: {types}" - assert "response.completed" in types, f"expected completion, got: {types}" + resp = client.post( + "/responses", + json={"model": "test", "input": "hi", "stream": True, "store": True, "background": True}, + ) + types = [e["type"] for e in _collect_sse_events(resp.text)] + # In-process fallback runs the handler to completion; no error surface. + assert "error" not in types, f"must not surface an error on opt-out, got: {types}" + assert "response.completed" in types, f"expected completion, got: {types}" From 0567ce36b9d1954679cc0263892be224070257db Mon Sep 17 00:00:00 2001 From: Naman Tyagi Date: Fri, 14 Aug 2026 16:38:22 +0530 Subject: [PATCH 5/9] [agentserver] Fix contract test isolation + actionable TaskManagerNotInitialized message - The switch-off contract tests asserted get_task_manager() raises, but the process-global TaskManager singleton can leak from a prior test in a shared pytest process (CI), causing DID-NOT-RAISE. The _switch_off fixture now also snapshots/resets/restores the manager singleton (set_task_manager(None)), so the no-manager path is exercised deterministically. - Make the TaskManagerNotInitialized message actionable: it now names set_resilient_tasks_enabled(True) and the responses resilient_background=True option instead of a generic 'ensure resilient tasks are enabled'. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../ai/agentserver/core/tasks/_manager.py | 7 +++++-- .../contract/test_resilient_start_failure.py | 20 +++++++++++++------ 2 files changed, 19 insertions(+), 8 deletions(-) diff --git a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_manager.py b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_manager.py index 1cff24b64b0a..aa1e4dad573c 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_manager.py +++ b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_manager.py @@ -297,8 +297,11 @@ def get_task_manager() -> TaskManager: ) raise TaskManagerNotInitialized( - "TaskManager not initialized. Ensure resilient tasks " - "are enabled on the AgentServerHost." # pylint: disable=implicit-str-concat + "TaskManager not initialized: the resilient task subsystem is not enabled. " + "Durable tasks and crash recovery are opt-in — call " + "set_resilient_tasks_enabled(True) before host startup (e.g. at import time), " + "or, for the responses protocol, construct the host with " + "ResponsesServerOptions(resilient_background=True)." # pylint: disable=implicit-str-concat ) return _manager diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py index 379419a2a082..e81371c8cb88 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py @@ -40,18 +40,26 @@ @pytest.fixture() def _switch_off() -> Any: - """Ensure the process-global resilient-tasks switch is OFF for the test. + """Ensure the process-global resilient-tasks state is OFF for the test. - The switch is process-global and other tests may have flipped it on (e.g. - by constructing a ``resilient_background=True`` host). Reset it to False so - the switch-off host path is genuinely exercised, and restore afterwards. + Both the enable switch and the ``TaskManager`` singleton are process-global + and other tests in a shared pytest process may have flipped the switch on or + installed a manager (e.g. by constructing a ``resilient_background=True`` + host). Snapshot, reset both to the switch-off / no-manager state so the + production opt-out path is genuinely exercised, then restore afterwards. """ - saved = resilient_tasks_enabled() + from azure.ai.agentserver.core.tasks import _manager as _mgr_mod # pylint: disable=import-outside-toplevel + from azure.ai.agentserver.core.tasks._manager import set_task_manager # pylint: disable=import-outside-toplevel + + saved_flag = resilient_tasks_enabled() + saved_mgr = _mgr_mod._manager # noqa: SLF001 # pylint: disable=protected-access set_resilient_tasks_enabled(False) + set_task_manager(None) try: yield finally: - set_resilient_tasks_enabled(saved) + set_task_manager(saved_mgr) + set_resilient_tasks_enabled(saved_flag) async def _noop_handler(request: Any, context: Any, cancellation_signal: asyncio.Event) -> AsyncIterator[Any]: From e3c728f801ca7a28d94d49ca1022a0c0d0b8942a Mon Sep 17 00:00:00 2001 From: Naman Tyagi Date: Fri, 14 Aug 2026 18:13:07 +0530 Subject: [PATCH 6/9] [agentserver] Make no-manager contract tests deterministic (fix CI flake) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The switch-off contract tests asserted get_task_manager() raises after entering the lifespan, but that depended on process-global manager/flag state (mutated by 1400+ other tests) and on which core build performs manager construction — making it fail deterministically in CI while passing locally. Force the no-manager condition explicitly via set_task_manager(None) inside the running lifespan so the responses SWALLOW path is exercised robustly. Core's gating of manager construction is covered separately by the core opt-in tests. Verified in a Linux (WSL) full-suite run: both tests pass among 1436 others. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../contract/test_resilient_start_failure.py | 28 +++++++++++-------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py index e81371c8cb88..7a71295dc280 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/contract/test_resilient_start_failure.py @@ -32,7 +32,7 @@ resilient_tasks_enabled, set_resilient_tasks_enabled, ) -from azure.ai.agentserver.core.tasks._manager import get_task_manager +from azure.ai.agentserver.core.tasks._manager import get_task_manager, set_task_manager from azure.ai.agentserver.responses import ResponsesAgentServerHost from azure.ai.agentserver.responses.hosting import _resilient_orchestrator as _ro from azure.ai.agentserver.responses.streaming._event_stream import ResponseEventStream @@ -49,7 +49,6 @@ def _switch_off() -> Any: production opt-out path is genuinely exercised, then restore afterwards. """ from azure.ai.agentserver.core.tasks import _manager as _mgr_mod # pylint: disable=import-outside-toplevel - from azure.ai.agentserver.core.tasks._manager import set_task_manager # pylint: disable=import-outside-toplevel saved_flag = resilient_tasks_enabled() saved_mgr = _mgr_mod._manager # noqa: SLF001 # pylint: disable=protected-access @@ -163,18 +162,23 @@ def test_no_manager_background_runs_handler_ok(self) -> None: class TestNoTaskManagerSwallowsAndRunsInProcess: - """Recovery is opt-in: with the switch OFF, the ASGI lifespan installs NO - TaskManager, so ``store=true`` work is NOT failed as a platform error — the - outer catch swallows ``TaskManagerNotInitialized`` and runs the handler - in-process (non-durable). The response still executes AND persists (GET - works). Enabling durability is the operator's explicit choice via - ``set_resilient_tasks_enabled(True)`` / ``resilient_background``.""" + """Recovery is opt-in: when no ``TaskManager`` is installed, ``store=true`` + work is NOT failed as a platform error — the responses outer catch swallows + ``TaskManagerNotInitialized`` and runs the handler in-process (non-durable). + The response still executes AND persists (GET works). + + These tests exercise the responses **swallow** behavior specifically. The + no-manager condition is forced deterministically via ``set_task_manager(None)`` + inside the running lifespan so the test does not depend on process-global + state left by other tests, nor on which core build performs (or skips) the + manager construction — that gating is covered by the core opt-in tests.""" def test_switch_off_no_manager_installed_and_response_completes(self, _switch_off: Any) -> None: - # Enter the lifespan with the switch explicitly OFF so this exercises the - # real production opt-out path (not the bare no-lifespan test client). with _build_client() as client: - # Lifespan ran but installed NO manager (switch off). + # Force the switch-off condition deterministically: whatever the + # lifespan installed, ensure no manager is present so the responses + # swallow path is the one under test. + set_task_manager(None) with pytest.raises(TaskManagerNotInitialized): get_task_manager() @@ -203,6 +207,8 @@ def test_switch_off_no_manager_installed_and_response_completes(self, _switch_of def test_switch_off_no_manager_streaming_runs_in_process(self, _switch_off: Any) -> None: with _build_client() as client: + # Force the switch-off condition deterministically (see above). + set_task_manager(None) with pytest.raises(TaskManagerNotInitialized): get_task_manager() From b600886150622253e770168d130c2296eb67ce44 Mon Sep 17 00:00:00 2001 From: Naman Tyagi Date: Fri, 14 Aug 2026 19:22:56 +0530 Subject: [PATCH 7/9] [agentserver] Fix pylint R0915: extract responses startup logging to a helper The added durability-mode startup log pushed ResponsesAgentServerHost.__init__ over the 50-statement pylint limit (53/50, R0915). Move the startup config + resilience-mode logging into a module-level _log_startup_configuration() helper, called once from __init__. Also correct the resilience note to reflect that only resilient_background auto-enables the subsystem (not steerable_conversations). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../agentserver/responses/hosting/_routing.py | 83 +++++++++++-------- 1 file changed, 47 insertions(+), 36 deletions(-) diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_routing.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_routing.py index 436bd208bb3c..c35257812fd2 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_routing.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_routing.py @@ -175,6 +175,51 @@ def _configure_streams_registry(runtime_options: ResponsesServerOptions) -> None ) +def _log_startup_configuration( + resolved_provider: "ResponseProviderProtocol", runtime_options: ResponsesServerOptions +) -> None: + """Log the responses startup configuration and durability mode. + + Emitted once at host construction. The resilience line reflects the final + resolved state (the resilient orchestrator auto-enables the switch when + ``resilient_background`` is set). When durability is disabled we log a + WARNING so operators are not surprised that a ``store=true`` response killed + mid-flight by an ungraceful crash stays ``in_progress`` on a later GET (no + crash recovery) — matching a plain stateless server. + + :param resolved_provider: The resolved response persistence provider. + :type resolved_provider: ResponseProviderProtocol + :param runtime_options: The responses server options. + :type runtime_options: ResponsesServerOptions + """ + logger.info( + "Responses protocol: storage_provider=%s, default_model=%s, " + "default_fetch_history_count=%s, shutdown_grace_period=%ss", + type(resolved_provider).__name__, + runtime_options.default_model or "(not set)", + runtime_options.default_fetch_history_count, + runtime_options.shutdown_grace_period_seconds, + ) + + from azure.ai.agentserver.core.tasks import ( # pylint: disable=import-outside-toplevel + resilient_tasks_enabled, + ) + + if resilient_tasks_enabled(): + logger.info( + "Responses resilience: ENABLED - store=true responses run inside durable " + "tasks with crash recovery (in-flight responses are recovered/marked failed " + "on restart)." + ) + else: + logger.warning( + "Responses resilience: DISABLED - store=true responses run in-process and are " + "NOT durable across an ungraceful crash: a response in-flight when the process " + "is hard-killed stays in_progress on a later GET (no mark-failed/recovery). " + "Enable durability via resilient_background, or set_resilient_tasks_enabled(True)." + ) + + def _validate_handler_signature(fn: Any) -> None: """Reject sync handlers and 2-arg signatures. @@ -486,42 +531,8 @@ def __init__( # Stash endpoint reference for request_shutdown() access. self._endpoint = endpoint - # --- Responses startup configuration logging --- - logger.info( - "Responses protocol: storage_provider=%s, default_model=%s, " - "default_fetch_history_count=%s, shutdown_grace_period=%ss", - type(resolved_provider).__name__, - runtime_options.default_model or "(not set)", - runtime_options.default_fetch_history_count, - runtime_options.shutdown_grace_period_seconds, - ) - - # Announce whether the durable-response subsystem is active. The - # resilient orchestrator (constructed above) auto-enables the switch - # when ``resilient_background`` / ``steerable_conversations`` is set, so - # this reflects the final resolved state. When disabled we emit a - # one-time startup WARNING so operators are not surprised that a - # ``store=true`` response killed mid-flight by an ungraceful crash stays - # ``in_progress`` (there is no crash recovery) — matching a plain - # stateless server. - from azure.ai.agentserver.core.tasks import ( # pylint: disable=import-outside-toplevel - resilient_tasks_enabled, - ) - - if resilient_tasks_enabled(): - logger.info( - "Responses resilience: ENABLED - store=true responses run inside durable " - "tasks with crash recovery (in-flight responses are recovered/marked failed " - "on restart)." - ) - else: - logger.warning( - "Responses resilience: DISABLED - store=true responses run in-process and are " - "NOT durable across an ungraceful crash: a response in-flight when the process " - "is hard-killed stays in_progress on a later GET (no mark-failed/recovery). " - "Enable durability via resilient_background / steerable_conversations, or " - "set_resilient_tasks_enabled(True)." - ) + # --- Responses startup configuration + durability-mode logging --- + _log_startup_configuration(resolved_provider, runtime_options) # ------------------------------------------------------------------ # Shutdown notification From 09a0f71318bea82bad4c6adf0f05dc9975209411 Mon Sep 17 00:00:00 2001 From: Naman Tyagi Date: Fri, 14 Aug 2026 19:56:01 +0530 Subject: [PATCH 8/9] [agentserver] Address review: fix ref leak, boot fail-fast, test isolation Copilot review fixes: - _RUNTIME_REFS leak: start_resilient() cached per-request refs keyed by response_id but only the resilient task body's finally evicts them. When primitive .start() raises (e.g. TaskManagerNotInitialized on the opt-out fallback), the body never runs, permanently retaining record/context/parsed/ cancel. Pop the entry on any start failure before propagating. - Ambiguous TaskManagerNotInitialized: instead of a per-request switch check (which broke non-lifespan test clients that auto-enable via resilient_background), AgentServerHost now FAILS THE LIFESPAN when resilient tasks are enabled but manager construction/startup fails - fail-fast at boot. A missing manager at request time therefore means opt-out, so the responses swallow stays unconditional (in-process non-durable fallback). - ASCII-only TaskManagerNotInitialized message (the em-dash broke latin-1 HTTP header encoding on the platform-error path). Test isolation: - Add an autouse fixture in the responses tests that snapshots/restores the process-global resilient-tasks switch + TaskManager singleton per test, so a resilient_background=True host in one test no longer leaks enabled state into later tests (was causing integration failures when run after resilient e2e). - Add a core test asserting the lifespan fails fast on boot startup failure. Verified on Linux (WSL): core optin 12, responses contract 6, and the full responses suite (minus the POSIX-only crash-harness dir) 1436 passed. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../azure/ai/agentserver/core/_base.py | 35 ++++++++------- .../ai/agentserver/core/tasks/_manager.py | 2 +- .../tests/tasks/test_task_manager_optin.py | 30 +++++++++++++ .../responses/hosting/_orchestrator.py | 44 +++++++++---------- .../hosting/_resilient_orchestrator.py | 12 ++++- .../tests/conftest.py | 32 ++++++++++++++ 6 files changed, 114 insertions(+), 41 deletions(-) diff --git a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py index 99267cc26556..a7c8d155bdb3 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py +++ b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py @@ -329,24 +329,25 @@ async def _lifespan(_app: Starlette) -> AsyncGenerator[None, None]: # noqa: RUF # an app that wants durable tasks / recovery must set the switch. task_manager = None if _resilient_tasks_enabled(): - try: - from .tasks._manager import ( # pylint: disable=import-outside-toplevel - TaskManager, - set_task_manager, - ) + # Resilient tasks were explicitly enabled. Construct the manager + # and run startup recovery. If EITHER fails, durability was + # explicitly requested, so fail the lifespan (fail-fast at boot) + # rather than start a server that would silently run store=true + # work non-durably. This guarantees that in a running deployment + # "enabled" always implies a live manager. + from .tasks._manager import ( # pylint: disable=import-outside-toplevel + TaskManager, + set_task_manager, + ) - task_manager = TaskManager( - config=cfg, - shutdown_event=asyncio.Event(), - shutdown_grace_seconds=_read_task_manager_shutdown_grace(), - ) - set_task_manager(task_manager) - await task_manager.startup() - logger.info("TaskManager initialized with startup recovery") - except ImportError: - pass # resilient module not available - except Exception: # pylint: disable=broad-exception-caught - logger.warning("Failed to initialize TaskManager", exc_info=True) + task_manager = TaskManager( + config=cfg, + shutdown_event=asyncio.Event(), + shutdown_grace_seconds=_read_task_manager_shutdown_grace(), + ) + set_task_manager(task_manager) + await task_manager.startup() + logger.info("TaskManager initialized with startup recovery") else: logger.info( "TaskManager NOT initialized (resilient tasks disabled; enable via " diff --git a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_manager.py b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_manager.py index aa1e4dad573c..0600ec43b670 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_manager.py +++ b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/tasks/_manager.py @@ -298,7 +298,7 @@ def get_task_manager() -> TaskManager: raise TaskManagerNotInitialized( "TaskManager not initialized: the resilient task subsystem is not enabled. " - "Durable tasks and crash recovery are opt-in — call " + "Durable tasks and crash recovery are opt-in - call " "set_resilient_tasks_enabled(True) before host startup (e.g. at import time), " "or, for the responses protocol, construct the host with " "ResponsesServerOptions(resilient_background=True)." # pylint: disable=implicit-str-concat diff --git a/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py b/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py index d5695a32becb..0ae89684a8f3 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py +++ b/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py @@ -228,3 +228,33 @@ async def _probe(ctx: "TaskContext[dict]") -> None: assert len(_fake_task_manager.instances) == 0 with pytest.raises(TaskManagerNotInitialized): get_task_manager() + + @pytest.mark.asyncio + async def test_switch_on_startup_failure_fails_lifespan( + self, _clean_state, monkeypatch: pytest.MonkeyPatch + ) -> None: + """When resilient tasks are ENABLED but manager startup fails, the + lifespan must fail fast (fail-fast at boot) rather than start a server + that would silently run store=true work non-durably.""" + from azure.ai.agentserver.core import AgentServerHost + + class _FailingManager: + def __init__(self, **kwargs) -> None: + pass + + async def startup(self) -> None: + raise RuntimeError("simulated boot failure") + + async def shutdown(self) -> None: + pass + + monkeypatch.setattr( + "azure.ai.agentserver.core.tasks._manager.TaskManager", + _FailingManager, + ) + set_resilient_tasks_enabled(True) + + app = AgentServerHost() + with pytest.raises(RuntimeError, match="simulated boot failure"): + async with app.router.lifespan_context(app): + pass diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py index c00fed495660..5f87013cc8e1 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py @@ -3908,16 +3908,16 @@ async def _start_resilient_background( Two outcomes when the resilient start cannot proceed: - - **No task subsystem installed** (the start raises - :class:`~azure.ai.agentserver.core.tasks.TaskManagerNotInitialized`). - This happens whenever resilient tasks were not enabled for the host - (the durable subsystem is opt-in via ``set_resilient_tasks_enabled`` / - ``resilient_background``), or in an in-process test client whose - lifespan never ran. The signal is **swallowed** and the handler runs - in-process via ``fallback_runner`` — there is no manager to recover - through, so this is the legitimate non-durable path, NOT a failure. - The response still executes and persists (GET works); it is simply not - crash-recoverable. This applies regardless of hosted vs local. + - **No manager installed** — the start raises + :class:`~azure.ai.agentserver.core.tasks.TaskManagerNotInitialized`. + Because ``AgentServerHost`` fails the lifespan when resilient tasks are + ENABLED but construction/startup fails, a missing manager in a running + deployment means resilient tasks are DISABLED (opt-out) — durability + was not requested (this also covers in-process test clients whose + lifespan never ran). The signal is **swallowed** and the handler runs + in-process via ``fallback_runner`` — the response still executes and + persists (GET works); it is simply not crash-recoverable. NOT a + failure. - **Subsystem present but the start fails**: fail immediately. The exception is tagged as a platform infrastructure error and re-raised (no silent degradation to a non-durable task — that would hide a real @@ -3928,7 +3928,7 @@ async def _start_resilient_background( :param record: The mutable execution record. :type record: ResponseExecution :param fallback_runner: The shielded runner coroutine function to run - in-process when no task subsystem is installed. + in-process when resilient tasks are disabled. :type fallback_runner: Any :keyword disposition: One of ``"re-invoke"`` (Row 1: resilient_bg+bg+store — task body re-runs handler on recovery) or ``"mark-failed"`` @@ -3940,7 +3940,7 @@ async def _start_resilient_background( :raises Exception: If the task subsystem is present and the resilient start fails (e.g. the task-store write is rejected). The exception is tagged with ``PLATFORM_ERROR_TAG`` so the endpoint surfaces - ``x-platform-error-source: platform``. A *missing* subsystem + ``x-platform-error-source: platform``. A missing subsystem (``TaskManagerNotInitialized``) is NOT raised — it is swallowed and handled via the in-process fallback (see above). """ @@ -3992,19 +3992,19 @@ async def _start_resilient_background( # surfaces HTTP 409 `conversation_fork_not_supported`. raise except TaskManagerNotInitialized: - # No resilient-task subsystem is installed in this process — the - # host did not enable resilient tasks - # (``set_resilient_tasks_enabled``), so recovery/durability is - # opt-out here. SWALLOW the signal and run the handler in-process: - # the response still executes and persists to the store (so GET - # works), it simply is not crash-recoverable. This is the deliberate - # non-durable path for a host that opted out of resilient tasks — - # NOT a failure. (When a manager IS present the start above succeeds - # and we use the resilient task with full recovery.) + # No manager is installed. Because ``AgentServerHost`` fails the + # lifespan when resilient tasks are ENABLED but construction/startup + # fails, a missing manager in a running deployment means resilient + # tasks are simply DISABLED (opt-out) — recovery/durability was not + # requested. (It also covers in-process test clients whose lifespan + # never ran.) SWALLOW and run the handler in-process: the response + # still executes and persists (GET works), it is simply not + # crash-recoverable. This is the deliberate non-durable path, NOT a + # failure. logger.info( "Resilient task subsystem not enabled for response %s; running handler " "in-process (non-durable). Enable via set_resilient_tasks_enabled(True) " - "for crash recovery.", + "(or resilient_background) for crash recovery.", ctx.response_id, ) record.execution_task = asyncio.create_task(fallback_runner()) diff --git a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py index c170e27997a5..32b34f6f6fc9 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_resilient_orchestrator.py @@ -1246,7 +1246,17 @@ async def start_resilient( # auto-queues against an in-flight chain and returns a TaskRun # whose ``is_queued`` is True (the public-surface detection signal). # See the queued-vs-fresh check below. - task_run = await picked_primitive.start(**start_kwargs) + try: + task_run = await picked_primitive.start(**start_kwargs) + except BaseException: + # If the primitive never started (e.g. ``TaskManagerNotInitialized`` + # when resilient tasks are disabled, or any start failure), the + # resilient task body — whose ``finally`` normally evicts the + # out-of-band refs — never runs. Drop the cache entry here so we do + # not permanently retain the record/context/parsed-request/cancel + # event for a response that fell back to in-process execution. + _RUNTIME_REFS.pop(response_id, None) + raise # Store the task run reference on the record for observability record.resilient_task_run = task_run # type: ignore[attr-defined] diff --git a/sdk/agentserver/azure-ai-agentserver-responses/tests/conftest.py b/sdk/agentserver/azure-ai-agentserver-responses/tests/conftest.py index b88f18a98698..84521fa950ae 100644 --- a/sdk/agentserver/azure-ai-agentserver-responses/tests/conftest.py +++ b/sdk/agentserver/azure-ai-agentserver-responses/tests/conftest.py @@ -30,6 +30,38 @@ def pytest_configure(config): ) +@pytest.fixture(autouse=True) +def _reset_resilient_tasks_global_state(): + """Contain process-global resilient-tasks state per test. + + The resilient-tasks enable switch and the ``TaskManager`` singleton are + process-global. A test that enables resilient tasks — e.g. constructs a + ``resilient_background=True`` host, which auto-enables the switch, or calls + ``set_resilient_tasks_enabled(True)`` — would otherwise leak that state into + subsequent tests in the same pytest process (a plain host would then build a + manager and behave durably). Snapshot both at test start and restore at the + end so each test is isolated. + """ + from azure.ai.agentserver.core.tasks import ( # pylint: disable=import-outside-toplevel + _manager as _mgr_mod, + ) + from azure.ai.agentserver.core.tasks import ( # pylint: disable=import-outside-toplevel + resilient_tasks_enabled, + set_resilient_tasks_enabled, + ) + from azure.ai.agentserver.core.tasks._manager import ( # pylint: disable=import-outside-toplevel + set_task_manager, + ) + + saved_flag = resilient_tasks_enabled() + saved_mgr = _mgr_mod._manager # noqa: SLF001 # pylint: disable=protected-access + try: + yield + finally: + set_task_manager(saved_mgr) + set_resilient_tasks_enabled(saved_flag) + + @pytest.fixture(autouse=True) def _isolated_resilient_tasks_root(tmp_path): """Isolate the LocalFileTaskProvider's default storage per test. From 1074aa5c73b5f35378671ee3bf5b52d1e36b98dd Mon Sep 17 00:00:00 2001 From: Naman Tyagi Date: Fri, 14 Aug 2026 22:54:05 +0530 Subject: [PATCH 9/9] [agentserver] Clear TaskManager singleton when boot startup fails When resilient tasks are enabled and TaskManager.startup() raises, the manager was already installed globally via set_task_manager() but the lifespan aborts before the shutdown block, leaving a failed/partially-started manager visible through get_task_manager(). Wrap startup with best-effort shutdown + clear the singleton before re-raising. The startup-failure test now asserts the singleton is cleared (and shutdown was invoked). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../azure/ai/agentserver/core/_base.py | 17 ++++++++++++++++- .../tests/tasks/test_task_manager_optin.py | 14 ++++++++++++-- 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py index a7c8d155bdb3..19eba248380b 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py +++ b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_base.py @@ -346,7 +346,22 @@ async def _lifespan(_app: Starlette) -> AsyncGenerator[None, None]: # noqa: RUF shutdown_grace_seconds=_read_task_manager_shutdown_grace(), ) set_task_manager(task_manager) - await task_manager.startup() + try: + await task_manager.startup() + except BaseException: + # ``set_task_manager`` above already installed the manager + # globally, but startup failed and the lifespan will now + # abort before the shutdown block runs. Best-effort tear the + # partially-started manager down and clear the singleton so a + # failed/half-initialized manager is not left visible through + # ``get_task_manager()``; then re-raise to fail the lifespan. + try: + await task_manager.shutdown() + except Exception: # pylint: disable=broad-exception-caught + logger.warning("Error shutting down TaskManager after startup failure", exc_info=True) + set_task_manager(None) + task_manager = None + raise logger.info("TaskManager initialized with startup recovery") else: logger.info( diff --git a/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py b/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py index 0ae89684a8f3..8d0150c78a57 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py +++ b/sdk/agentserver/azure-ai-agentserver-core/tests/tasks/test_task_manager_optin.py @@ -235,8 +235,13 @@ async def test_switch_on_startup_failure_fails_lifespan( ) -> None: """When resilient tasks are ENABLED but manager startup fails, the lifespan must fail fast (fail-fast at boot) rather than start a server - that would silently run store=true work non-durably.""" + that would silently run store=true work non-durably. The partially + started manager must be torn down and the singleton cleared so it is not + left visible through ``get_task_manager()``.""" from azure.ai.agentserver.core import AgentServerHost + from azure.ai.agentserver.core.tasks._manager import get_task_manager + + shutdown_calls: list[int] = [] class _FailingManager: def __init__(self, **kwargs) -> None: @@ -246,7 +251,7 @@ async def startup(self) -> None: raise RuntimeError("simulated boot failure") async def shutdown(self) -> None: - pass + shutdown_calls.append(1) monkeypatch.setattr( "azure.ai.agentserver.core.tasks._manager.TaskManager", @@ -258,3 +263,8 @@ async def shutdown(self) -> None: with pytest.raises(RuntimeError, match="simulated boot failure"): async with app.router.lifespan_context(app): pass + + # The partially started manager was torn down and the singleton cleared. + assert shutdown_calls == [1] + with pytest.raises(TaskManagerNotInitialized): + get_task_manager()