diff --git a/content/develop/ai/context-engine/agent-mesh/_index.md b/content/develop/ai/context-engine/agent-mesh/_index.md index b2f0fcee74..6613fbe33c 100644 --- a/content/develop/ai/context-engine/agent-mesh/_index.md +++ b/content/develop/ai/context-engine/agent-mesh/_index.md @@ -32,7 +32,7 @@ Resumable streams let you: - **Reconnect to the same work.** The application can keep publishing while the client is disconnected. The client returns with its saved cursor to receive retained updates it missed and follow new ones. - **Resume through another serving replica.** Redis gives replicas a shared event log. Replicas with the same Redis configuration and access policy can serve a reconnecting client, independently of which process produced the events. -Resumption recovers event delivery. Your application remains responsible for restarting work if the process executing it stops. +Resumption recovers event delivery. Your application remains responsible for restarting work if the process executing it stops. With A2A, another process can [take over a Task](/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md#hand-off-a-task-to-another-process) and finish it on the same event log. ## Choose an integration diff --git a/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md b/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md index 934a6b78a4..26fdf7eb36 100644 --- a/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md +++ b/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md @@ -120,6 +120,8 @@ A missing version header is interpreted as A2A 0.3 and rejected by the resumable Message-only replies and error frames have no Task-version ID. They must not advance the saved cursor. Task versions are not raw Redis stream entry IDs. +The route factory's `credential_expiry` callback and `max_connection_duration` limit also bound the wait for the first frame. If credentials expire before the server sends response headers, the request returns HTTP 401; refresh credentials before reconnecting. If the maximum connection duration expires first, the request returns HTTP 503 with `Retry-After: 1`; wait one second, then reconnect with your saved cursor. + ## Handle replay and completion | Condition | Behavior | @@ -141,10 +143,76 @@ For cancellation, use the official handler's cancellation path. A cancellation c `store.request_handler()` applies `store.heartbeat(executor, interval=5.0, alive_ttl=15.0)` to track the running agent under a process lease. It checks stored terminal state and stops local execution when another replica cancels or finishes the Task. Set `alive_ttl` greater than `interval`. -A store-backed subscriber or `get()` marks a submitted or working Task failed with a `producer_lost` status message when its producer lease expires. Tasks waiting for input or authentication, and tasks without tracked heartbeats, are excluded. +A store-backed subscriber or `get()` marks a submitted or working Task failed with a `producer_lost` status message when its producer lease expires. Tasks waiting for input or authentication, and tasks without tracked heartbeats, are excluded. To resume such a Task in another process instead, see [Hand off a task to another process](#hand-off-a-task-to-another-process). The wrapper does not restart computation. Without it, an agent on another replica learns of cancellation at its next write. With A2A SDK 1.2.3, continued emission after that cancellation can also leave handler shutdown waiting indefinitely; the wrapper stops the agent first. +## Hand off a task to another process + +One process can start a Task and another process can finish it. Both write to the same Task and its ordered event log, so subscribers on any replica see one sequence of updates, whichever process produced them. + +Before the heartbeat wrapper runs your executor, it claims the Task for its process. Each save carries that claim. After another process claims the Task, the store rejects saves from the previous process with `ConcurrentTaskModificationError`, so a process that stalls and later wakes up cannot overwrite newer work. That process also stops its executor at its next successful heartbeat. + +The wrapper grants a claim when the Task is new, when no process runs it, when it waits for input or authentication, or when the process that ran it stopped sending heartbeats. It refuses a claim for a terminal or deleted Task, a Task the caller doesn't own, and a Task that a process sending heartbeats is running or has just claimed. After a refused claim, the executor doesn't run, the Task keeps its current state, and the message is not added to the Task's history. If your executor's first update on a resumed Task is a status or artifact update, and you don't limit `history_length`, check the returned Task's history to tell whether your message was accepted. + +### Continue a waiting task in another process + +A Task that waits for input or authentication is not running anywhere. The next message can reach any replica. That replica's wrapper claims the Task and runs your executor with the stored Task as `context.current_task`. No extra configuration is required. While one replica is processing a reply, other replicas refuse a second reply to the same Task. + +### Take over a long-running task + +By default, a reader that notices a lost heartbeat marks a submitted or working Task failed with `producer_lost`. This includes the task read that a resume request makes. To let another process resume the Task instead, create the store with `takeover=True` on every replica: + +```python +store = RedisTaskStore(commands, namespace="myapp:a2a", takeover=True) +``` + +To resume a Task, send it a message through the official request handler in the process that takes it over. The handler reloads the Task, the wrapper claims it, and your executor runs with the stored Task. A worker can make that call in-process, using a `ServerCallContext` that carries the Task owner's verified identity: + +```python +import uuid + +from a2a.types import Message, Part, Role, SendMessageRequest + + +async def resume(handler, task_id, context_id, owner_context): + request = SendMessageRequest( + message=Message( + message_id=f"resume-{uuid.uuid4()}", + role=Role.ROLE_USER, + task_id=task_id, + context_id=context_id, + parts=[Part(text="resume")], + ) + ) + request.configuration.return_immediately = True + return await handler.on_message_send(request, owner_context) +``` + +Build `owner_context` in trusted worker code from the identity that owns the Task. Never build it from request input. The message becomes part of the Task's history. + +Have your executor record its progress in the Task, for example as one artifact per completed step, and skip completed steps when `context.current_task` already contains them: + +```python +from a2a.server.agent_execution import AgentExecutor + + +class TripPlanner(AgentExecutor): + async def execute(self, context, event_queue): + task = context.current_task + done = {artifact.artifact_id for artifact in task.artifacts} if task else set() + if "flights" not in done: + ... # book flights, then emit the "flights" artifact + if "hotels" not in done: + ... # book hotels, then emit the "hotels" artifact + ... # emit the completed status + + async def cancel(self, context, event_queue): + ... # emit the canceled status +``` + +The integration does not detect stopped processes or schedule resumption. Track the Tasks your workers run, for example in your job queue, and resume or cancel those whose process stopped. With `takeover=True`, a running Task that nobody resumes stays working until `active_ttl` expires, and its subscribers stay connected until then. A step that the first process performed but did not record runs again in the next process, so make external effects safe to repeat. + ## Configure retention and limits Pass these options to `RedisTaskStore`: @@ -159,6 +227,7 @@ Pass these options to `RedisTaskStore`: | `max_event_bytes` | 1,048,576 | Serialized event size limit. | | `max_events` | 10,000 | Approximate retained log length. | | `max_attempts` | 5 | Maximum attempts under transaction contention. | +| `takeover` | `False` | Keep a submitted or working Task claimable after its process stops sending heartbeats, instead of marking it failed. See [Hand off a task to another process](#hand-off-a-task-to-another-process). | Heartbeats do not extend Task retention. Set `active_ttl` longer than the longest interval without a saved update. @@ -172,7 +241,7 @@ Owner metadata prevents another owner from reusing a Task ID until it expires. D | `store.request_handler()` | Build the official handler with paired events and heartbeat tracking. | | `store.event_stream()` | Build the matching event reader, with producer-loss recovery. | | `RedisTaskEventStream` | Configure a reader explicitly with `store=store` and a separate client. | -| `store.heartbeat()` | Track execution liveness and notice remote cancellation. | +| `store.heartbeat()` | Track execution liveness, claim Tasks for hand-off, and notice remote cancellation. | | `create_resumable_rest_routes()` | Expose cursor-enabled REST streaming. | | `resumable_streams_extension()` | Advertise cursor support in the agent card. |