feat(server): recover tasks abandoned by a crashed replica

samples/clustermode/README.md

@@ -46,6 +46,27 @@
 OK: one task was started, observed, and cancelled across three different replicas.
 ```
 
+## Recovering from a crashed replica
+
+Each turn holds a lease in the shared `task_leases` table, renewed every
+`--heartbeat-interval` seconds (2 s here). If a replica dies mid-turn, its lease
+expires after `--lease-ttl` seconds (10 s here) and the `AbandonedTaskSweeper`
+on a surviving replica marks the task FAILED. A follow-up sent while another
+replica holds the lease is rejected instead of starting a second executor.
+
+To see it, start the cluster with slower ticks, send a message to one replica,
+then kill that replica's process:
+
+```bash
+python -m samples.clustermode.run_cluster --replicas 2 --ticks 60
+pkill -9 -f 'samples.clustermode.server.*--port 41241'
+```
+
+A `GetTask` against port 41242 shows the task in WORKING until the lease
+expires, then FAILED with the status message "The replica running this task
+stopped responding." When a replica is stopped normally, `handler.aclose()`
+fails its running turns right away.
+
 ## Files
 
 | File | Purpose |
@@ -66,9 +87,13 @@
     task_store=VersionedDatabaseTaskStore(engine=..., create_table=False),
     agent_card=agent_card,
     event_stream=DatabaseTaskEventStream(engine=..., create_table=False),
+    lease_store=DatabaseTaskLeaseStore(engine=..., create_table=False),
 )
 ```
 
