A2A Task Lifecycle States, in depth: the nine TaskState values, what the spec fixes versus server policy, how each channel shows transitions, a client reducer, watchdogs and conformance tests

The nine states and their three classes

A2A 1.0 defines the TaskState enum in its protobuf definition with nine values. One is the protobuf zero value; the other eight fall into three classes that decide what a client may do next.

The 0.3 specification used lowercase strings for the same idea (submitted, input-required, auth-required and so on) and also had an unknown state. If you talk to older agents, normalise both spellings at the edge of your client and keep the rest of the code on one vocabulary.

What the specification fixes, and what it leaves to the server

It is tempting to treat a state diagram as part of the protocol. The 1.0 specification is narrower, and the gap matters when you write clients that must work with agents you did not build. The spec fixes these points:

• Which states are terminal (completed, failed, canceled, rejected) and which are interrupted (input-required, auth-required).
• A task in a terminal state cannot accept further messages; sending one returns UnsupportedOperationError, and so does SubscribeToTask on a terminal task.
• CancelTask on a task that is not cancelable, for example one already completed, failed or canceled, returns TaskNotCancelableError.
• A SendStreamingMessage stream begins with the Task object (or contains exactly one Message if the agent replied directly), and it MUST close when the task reaches a terminal state. Subscription streams also end at a terminal state.
• Events are delivered in the order they were generated, on every protocol binding.
• A non-streaming SendMessage with returnImmediately false waits until the task is terminal or interrupted before returning.

What the spec does not give you is an edge table. Whether a server may go straight from submitted to failed, whether auth-required can be entered from submitted, or whether a cancel is honoured while input-required, are server decisions. It is also silent on whether a stream closes when the task becomes interrupted; some servers keep it open, others end it. A robust client accepts any move between non-terminal states, and treats only two things as violations: leaving a terminal state and the unspecified value.

How each channel shows the lifecycle

Note what 1.0's TaskStatusUpdateEvent contains: a task id, a context id, a TaskStatus (state, optional message, timestamp) and metadata. There is no end-of-stream flag on status updates, so a client detects the end from a terminal state or from the stream closing. Artifact updates carry append and lastChunk so a large artifact can arrive in pieces. Streaming details are in A2A streaming architecture and webhook handling in A2A push notifications.

A client-side reducer

The cleanest way to consume any of these channels is one reducer that folds every response into a local view. It replaces the view on a Task snapshot, applies status and artifact updates, and raises on the two genuine violations. Field names follow the 1.0 JSON form of StreamResponse.

TERMINAL = {"TASK_STATE_COMPLETED", "TASK_STATE_FAILED",
            "TASK_STATE_CANCELED", "TASK_STATE_REJECTED"}
INTERRUPTED = {"TASK_STATE_INPUT_REQUIRED", "TASK_STATE_AUTH_REQUIRED"}
ACTIVE = {"TASK_STATE_SUBMITTED", "TASK_STATE_WORKING"}

class ProtocolViolation(Exception):
    pass

class TaskView:
    """Client-side view of one task, built only from what the server sent."""
    def __init__(self):
        self.task_id = self.state = self.status_message = None
        self.artifacts = {}          # artifactId -> list of parts
        self.complete_artifacts = set()

    def apply(self, resp: dict):
        if "message" in resp:                  # direct reply, no task was created
            return "message"
        if "task" in resp:                     # snapshot: replaces local view
            t = resp["task"]
            self.task_id = t["id"]
            self._set_state(t["status"], snapshot=True)
            for a in t.get("artifacts", []):
                self.artifacts[a["artifactId"]] = list(a.get("parts", []))
        elif "statusUpdate" in resp:
            ev = resp["statusUpdate"]
            self._check_task(ev["taskId"])
            self._set_state(ev["status"])
        elif "artifactUpdate" in resp:
            ev = resp["artifactUpdate"]
            self._check_task(ev["taskId"])
            a = ev["artifact"]
            parts = self.artifacts.setdefault(a["artifactId"], [])
            if not ev.get("append"):
                parts.clear()                  # a non-append update replaces the artifact
            parts.extend(a.get("parts", []))
            if ev.get("lastChunk"):
                self.complete_artifacts.add(a["artifactId"])
        return self.state

    def _check_task(self, task_id):
        if self.task_id is not None and task_id != self.task_id:
            raise ProtocolViolation(f"event for {task_id} on stream for {self.task_id}")

    def _set_state(self, status, snapshot=False):
        new = status["state"]
        if new == "TASK_STATE_UNSPECIFIED":
            raise ProtocolViolation("server sent the unspecified state")
        if self.state in TERMINAL and new != self.state:
            raise ProtocolViolation(f"left terminal state {self.state} for {new}")
        self.state, self.status_message = new, status.get("message")

    @property
    def done(self):
        return self.state in TERMINAL

    @property
    def needs_caller(self):
        return self.state in INTERRUPTED

The same reducer handles a blocking response (one Task snapshot), a stream (snapshot then updates), a polled GetTask result wrapped as {"task": result}, and push payloads once normalised to the same shape, so the rest of the application asks only view.done and view.needs_caller. Replaying a status update or snapshot is harmless, which matters because push deliveries can repeat. Replaying an artifact update with append set is not: it would add the parts twice, so deduplicate such events or reconcile with a fresh GetTask snapshot when duplicates are possible.

Worked example: a booking with a question in the middle

An orchestrator delegates 'book the cheapest morning flight to Lisbon on 14 October' to a travel agent over a stream. The agent finds two equal fares and asks which departure the user prefers. The trace below shows the JSON-RPC results the caller receives, trimmed to the lifecycle fields.

