From ded7a237c9aa130de2ab08e95c3f6be2f06aa924 Mon Sep 17 00:00:00 2001
From: "Liu Juanjuan (Helen)"
 <31027645+liujuanjuan1984@users.noreply.github.com>
Date: Tue, 29 Sep 2026 12:29:02 +0000
Subject: [PATCH 1/5] =?UTF-8?q?fix(server):=20=E4=BF=AE=E5=A4=8D=E4=BB=BB?=
 =?UTF-8?q?=E5=8A=A1=E5=A4=B1=E8=B4=A5=E6=8C=81=E4=B9=85=E5=8C=96=E4=B8=8E?=
 =?UTF-8?q?=E5=85=B3=E9=97=AD=E6=B8=85=E7=90=86=20(#533)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 docs/compatibility.md                        |  32 ++
 docs/guide.md                                |  25 +
 docs/maintainer-architecture.md              |   1 +
 src/opencode_a2a/server/application.py       |  58 +-
 src/opencode_a2a/server/handler_lifecycle.py | 159 ++++++
 src/opencode_a2a/server/lifespan.py          |  22 +-
 tests/server/test_handler_lifecycle.py       | 563 +++++++++++++++++++
 7 files changed, 806 insertions(+), 54 deletions(-)
 create mode 100644 src/opencode_a2a/server/handler_lifecycle.py
 create mode 100644 tests/server/test_handler_lifecycle.py

diff --git a/docs/compatibility.md b/docs/compatibility.md
index 47891d5..3418ea8 100644
--- a/docs/compatibility.md
+++ b/docs/compatibility.md
@@ -28,6 +28,38 @@ The [SDK 1.1.2 to 1.1.5 changes](https://github.com/a2aproject/a2a-python/compar
 
 The missing-import typing overrides for `google.protobuf.*` and `jsonrpc.*` remain limited to those third-party modules. Runtime support is validated by the SDK and transport regressions, not inferred from the typing overrides. Ruff's local pre-commit hooks run through `uv run --locked --extra dev`, so `uv.lock` is the single version source for linting and formatting.
 
+### Handler Lifecycle and Legacy Queue Migration
+
+The adapter still uses the SDK's `LegacyRequestHandler`, through the internal
+`ManagedLegacyRequestHandler` lifecycle boundary. It does not use the newer
+`DefaultRequestHandler`/`ActiveTaskRegistry`; the latter's early-failure handling
+and `aclose()` do not apply automatically to this adapter.
+
+Unhandled producer errors are persisted directly as `FAILED` under the original
+request identity, including when no task row exists yet. Existing terminal
+snapshots remain immutable. Failure to persist is logged without storage error
+payloads and does not replace the execution error or skip queue cleanup. Both
+send paths explicitly check completed producers so queue closure cannot hide an
+execution exception. `GetTask` delegates parameter validation and history slicing
+to the pinned SDK, retaining adapter output negotiation and storage-error mapping.
+Empty IDs and negative history lengths are rejected as invalid parameters.
+
+Application shutdown closes the handler before its clients and stores. See the
+[shutdown contract](guide.md#execution-failure-and-shutdown) for local cancellation
+and upstream-session limits. Normal client disconnects still allow background
+execution and persistence.
+
+SDK 1.1.5 still provides `EventQueueLegacy`, but documents its future removal.
+Replacing one queue constructor or changing the handler's parent class is not a
+complete migration: the legacy queue manager, consumer, cancellation, subscription,
+and output aggregation paths are coupled. Retain the pinned implementation until
+a deliberate SDK upgrade or a separately reviewed migration replaces these paths.
+Before accepting an SDK version that removes legacy interfaces, prove parity for
+identity isolation, output/extension negotiation, cancel idempotency, first-terminal
+persistence, nonblocking sends, subscription, and disconnect/shutdown behavior on
+Python 3.11–3.14 with both memory and SQLite stores. Do not copy the SDK's queue
+implementation into this repository merely to suppress the deprecation.
+
 ## Contract Honesty
 
 Machine-readable discovery surfaces must reflect actual runtime behavior:
diff --git a/docs/guide.md b/docs/guide.md
index 2d4b4e9..c3ee8f3 100644
--- a/docs/guide.md
+++ b/docs/guide.md
@@ -217,6 +217,31 @@ The package declares `sqlalchemy[asyncio]` so standalone installs such as `uv to
 
 Published package metadata also enforces the security floors `click>=8.3.3` and `pyasn1>=0.6.4`. These apply to standalone wheel/sdist installs as well as repository environments; no separate uv constraint configuration is needed.
 
+### Execution Failure and Shutdown
+
+An unhandled execution error writes `FAILED` directly to the task store using the
+original request identity, then closes the producer queue and removes its runtime
+registration. An already persisted terminal task is never overwritten. If storage
+is unavailable, cleanup still runs, but a durable failure state cannot be promised.
+This also covers failures before the first task event; ordinary provider errors
+continue to use the executor's existing error events.
+
+At application shutdown, the handler rejects new execution setup and waits for
+setup already in progress to register its producers. It cancels and awaits local
+producers, marks interrupted nonterminal executions `FAILED` when storage is
+available, and drains/cancels tracked background consumers and cleanup tasks before
+closing outbound clients, the OpenCode HTTP client, state stores, and the database
+engine. Cleanup of the remaining resources is attempted even if a close operation
+fails. Shutdown is idempotent and does not wait for model generation to finish,
+though executor cleanup and storage I/O must still complete. Supervisor shutdown
+budgets must allow that cleanup; a hard process kill cannot provide these guarantees.
+
+This policy cancels adapter execution, not the independently running OpenCode
+session. To request a best-effort upstream session abort, use `CancelTask` before
+stopping the service. That explicit cancel path retains its `CANCELED` state and
+idempotency. An SSE client disconnect alone does not trigger application shutdown:
+the producer continues in the background and persists its remaining output.
+
 ### SQLite Persistence Hardening
 
 File-backed SQLite databases are hardened at startup and on every new connection on POSIX systems:
diff --git a/docs/maintainer-architecture.md b/docs/maintainer-architecture.md
index 114d107..79e0b50 100644
--- a/docs/maintainer-architecture.md
+++ b/docs/maintainer-architecture.md
@@ -74,6 +74,7 @@ flowchart TD
 ### Server Layer
 
 - `server/application.py`: app assembly, route wiring, request handler customization, and top-level lifecycle integration
+- `server/handler_lifecycle.py`: legacy SDK producer ownership, direct failure persistence, setup admission, and handler shutdown
 - `server/middleware.py`: auth, protocol negotiation, payload/body guards, logging, and response decoration
 - `server/agent_card.py` / `server/openapi.py`: machine-readable contract publication
 - `server/rest_tasks.py`: SDK-owned REST task routes plus adapter-specific list behavior
diff --git a/src/opencode_a2a/server/application.py b/src/opencode_a2a/server/application.py
index 96510e9..cca1f8d 100644
--- a/src/opencode_a2a/server/application.py
+++ b/src/opencode_a2a/server/application.py
@@ -9,16 +9,11 @@
 import uvicorn
 from a2a.server.agent_execution import RequestContext
 from a2a.server.events import EventConsumer, EventQueueLegacy
-from a2a.server.request_handlers.default_request_handler import (
-    TERMINAL_TASK_STATES,
-    LegacyRequestHandler,
-)
 from a2a.server.request_handlers.response_helpers import agent_card_to_dict
 from a2a.server.routes.common import DefaultServerCallContextBuilder
 from a2a.server.routes.rest_dispatcher import RestDispatcher
 from a2a.server.tasks import ResultAggregator, TaskManager
 from a2a.types import (
-    AgentCard,
     Artifact,
     CancelTaskRequest,
     GetTaskRequest,
@@ -46,7 +41,7 @@
     A2AError,
     ErrorMapping,
 )
-from a2a.utils.task import apply_history_length, validate_history_length
+from a2a.utils.task import apply_history_length
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse, PlainTextResponse
 from google.protobuf.json_format import MessageToDict, ParseDict, ParseError
@@ -88,6 +83,7 @@
 )
 from ..profile.runtime import build_runtime_profile
 from ..protocol_versions import A2A_PROTOCOL_VERSION
+from ..task_states import TERMINAL_TASK_STATES
 from ..trace_context import install_log_record_factory
 from .agent_card import (
     _CHAT_OUTPUT_MODES,
@@ -96,6 +92,7 @@
 from .client_manager import A2AClientManager
 from .context_helpers import AuthenticatedIdentityUser
 from .database import build_database_engine
+from .handler_lifecycle import ManagedLegacyRequestHandler
 from .lifespan import build_lifespan
 from .middleware import (
     build_agent_card_etag,
@@ -234,46 +231,15 @@ def _parse_rest_send_message_request(body: bytes):
 
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Awaitable, Callable
+    from collections.abc import AsyncIterator, Callable
 
-    from a2a.server.agent_execution import AgentExecutor, RequestContextBuilder
     from a2a.server.context import ServerCallContext
     from a2a.server.request_handlers.request_handler import RequestHandler
     from a2a.server.routes.common import ServerCallContextBuilder
-    from a2a.server.tasks import (
-        PushNotificationConfigStore,
-        PushNotificationSender,
-        TaskStore,
-    )
-
 
-class OpencodeRequestHandler(LegacyRequestHandler):
-    """Custom request handler to gracefully handle client disconnects and prevent dead loops."""
 
-    def __init__(  # noqa: PLR0913
-        self,
-        agent_executor: AgentExecutor,
-        task_store: TaskStore,
-        agent_card: AgentCard,
-        queue_manager: Any | None = None,
-        push_config_store: PushNotificationConfigStore | None = None,
-        push_sender: PushNotificationSender | None = None,
-        request_context_builder: RequestContextBuilder | None = None,
-        extended_agent_card: AgentCard | None = None,
-        extended_card_modifier: Callable[[AgentCard, ServerCallContext], Awaitable[AgentCard]]
-        | None = None,
-    ) -> None:
-        super().__init__(
-            agent_executor=agent_executor,
-            task_store=task_store,
-            agent_card=agent_card,
-            queue_manager=queue_manager,
-            push_config_store=push_config_store,
-            push_sender=push_sender,
-            request_context_builder=request_context_builder,
-            extended_agent_card=extended_agent_card,
-            extended_card_modifier=extended_card_modifier,
-        )
+class OpencodeRequestHandler(ManagedLegacyRequestHandler):
+    """Preserve adapter contracts around the SDK's legacy execution pipeline."""
 
     @staticmethod
     def _task_store_failure_message(operation: str) -> str:
@@ -522,11 +488,8 @@ async def on_get_task(
         context=None,
     ) -> Task | None:
         try:
-            validate_history_length(params)
-            task = await self.task_store.get(params.id, context)
-            if not task:
-                raise TaskNotFoundError()
-            return self._apply_task_output_negotiation(apply_history_length(task, params), context)
+            task = await super().on_get_task(params, context)
+            return self._apply_task_output_negotiation(task, context) if task is not None else None
         except TaskStoreOperationError as exc:
             raise self._task_store_server_error(exc) from exc
 
@@ -682,6 +645,8 @@ async def on_message_send_stream(self, params, context=None):
                     self._validate_task_id_match(task_id, event.id)
                 await self._send_push_notification_if_needed(task_id, event)
                 yield event
+            if producer_task.done() and not producer_task.cancelled():
+                producer_task.result()
             stream_completed = True
         except TaskStoreOperationError as exc:
             logger.exception(
@@ -793,6 +758,8 @@ async def push_notification_callback() -> None:
                 except asyncio.CancelledError:
                     pass
 
+        if producer_task.done() and not producer_task.cancelled():
+            producer_task.result()
         if not result:
             raise InternalError()
 
@@ -970,6 +937,7 @@ def create_app(settings: Settings) -> FastAPI:
     extended_card_etag = build_agent_card_etag(extended_agent_card)
     persistence_summary = describe_lightweight_persistence_backend(settings)
     lifespan = build_lifespan(
+        request_handler=handler,
         database_engine=database_engine,
         task_store_runtime=task_store_runtime,
         runtime_state_runtime=runtime_state_runtime,
diff --git a/src/opencode_a2a/server/handler_lifecycle.py b/src/opencode_a2a/server/handler_lifecycle.py
new file mode 100644
index 0000000..0efd5fa
--- /dev/null
+++ b/src/opencode_a2a/server/handler_lifecycle.py
@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Awaitable, Callable
+from contextlib import suppress
+from typing import Any
+
+from a2a.server.agent_execution import AgentExecutor, RequestContext, RequestContextBuilder
+from a2a.server.context import ServerCallContext
+from a2a.server.events import EventQueueLegacy
+from a2a.server.events.queue_manager import NoTaskQueue
+from a2a.server.request_handlers.default_request_handler import LegacyRequestHandler
+from a2a.server.tasks import PushNotificationConfigStore, PushNotificationSender, TaskStore
+from a2a.types import AgentCard, InternalError, Task, TaskState, TaskStatus
+
+from ..task_states import TERMINAL_TASK_STATES
+
+logger = logging.getLogger(__name__)
+
+
+class ManagedLegacyRequestHandler(LegacyRequestHandler):
+    """Own producer and background-task lifetimes around the pinned SDK legacy handler."""
+
+    def __init__(  # noqa: PLR0913
+        self,
+        agent_executor: AgentExecutor,
+        task_store: TaskStore,
+        agent_card: AgentCard,
+        queue_manager: Any | None = None,
+        push_config_store: PushNotificationConfigStore | None = None,
+        push_sender: PushNotificationSender | None = None,
+        request_context_builder: RequestContextBuilder | None = None,
+        extended_agent_card: AgentCard | None = None,
+        extended_card_modifier: Callable[[AgentCard, ServerCallContext], Awaitable[AgentCard]]
+        | None = None,
+    ) -> None:
+        super().__init__(
+            agent_executor=agent_executor,
+            task_store=task_store,
+            agent_card=agent_card,
+            queue_manager=queue_manager,
+            push_config_store=push_config_store,
+            push_sender=push_sender,
+            request_context_builder=request_context_builder,
+            extended_agent_card=extended_agent_card,
+            extended_card_modifier=extended_card_modifier,
+        )
+        self._setup_lock = asyncio.Lock()
+        self._closing = False
+        self._close_task: asyncio.Task[None] | None = None
+        self._managed_producers: dict[asyncio.Task, str] = {}
+
+    async def _persist_execution_failure(self, request: RequestContext) -> None:
+        """Write independently of the closing queue, preserving terminal snapshots."""
+        if not request.task_id or not request.context_id:
+            return
+        try:
+            existing = await self.task_store.get(request.task_id, request.call_context)
+            if existing and existing.status.state in TERMINAL_TASK_STATES:
+                return
+            task = Task(id=request.task_id, context_id=request.context_id)
+            if existing:
+                task.CopyFrom(existing)
+            if request.message is not None and not any(
+                message.message_id == request.message.message_id for message in task.history
+            ):
+                task.history.append(request.message)
+            task.status.CopyFrom(TaskStatus(state=TaskState.TASK_STATE_FAILED))
+            task.status.timestamp.GetCurrentTime()
+            await self.task_store.save(task, request.call_context)
+        except Exception as exc:
+            # Preserve the producer error without logging storage payloads or credentials.
+            logger.error(
+                "Could not persist execution failure task_id=%s error_type=%s",
+                request.task_id,
+                type(exc).__name__,
+            )
+
+    async def _run_event_stream(self, request: RequestContext, queue: EventQueueLegacy) -> None:
+        try:
+            await self.agent_executor.execute(request, queue)
+        except asyncio.CancelledError:
+            if self._closing:
+                await self._persist_execution_failure(request)
+            raise
+        except Exception as exc:
+            logger.error(
+                "Agent producer failed task_id=%s error_type=%s",
+                request.task_id,
+                type(exc).__name__,
+            )
+            await self._persist_execution_failure(request)
+            raise
+        finally:
+            await queue.close()
+
+    async def _cleanup_producer(self, producer_task: asyncio.Task, task_id: str) -> None:
+        try:
+            # The consumer reports execution failures. They must not skip cleanup.
+            with suppress(asyncio.CancelledError, Exception):
+                await producer_task
+        finally:
+            try:
+                with suppress(NoTaskQueue):
+                    await self._queue_manager.close(task_id)
+            finally:
+                with self._running_agents_lock:
+                    if self._running_agents.get(task_id) is producer_task:
+                        self._running_agents.pop(task_id, None)
+                self._managed_producers.pop(producer_task, None)
+
+    def _track_background_task(self, task: asyncio.Task) -> None:
+        super()._track_background_task(task)
+        if self._closing:
+            task.cancel()
+
+    async def aclose(self) -> None:
+        """Stop local execution before its clients and stores are closed."""
+        if self._close_task is None:
+            self._closing = True
+            self._close_task = asyncio.create_task(self._drain_execution())
+        try:
+            await asyncio.shield(self._close_task)
+        except asyncio.CancelledError:
+            # Do not allow lifespan cancellation to close dependencies mid-drain.
+            await asyncio.shield(self._close_task)
+            raise
+
+    async def _drain_execution(self) -> None:
+        # A setup already in flight must register its producer before the snapshot.
+        async with self._setup_lock:
+            producers = dict(self._managed_producers)
+            producers.update((task, task_id) for task_id, task in self._running_agents.items())
+        for task in producers:
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(*producers, return_exceptions=True)
+        results = await asyncio.gather(
+            *(self._cleanup_producer(task, task_id) for task, task_id in producers.items()),
+            return_exceptions=True,
+        )
+        # Cancellation of a consumer may itself register a deferred cleanup.
+        while pending := [task for task in self._background_tasks if not task.done()]:
+            for task in pending:
+                task.cancel()
+            await asyncio.gather(*pending, return_exceptions=True)
+        errors = [result for result in results if isinstance(result, Exception)]
+        if errors:
+            raise ExceptionGroup("Request handler cleanup failed", errors)
+
+    async def _setup_message_execution(self, params, context=None):  # noqa: ANN001
+        async with self._setup_lock:
+            if self._closing:
+                raise InternalError(message="Server is shutting down.")
+            result = await super()._setup_message_execution(params, context)
+            _manager, task_id, _queue, _aggregator, producer_task = result
+            self._managed_producers[producer_task] = task_id
+            return result
diff --git a/src/opencode_a2a/server/lifespan.py b/src/opencode_a2a/server/lifespan.py
index fec0605..c56e079 100644
--- a/src/opencode_a2a/server/lifespan.py
+++ b/src/opencode_a2a/server/lifespan.py
@@ -2,13 +2,14 @@
 
 import logging
 from collections.abc import Mapping
-from contextlib import asynccontextmanager
+from contextlib import AsyncExitStack, asynccontextmanager
 
 logger = logging.getLogger(__name__)
 
 
 def build_lifespan(
     *,
+    request_handler,
     database_engine,
     task_store_runtime,
     runtime_state_runtime,
@@ -36,13 +37,16 @@ async def lifespan(_app):
             runtime_state_started = True
             yield
         finally:
-            await client_manager.close_all()
-            await upstream_client.close()
-            if runtime_state_started:
-                await runtime_state_runtime.shutdown()
-            if task_store_started:
-                await task_store_runtime.shutdown()
-            if database_engine is not None:
-                await database_engine.dispose()
+            # ExitStack runs every callback even if an earlier cleanup raises.
+            async with AsyncExitStack() as cleanup:
+                if database_engine is not None:
+                    cleanup.push_async_callback(database_engine.dispose)
+                if task_store_started:
+                    cleanup.push_async_callback(task_store_runtime.shutdown)
+                if runtime_state_started:
+                    cleanup.push_async_callback(runtime_state_runtime.shutdown)
+                cleanup.push_async_callback(upstream_client.close)
+                cleanup.push_async_callback(client_manager.close_all)
+                cleanup.push_async_callback(request_handler.aclose)
 
     return lifespan
diff --git a/tests/server/test_handler_lifecycle.py b/tests/server/test_handler_lifecycle.py
new file mode 100644
index 0000000..ebf3162
--- /dev/null
+++ b/tests/server/test_handler_lifecycle.py
@@ -0,0 +1,563 @@
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import pytest_asyncio
+from a2a.server.context import ServerCallContext
+from a2a.types import (
+    AgentCapabilities,
+    AgentCard,
+    GetTaskRequest,
+    Message,
+    Part,
+    Role,
+    SendMessageRequest,
+    Task,
+    TaskState,
+    TaskStatus,
+)
+from a2a.utils.errors import InvalidParamsError
+
+from opencode_a2a.server.application import OpencodeRequestHandler
+from opencode_a2a.server.lifespan import build_lifespan
+from opencode_a2a.server.task_store import build_task_store_runtime
+from tests.support.settings import make_settings
+
+
+def _context(identity="owner"):
+    return ServerCallContext(state={"identity": identity})
+
+
+def _request(task_id=""):
+    return SendMessageRequest(
+        message=Message(
+            message_id="input-message",
+            task_id=task_id,
+            role=Role.ROLE_USER,
+            parts=[Part(text="hello")],
+        )
+    )
+
+
+def _handler(store, execute):
+    return OpencodeRequestHandler(
+        agent_executor=SimpleNamespace(execute=execute),
+        task_store=store,
+        agent_card=AgentCard(name="test", capabilities=AgentCapabilities(streaming=True)),
+    )
+
+
+@pytest_asyncio.fixture(params=["memory", "database"])
+async def store(request, tmp_path):
+    runtime = build_task_store_runtime(
+        make_settings(
+            a2a_task_store_backend=request.param,
+            a2a_task_store_database_url=f"sqlite+aiosqlite:///{tmp_path}/tasks.db",
+        )
+    )
+    await runtime.startup()
+    try:
+        yield runtime.task_store
+    finally:
+        await runtime.shutdown()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("existing", [False, True])
+async def test_unhandled_producer_failure_persists_failed_and_releases_queue(store, existing):
+    async def execute(context, queue):
+        raise RuntimeError("injected producer failure")
+
+    handler = _handler(store, execute)
+    params = _request("existing" if existing else "")
+    context = _context()
+    if existing:
+        await store.save(
+            Task(
+                id="existing",
+                context_id="ctx",
+                status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+            ),
+            context,
+        )
+    try:
+        with pytest.raises(RuntimeError, match="injected producer failure"):
+            await asyncio.wait_for(handler.on_message_send(params, context), 2)
+        task_id = params.message.task_id
+        stored = await store.get(task_id, context)
+        assert stored is not None
+        assert stored.status.state == TaskState.TASK_STATE_FAILED
+        assert "injected producer failure" not in str(stored)
+        assert await store.get(task_id, _context("other")) is None
+        assert not handler._running_agents
+        assert await handler._queue_manager.get(task_id) is None
+    finally:
+        await handler.aclose()
+
+
+@pytest.mark.asyncio
+async def test_shutdown_drains_nonblocking_work_before_closing_dependencies(store):
+    async def execute(context, queue):
+        await queue.enqueue_event(
+            Task(
+                id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+            )
+        )
+        await asyncio.Event().wait()
+
+    handler = _handler(store, execute)
+    closed = []
+
+    async def upstream_close():
+        assert not handler._running_agents
+        assert not any(not task.done() for task in handler._background_tasks)
+        stored = await store.get(params.message.task_id, _context())
+        assert stored.status.state == TaskState.TASK_STATE_FAILED
+        closed.append("upstream")
+
+    deps = SimpleNamespace(startup=AsyncMock(), shutdown=AsyncMock())
+    lifespan = build_lifespan(
+        request_handler=handler,
+        database_engine=None,
+        task_store_runtime=deps,
+        runtime_state_runtime=deps,
+        client_manager=SimpleNamespace(close_all=AsyncMock()),
+        upstream_client=SimpleNamespace(close=upstream_close),
+    )
+    params = _request()
+    params.configuration.return_immediately = True
+    try:
+        async with lifespan(None):
+            await asyncio.wait_for(handler.on_message_send(params, _context()), 2)
+        assert closed == ["upstream"]
+        await handler.aclose()
+    finally:
+        await handler.aclose()
+
+
+@pytest.mark.asyncio
+async def test_get_task_rejects_empty_id_before_store_access():
+    store = AsyncMock()
+    handler = _handler(store, AsyncMock())
+    with pytest.raises(InvalidParamsError):
+        await handler.on_get_task(GetTaskRequest(id=""), _context())
+    store.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_stream_failure_after_disconnect_is_persisted_and_cleaned(store):
+    fail = asyncio.Event()
+
+    async def execute(context, queue):
+        await queue.enqueue_event(
+            Task(
+                id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+            )
+        )
+        await fail.wait()
+        raise RuntimeError("detached failure")
+
+    handler = _handler(store, execute)
+    params = _request()
+    stream = handler.on_message_send_stream(params, _context())
+    try:
+        await asyncio.wait_for(anext(stream), 2)
+        await stream.aclose()
+        assert handler._running_agents  # Disconnect must not cancel execution.
+        fail.set()
+        await asyncio.wait_for(asyncio.gather(*handler._background_tasks), 2)
+        stored = await store.get(params.message.task_id, _context())
+        assert stored.status.state == TaskState.TASK_STATE_FAILED
+        assert not handler._running_agents
+        assert not handler._managed_producers
+        assert await handler._queue_manager.get(params.message.task_id) is None
+    finally:
+        await stream.aclose()
+        await handler.aclose()
+
+
+@pytest.mark.asyncio
+async def test_stream_without_events_propagates_failure(store):
+    handler = _handler(store, AsyncMock(side_effect=RuntimeError("stream failure")))
+    params = _request()
+    try:
+        with pytest.raises(RuntimeError, match="stream failure"):
+            async for _ in handler.on_message_send_stream(params, _context()):
+                pass
+        await asyncio.gather(*handler._background_tasks)
+        stored = await store.get(params.message.task_id, _context())
+        assert stored.status.state == TaskState.TASK_STATE_FAILED
+        assert not handler._running_agents
+    finally:
+        await handler.aclose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "terminal", [TaskState.TASK_STATE_COMPLETED, TaskState.TASK_STATE_CANCELED]
+)
+async def test_failure_does_not_overwrite_terminal_task(store, terminal):
+    saved = None
+
+    async def execute(context, queue):
+        nonlocal saved
+        saved = Task(
+            id=context.task_id,
+            context_id=context.context_id,
+            status=TaskStatus(state=terminal),
+            metadata={"keep": "original"},
+        )
+        await store.save(saved, context.call_context)
+        raise RuntimeError("late failure")
+
+    handler = _handler(store, execute)
+    params = _request()
+    try:
+        with pytest.raises(RuntimeError):
+            await handler.on_message_send(params, _context())
+        assert await store.get(params.message.task_id, _context()) == saved
+        assert not handler._running_agents
+    finally:
+        await handler.aclose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["get", "save"])
+async def test_failure_to_persist_does_not_mask_producer_error_or_skip_cleanup(store, operation):
+    original = getattr(store, operation)
+
+    async def execute(context, queue):
+        setattr(store, operation, AsyncMock(side_effect=RuntimeError("storage unavailable")))
+        raise RuntimeError("original execution failure")
+
+    handler = _handler(store, execute)
+    params = _request()
+    try:
+        with pytest.raises(RuntimeError, match="original execution failure"):
+            await handler.on_message_send(params, _context())
+        assert not handler._running_agents
+        assert not handler._managed_producers
+        assert await handler._queue_manager.get(params.message.task_id) is None
+    finally:
+        setattr(store, operation, original)
+        await handler.aclose()
+
+
+@pytest.mark.asyncio
+async def test_close_waits_for_setup_and_rejects_new_requests(store):
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    handler = _handler(store, AsyncMock())
+    original = handler._request_context_builder.build
+
+    async def blocked_build(**kwargs):
+        entered.set()
+        await release.wait()
+        return await original(**kwargs)
+
+    handler._request_context_builder.build = blocked_build
+    setup = asyncio.create_task(handler._setup_message_execution(_request(), _context()))
+    await entered.wait()
+    close = asyncio.create_task(handler.aclose())
+    await asyncio.sleep(0)
+    assert not close.done()
+    release.set()
+    _, task_id, queue, _, producer = await setup
+    await asyncio.wait_for(close, 2)
+    assert producer.done()
+    assert queue.is_closed()
+    assert not handler._running_agents
+    assert not handler._managed_producers
+    assert await handler._queue_manager.get(task_id) is None
+    from a2a.types import InternalError
+
+    with pytest.raises(InternalError, match="shutting down"):
+        await handler.on_message_send(_request(), _context())
+    await handler.aclose()
+
+
+@pytest.mark.asyncio
+async def test_cancelled_close_waiter_still_waits_for_drain(store):
+    started = asyncio.Event()
+    cancelled = asyncio.Event()
+    release = asyncio.Event()
+
+    async def execute(context, queue):
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cancelled.set()
+            await release.wait()
+
+    handler = _handler(store, execute)
+    await handler._setup_message_execution(_request(), _context())
+    await started.wait()
+    close = asyncio.create_task(handler.aclose())
+    await cancelled.wait()
+    close.cancel()
+    await asyncio.sleep(0)
+    assert not close.done()
+    release.set()
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(close, 2)
+    assert not handler._running_agents
+    await handler.aclose()
+
+
+@pytest.mark.asyncio
+async def test_lifespan_attempts_all_cleanup_after_failure():
+    calls = []
+
+    def callback(name, fail=False):
+        async def run():
+            calls.append(name)
+            if fail:
+                raise RuntimeError(name)
+
+        return run
+
+    lifespan = build_lifespan(
+        request_handler=SimpleNamespace(aclose=callback("handler", fail=True)),
+        database_engine=SimpleNamespace(dispose=callback("engine")),
+        task_store_runtime=SimpleNamespace(
+            startup=callback("task-start"), shutdown=callback("task-stop")
+        ),
+        runtime_state_runtime=SimpleNamespace(
+            startup=callback("state-start"), shutdown=callback("state-stop")
+        ),
+        client_manager=SimpleNamespace(close_all=callback("client", fail=True)),
+        upstream_client=SimpleNamespace(close=callback("upstream")),
+    )
+    with pytest.raises(RuntimeError, match="client"):
+        async with lifespan(None):
+            pass
+    assert calls == [
+        "task-start",
+        "state-start",
+        "handler",
+        "client",
+        "upstream",
+        "state-stop",
+        "task-stop",
+        "engine",
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("history_length", [None, 0, 1, 10])
+async def test_get_task_preserves_stored_history_and_identity(store, history_length):
+    from a2a.types import TaskNotFoundError
+
+    context = _context()
+    original = Task(
+        id="history-task",
+        context_id="ctx",
+        status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
+        history=[
+            Message(message_id=f"msg-{i}", role=Role.ROLE_USER, parts=[Part(text=str(i))])
+            for i in range(3)
+        ],
+    )
+    await store.save(original, context)
+    handler = _handler(store, AsyncMock())
+    params = GetTaskRequest(id=original.id)
+    if history_length is not None:
+        params.history_length = history_length
+    result = await handler.on_get_task(params, context)
+    expected_length = 3 if history_length is None else min(3, history_length)
+    assert len(result.history) == expected_length
+    if history_length == 1:
+        assert result.history[0].message_id == "msg-2"
+    assert await store.get(original.id, context) == original
+    with pytest.raises(TaskNotFoundError):
+        await handler.on_get_task(params, _context("other"))
+
+
+@pytest.mark.asyncio
+async def test_get_task_rejects_negative_history_before_store_access():
+    store = AsyncMock()
+    handler = _handler(store, AsyncMock())
+    with pytest.raises(InvalidParamsError):
+        await handler.on_get_task(GetTaskRequest(id="task", history_length=-1), _context())
+    store.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_handler_close_cleans_other_tasks_when_queue_cleanup_fails(store):
+    started = asyncio.Event()
+
+    async def execute(context, queue):
+        started.set()
+        await asyncio.Event().wait()
+
+    handler = _handler(store, execute)
+    await handler._setup_message_execution(_request(), _context())
+    await started.wait()
+    background = asyncio.create_task(asyncio.Event().wait())
+    handler._track_background_task(background)
+    original = handler._queue_manager.close
+
+    async def failing_close(task_id):
+        await original(task_id)
+        raise RuntimeError("queue cleanup failed")
+
+    handler._queue_manager.close = failing_close
+    with pytest.raises(ExceptionGroup, match="Request handler cleanup failed"):
+        await handler.aclose()
+    assert background.done()
+    assert not handler._running_agents
+    assert not handler._managed_producers
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failed_start", ["task", "state"])
+async def test_partial_startup_still_closes_resources(failed_start):
+    task_runtime = SimpleNamespace(startup=AsyncMock(), shutdown=AsyncMock())
+    state_runtime = SimpleNamespace(startup=AsyncMock(), shutdown=AsyncMock())
+    runtime = task_runtime if failed_start == "task" else state_runtime
+    runtime.startup.side_effect = RuntimeError("startup failed")
+    handler, client, upstream, engine = AsyncMock(), AsyncMock(), AsyncMock(), AsyncMock()
+    lifespan = build_lifespan(
+        request_handler=handler,
+        database_engine=engine,
+        task_store_runtime=task_runtime,
+        runtime_state_runtime=state_runtime,
+        client_manager=client,
+        upstream_client=upstream,
+    )
+    with pytest.raises(RuntimeError, match="startup failed"):
+        async with lifespan(None):
+            pytest.fail("Startup must fail before yielding")
+    handler.aclose.assert_awaited_once()
+    client.close_all.assert_awaited_once()
+    upstream.close.assert_awaited_once()
+    engine.dispose.assert_awaited_once()
+    assert task_runtime.shutdown.await_count == (failed_start == "state")
+    state_runtime.shutdown.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_explicit_cancel_remains_canceled_and_idempotent(store):
+    from a2a.types import CancelTaskRequest, TaskStatusUpdateEvent
+
+    async def execute(context, queue):
+        await queue.enqueue_event(
+            Task(
+                id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+            )
+        )
+        await asyncio.Event().wait()
+
+    async def cancel(context, queue):
+        await queue.enqueue_event(
+            TaskStatusUpdateEvent(
+                task_id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_CANCELED),
+            )
+        )
+
+    handler = _handler(store, execute)
+    handler.agent_executor.cancel = cancel
+    params = _request()
+    params.configuration.return_immediately = True
+    try:
+        task = await handler.on_message_send(params, _context())
+        request = CancelTaskRequest(id=task.id)
+        result = await handler.on_cancel_task(request, _context())
+        assert result.status.state == TaskState.TASK_STATE_CANCELED
+        assert await handler.on_cancel_task(request, _context()) == result
+        await handler.aclose()
+        stored = await store.get(task.id, _context())
+        assert stored.status.state == TaskState.TASK_STATE_CANCELED
+    finally:
+        await handler.aclose()
+
+
+@pytest.mark.asyncio
+async def test_app_wires_shutdown_and_get_task_validation(monkeypatch):
+    import httpx
+
+    from opencode_a2a.server.application import create_app
+
+    app = create_app(make_settings(a2a_task_store_backend="memory"))
+    handler = app.state._jsonrpc_app._http_handler
+
+    async def execute(context, queue):
+        await queue.enqueue_event(
+            Task(
+                id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+            )
+        )
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(app.state.agent_executor, "execute", execute)
+    async with app.router.lifespan_context(app):
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),
+            base_url="http://test",
+            headers={"Authorization": "Bearer test-token"},
+        ) as client:
+            invalid = await client.post(
+                "/",
+                json={
+                    "jsonrpc": "2.0",
+                    "id": 1,
+                    "method": "GetTask",
+                    "params": {"id": ""},
+                },
+            )
+            assert invalid.json()["error"]["code"] == -32602
+            response = await client.post(
+                "/message:send",
+                json={
+                    "message": {
+                        "messageId": "app-request",
+                        "role": "ROLE_USER",
+                        "parts": [{"text": "hello"}],
+                    },
+                    "configuration": {"returnImmediately": True},
+                },
+            )
+            assert response.status_code == 200, response.text
+            assert handler._running_agents
+    assert not handler._running_agents
+    assert not handler._managed_producers
+    assert not any(not task.done() for task in handler._background_tasks)
+
+
+@pytest.mark.asyncio
+async def test_shutdown_drains_background_tasks_created_during_cancellation(store):
+    handler = _handler(store, AsyncMock())
+    started = asyncio.Event()
+    late_tasks = []
+
+    async def consume():
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            late = asyncio.create_task(asyncio.Event().wait())
+            late_tasks.append(late)
+            handler._track_background_task(late)
+
+    consumer = asyncio.create_task(consume())
+    handler._track_background_task(consumer)
+    await started.wait()
+    await asyncio.wait_for(asyncio.gather(handler.aclose(), handler.aclose()), 2)
+    assert consumer.done()
+    assert len(late_tasks) == 1
+    assert late_tasks[0].done()
+    assert not any(not task.done() for task in handler._background_tasks)

From 805c913097bd91c023059407607d7f790d1126ba Mon Sep 17 00:00:00 2001
From: "Liu Juanjuan (Helen)"
 <31027645+liujuanjuan1984@users.noreply.github.com>
Date: Tue, 29 Sep 2026 12:41:23 +0000
Subject: [PATCH 2/5] fix(server): preserve terminal drains and shutdown
 barriers (#533)

---
 docs/compatibility.md                        |   3 +-
 docs/guide.md                                |  11 +-
 src/opencode_a2a/server/application.py       |  77 ++++++-------
 src/opencode_a2a/server/handler_lifecycle.py |  38 +++++--
 tests/conftest.py                            |  20 +++-
 tests/server/test_handler_lifecycle.py       | 107 +++++++++++++++++++
 6 files changed, 202 insertions(+), 54 deletions(-)

diff --git a/docs/compatibility.md b/docs/compatibility.md
index 3418ea8..893569f 100644
--- a/docs/compatibility.md
+++ b/docs/compatibility.md
@@ -40,7 +40,8 @@ request identity, including when no task row exists yet. Existing terminal
 snapshots remain immutable. Failure to persist is logged without storage error
 payloads and does not replace the execution error or skip queue cleanup. Both
 send paths explicitly check completed producers so queue closure cannot hide an
-execution exception. `GetTask` delegates parameter validation and history slicing
+execution exception; producer storage errors still use the adapter's stable error
+mapping. `GetTask` delegates parameter validation and history slicing
 to the pinned SDK, retaining adapter output negotiation and storage-error mapping.
 Empty IDs and negative history lengths are rejected as invalid parameters.
 
diff --git a/docs/guide.md b/docs/guide.md
index c3ee8f3..e5566c1 100644
--- a/docs/guide.md
+++ b/docs/guide.md
@@ -229,10 +229,13 @@ continue to use the executor's existing error events.
 At application shutdown, the handler rejects new execution setup and waits for
 setup already in progress to register its producers. It cancels and awaits local
 producers, marks interrupted nonterminal executions `FAILED` when storage is
-available, and drains/cancels tracked background consumers and cleanup tasks before
-closing outbound clients, the OpenCode HTTP client, state stores, and the database
-engine. Cleanup of the remaining resources is attempted even if a close operation
-fails. Shutdown is idempotent and does not wait for model generation to finish,
+available, and lets tracked consumers finish persisting buffered output from closed
+queues before cancelling other background cleanup tasks. This also preserves a
+completed producer's terminal output when its consumer is still writing to storage.
+Only then does it close outbound clients, the OpenCode HTTP client, state stores,
+and the database engine. Cleanup of the remaining resources is attempted even if a close operation
+fails. Repeated cancellation of the shutdown waiter does not bypass this ordering.
+Shutdown is idempotent and does not wait for model generation to finish,
 though executor cleanup and storage I/O must still complete. Supervisor shutdown
 budgets must allow that cleanup; a hard process kill cannot provide these guarantees.
 
diff --git a/src/opencode_a2a/server/application.py b/src/opencode_a2a/server/application.py
index cca1f8d..32f26a9 100644
--- a/src/opencode_a2a/server/application.py
+++ b/src/opencode_a2a/server/application.py
@@ -673,7 +673,7 @@ async def on_message_send_stream(self, params, context=None):
                 )
             )
             detached_task.set_name(f"continue_stream_after_disconnect:{task_id}")
-            self._track_background_task(detached_task)
+            self._track_consumer_task(detached_task)
             stream_detached = True
             raise
         finally:
@@ -709,22 +709,47 @@ async def on_message_send(self, params, context=None):
         interrupted_or_non_blocking = False
         bg_consume_task: asyncio.Task | None = None
         try:
+            try:
 
-            async def push_notification_callback() -> None:
-                await self._send_push_notification_if_needed(task_id, result_aggregator)
-
-            (
-                result,
-                interrupted_or_non_blocking,
-                bg_consume_task,
-            ) = await result_aggregator.consume_and_break_on_interrupt(
-                consumer,
-                blocking=blocking,
-                event_callback=push_notification_callback,
-            )
-            if bg_consume_task is not None:
-                bg_consume_task.set_name(f"continue_consuming:{task_id}")
-                self._track_background_task(bg_consume_task)
+                async def push_notification_callback() -> None:
+                    await self._send_push_notification_if_needed(task_id, result_aggregator)
+
+                (
+                    result,
+                    interrupted_or_non_blocking,
+                    bg_consume_task,
+                ) = await result_aggregator.consume_and_break_on_interrupt(
+                    consumer,
+                    blocking=blocking,
+                    event_callback=push_notification_callback,
+                )
+                if bg_consume_task is not None:
+                    bg_consume_task.set_name(f"continue_consuming:{task_id}")
+                    self._track_consumer_task(bg_consume_task)
+            finally:
+                if interrupted_or_non_blocking:
+                    cleanup_task = asyncio.create_task(
+                        self._cleanup_producer(producer_task, task_id)
+                    )
+                    cleanup_task.set_name(f"cleanup_producer:{task_id}")
+                    self._track_background_task(cleanup_task)
+                else:
+                    try:
+                        current_task = asyncio.current_task()
+                        if current_task is not None and current_task.cancelled():
+                            logger.debug(
+                                "Client disconnected from message request. Cancelling task %s",
+                                task_id,
+                            )
+                            producer_task.cancel()
+                            await queue.close(immediate=True)
+
+                        await asyncio.shield(self._cleanup_producer(producer_task, task_id))
+                    except asyncio.CancelledError:
+                        pass
+
+            if producer_task.done() and not producer_task.cancelled():
+                producer_task.result()
         except TaskStoreOperationError as exc:
             logger.exception(
                 "Task store operation failed during SendMessage task_id=%s operation=%s",
@@ -739,27 +764,7 @@ async def push_notification_callback() -> None:
         except Exception:
             logger.exception("Agent execution failed")
             raise
-        finally:
-            if interrupted_or_non_blocking:
-                cleanup_task = asyncio.create_task(self._cleanup_producer(producer_task, task_id))
-                cleanup_task.set_name(f"cleanup_producer:{task_id}")
-                self._track_background_task(cleanup_task)
-            else:
-                try:
-                    current_task = asyncio.current_task()
-                    if current_task is not None and current_task.cancelled():
-                        logger.debug(
-                            "Client disconnected from message request. Cancelling task %s", task_id
-                        )
-                        producer_task.cancel()
-                        await queue.close(immediate=True)
-
-                    await asyncio.shield(self._cleanup_producer(producer_task, task_id))
-                except asyncio.CancelledError:
-                    pass
 
-        if producer_task.done() and not producer_task.cancelled():
-            producer_task.result()
         if not result:
             raise InternalError()
 
diff --git a/src/opencode_a2a/server/handler_lifecycle.py b/src/opencode_a2a/server/handler_lifecycle.py
index 0efd5fa..dfc535f 100644
--- a/src/opencode_a2a/server/handler_lifecycle.py
+++ b/src/opencode_a2a/server/handler_lifecycle.py
@@ -50,6 +50,7 @@ def __init__(  # noqa: PLR0913
         self._closing = False
         self._close_task: asyncio.Task[None] | None = None
         self._managed_producers: dict[asyncio.Task, str] = {}
+        self._draining_consumers: set[asyncio.Task] = set()
 
     async def _persist_execution_failure(self, request: RequestContext) -> None:
         """Write independently of the closing queue, preserving terminal snapshots."""
@@ -110,9 +111,15 @@ async def _cleanup_producer(self, producer_task: asyncio.Task, task_id: str) ->
                         self._running_agents.pop(task_id, None)
                 self._managed_producers.pop(producer_task, None)
 
+    def _track_consumer_task(self, task: asyncio.Task) -> None:
+        # These consumers can still be persisting output after the producer exits.
+        self._draining_consumers.add(task)
+        task.add_done_callback(self._draining_consumers.discard)
+        self._track_background_task(task)
+
     def _track_background_task(self, task: asyncio.Task) -> None:
         super()._track_background_task(task)
-        if self._closing:
+        if self._closing and task not in self._draining_consumers:
             task.cancel()
 
     async def aclose(self) -> None:
@@ -120,12 +127,18 @@ async def aclose(self) -> None:
         if self._close_task is None:
             self._closing = True
             self._close_task = asyncio.create_task(self._drain_execution())
-        try:
-            await asyncio.shield(self._close_task)
-        except asyncio.CancelledError:
-            # Do not allow lifespan cancellation to close dependencies mid-drain.
-            await asyncio.shield(self._close_task)
-            raise
+        cancelled = False
+        while True:
+            try:
+                await asyncio.shield(self._close_task)
+                break
+            except asyncio.CancelledError:
+                if self._close_task.cancelled():
+                    raise
+                # Repeated cancellation must not let lifespan close dependencies early.
+                cancelled = True
+        if cancelled:
+            raise asyncio.CancelledError
 
     async def _drain_execution(self) -> None:
         # A setup already in flight must register its producer before the snapshot.
@@ -140,10 +153,15 @@ async def _drain_execution(self) -> None:
             *(self._cleanup_producer(task, task_id) for task, task_id in producers.items()),
             return_exceptions=True,
         )
-        # Cancellation of a consumer may itself register a deferred cleanup.
-        while pending := [task for task in self._background_tasks if not task.done()]:
+        # Queues are closed, so consumers finish naturally after persisting buffered
+        # output. Cancelling them here could leave a successful task stuck WORKING.
+        # Other background work can be cancelled and may register further cleanup.
+        while pending := [
+            task for task in self._background_tasks | self._draining_consumers if not task.done()
+        ]:
             for task in pending:
-                task.cancel()
+                if task not in self._draining_consumers:
+                    task.cancel()
             await asyncio.gather(*pending, return_exceptions=True)
         errors = [result for result in results if isinstance(result, Exception)]
         if errors:
diff --git a/tests/conftest.py b/tests/conftest.py
index eb13b14..62380ef 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import AsyncGenerator
+from contextlib import AsyncExitStack
 
 import pytest
 import pytest_asyncio
@@ -8,10 +9,12 @@
 
 
 @pytest_asyncio.fixture(autouse=True)
-async def dispose_app_database_engines(monkeypatch: pytest.MonkeyPatch) -> AsyncGenerator[None]:
+async def dispose_app_resources(monkeypatch: pytest.MonkeyPatch) -> AsyncGenerator[None]:
     import opencode_a2a.server.application as app_module
 
     tracked_engines: dict[int, AsyncEngine] = {}
+    tracked_handlers: list[app_module.OpencodeRequestHandler] = []
+    original_handler = app_module.OpencodeRequestHandler
     original_build_database_engine = app_module.build_database_engine
 
     def _build_database_engine(settings):  # noqa: ANN001
@@ -19,8 +22,19 @@ def _build_database_engine(settings):  # noqa: ANN001
         tracked_engines[id(engine)] = engine
         return engine
 
+    def _build_handler(*args, **kwargs):  # noqa: ANN002, ANN003
+        handler = original_handler(*args, **kwargs)
+        tracked_handlers.append(handler)
+        return handler
+
     monkeypatch.setattr(app_module, "build_database_engine", _build_database_engine)
+    monkeypatch.setattr(app_module, "OpencodeRequestHandler", _build_handler)
     yield
 
-    for engine in tracked_engines.values():
-        await engine.dispose()
+    # ASGITransport does not run lifespan. Match the application's shutdown order
+    # so background writes cannot keep checked-out connections past loop teardown.
+    async with AsyncExitStack() as cleanup:
+        for engine in tracked_engines.values():
+            cleanup.push_async_callback(engine.dispose)
+        for handler in tracked_handlers:
+            cleanup.push_async_callback(handler.aclose)
diff --git a/tests/server/test_handler_lifecycle.py b/tests/server/test_handler_lifecycle.py
index ebf3162..9ddb440 100644
--- a/tests/server/test_handler_lifecycle.py
+++ b/tests/server/test_handler_lifecycle.py
@@ -561,3 +561,110 @@ async def consume():
     assert len(late_tasks) == 1
     assert late_tasks[0].done()
     assert not any(not task.done() for task in handler._background_tasks)
+
+
+@pytest.mark.asyncio
+async def test_repeated_cancellation_cannot_bypass_shutdown_barrier(store):
+    started = asyncio.Event()
+    stopping = asyncio.Event()
+    release = asyncio.Event()
+
+    async def execute(context, queue):
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            stopping.set()
+            await release.wait()
+
+    handler = _handler(store, execute)
+    await handler._setup_message_execution(_request(), _context())
+    await started.wait()
+    close = asyncio.create_task(handler.aclose())
+    try:
+        await stopping.wait()
+        for _ in range(3):
+            close.cancel()
+            await asyncio.sleep(0)
+        assert not close.done()
+    finally:
+        release.set()
+        await asyncio.gather(close, return_exceptions=True)
+        await handler.aclose()
+    assert not handler._managed_producers
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streaming", [False, True])
+async def test_producer_storage_error_uses_stable_error_mapping(store, streaming):
+    from opencode_a2a.server.task_store import TaskStoreOperationError
+
+    handler = _handler(store, AsyncMock(side_effect=TaskStoreOperationError("save", "task")))
+    params = _request()
+    try:
+        if streaming:
+            events = [event async for event in handler.on_message_send_stream(params, _context())]
+            failed = events[-1]
+        else:
+            failed = await handler.on_message_send(params, _context())
+        assert failed.status.state == TaskState.TASK_STATE_FAILED
+        assert failed.metadata["opencode"]["error"]["type"] == "TASK_STORE_UNAVAILABLE"
+        assert failed.metadata["opencode"]["error"]["operation"] == "save"
+    finally:
+        await handler.aclose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streaming", [False, True])
+async def test_shutdown_drains_completed_producer_output_before_closing_store(
+    store, monkeypatch, streaming
+):
+    from a2a.types import TaskStatusUpdateEvent
+
+    writing_terminal = asyncio.Event()
+    release_write = asyncio.Event()
+    original_save = store.save
+
+    async def delayed_save(task, context=None):
+        if task.status.state == TaskState.TASK_STATE_COMPLETED:
+            writing_terminal.set()
+            await release_write.wait()
+        await original_save(task, context)
+
+    monkeypatch.setattr(store, "save", delayed_save)
+
+    async def execute(context, queue):
+        await queue.enqueue_event(
+            Task(
+                id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+            )
+        )
+        await queue.enqueue_event(
+            TaskStatusUpdateEvent(
+                task_id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
+            )
+        )
+
+    handler = _handler(store, execute)
+    params = _request()
+    params.configuration.return_immediately = True
+    if streaming:
+        stream = handler.on_message_send_stream(params, _context())
+        await anext(stream)
+        await stream.aclose()
+    else:
+        await handler.on_message_send(params, _context())
+    await writing_terminal.wait()
+    close = asyncio.create_task(handler.aclose())
+    try:
+        with pytest.raises(TimeoutError):
+            await asyncio.wait_for(asyncio.shield(close), 0.05)
+    finally:
+        release_write.set()
+        await close
+    saved = await store.get(params.message.task_id, _context())
+    assert saved.status.state == TaskState.TASK_STATE_COMPLETED

From c69d5639da74c4ed3d0f31503c9e19c8fcc3bae9 Mon Sep 17 00:00:00 2001
From: "Liu Juanjuan (Helen)"
 <31027645+liujuanjuan1984@users.noreply.github.com>
Date: Tue, 29 Sep 2026 12:57:39 +0000
Subject: [PATCH 3/5] fix(storage): make historical database upgrades
 reproducible (#533)

---
 README.md                                     |   4 +
 docs/compatibility.md                         |  22 +-
 docs/guide.md                                 |  56 ++++-
 pyproject.toml                                |   3 +
 src/opencode_a2a/server/task_store.py         |   2 +
 .../server/task_store_sdk_compat.py           |  59 +++++-
 tests/server/test_task_store_upgrade.py       | 192 ++++++++++++++++++
 uv.lock                                       | 111 +++++++++-
 8 files changed, 440 insertions(+), 9 deletions(-)
 create mode 100644 tests/server/test_task_store_upgrade.py

diff --git a/README.md b/README.md
index 7b52ff5..19e7ec2 100644
--- a/README.md
+++ b/README.md
@@ -59,6 +59,10 @@ Upgrade later with:
 uv tool upgrade opencode-a2a
 ```
 
+If you retain an older SQLite database, follow
+[Upgrading an Existing Database](docs/guide.md#upgrading-an-existing-database)
+for version-matched tooling, task ownership, and rollback instructions.
+
 Make sure provider credentials and a default model are configured on the OpenCode side, then start OpenCode:
 
 ```bash
diff --git a/docs/compatibility.md b/docs/compatibility.md
index 893569f..8ce448f 100644
--- a/docs/compatibility.md
+++ b/docs/compatibility.md
@@ -53,14 +53,30 @@ execution and persistence.
 SDK 1.1.5 still provides `EventQueueLegacy`, but documents its future removal.
 Replacing one queue constructor or changing the handler's parent class is not a
 complete migration: the legacy queue manager, consumer, cancellation, subscription,
-and output aggregation paths are coupled. Retain the pinned implementation until
-a deliberate SDK upgrade or a separately reviewed migration replaces these paths.
-Before accepting an SDK version that removes legacy interfaces, prove parity for
+and output aggregation paths are coupled. Full replacement remains unfinished
+code migration work, tracked in [#533](https://github.com/Intelligent-Internet/opencode-a2a/issues/533).
+It is not a database migration and is not waiting on production observation.
+The pinned SDK still supports the current implementation; this lifecycle patch
+does not upgrade that SDK or claim to remove its legacy dependency.
+
+In SDK 1.1.5, the newer handler ignores a second full `Task` snapshot after the
+initial task, whereas current executor paths use such snapshots. Its forced
+`aclose()` cancels consumers and does not itself persist interrupted execution as
+FAILED. Isolated probes with an initial WORKING task reproduce a retained WORKING
+state both after a final full COMPLETED task and after active execution shutdown.
+Migration must adapt these event and lifecycle contracts, not just swap imports.
+Before completing the migration or accepting an SDK that removes legacy APIs, prove parity for
 identity isolation, output/extension negotiation, cancel idempotency, first-terminal
 persistence, nonblocking sends, subscription, and disconnect/shutdown behavior on
 Python 3.11–3.14 with both memory and SQLite stores. Do not copy the SDK's queue
 implementation into this repository merely to suppress the deprecation.
 
+User upgrades are a separate contract: ship compatible code, version-matched
+migration tooling, and automated tests against historical data. See
+[Upgrading an Existing Database](guide.md#upgrading-an-existing-database).
+The acceptance gate is reproducible repository validation, not access to a live
+deployment. `migrations` is an optional install extra tied to the runtime SDK pin.
+
 ## Contract Honesty
 
 Machine-readable discovery surfaces must reflect actual runtime behavior:
diff --git a/docs/guide.md b/docs/guide.md
index e5566c1..142f0a0 100644
--- a/docs/guide.md
+++ b/docs/guide.md
@@ -269,7 +269,7 @@ Database-backed task persistence also keeps the existing first-terminal-state-wi
 
 At startup, the runtime logs a concise persistence summary covering the active backend, the redacted database URL when applicable, the shared persistence scope, and whether the SQLite local durability profile is active.
 
-The adapter-owned state tables listed above remain managed by the internal migration runner. The SDK-owned `tasks` table does not use runtime auto-migration here; upgrade existing SDK task schemas explicitly with upstream `a2a-db` before starting the service after an SDK schema change. If `a2a-db` is unavailable in your environment, install the `a2a-sdk[db-cli]` extra first.
+The adapter-owned state tables listed above remain managed by the internal migration runner. The SDK-owned `tasks` table does not use runtime **schema** auto-migration here. Upgrade an old SDK task schema with the matching release's migration tool as described below. Once the schema is compatible, startup normalizes legacy task payloads before accepting requests.
 
 In-flight asyncio locks, outbound A2A client caches, and stream-local aggregation buffers remain process-local runtime state.
 
@@ -279,6 +279,60 @@ To opt into an ephemeral development profile, set:
 A2A_TASK_STORE_BACKEND=memory
 ```
 
+### Upgrading an Existing Database
+
+Queue/handler implementation changes concern process-local Python objects; they
+do not require a database migration. This lifecycle update keeps SDK 1.1.5 and
+does not introduce a new schema. A fresh installation or the `memory` backend
+needs no migration command. An existing compatible v1 database also needs no
+manual schema change.
+
+For an older SQLite database that startup rejects for missing `owner`,
+`last_updated`, `protocol_version`, or the owner/time index:
+
+1. Stop the application, make a consistent backup of its SQLite database, and
+   keep that backup with the previous application version for rollback. Use an
+   absolute database path to avoid upgrading a different file by accident.
+2. Install the target release. Run its migration tooling using the **same
+   application version**, so the runtime and migrations use the same pinned SDK:
+
+   ```bash
+   OPENCODE_A2A_VERSION="$(opencode-a2a --version | awk '{print $2}')"
+   uvx --from "opencode-a2a[migrations]==${OPENCODE_A2A_VERSION}" a2a-db upgrade head \
+     --database-url 'sqlite+aiosqlite:////absolute/path/opencode-a2a.db' \
+     --add_columns_owner_last_updated-default-owner automation
+   ```
+
+   For a source checkout, use `uv run --extra migrations a2a-db` with the same
+   arguments. For an existing Python virtual environment, install that release
+   with the `opencode-a2a[migrations]` extra and run its `a2a-db` executable.
+   Do not install a floating SDK release separately into the runtime environment.
+
+   Replace `automation` with the intended bearer credential's `principal` or
+   Basic credential's username. The upstream default,
+   `legacy_v03_no_user_info`, does not match the normal authenticated identity
+   and can make old tasks appear missing. This option assigns **all rows lacking
+   an owner column** to one identity; it cannot infer historical ownership.
+   If the old data needs multiple owners, establish the ownership mapping before
+   migration. Existing owner columns are preserved, and rerunning the command
+   with another owner does not reassign them.
+3. Start the new application. Before serving requests, it converts rows marked
+   `0.3` or with no protocol marker to the SDK's v1 task representation, preserves
+   owners and task contents, and backfills missing query timestamps from task
+   status timestamps. Rows without any historical timestamp remain undated.
+   Conversion uses one transaction, reads in bounded batches, and skips existing
+   v1 rows on subsequent starts. Malformed legacy payloads fail startup and roll
+   back the payload conversion; the completed schema upgrade remains in place.
+4. Verify a known task with `GetTask` and `ListTasks` under its original identity,
+   including status/time filters. Terminal snapshots remain immutable. To roll
+   back to an older application, restore the stopped database backup rather than
+   relying on a schema downgrade to undo converted JSON payloads.
+
+These upgrade paths are tested against disposable historical databases in
+`tests/server/test_task_store_upgrade.py`; no running deployment or live database
+is required for repository acceptance. Neither queue replacement nor payload
+conversion resumes executions that were interrupted before this release.
+
 ## Troubleshooting Provider Auth State
 
 If one deployment works while another fails against the same upstream provider, check the deployed OpenCode user's local state before assuming the difference comes from the `opencode-a2a` package itself.
diff --git a/pyproject.toml b/pyproject.toml
index d9ed8f9..41de429 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -43,6 +43,9 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
+migrations = [
+    "a2a-sdk[db-cli]==1.1.5",
+]
 dev = [
     "mypy>=1.19.1,<3.0",
     "pip-audit>=2.9,<3.0",
diff --git a/src/opencode_a2a/server/task_store.py b/src/opencode_a2a/server/task_store.py
index 612ffb2..243093f 100644
--- a/src/opencode_a2a/server/task_store.py
+++ b/src/opencode_a2a/server/task_store.py
@@ -300,6 +300,7 @@ async def _startup() -> None:
         compat = DatabaseTaskStoreCompat(raw_task_store)
         await compat.validate_schema()
         await compat.initialize()
+        await compat.migrate_legacy_rows()
 
     async def _shutdown() -> None:
         if engine is None:
@@ -352,6 +353,7 @@ async def initialize_task_store(task_store: TaskStore) -> None:
         compat = DatabaseTaskStoreCompat(raw_task_store)
         await compat.validate_schema()
         await compat.initialize()
+        await compat.migrate_legacy_rows()
         return
     initialize = getattr(task_store, "initialize", None)
     if callable(initialize):
diff --git a/src/opencode_a2a/server/task_store_sdk_compat.py b/src/opencode_a2a/server/task_store_sdk_compat.py
index ab64f43..0e0d6aa 100644
--- a/src/opencode_a2a/server/task_store_sdk_compat.py
+++ b/src/opencode_a2a/server/task_store_sdk_compat.py
@@ -9,13 +9,20 @@
 from a2a.server.tasks.database_task_store import DatabaseTaskStore
 from a2a.types import Task, TaskState
 from google.protobuf.json_format import MessageToDict, ParseDict
-from sqlalchemy import inspect, or_, select
+from sqlalchemy import inspect, or_, select, update
 from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 
 from ..task_states import TERMINAL_TASK_STATES
 from .database import redact_database_url_for_logs
 
-_TERMINAL_TASK_STATE_VALUES = tuple(TaskState.Name(int(state)) for state in TERMINAL_TASK_STATES)
+_TERMINAL_TASK_STATE_VALUES = tuple(
+    value
+    for state in TERMINAL_TASK_STATES
+    for value in (
+        TaskState.Name(int(state)),
+        TaskState.Name(int(state)).removeprefix("TASK_STATE_").lower(),
+    )
+)
 _REQUIRED_TASK_MODEL_COLUMNS = frozenset(
     {
         "id",
@@ -57,6 +64,47 @@ def __init__(self, task_store: DatabaseTaskStore) -> None:
     async def initialize(self) -> None:
         await self._task_store.initialize()
 
+    async def migrate_legacy_rows(self) -> None:
+        """Normalize historical payloads before serving queries, in one transaction.
+
+        The SDK schema migrations retain v0.3 JSON and leave last_updated empty.
+        Read-time conversion alone cannot fix SQL status/time filters. The protocol
+        marker makes this resumable and leaves current rows and ownership unchanged.
+        """
+        model = self._shape.task_model
+        table = model.__table__
+        async with self._shape.session_maker.begin() as session:
+            after_id: str | None = None
+            while True:
+                statement = select(model).where(
+                    or_(model.protocol_version.is_(None), model.protocol_version == "0.3")
+                )
+                if after_id is not None:
+                    statement = statement.where(model.id > after_id)
+                rows = (
+                    (await session.execute(statement.order_by(model.id).limit(100))).scalars().all()
+                )
+                if not rows:
+                    return
+                for row in rows:
+                    try:
+                        task = _task_model_to_core(
+                            row,
+                            model_to_core_conversion=self._shape.model_to_core_conversion,
+                        )
+                    except Exception:
+                        raise TaskStoreSchemaCompatibilityError(
+                            "Cannot convert a legacy SDK task payload; restore the pre-upgrade "
+                            "backup or repair the legacy data before restarting."
+                        ) from None
+                    values = _task_row_values(task, owner=row.owner)
+                    if row.last_updated is not None:
+                        values["last_updated"] = row.last_updated
+                    await session.execute(
+                        update(table).where(table.c.id == row.id).values(**values)
+                    )
+                after_id = rows[-1].id
+
     async def validate_schema(self) -> None:
         database_url = redact_database_url_for_logs(
             self._task_store.engine.url.render_as_string(hide_password=True)
@@ -249,6 +297,9 @@ def _validate_sdk_task_table_schema(
     raise TaskStoreSchemaCompatibilityError(
         f"Legacy SDK task table schema detected for '{table_name}' "
         f"({'; '.join(details)}). Run "
-        f"`a2a-db --database-url {database_url}` before starting the service. "
-        "If `a2a-db` is unavailable, install the `a2a-sdk[db-cli]` extra first."
+        "the matching release's `a2a-db upgrade head` before starting the service; "
+        "install this release with the `opencode-a2a[migrations]` extra. "
+        "When adding owner, explicitly set "
+        "`--add_columns_owner_last_updated-default-owner` to the intended auth principal. "
+        f"Database: {database_url}. See docs/guide.md#upgrading-an-existing-database."
     )
diff --git a/tests/server/test_task_store_upgrade.py b/tests/server/test_task_store_upgrade.py
new file mode 100644
index 0000000..bfbc6cc
--- /dev/null
+++ b/tests/server/test_task_store_upgrade.py
@@ -0,0 +1,192 @@
+from __future__ import annotations
+
+import json
+import sqlite3
+import subprocess
+import sys
+from contextlib import closing
+from pathlib import Path
+
+import pytest
+from a2a.server.context import ServerCallContext
+from a2a.types import ListTasksRequest, Task, TaskState, TaskStatus
+
+from opencode_a2a.server.task_store import build_task_store_runtime, unwrap_task_store
+from opencode_a2a.server.task_store_sdk_compat import TaskStoreSchemaCompatibilityError
+from tests.support.settings import make_settings
+
+
+def _legacy_database(path: Path, state: str) -> None:
+    with closing(sqlite3.connect(path)) as connection, connection:
+        connection.execute(
+            "CREATE TABLE tasks (id VARCHAR(36) PRIMARY KEY, context_id VARCHAR(36), "
+            "kind VARCHAR(16), status JSON, artifacts JSON, history JSON, metadata JSON)"
+        )
+        connection.execute(
+            "INSERT INTO tasks VALUES (?, ?, ?, ?, ?, ?, ?)",
+            (
+                "legacy-task",
+                "legacy-context",
+                "task",
+                json.dumps({"state": state, "timestamp": "2026-01-02T03:04:05Z"}),
+                json.dumps([{"artifactId": "result", "parts": [{"kind": "text", "text": "kept"}]}]),
+                json.dumps(
+                    [
+                        {
+                            "kind": "message",
+                            "messageId": "input",
+                            "role": "user",
+                            "parts": [{"kind": "text", "text": "hello"}],
+                        }
+                    ]
+                ),
+                json.dumps({"opencode": {"session_id": "preserved-session"}}),
+            ),
+        )
+
+
+def _upstream_upgrade(path: Path) -> None:
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            "from a2a.a2a_db_cli import run_migrations; run_migrations()",
+            "upgrade",
+            "head",
+            "--database-url",
+            f"sqlite+aiosqlite:///{path}",
+            "--add_columns_owner_last_updated-default-owner",
+            "automation",
+        ],
+        capture_output=True,
+        text=True,
+        timeout=30,
+    )
+    assert result.returncode == 0, result.stderr
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("state", ["completed", "failed", "canceled", "rejected"])
+@pytest.mark.parametrize("normalize_rows", [False, True])
+async def test_upstream_migrated_terminal_task_cannot_be_overwritten(
+    tmp_path: Path, state: str, normalize_rows: bool
+) -> None:
+    path = tmp_path / "legacy.db"
+    _legacy_database(path, state)
+    _upstream_upgrade(path)
+    runtime = build_task_store_runtime(
+        make_settings(a2a_task_store_database_url=f"sqlite+aiosqlite:///{path}")
+    )
+    context = ServerCallContext(state={"identity": "automation"})
+    try:
+        if normalize_rows:
+            await runtime.startup()
+        else:
+            # Protect callers of the store factory even before lifespan normalization.
+            await unwrap_task_store(runtime.task_store).initialize()
+        before = await runtime.task_store.get("legacy-task", context)
+        assert before is not None
+        await runtime.task_store.save(
+            Task(
+                id=before.id,
+                context_id=before.context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+            ),
+            context,
+        )
+        assert await runtime.task_store.get(before.id, context) == before
+    finally:
+        await runtime.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_upgrade_preserves_contents_owner_and_is_repeatable(tmp_path: Path) -> None:
+    path = tmp_path / "legacy.db"
+    _legacy_database(path, "completed")
+    _upstream_upgrade(path)
+    _upstream_upgrade(path)
+    settings = make_settings(a2a_task_store_database_url=f"sqlite+aiosqlite:///{path}")
+    context = ServerCallContext(state={"identity": "automation"})
+    other_context = ServerCallContext(state={"identity": "another-user"})
+    runtime = build_task_store_runtime(settings)
+    try:
+        before = await runtime.task_store.get("legacy-task", context)
+        assert before is not None
+        assert before.artifacts[0].parts[0].text == "kept"
+        assert before.history[0].parts[0].text == "hello"
+        await runtime.startup()
+        assert await runtime.task_store.get("legacy-task", context) == before
+        assert await runtime.task_store.get("legacy-task", other_context) is None
+        assert not (await runtime.task_store.list(ListTasksRequest(), other_context)).tasks
+    finally:
+        await runtime.shutdown()
+
+    with closing(sqlite3.connect(path)) as connection:
+        after_first = connection.execute("SELECT * FROM tasks").fetchall()
+        assert connection.execute("SELECT protocol_version FROM tasks").fetchone() == ("1.0",)
+
+    restarted = build_task_store_runtime(settings)
+    try:
+        await restarted.startup()
+        assert await restarted.task_store.get("legacy-task", context) == before
+    finally:
+        await restarted.shutdown()
+    with closing(sqlite3.connect(path)) as connection:
+        assert connection.execute("SELECT * FROM tasks").fetchall() == after_first
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("invalid_last_row", [False, True])
+async def test_legacy_row_conversion_batches_and_rolls_back_as_one_transaction(
+    tmp_path: Path, invalid_last_row: bool
+) -> None:
+    path = tmp_path / "legacy.db"
+    _legacy_database(path, "working")
+    with closing(sqlite3.connect(path)) as connection, connection:
+        for index in range(101):
+            connection.execute(
+                "INSERT INTO tasks SELECT ?, context_id, kind, status, artifacts, history, "
+                "metadata FROM tasks WHERE id = 'legacy-task'",
+                (f"task-{index:03}",),
+            )
+        if invalid_last_row:
+            connection.execute(
+                "UPDATE tasks SET status = ? WHERE id = 'task-100'",
+                (json.dumps({"state": "invalid-state"}),),
+            )
+    _upstream_upgrade(path)
+    runtime = build_task_store_runtime(
+        make_settings(a2a_task_store_database_url=f"sqlite+aiosqlite:///{path}")
+    )
+    try:
+        if invalid_last_row:
+            with pytest.raises(TaskStoreSchemaCompatibilityError, match="legacy SDK task payload"):
+                await runtime.startup()
+        else:
+            await runtime.startup()
+    finally:
+        await runtime.shutdown()
+    with closing(sqlite3.connect(path)) as connection:
+        count = connection.execute(
+            "SELECT COUNT(*) FROM tasks WHERE protocol_version = '1.0'"
+        ).fetchone()[0]
+        assert count == (0 if invalid_last_row else 102)
+
+
+@pytest.mark.asyncio
+async def test_migrated_task_supports_status_and_timestamp_filters(tmp_path: Path) -> None:
+    path = tmp_path / "legacy.db"
+    _legacy_database(path, "completed")
+    _upstream_upgrade(path)
+    runtime = build_task_store_runtime(
+        make_settings(a2a_task_store_database_url=f"sqlite+aiosqlite:///{path}")
+    )
+    context = ServerCallContext(state={"identity": "automation"})
+    try:
+        await runtime.startup()
+        request = ListTasksRequest(status=TaskState.TASK_STATE_COMPLETED)
+        request.status_timestamp_after.FromJsonString("2026-01-01T00:00:00Z")
+        result = await runtime.task_store.list(request, context)
+        assert [task.id for task in result.tasks] == ["legacy-task"]
+    finally:
+        await runtime.shutdown()
diff --git a/uv.lock b/uv.lock
index 49d302b..5360519 100644
--- a/uv.lock
+++ b/uv.lock
@@ -26,6 +26,11 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/0b/91/087e4281bae91e19c4bb0736dc0807725e2e1f641dcc1ac86e625137c32c/a2a_sdk-1.1.5-py3-none-any.whl", hash = "sha256:c1d63d36b79a097c62dd9be001fce21d5939f5a811a9c5afd0ba8082f13f68bf", size = 255110, upload-time = "2026-09-21T09:52:24.741Z" },
 ]
 
+[package.optional-dependencies]
+db-cli = [
+    { name = "alembic" },
+]
+
 [[package]]
 name = "aiologic"
 version = "0.17.1"
@@ -49,6 +54,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/00/b7/e3bf5133d697a08128598c8d0abc5e16377b51465a33756de24fa7dee953/aiosqlite-0.22.1-py3-none-any.whl", hash = "sha256:21c002eb13823fad740196c5a2e9d8e62f6243bd9e7e4a1f87fb5e44ecb4fceb", size = 17405, upload-time = "2025-12-23T19:25:42.139Z" },
 ]
 
+[[package]]
+name = "alembic"
+version = "1.20.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "mako" },
+    { name = "sqlalchemy" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/ed/aa/02910bdb8e2f1444f6654d5b296cd827d126f82209050ee7b1000f92ac4b/alembic-1.20.0.tar.gz", hash = "sha256:db505480647bc60386c5369402f4a57a506b7539c9e9ef5e270d45cbbe4939bf", size = 2093272, upload-time = "2026-09-11T19:09:11.126Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/3f/27/78a89b55b0904d222183164e079b4ca56208e94eff1d35ad1f1ad5be9b06/alembic-1.20.0-py3-none-any.whl", hash = "sha256:77eb101048d95f982c0353e9233404889dcd7a6fc244c107836c0e2fc9cf7d9d", size = 268719, upload-time = "2026-09-11T19:09:12.88Z" },
+]
+
 [[package]]
 name = "annotated-doc"
 version = "0.0.5"
@@ -1018,6 +1037,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/af/40/791891d4c0c4dab4c5e187c17261cedc26285fd41541577f900470a45a4d/license_expression-30.4.4-py3-none-any.whl", hash = "sha256:421788fdcadb41f049d2dc934ce666626265aeccefddd25e162a26f23bcbf8a4", size = 120615, upload-time = "2025-07-22T11:13:31.217Z" },
 ]
 
+[[package]]
+name = "mako"
+version = "1.4.3"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "markupsafe" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/5a/09/e07c4b5579a79f4b16f8d4f29f6c54514ac787c4ad506b8c4f28a0e6b0bf/mako-1.4.3.tar.gz", hash = "sha256:cd6537fe88d5fec315c55c2f8529bc4ce7a9a352ad7db3eeaa6a66e2dd4ec37a", size = 412799, upload-time = "2026-09-22T20:54:31.509Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/6d/a0/053d6af3e8f871e0073b4a36732d9e65be77a72e5434c31b94f6af78a6bb/mako-1.4.3-py3-none-any.whl", hash = "sha256:723296007c870bfd6b3f0c3230dba7198096e5269297ebf5e4eff9e7ffa39d4f", size = 80164, upload-time = "2026-09-22T20:54:33.128Z" },
+]
+
 [[package]]
 name = "markdown-it-py"
 version = "4.2.0"
@@ -1030,6 +1061,80 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" },
 ]
 
+[[package]]
+name = "markupsafe"
+version = "3.0.3"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/7e/99/7690b6d4034fffd95959cbe0c02de8deb3098cc577c67bb6a24fe5d7caa7/markupsafe-3.0.3.tar.gz", hash = "sha256:722695808f4b6457b320fdc131280796bdceb04ab50fe1795cd540799ebe1698", size = 80313, upload-time = "2025-09-27T18:37:40.426Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/08/db/fefacb2136439fc8dd20e797950e749aa1f4997ed584c62cfb8ef7c2be0e/markupsafe-3.0.3-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:1cc7ea17a6824959616c525620e387f6dd30fec8cb44f649e31712db02123dad", size = 11631, upload-time = "2025-09-27T18:36:18.185Z" },
+    { url = "https://files.pythonhosted.org/packages/e1/2e/5898933336b61975ce9dc04decbc0a7f2fee78c30353c5efba7f2d6ff27a/markupsafe-3.0.3-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4bd4cd07944443f5a265608cc6aab442e4f74dff8088b0dfc8238647b8f6ae9a", size = 12058, upload-time = "2025-09-27T18:36:19.444Z" },
+    { url = "https://files.pythonhosted.org/packages/1d/09/adf2df3699d87d1d8184038df46a9c80d78c0148492323f4693df54e17bb/markupsafe-3.0.3-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:6b5420a1d9450023228968e7e6a9ce57f65d148ab56d2313fcd589eee96a7a50", size = 24287, upload-time = "2025-09-27T18:36:20.768Z" },
+    { url = "https://files.pythonhosted.org/packages/30/ac/0273f6fcb5f42e314c6d8cd99effae6a5354604d461b8d392b5ec9530a54/markupsafe-3.0.3-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0bf2a864d67e76e5c9a34dc26ec616a66b9888e25e7b9460e1c76d3293bd9dbf", size = 22940, upload-time = "2025-09-27T18:36:22.249Z" },
+    { url = "https://files.pythonhosted.org/packages/19/ae/31c1be199ef767124c042c6c3e904da327a2f7f0cd63a0337e1eca2967a8/markupsafe-3.0.3-cp311-cp311-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:bc51efed119bc9cfdf792cdeaa4d67e8f6fcccab66ed4bfdd6bde3e59bfcbb2f", size = 21887, upload-time = "2025-09-27T18:36:23.535Z" },
+    { url = "https://files.pythonhosted.org/packages/b2/76/7edcab99d5349a4532a459e1fe64f0b0467a3365056ae550d3bcf3f79e1e/markupsafe-3.0.3-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:068f375c472b3e7acbe2d5318dea141359e6900156b5b2ba06a30b169086b91a", size = 23692, upload-time = "2025-09-27T18:36:24.823Z" },
+    { url = "https://files.pythonhosted.org/packages/a4/28/6e74cdd26d7514849143d69f0bf2399f929c37dc2b31e6829fd2045b2765/markupsafe-3.0.3-cp311-cp311-musllinux_1_2_riscv64.whl", hash = "sha256:7be7b61bb172e1ed687f1754f8e7484f1c8019780f6f6b0786e76bb01c2ae115", size = 21471, upload-time = "2025-09-27T18:36:25.95Z" },
+    { url = "https://files.pythonhosted.org/packages/62/7e/a145f36a5c2945673e590850a6f8014318d5577ed7e5920a4b3448e0865d/markupsafe-3.0.3-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:f9e130248f4462aaa8e2552d547f36ddadbeaa573879158d721bbd33dfe4743a", size = 22923, upload-time = "2025-09-27T18:36:27.109Z" },
+    { url = "https://files.pythonhosted.org/packages/0f/62/d9c46a7f5c9adbeeeda52f5b8d802e1094e9717705a645efc71b0913a0a8/markupsafe-3.0.3-cp311-cp311-win32.whl", hash = "sha256:0db14f5dafddbb6d9208827849fad01f1a2609380add406671a26386cdf15a19", size = 14572, upload-time = "2025-09-27T18:36:28.045Z" },
+    { url = "https://files.pythonhosted.org/packages/83/8a/4414c03d3f891739326e1783338e48fb49781cc915b2e0ee052aa490d586/markupsafe-3.0.3-cp311-cp311-win_amd64.whl", hash = "sha256:de8a88e63464af587c950061a5e6a67d3632e36df62b986892331d4620a35c01", size = 15077, upload-time = "2025-09-27T18:36:29.025Z" },
+    { url = "https://files.pythonhosted.org/packages/35/73/893072b42e6862f319b5207adc9ae06070f095b358655f077f69a35601f0/markupsafe-3.0.3-cp311-cp311-win_arm64.whl", hash = "sha256:3b562dd9e9ea93f13d53989d23a7e775fdfd1066c33494ff43f5418bc8c58a5c", size = 13876, upload-time = "2025-09-27T18:36:29.954Z" },
+    { url = "https://files.pythonhosted.org/packages/5a/72/147da192e38635ada20e0a2e1a51cf8823d2119ce8883f7053879c2199b5/markupsafe-3.0.3-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:d53197da72cc091b024dd97249dfc7794d6a56530370992a5e1a08983ad9230e", size = 11615, upload-time = "2025-09-27T18:36:30.854Z" },
+    { url = "https://files.pythonhosted.org/packages/9a/81/7e4e08678a1f98521201c3079f77db69fb552acd56067661f8c2f534a718/markupsafe-3.0.3-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:1872df69a4de6aead3491198eaf13810b565bdbeec3ae2dc8780f14458ec73ce", size = 12020, upload-time = "2025-09-27T18:36:31.971Z" },
+    { url = "https://files.pythonhosted.org/packages/1e/2c/799f4742efc39633a1b54a92eec4082e4f815314869865d876824c257c1e/markupsafe-3.0.3-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:3a7e8ae81ae39e62a41ec302f972ba6ae23a5c5396c8e60113e9066ef893da0d", size = 24332, upload-time = "2025-09-27T18:36:32.813Z" },
+    { url = "https://files.pythonhosted.org/packages/3c/2e/8d0c2ab90a8c1d9a24f0399058ab8519a3279d1bd4289511d74e909f060e/markupsafe-3.0.3-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:d6dd0be5b5b189d31db7cda48b91d7e0a9795f31430b7f271219ab30f1d3ac9d", size = 22947, upload-time = "2025-09-27T18:36:33.86Z" },
+    { url = "https://files.pythonhosted.org/packages/2c/54/887f3092a85238093a0b2154bd629c89444f395618842e8b0c41783898ea/markupsafe-3.0.3-cp312-cp312-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:94c6f0bb423f739146aec64595853541634bde58b2135f27f61c1ffd1cd4d16a", size = 21962, upload-time = "2025-09-27T18:36:35.099Z" },
+    { url = "https://files.pythonhosted.org/packages/c9/2f/336b8c7b6f4a4d95e91119dc8521402461b74a485558d8f238a68312f11c/markupsafe-3.0.3-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:be8813b57049a7dc738189df53d69395eba14fb99345e0a5994914a3864c8a4b", size = 23760, upload-time = "2025-09-27T18:36:36.001Z" },
+    { url = "https://files.pythonhosted.org/packages/32/43/67935f2b7e4982ffb50a4d169b724d74b62a3964bc1a9a527f5ac4f1ee2b/markupsafe-3.0.3-cp312-cp312-musllinux_1_2_riscv64.whl", hash = "sha256:83891d0e9fb81a825d9a6d61e3f07550ca70a076484292a70fde82c4b807286f", size = 21529, upload-time = "2025-09-27T18:36:36.906Z" },
+    { url = "https://files.pythonhosted.org/packages/89/e0/4486f11e51bbba8b0c041098859e869e304d1c261e59244baa3d295d47b7/markupsafe-3.0.3-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:77f0643abe7495da77fb436f50f8dab76dbc6e5fd25d39589a0f1fe6548bfa2b", size = 23015, upload-time = "2025-09-27T18:36:37.868Z" },
+    { url = "https://files.pythonhosted.org/packages/2f/e1/78ee7a023dac597a5825441ebd17170785a9dab23de95d2c7508ade94e0e/markupsafe-3.0.3-cp312-cp312-win32.whl", hash = "sha256:d88b440e37a16e651bda4c7c2b930eb586fd15ca7406cb39e211fcff3bf3017d", size = 14540, upload-time = "2025-09-27T18:36:38.761Z" },
+    { url = "https://files.pythonhosted.org/packages/aa/5b/bec5aa9bbbb2c946ca2733ef9c4ca91c91b6a24580193e891b5f7dbe8e1e/markupsafe-3.0.3-cp312-cp312-win_amd64.whl", hash = "sha256:26a5784ded40c9e318cfc2bdb30fe164bdb8665ded9cd64d500a34fb42067b1c", size = 15105, upload-time = "2025-09-27T18:36:39.701Z" },
+    { url = "https://files.pythonhosted.org/packages/e5/f1/216fc1bbfd74011693a4fd837e7026152e89c4bcf3e77b6692fba9923123/markupsafe-3.0.3-cp312-cp312-win_arm64.whl", hash = "sha256:35add3b638a5d900e807944a078b51922212fb3dedb01633a8defc4b01a3c85f", size = 13906, upload-time = "2025-09-27T18:36:40.689Z" },
+    { url = "https://files.pythonhosted.org/packages/38/2f/907b9c7bbba283e68f20259574b13d005c121a0fa4c175f9bed27c4597ff/markupsafe-3.0.3-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:e1cf1972137e83c5d4c136c43ced9ac51d0e124706ee1c8aa8532c1287fa8795", size = 11622, upload-time = "2025-09-27T18:36:41.777Z" },
+    { url = "https://files.pythonhosted.org/packages/9c/d9/5f7756922cdd676869eca1c4e3c0cd0df60ed30199ffd775e319089cb3ed/markupsafe-3.0.3-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:116bb52f642a37c115f517494ea5feb03889e04df47eeff5b130b1808ce7c219", size = 12029, upload-time = "2025-09-27T18:36:43.257Z" },
+    { url = "https://files.pythonhosted.org/packages/00/07/575a68c754943058c78f30db02ee03a64b3c638586fba6a6dd56830b30a3/markupsafe-3.0.3-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:133a43e73a802c5562be9bbcd03d090aa5a1fe899db609c29e8c8d815c5f6de6", size = 24374, upload-time = "2025-09-27T18:36:44.508Z" },
+    { url = "https://files.pythonhosted.org/packages/a9/21/9b05698b46f218fc0e118e1f8168395c65c8a2c750ae2bab54fc4bd4e0e8/markupsafe-3.0.3-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ccfcd093f13f0f0b7fdd0f198b90053bf7b2f02a3927a30e63f3ccc9df56b676", size = 22980, upload-time = "2025-09-27T18:36:45.385Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/71/544260864f893f18b6827315b988c146b559391e6e7e8f7252839b1b846a/markupsafe-3.0.3-cp313-cp313-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:509fa21c6deb7a7a273d629cf5ec029bc209d1a51178615ddf718f5918992ab9", size = 21990, upload-time = "2025-09-27T18:36:46.916Z" },
+    { url = "https://files.pythonhosted.org/packages/c2/28/b50fc2f74d1ad761af2f5dcce7492648b983d00a65b8c0e0cb457c82ebbe/markupsafe-3.0.3-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:a4afe79fb3de0b7097d81da19090f4df4f8d3a2b3adaa8764138aac2e44f3af1", size = 23784, upload-time = "2025-09-27T18:36:47.884Z" },
+    { url = "https://files.pythonhosted.org/packages/ed/76/104b2aa106a208da8b17a2fb72e033a5a9d7073c68f7e508b94916ed47a9/markupsafe-3.0.3-cp313-cp313-musllinux_1_2_riscv64.whl", hash = "sha256:795e7751525cae078558e679d646ae45574b47ed6e7771863fcc079a6171a0fc", size = 21588, upload-time = "2025-09-27T18:36:48.82Z" },
+    { url = "https://files.pythonhosted.org/packages/b5/99/16a5eb2d140087ebd97180d95249b00a03aa87e29cc224056274f2e45fd6/markupsafe-3.0.3-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:8485f406a96febb5140bfeca44a73e3ce5116b2501ac54fe953e488fb1d03b12", size = 23041, upload-time = "2025-09-27T18:36:49.797Z" },
+    { url = "https://files.pythonhosted.org/packages/19/bc/e7140ed90c5d61d77cea142eed9f9c303f4c4806f60a1044c13e3f1471d0/markupsafe-3.0.3-cp313-cp313-win32.whl", hash = "sha256:bdd37121970bfd8be76c5fb069c7751683bdf373db1ed6c010162b2a130248ed", size = 14543, upload-time = "2025-09-27T18:36:51.584Z" },
+    { url = "https://files.pythonhosted.org/packages/05/73/c4abe620b841b6b791f2edc248f556900667a5a1cf023a6646967ae98335/markupsafe-3.0.3-cp313-cp313-win_amd64.whl", hash = "sha256:9a1abfdc021a164803f4d485104931fb8f8c1efd55bc6b748d2f5774e78b62c5", size = 15113, upload-time = "2025-09-27T18:36:52.537Z" },
+    { url = "https://files.pythonhosted.org/packages/f0/3a/fa34a0f7cfef23cf9500d68cb7c32dd64ffd58a12b09225fb03dd37d5b80/markupsafe-3.0.3-cp313-cp313-win_arm64.whl", hash = "sha256:7e68f88e5b8799aa49c85cd116c932a1ac15caaa3f5db09087854d218359e485", size = 13911, upload-time = "2025-09-27T18:36:53.513Z" },
+    { url = "https://files.pythonhosted.org/packages/e4/d7/e05cd7efe43a88a17a37b3ae96e79a19e846f3f456fe79c57ca61356ef01/markupsafe-3.0.3-cp313-cp313t-macosx_10_13_x86_64.whl", hash = "sha256:218551f6df4868a8d527e3062d0fb968682fe92054e89978594c28e642c43a73", size = 11658, upload-time = "2025-09-27T18:36:54.819Z" },
+    { url = "https://files.pythonhosted.org/packages/99/9e/e412117548182ce2148bdeacdda3bb494260c0b0184360fe0d56389b523b/markupsafe-3.0.3-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:3524b778fe5cfb3452a09d31e7b5adefeea8c5be1d43c4f810ba09f2ceb29d37", size = 12066, upload-time = "2025-09-27T18:36:55.714Z" },
+    { url = "https://files.pythonhosted.org/packages/bc/e6/fa0ffcda717ef64a5108eaa7b4f5ed28d56122c9a6d70ab8b72f9f715c80/markupsafe-3.0.3-cp313-cp313t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:4e885a3d1efa2eadc93c894a21770e4bc67899e3543680313b09f139e149ab19", size = 25639, upload-time = "2025-09-27T18:36:56.908Z" },
+    { url = "https://files.pythonhosted.org/packages/96/ec/2102e881fe9d25fc16cb4b25d5f5cde50970967ffa5dddafdb771237062d/markupsafe-3.0.3-cp313-cp313t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8709b08f4a89aa7586de0aadc8da56180242ee0ada3999749b183aa23df95025", size = 23569, upload-time = "2025-09-27T18:36:57.913Z" },
+    { url = "https://files.pythonhosted.org/packages/4b/30/6f2fce1f1f205fc9323255b216ca8a235b15860c34b6798f810f05828e32/markupsafe-3.0.3-cp313-cp313t-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:b8512a91625c9b3da6f127803b166b629725e68af71f8184ae7e7d54686a56d6", size = 23284, upload-time = "2025-09-27T18:36:58.833Z" },
+    { url = "https://files.pythonhosted.org/packages/58/47/4a0ccea4ab9f5dcb6f79c0236d954acb382202721e704223a8aafa38b5c8/markupsafe-3.0.3-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:9b79b7a16f7fedff2495d684f2b59b0457c3b493778c9eed31111be64d58279f", size = 24801, upload-time = "2025-09-27T18:36:59.739Z" },
+    { url = "https://files.pythonhosted.org/packages/6a/70/3780e9b72180b6fecb83a4814d84c3bf4b4ae4bf0b19c27196104149734c/markupsafe-3.0.3-cp313-cp313t-musllinux_1_2_riscv64.whl", hash = "sha256:12c63dfb4a98206f045aa9563db46507995f7ef6d83b2f68eda65c307c6829eb", size = 22769, upload-time = "2025-09-27T18:37:00.719Z" },
+    { url = "https://files.pythonhosted.org/packages/98/c5/c03c7f4125180fc215220c035beac6b9cb684bc7a067c84fc69414d315f5/markupsafe-3.0.3-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:8f71bc33915be5186016f675cd83a1e08523649b0e33efdb898db577ef5bb009", size = 23642, upload-time = "2025-09-27T18:37:01.673Z" },
+    { url = "https://files.pythonhosted.org/packages/80/d6/2d1b89f6ca4bff1036499b1e29a1d02d282259f3681540e16563f27ebc23/markupsafe-3.0.3-cp313-cp313t-win32.whl", hash = "sha256:69c0b73548bc525c8cb9a251cddf1931d1db4d2258e9599c28c07ef3580ef354", size = 14612, upload-time = "2025-09-27T18:37:02.639Z" },
+    { url = "https://files.pythonhosted.org/packages/2b/98/e48a4bfba0a0ffcf9925fe2d69240bfaa19c6f7507b8cd09c70684a53c1e/markupsafe-3.0.3-cp313-cp313t-win_amd64.whl", hash = "sha256:1b4b79e8ebf6b55351f0d91fe80f893b4743f104bff22e90697db1590e47a218", size = 15200, upload-time = "2025-09-27T18:37:03.582Z" },
+    { url = "https://files.pythonhosted.org/packages/0e/72/e3cc540f351f316e9ed0f092757459afbc595824ca724cbc5a5d4263713f/markupsafe-3.0.3-cp313-cp313t-win_arm64.whl", hash = "sha256:ad2cf8aa28b8c020ab2fc8287b0f823d0a7d8630784c31e9ee5edea20f406287", size = 13973, upload-time = "2025-09-27T18:37:04.929Z" },
+    { url = "https://files.pythonhosted.org/packages/33/8a/8e42d4838cd89b7dde187011e97fe6c3af66d8c044997d2183fbd6d31352/markupsafe-3.0.3-cp314-cp314-macosx_10_13_x86_64.whl", hash = "sha256:eaa9599de571d72e2daf60164784109f19978b327a3910d3e9de8c97b5b70cfe", size = 11619, upload-time = "2025-09-27T18:37:06.342Z" },
+    { url = "https://files.pythonhosted.org/packages/b5/64/7660f8a4a8e53c924d0fa05dc3a55c9cee10bbd82b11c5afb27d44b096ce/markupsafe-3.0.3-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:c47a551199eb8eb2121d4f0f15ae0f923d31350ab9280078d1e5f12b249e0026", size = 12029, upload-time = "2025-09-27T18:37:07.213Z" },
+    { url = "https://files.pythonhosted.org/packages/da/ef/e648bfd021127bef5fa12e1720ffed0c6cbb8310c8d9bea7266337ff06de/markupsafe-3.0.3-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:f34c41761022dd093b4b6896d4810782ffbabe30f2d443ff5f083e0cbbb8c737", size = 24408, upload-time = "2025-09-27T18:37:09.572Z" },
+    { url = "https://files.pythonhosted.org/packages/41/3c/a36c2450754618e62008bf7435ccb0f88053e07592e6028a34776213d877/markupsafe-3.0.3-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:457a69a9577064c05a97c41f4e65148652db078a3a509039e64d3467b9e7ef97", size = 23005, upload-time = "2025-09-27T18:37:10.58Z" },
+    { url = "https://files.pythonhosted.org/packages/bc/20/b7fdf89a8456b099837cd1dc21974632a02a999ec9bf7ca3e490aacd98e7/markupsafe-3.0.3-cp314-cp314-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:e8afc3f2ccfa24215f8cb28dcf43f0113ac3c37c2f0f0806d8c70e4228c5cf4d", size = 22048, upload-time = "2025-09-27T18:37:11.547Z" },
+    { url = "https://files.pythonhosted.org/packages/9a/a7/591f592afdc734f47db08a75793a55d7fbcc6902a723ae4cfbab61010cc5/markupsafe-3.0.3-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:ec15a59cf5af7be74194f7ab02d0f59a62bdcf1a537677ce67a2537c9b87fcda", size = 23821, upload-time = "2025-09-27T18:37:12.48Z" },
+    { url = "https://files.pythonhosted.org/packages/7d/33/45b24e4f44195b26521bc6f1a82197118f74df348556594bd2262bda1038/markupsafe-3.0.3-cp314-cp314-musllinux_1_2_riscv64.whl", hash = "sha256:0eb9ff8191e8498cca014656ae6b8d61f39da5f95b488805da4bb029cccbfbaf", size = 21606, upload-time = "2025-09-27T18:37:13.485Z" },
+    { url = "https://files.pythonhosted.org/packages/ff/0e/53dfaca23a69fbfbbf17a4b64072090e70717344c52eaaaa9c5ddff1e5f0/markupsafe-3.0.3-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:2713baf880df847f2bece4230d4d094280f4e67b1e813eec43b4c0e144a34ffe", size = 23043, upload-time = "2025-09-27T18:37:14.408Z" },
+    { url = "https://files.pythonhosted.org/packages/46/11/f333a06fc16236d5238bfe74daccbca41459dcd8d1fa952e8fbd5dccfb70/markupsafe-3.0.3-cp314-cp314-win32.whl", hash = "sha256:729586769a26dbceff69f7a7dbbf59ab6572b99d94576a5592625d5b411576b9", size = 14747, upload-time = "2025-09-27T18:37:15.36Z" },
+    { url = "https://files.pythonhosted.org/packages/28/52/182836104b33b444e400b14f797212f720cbc9ed6ba34c800639d154e821/markupsafe-3.0.3-cp314-cp314-win_amd64.whl", hash = "sha256:bdc919ead48f234740ad807933cdf545180bfbe9342c2bb451556db2ed958581", size = 15341, upload-time = "2025-09-27T18:37:16.496Z" },
+    { url = "https://files.pythonhosted.org/packages/6f/18/acf23e91bd94fd7b3031558b1f013adfa21a8e407a3fdb32745538730382/markupsafe-3.0.3-cp314-cp314-win_arm64.whl", hash = "sha256:5a7d5dc5140555cf21a6fefbdbf8723f06fcd2f63ef108f2854de715e4422cb4", size = 14073, upload-time = "2025-09-27T18:37:17.476Z" },
+    { url = "https://files.pythonhosted.org/packages/3c/f0/57689aa4076e1b43b15fdfa646b04653969d50cf30c32a102762be2485da/markupsafe-3.0.3-cp314-cp314t-macosx_10_13_x86_64.whl", hash = "sha256:1353ef0c1b138e1907ae78e2f6c63ff67501122006b0f9abad68fda5f4ffc6ab", size = 11661, upload-time = "2025-09-27T18:37:18.453Z" },
+    { url = "https://files.pythonhosted.org/packages/89/c3/2e67a7ca217c6912985ec766c6393b636fb0c2344443ff9d91404dc4c79f/markupsafe-3.0.3-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:1085e7fbddd3be5f89cc898938f42c0b3c711fdcb37d75221de2666af647c175", size = 12069, upload-time = "2025-09-27T18:37:19.332Z" },
+    { url = "https://files.pythonhosted.org/packages/f0/00/be561dce4e6ca66b15276e184ce4b8aec61fe83662cce2f7d72bd3249d28/markupsafe-3.0.3-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1b52b4fb9df4eb9ae465f8d0c228a00624de2334f216f178a995ccdcf82c4634", size = 25670, upload-time = "2025-09-27T18:37:20.245Z" },
+    { url = "https://files.pythonhosted.org/packages/50/09/c419f6f5a92e5fadde27efd190eca90f05e1261b10dbd8cbcb39cd8ea1dc/markupsafe-3.0.3-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:fed51ac40f757d41b7c48425901843666a6677e3e8eb0abcff09e4ba6e664f50", size = 23598, upload-time = "2025-09-27T18:37:21.177Z" },
+    { url = "https://files.pythonhosted.org/packages/22/44/a0681611106e0b2921b3033fc19bc53323e0b50bc70cffdd19f7d679bb66/markupsafe-3.0.3-cp314-cp314t-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:f190daf01f13c72eac4efd5c430a8de82489d9cff23c364c3ea822545032993e", size = 23261, upload-time = "2025-09-27T18:37:22.167Z" },
+    { url = "https://files.pythonhosted.org/packages/5f/57/1b0b3f100259dc9fffe780cfb60d4be71375510e435efec3d116b6436d43/markupsafe-3.0.3-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:e56b7d45a839a697b5eb268c82a71bd8c7f6c94d6fd50c3d577fa39a9f1409f5", size = 24835, upload-time = "2025-09-27T18:37:23.296Z" },
+    { url = "https://files.pythonhosted.org/packages/26/6a/4bf6d0c97c4920f1597cc14dd720705eca0bf7c787aebc6bb4d1bead5388/markupsafe-3.0.3-cp314-cp314t-musllinux_1_2_riscv64.whl", hash = "sha256:f3e98bb3798ead92273dc0e5fd0f31ade220f59a266ffd8a4f6065e0a3ce0523", size = 22733, upload-time = "2025-09-27T18:37:24.237Z" },
+    { url = "https://files.pythonhosted.org/packages/14/c7/ca723101509b518797fedc2fdf79ba57f886b4aca8a7d31857ba3ee8281f/markupsafe-3.0.3-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:5678211cb9333a6468fb8d8be0305520aa073f50d17f089b5b4b477ea6e67fdc", size = 23672, upload-time = "2025-09-27T18:37:25.271Z" },
+    { url = "https://files.pythonhosted.org/packages/fb/df/5bd7a48c256faecd1d36edc13133e51397e41b73bb77e1a69deab746ebac/markupsafe-3.0.3-cp314-cp314t-win32.whl", hash = "sha256:915c04ba3851909ce68ccc2b8e2cd691618c4dc4c4232fb7982bca3f41fd8c3d", size = 14819, upload-time = "2025-09-27T18:37:26.285Z" },
+    { url = "https://files.pythonhosted.org/packages/1a/8a/0402ba61a2f16038b48b39bccca271134be00c5c9f0f623208399333c448/markupsafe-3.0.3-cp314-cp314t-win_amd64.whl", hash = "sha256:4faffd047e07c38848ce017e8725090413cd80cbc23d86e55c587bf979e579c9", size = 15426, upload-time = "2025-09-27T18:37:27.316Z" },
+    { url = "https://files.pythonhosted.org/packages/70/bc/6f1c2f612465f5fa89b95bead1f44dcb607670fd42891d8fdcd5d039f4f4/markupsafe-3.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:32001d6a8fc98c8cb5c947787c5d08b0a50663d139f1305bac5885d98d9b40fa", size = 14146, upload-time = "2025-09-27T18:37:28.327Z" },
+]
+
 [[package]]
 name = "mdurl"
 version = "0.1.2"
@@ -1209,10 +1314,14 @@ dev = [
     { name = "pytest-cov" },
     { name = "ruff" },
 ]
+migrations = [
+    { name = "a2a-sdk", extra = ["db-cli"] },
+]
 
 [package.metadata]
 requires-dist = [
     { name = "a2a-sdk", specifier = "==1.1.5" },
+    { name = "a2a-sdk", extras = ["db-cli"], marker = "extra == 'migrations'", specifier = "==1.1.5" },
     { name = "aiosqlite", specifier = ">=0.20,<1.0" },
     { name = "click", specifier = ">=8.3.3" },
     { name = "fastapi", specifier = ">=0.139,<1.0" },
@@ -1233,7 +1342,7 @@ requires-dist = [
     { name = "sse-starlette", specifier = ">=2.1,<4.0" },
     { name = "uvicorn", specifier = ">=0.29,<1.0" },
 ]
-provides-extras = ["dev"]
+provides-extras = ["migrations", "dev"]
 
 [[package]]
 name = "packageurl-python"

From 1f1d67a237d67131a8f3fbb99f11fa31f42805f2 Mon Sep 17 00:00:00 2001
From: "Liu Juanjuan (Helen)"
 <31027645+liujuanjuan1984@users.noreply.github.com>
Date: Tue, 29 Sep 2026 13:08:57 +0000
Subject: [PATCH 4/5] docs: condense lifecycle and upgrade guidance (#533)

---
 README.md                       |  4 --
 docs/compatibility.md           | 49 -----------------
 docs/guide.md                   | 93 +++++----------------------------
 docs/maintainer-architecture.md |  1 -
 4 files changed, 14 insertions(+), 133 deletions(-)

diff --git a/README.md b/README.md
index 19e7ec2..7b52ff5 100644
--- a/README.md
+++ b/README.md
@@ -59,10 +59,6 @@ Upgrade later with:
 uv tool upgrade opencode-a2a
 ```
 
-If you retain an older SQLite database, follow
-[Upgrading an Existing Database](docs/guide.md#upgrading-an-existing-database)
-for version-matched tooling, task ownership, and rollback instructions.
-
 Make sure provider credentials and a default model are configured on the OpenCode side, then start OpenCode:
 
 ```bash
diff --git a/docs/compatibility.md b/docs/compatibility.md
index 8ce448f..47891d5 100644
--- a/docs/compatibility.md
+++ b/docs/compatibility.md
@@ -28,55 +28,6 @@ The [SDK 1.1.2 to 1.1.5 changes](https://github.com/a2aproject/a2a-python/compar
 
 The missing-import typing overrides for `google.protobuf.*` and `jsonrpc.*` remain limited to those third-party modules. Runtime support is validated by the SDK and transport regressions, not inferred from the typing overrides. Ruff's local pre-commit hooks run through `uv run --locked --extra dev`, so `uv.lock` is the single version source for linting and formatting.
 
-### Handler Lifecycle and Legacy Queue Migration
-
-The adapter still uses the SDK's `LegacyRequestHandler`, through the internal
-`ManagedLegacyRequestHandler` lifecycle boundary. It does not use the newer
-`DefaultRequestHandler`/`ActiveTaskRegistry`; the latter's early-failure handling
-and `aclose()` do not apply automatically to this adapter.
-
-Unhandled producer errors are persisted directly as `FAILED` under the original
-request identity, including when no task row exists yet. Existing terminal
-snapshots remain immutable. Failure to persist is logged without storage error
-payloads and does not replace the execution error or skip queue cleanup. Both
-send paths explicitly check completed producers so queue closure cannot hide an
-execution exception; producer storage errors still use the adapter's stable error
-mapping. `GetTask` delegates parameter validation and history slicing
-to the pinned SDK, retaining adapter output negotiation and storage-error mapping.
-Empty IDs and negative history lengths are rejected as invalid parameters.
-
-Application shutdown closes the handler before its clients and stores. See the
-[shutdown contract](guide.md#execution-failure-and-shutdown) for local cancellation
-and upstream-session limits. Normal client disconnects still allow background
-execution and persistence.
-
-SDK 1.1.5 still provides `EventQueueLegacy`, but documents its future removal.
-Replacing one queue constructor or changing the handler's parent class is not a
-complete migration: the legacy queue manager, consumer, cancellation, subscription,
-and output aggregation paths are coupled. Full replacement remains unfinished
-code migration work, tracked in [#533](https://github.com/Intelligent-Internet/opencode-a2a/issues/533).
-It is not a database migration and is not waiting on production observation.
-The pinned SDK still supports the current implementation; this lifecycle patch
-does not upgrade that SDK or claim to remove its legacy dependency.
-
-In SDK 1.1.5, the newer handler ignores a second full `Task` snapshot after the
-initial task, whereas current executor paths use such snapshots. Its forced
-`aclose()` cancels consumers and does not itself persist interrupted execution as
-FAILED. Isolated probes with an initial WORKING task reproduce a retained WORKING
-state both after a final full COMPLETED task and after active execution shutdown.
-Migration must adapt these event and lifecycle contracts, not just swap imports.
-Before completing the migration or accepting an SDK that removes legacy APIs, prove parity for
-identity isolation, output/extension negotiation, cancel idempotency, first-terminal
-persistence, nonblocking sends, subscription, and disconnect/shutdown behavior on
-Python 3.11–3.14 with both memory and SQLite stores. Do not copy the SDK's queue
-implementation into this repository merely to suppress the deprecation.
-
-User upgrades are a separate contract: ship compatible code, version-matched
-migration tooling, and automated tests against historical data. See
-[Upgrading an Existing Database](guide.md#upgrading-an-existing-database).
-The acceptance gate is reproducible repository validation, not access to a live
-deployment. `migrations` is an optional install extra tied to the runtime SDK pin.
-
 ## Contract Honesty
 
 Machine-readable discovery surfaces must reflect actual runtime behavior:
diff --git a/docs/guide.md b/docs/guide.md
index 142f0a0..877e2b6 100644
--- a/docs/guide.md
+++ b/docs/guide.md
@@ -217,33 +217,7 @@ The package declares `sqlalchemy[asyncio]` so standalone installs such as `uv to
 
 Published package metadata also enforces the security floors `click>=8.3.3` and `pyasn1>=0.6.4`. These apply to standalone wheel/sdist installs as well as repository environments; no separate uv constraint configuration is needed.
 
-### Execution Failure and Shutdown
-
-An unhandled execution error writes `FAILED` directly to the task store using the
-original request identity, then closes the producer queue and removes its runtime
-registration. An already persisted terminal task is never overwritten. If storage
-is unavailable, cleanup still runs, but a durable failure state cannot be promised.
-This also covers failures before the first task event; ordinary provider errors
-continue to use the executor's existing error events.
-
-At application shutdown, the handler rejects new execution setup and waits for
-setup already in progress to register its producers. It cancels and awaits local
-producers, marks interrupted nonterminal executions `FAILED` when storage is
-available, and lets tracked consumers finish persisting buffered output from closed
-queues before cancelling other background cleanup tasks. This also preserves a
-completed producer's terminal output when its consumer is still writing to storage.
-Only then does it close outbound clients, the OpenCode HTTP client, state stores,
-and the database engine. Cleanup of the remaining resources is attempted even if a close operation
-fails. Repeated cancellation of the shutdown waiter does not bypass this ordering.
-Shutdown is idempotent and does not wait for model generation to finish,
-though executor cleanup and storage I/O must still complete. Supervisor shutdown
-budgets must allow that cleanup; a hard process kill cannot provide these guarantees.
-
-This policy cancels adapter execution, not the independently running OpenCode
-session. To request a best-effort upstream session abort, use `CancelTask` before
-stopping the service. That explicit cancel path retains its `CANCELED` state and
-idempotency. An SSE client disconnect alone does not trigger application shutdown:
-the producer continues in the background and persists its remaining output.
+Unhandled execution errors and shutdown interruptions persist `FAILED` when storage is available, preserving existing terminal states. Shutdown drains buffered output before closing clients and stores. SSE disconnects allow execution to continue; use `CancelTask` to request an upstream abort before stopping the application.
 
 ### SQLite Persistence Hardening
 
@@ -269,7 +243,7 @@ Database-backed task persistence also keeps the existing first-terminal-state-wi
 
 At startup, the runtime logs a concise persistence summary covering the active backend, the redacted database URL when applicable, the shared persistence scope, and whether the SQLite local durability profile is active.
 
-The adapter-owned state tables listed above remain managed by the internal migration runner. The SDK-owned `tasks` table does not use runtime **schema** auto-migration here. Upgrade an old SDK task schema with the matching release's migration tool as described below. Once the schema is compatible, startup normalizes legacy task payloads before accepting requests.
+SDK task schemas require an explicit [upgrade](#upgrading-an-existing-database) when incompatible; startup then converts legacy task payloads transactionally. Existing v1 rows and task ownership are preserved.
 
 In-flight asyncio locks, outbound A2A client caches, and stream-local aggregation buffers remain process-local runtime state.
 
@@ -281,57 +255,18 @@ A2A_TASK_STORE_BACKEND=memory
 
 ### Upgrading an Existing Database
 
-Queue/handler implementation changes concern process-local Python objects; they
-do not require a database migration. This lifecycle update keeps SDK 1.1.5 and
-does not introduce a new schema. A fresh installation or the `memory` backend
-needs no migration command. An existing compatible v1 database also needs no
-manual schema change.
-
-For an older SQLite database that startup rejects for missing `owner`,
-`last_updated`, `protocol_version`, or the owner/time index:
-
-1. Stop the application, make a consistent backup of its SQLite database, and
-   keep that backup with the previous application version for rollback. Use an
-   absolute database path to avoid upgrading a different file by accident.
-2. Install the target release. Run its migration tooling using the **same
-   application version**, so the runtime and migrations use the same pinned SDK:
-
-   ```bash
-   OPENCODE_A2A_VERSION="$(opencode-a2a --version | awk '{print $2}')"
-   uvx --from "opencode-a2a[migrations]==${OPENCODE_A2A_VERSION}" a2a-db upgrade head \
-     --database-url 'sqlite+aiosqlite:////absolute/path/opencode-a2a.db' \
-     --add_columns_owner_last_updated-default-owner automation
-   ```
-
-   For a source checkout, use `uv run --extra migrations a2a-db` with the same
-   arguments. For an existing Python virtual environment, install that release
-   with the `opencode-a2a[migrations]` extra and run its `a2a-db` executable.
-   Do not install a floating SDK release separately into the runtime environment.
-
-   Replace `automation` with the intended bearer credential's `principal` or
-   Basic credential's username. The upstream default,
-   `legacy_v03_no_user_info`, does not match the normal authenticated identity
-   and can make old tasks appear missing. This option assigns **all rows lacking
-   an owner column** to one identity; it cannot infer historical ownership.
-   If the old data needs multiple owners, establish the ownership mapping before
-   migration. Existing owner columns are preserved, and rerunning the command
-   with another owner does not reassign them.
-3. Start the new application. Before serving requests, it converts rows marked
-   `0.3` or with no protocol marker to the SDK's v1 task representation, preserves
-   owners and task contents, and backfills missing query timestamps from task
-   status timestamps. Rows without any historical timestamp remain undated.
-   Conversion uses one transaction, reads in bounded batches, and skips existing
-   v1 rows on subsequent starts. Malformed legacy payloads fail startup and roll
-   back the payload conversion; the completed schema upgrade remains in place.
-4. Verify a known task with `GetTask` and `ListTasks` under its original identity,
-   including status/time filters. Terminal snapshots remain immutable. To roll
-   back to an older application, restore the stopped database backup rather than
-   relying on a schema downgrade to undo converted JSON payloads.
-
-These upgrade paths are tested against disposable historical databases in
-`tests/server/test_task_store_upgrade.py`; no running deployment or live database
-is required for repository acceptance. Neither queue replacement nor payload
-conversion resumes executions that were interrupted before this release.
+If startup reports an incompatible task schema, stop the application and back up SQLite, then use the installed release's migration tool:
+
+```bash
+OPENCODE_A2A_VERSION="$(opencode-a2a --version | awk '{print $2}')"
+uvx --from "opencode-a2a[migrations]==${OPENCODE_A2A_VERSION}" a2a-db upgrade head \
+  --database-url 'sqlite+aiosqlite:////absolute/path/opencode-a2a.db' \
+  --add_columns_owner_last_updated-default-owner automation
+```
+
+For source checkouts, use `uv run --extra migrations a2a-db` with the same arguments. Replace `automation` with the intended bearer principal or Basic username: adding an owner column assigns all old tasks to that identity; existing owners are unchanged. Multiple historical owners require an explicit mapping before migration.
+
+Restart and verify an old task under its owner. Startup converts legacy payloads and fills missing query timestamps; malformed payloads roll back conversion and prevent startup. Restore the backup to roll back the upgrade. Fresh installations and compatible schemas need no migration command.
 
 ## Troubleshooting Provider Auth State
 
diff --git a/docs/maintainer-architecture.md b/docs/maintainer-architecture.md
index 79e0b50..114d107 100644
--- a/docs/maintainer-architecture.md
+++ b/docs/maintainer-architecture.md
@@ -74,7 +74,6 @@ flowchart TD
 ### Server Layer
 
 - `server/application.py`: app assembly, route wiring, request handler customization, and top-level lifecycle integration
-- `server/handler_lifecycle.py`: legacy SDK producer ownership, direct failure persistence, setup admission, and handler shutdown
 - `server/middleware.py`: auth, protocol negotiation, payload/body guards, logging, and response decoration
 - `server/agent_card.py` / `server/openapi.py`: machine-readable contract publication
 - `server/rest_tasks.py`: SDK-owned REST task routes plus adapter-specific list behavior

From 34a9f814afbb1c48cc72725119fc731de5708dc0 Mon Sep 17 00:00:00 2001
From: "Liu Juanjuan (Helen)"
 <31027645+liujuanjuan1984@users.noreply.github.com>
Date: Tue, 29 Sep 2026 13:13:47 +0000
Subject: [PATCH 5/5] fix(server): drain buffered output before shutdown
 failure (#533)

---
 src/opencode_a2a/server/handler_lifecycle.py | 13 +++-
 tests/server/test_handler_lifecycle.py       | 68 +++++++++++++++++++-
 2 files changed, 78 insertions(+), 3 deletions(-)

diff --git a/src/opencode_a2a/server/handler_lifecycle.py b/src/opencode_a2a/server/handler_lifecycle.py
index dfc535f..3424c51 100644
--- a/src/opencode_a2a/server/handler_lifecycle.py
+++ b/src/opencode_a2a/server/handler_lifecycle.py
@@ -51,6 +51,7 @@ def __init__(  # noqa: PLR0913
         self._close_task: asyncio.Task[None] | None = None
         self._managed_producers: dict[asyncio.Task, str] = {}
         self._draining_consumers: set[asyncio.Task] = set()
+        self._shutdown_failures: list[RequestContext] = []
 
     async def _persist_execution_failure(self, request: RequestContext) -> None:
         """Write independently of the closing queue, preserving terminal snapshots."""
@@ -83,7 +84,7 @@ async def _run_event_stream(self, request: RequestContext, queue: EventQueueLega
             await self.agent_executor.execute(request, queue)
         except asyncio.CancelledError:
             if self._closing:
-                await self._persist_execution_failure(request)
+                self._shutdown_failures.append(request)
             raise
         except Exception as exc:
             logger.error(
@@ -91,7 +92,10 @@ async def _run_event_stream(self, request: RequestContext, queue: EventQueueLega
                 request.task_id,
                 type(exc).__name__,
             )
-            await self._persist_execution_failure(request)
+            if self._closing:
+                self._shutdown_failures.append(request)
+            else:
+                await self._persist_execution_failure(request)
             raise
         finally:
             await queue.close()
@@ -163,6 +167,11 @@ async def _drain_execution(self) -> None:
                 if task not in self._draining_consumers:
                     task.cancel()
             await asyncio.gather(*pending, return_exceptions=True)
+        # First-terminal-wins must not freeze a partial snapshot before buffered
+        # artifacts or a completed status have reached the store.
+        for request in self._shutdown_failures:
+            await self._persist_execution_failure(request)
+        self._shutdown_failures.clear()
         errors = [result for result in results if isinstance(result, Exception)]
         if errors:
             raise ExceptionGroup("Request handler cleanup failed", errors)
diff --git a/tests/server/test_handler_lifecycle.py b/tests/server/test_handler_lifecycle.py
index 9ddb440..65e7355 100644
--- a/tests/server/test_handler_lifecycle.py
+++ b/tests/server/test_handler_lifecycle.py
@@ -616,8 +616,9 @@ async def test_producer_storage_error_uses_stable_error_mapping(store, streaming
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("streaming", [False, True])
+@pytest.mark.parametrize("producer_exit", ["completed", "canceled", "error"])
 async def test_shutdown_drains_completed_producer_output_before_closing_store(
-    store, monkeypatch, streaming
+    store, monkeypatch, streaming, producer_exit
 ):
     from a2a.types import TaskStatusUpdateEvent
 
@@ -648,6 +649,13 @@ async def execute(context, queue):
                 status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
             )
         )
+        if producer_exit != "completed":
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                if producer_exit == "error":
+                    raise RuntimeError("executor cleanup failed") from None
+                raise
 
     handler = _handler(store, execute)
     params = _request()
@@ -668,3 +676,61 @@ async def execute(context, queue):
         await close
     saved = await store.get(params.message.task_id, _context())
     assert saved.status.state == TaskState.TASK_STATE_COMPLETED
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streaming", [False, True])
+async def test_shutdown_preserves_buffered_artifacts_before_marking_failed(
+    store, monkeypatch, streaming
+):
+    from a2a.types import Artifact, TaskArtifactUpdateEvent
+
+    writing_artifact = asyncio.Event()
+    release_write = asyncio.Event()
+    original_save = store.save
+
+    async def delayed_save(task, context=None):
+        if task.artifacts:
+            writing_artifact.set()
+            await release_write.wait()
+        await original_save(task, context)
+
+    monkeypatch.setattr(store, "save", delayed_save)
+
+    async def execute(context, queue):
+        await queue.enqueue_event(
+            Task(
+                id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+            )
+        )
+        await queue.enqueue_event(
+            TaskArtifactUpdateEvent(
+                task_id=context.task_id,
+                context_id=context.context_id,
+                artifact=Artifact(artifact_id="partial", parts=[Part(text="keep this output")]),
+            )
+        )
+        await asyncio.Event().wait()
+
+    handler = _handler(store, execute)
+    params = _request()
+    params.configuration.return_immediately = True
+    if streaming:
+        stream = handler.on_message_send_stream(params, _context())
+        await anext(stream)
+        await stream.aclose()
+    else:
+        await handler.on_message_send(params, _context())
+    await writing_artifact.wait()
+    close = asyncio.create_task(handler.aclose())
+    try:
+        with pytest.raises(TimeoutError):
+            await asyncio.wait_for(asyncio.shield(close), 0.05)
+    finally:
+        release_write.set()
+        await close
+    saved = await store.get(params.message.task_id, _context())
+    assert saved.status.state == TaskState.TASK_STATE_FAILED
+    assert saved.artifacts[0].parts[0].text == "keep this output"