+The lease store is optional; without it, a task whose replica crashes mid-turn
+stays WORKING.
+
 Omit `event_stream` (and use a plain `InMemoryTaskStore`) and you get the ordinary
 single-process behaviour — the multi-server path is fully opt-in.
 ```

samples/clustermode/cluster_common.py

@@ -24,7 +24,10 @@
 from a2a.helpers.proto_helpers import new_task_from_user_message
 from a2a.server.agent_execution.agent_executor import AgentExecutor
 from a2a.server.agent_execution.context import RequestContext
-from a2a.server.cluster import VersionedDatabaseTaskStore
+from a2a.server.cluster import (
+    DatabaseTaskLeaseStore,
+    VersionedDatabaseTaskStore,
+)
 from a2a.server.cluster.database_event_stream import DatabaseTaskEventStream
 from a2a.server.events.event_queue import EventQueue
 from a2a.server.models import Base
@@ -60,7 +63,7 @@ def default_dsn() -> str:
 
 
 async def init_schema(dsn: str) -> None:
-    """Creates the shared tables (task + task_events) once, up front."""
+    """Creates the shared tables (tasks, task_events, task_leases) once."""
     engine = create_async_engine(dsn)
     try:
         async with engine.begin() as conn:
@@ -86,6 +89,13 @@ def build_cluster_backends(
     return store, stream
 
 
+def build_lease_store(dsn: str) -> DatabaseTaskLeaseStore:
+    """Builds the shared turn-lease store used to recover crashed turns."""
+    return DatabaseTaskLeaseStore(
+        engine=create_async_engine(dsn), create_table=False
+    )
+
+
 def build_agent_card(base_url: str) -> AgentCard:
     """The agent card advertised by every replica (same logical agent)."""
     return AgentCard(

samples/clustermode/server.py

@@ -14,6 +14,7 @@
 
 from fastapi import FastAPI
 
+from a2a.server.cluster.recovery import AbandonedTaskSweeper
 from a2a.server.request_handlers import DefaultRequestHandler
 from a2a.server.routes import (
     add_a2a_routes_to_fastapi,
@@ -24,6 +25,7 @@
     SlowEchoAgent,
     build_agent_card,
     build_cluster_backends,
+    build_lease_store,
     default_dsn,
     init_schema,
 )
@@ -32,7 +34,14 @@
 logger = logging.getLogger(__name__)
 
 
-async def serve(host: str, port: int, replica_id: str, ticks: int) -> None:
+async def serve(  # noqa: PLR0913
+    host: str,
+    port: int,
+    replica_id: str,
+    ticks: int,
+    lease_ttl: float,
+    heartbeat_interval: float,
+) -> None:
     """Runs one replica bound to a shared store + event stream."""
     dsn = default_dsn()
     await init_schema(dsn)
@@ -42,12 +51,21 @@ async def serve(host: str, port: int, replica_id: str, ticks: int) -> None:
     store, stream = build_cluster_backends(dsn)
     await store.initialize()
     await stream.initialize()
+    leases = build_lease_store(dsn)
+    await leases.initialize()
 
     handler = DefaultRequestHandler(
         agent_executor=SlowEchoAgent(replica_id, ticks=ticks),
         task_store=store,
         agent_card=agent_card,
         event_stream=stream,
+        lease_store=leases,
+        lease_ttl=lease_ttl,
+        heartbeat_interval=heartbeat_interval,
+    )
+    # Fails tasks whose replica died mid-turn, once their lease expires.
+    sweeper = AbandonedTaskSweeper(
+        leases, store, interval=heartbeat_interval, event_stream=stream
     )
 
     jsonrpc_routes = create_jsonrpc_routes(
@@ -67,7 +85,14 @@ async def serve(host: str, port: int, replica_id: str, ticks: int) -> None:
         'Replica %s listening on %s (dsn=%s)', replica_id, base_url, dsn
     )
     config = uvicorn.Config(app, host=host, port=port, log_level='warning')
-    await uvicorn.Server(config).serve()
+    sweeper.start()
+    try:
+        await uvicorn.Server(config).serve()
+    finally:
+        await sweeper.aclose()
+        # Records FAILED for turns still running here, instead of leaving
+        # them for the sweeper on another replica.
+        await handler.aclose()
 
 
 def main() -> None:
@@ -78,9 +103,20 @@ def main() -> None:
     parser.add_argument('--port', type=int, default=41241)
     parser.add_argument('--replica-id', default='replica-0')
     parser.add_argument('--ticks', type=int, default=10)
+    parser.add_argument('--lease-ttl', type=float, default=10.0)
+    parser.add_argument('--heartbeat-interval', type=float, default=2.0)
     args = parser.parse_args()
     with contextlib.suppress(KeyboardInterrupt):
-        asyncio.run(serve(args.host, args.port, args.replica_id, args.ticks))
+        asyncio.run(
+            serve(
+                args.host,
+                args.port,
+                args.replica_id,
+                args.ticks,
+                args.lease_ttl,
+                args.heartbeat_interval,
+            )
+        )
 
 
 if __name__ == '__main__':

src/a2a/migrations/README.md

@@ -40,6 +40,8 @@ Always run this command after installing or upgrading the SDK to ensure your dat
 
 Revision `b5e3d1c8a2f7` creates two additive tables, `task_versions` and `task_events`, used only by the clustered multi-server stores (`a2a.server.cluster`). It leaves the `tasks` table untouched, so deployments that use the default `DatabaseTaskStore` need not run it and are unaffected. Run it only when adopting a `VersionedTaskStore` for multi-server deployment.
 
+Revision `b0f01b04ff26` creates the additive `task_leases` table, used only by `DatabaseTaskLeaseStore` (`a2a.server.cluster`) to recover tasks whose replica stopped while running them. Run it only when adopting a lease store.
+
 ```bash
 uv run a2a-db
 ```

src/a2a/migrations/versions/b0f01b04ff26_add_task_leases.py

@@ -0,0 +1,71 @@
+"""add task_leases table
+
+Revision ID: b0f01b04ff26
+Revises: b5e3d1c8a2f7
+Create Date: 2026-10-23 16:00:00.000000
+
+"""
+
+import logging
+
+from collections.abc import Sequence
+from typing import Union
+
+import sqlalchemy as sa
+
+
+try:
+    from alembic import context, op
+except ImportError as e:
+    raise ImportError(
+        "A2A migrations require the 'db-cli' extra. Install with: 'pip install a2a-sdk[db-cli]'."
+    ) from e
+
+from a2a.migrations.migration_utils import table_exists
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'b0f01b04ff26'
+down_revision: Union[str, Sequence[str], None] = 'b5e3d1c8a2f7'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema: create the task_leases table."""
+    leases_table = context.config.get_main_option(
+        'task_leases_table', 'task_leases'
+    )
+
+    if context.is_offline_mode() or not table_exists(leases_table):
+        op.create_table(
+            leases_table,
+            sa.Column('task_id', sa.String(36), primary_key=True),
+            sa.Column('owner', sa.String(255), nullable=True),
+            sa.Column('token', sa.String(32), nullable=False),
+            sa.Column('expires_at_ms', sa.BigInteger(), nullable=False),
+        )
+        op.create_index(
+            f'ix_{leases_table}_expires_at_ms',
+            leases_table,
+            ['expires_at_ms'],
+        )
+    else:
+        logging.info(
+            "Table '%s' already exists. Skipping creation.", leases_table
+        )
+
+
+def downgrade() -> None:
+    """Downgrade schema: drop the task_leases table."""
+    leases_table = context.config.get_main_option(
+        'task_leases_table', 'task_leases'
+    )
+
+    if context.is_offline_mode() or table_exists(leases_table):
+        op.drop_index(
+            f'ix_{leases_table}_expires_at_ms', table_name=leases_table
+        )
+        op.drop_table(leases_table)
+    else:
+        logging.info("Table '%s' does not exist. Skipping drop.", leases_table)