{"result": {"task": {"id": "tsk_7", "contextId": "ctx_3",
  "status": {"state": "TASK_STATE_SUBMITTED"}}}}
{"result": {"statusUpdate": {"taskId": "tsk_7", "contextId": "ctx_3",
  "status": {"state": "TASK_STATE_WORKING"}}}}
{"result": {"statusUpdate": {"taskId": "tsk_7", "contextId": "ctx_3",
  "status": {"state": "TASK_STATE_INPUT_REQUIRED",
    "message": {"role": "ROLE_AGENT", "messageId": "m-2",
      "parts": [{"text": "Two flights match. Depart 07:10 or 09:45?"}]}}}}}
-- caller answers with SendStreamingMessage carrying taskId "tsk_7" --
{"result": {"task": {"id": "tsk_7", "contextId": "ctx_3",
  "status": {"state": "TASK_STATE_WORKING"}}}}
{"result": {"artifactUpdate": {"taskId": "tsk_7", "contextId": "ctx_3",
  "artifact": {"artifactId": "itin", "parts": [{"text": "Booked 09:45, ref Q7XK2"}]},
  "lastChunk": true}}}
{"result": {"statusUpdate": {"taskId": "tsk_7", "contextId": "ctx_3",
  "status": {"state": "TASK_STATE_COMPLETED"}}}}
-- stream closes --

Four lifecycle facts are visible. The first event is the Task snapshot, as the spec requires. The question arrives in the status message of the input-required update, which is where the caller should look for it; render it to the user or answer it from context. The answer is a new message that names the same taskId and context, not a new task. And completion is signalled by the terminal status update, after which the stream closes; the itinerary artifact arrived complete, flagged with lastChunk. Had the stream closed at input-required on this server, the caller would see the same events split across two streams, which the reducer handles identically.

Watchdogs for every non-terminal state

The protocol has no built-in timeouts, so both sides need watchdogs or tasks leak. Typical limits, to be tuned per agent:

The important rule is that the server should end every task it can no longer complete by moving it to a terminal state with a reason, rather than letting it sit. A client cannot distinguish 'still working' from 'worker died' without help.

Races the caller must expect

• Cancel versus completion. The caller sends CancelTask just as the agent finishes. Exactly one wins. If the caller gets TaskNotCancelableError, it must call GetTask and accept the actual terminal state, possibly completed with artifacts. Design details are in A2A task cancellation.
• Resume versus expiry. The caller answers an input-required question just after the server's TTL failed the task. The reply returns UnsupportedOperationError; the caller starts a new task in the same context, ideally quoting the earlier answer.
• Duplicate resume. A retried continuation message must not answer the question twice. Reuse the same messageId on retries so the server can deduplicate; see A2A idempotency.
• Push before snapshot. A webhook can arrive before the caller has stored the task id from the response. Buffer unknown-task notifications briefly, then reconcile with GetTask.

Conformance tests for a server

Before you depend on an agent, test its lifecycle behaviour, not just its happy path. These checks map directly to the specification's rules; the client and fixture names are placeholders for your SDK.

import pytest

async def test_terminal_task_rejects_messages(client, finished_task):
    with pytest.raises(UnsupportedOperationError):
        await client.send_message(task_id=finished_task.id, text="one more thing")

async def test_cancel_after_completion(client, finished_task):
    with pytest.raises(TaskNotCancelableError):
        await client.cancel_task(finished_task.id)
    assert (await client.get_task(finished_task.id)).status.state == "TASK_STATE_COMPLETED"

async def test_stream_starts_with_task_and_closes_on_terminal(client):
    events = [e async for e in client.send_streaming_message(text="quick job")]
    assert "task" in events[0] or "message" in events[0]
    view = TaskView()
    for e in events:
        view.apply(e)                       # raises on any illegal sequence
    assert view.done or "message" in events[0]

async def test_subscribe_to_terminal_task_is_refused(client, finished_task):
    with pytest.raises(UnsupportedOperationError):
        [e async for e in client.subscribe_to_task(finished_task.id)]

Add policy tests for the agent you run: a stuck worker ends in failed within the watchdog limit, an input-required task expires as documented, and failed tasks carry a status message a caller can act on. Error codes and their meanings are listed in A2A error handling.

Failure modes

• Treating interrupted as done. A blocking call returns at input-required, the caller reads empty artifacts and reports success. Always branch on the state class.
• Reopening finished tasks. Follow-ups after completion are new tasks in the same context. Code that sends them to the old task id fails on every compliant server.
• Hanging on an open stream. A client that waits for the stream to close while the task is input-required waits forever on servers that keep it open. Act on the state, not on the socket.
• Unbounded working. Servers without executor heartbeats leave tasks in working after a crash. Callers see silence and either hang or retry blindly.
• Version skew. A client that only knows input-required never matches TASK_STATE_INPUT_REQUIRED. Normalise names in one place.

What to do next

1. Write the state-class branch (active, interrupted, terminal) into every call site that reads a task.
2. Adopt one reducer for snapshots, stream events, polls and push payloads, raising only on terminal exits and the unspecified state.
3. Decide and document your server's edge policy, including whether streams close on interrupted states.
4. Add server watchdogs for submitted, working, input-required and auth-required, each ending in a terminal state with a reason.
5. Handle the cancel and completion race by reading GetTask after TaskNotCancelableError.
6. Run the conformance tests against every agent you depend on, and again on each version upgrade.