diff --git a/docs/guide.md b/docs/guide.md
index 2d4b4e9..877e2b6 100644
--- a/docs/guide.md
+++ b/docs/guide.md
@@ -217,6 +217,8 @@ 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.
 
+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
 
 File-backed SQLite databases are hardened at startup and on every new connection on POSIX systems:
@@ -241,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 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.
+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.
 
@@ -251,6 +253,21 @@ To opt into an ephemeral development profile, set:
 A2A_TASK_STORE_BACKEND=memory
 ```
 
+### Upgrading an Existing Database
+
+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
 
 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/application.py b/src/opencode_a2a/server/application.py
index 96510e9..32f26a9 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(
@@ -708,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:
@@ -744,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",
@@ -774,24 +764,6 @@ 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 not result:
             raise InternalError()
@@ -970,6 +942,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..3424c51
--- /dev/null
+++ b/src/opencode_a2a/server/handler_lifecycle.py
@@ -0,0 +1,186 @@
+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] = {}
+        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."""
+        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:
+                self._shutdown_failures.append(request)
+            raise
+        except Exception as exc:
+            logger.error(
+                "Agent producer failed task_id=%s error_type=%s",
+                request.task_id,
+                type(exc).__name__,
+            )
+            if self._closing:
+                self._shutdown_failures.append(request)
+            else:
+                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_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 and task not in self._draining_consumers:
+            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())
+        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.
+        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,
+        )
+        # 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:
+                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)
+
+    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/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/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
new file mode 100644
index 0000000..65e7355
--- /dev/null
+++ b/tests/server/test_handler_lifecycle.py
@@ -0,0 +1,736 @@
+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)
+
+
+@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])
+@pytest.mark.parametrize("producer_exit", ["completed", "canceled", "error"])
+async def test_shutdown_drains_completed_producer_output_before_closing_store(
+    store, monkeypatch, streaming, producer_exit
+):
+    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),
+            )
+        )
+        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()
+    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
+
+
+@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"
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"