Skip to content

fix(server): let SubscribeToTask follow a task that awaits input - #395

Open
gulldan wants to merge 7 commits into
a2aproject:mainfrom
gulldan:fix/subscribe-interrupted-task
Open

gulldan wants to merge 7 commits into
a2aproject:mainfrom
gulldan:fix/subscribe-interrupted-task

Conversation

@gulldan

@gulldan gulldan commented Oct 29, 2026 •

Copy link
Copy Markdown
Contributor

Stack: #406 → #405 → #407 → #411 → #395. Stacked on #411: review only the last commit.

Fixes #393.

Problem

subscribe_to_task resubscribes only to a task whose execution is still registered. An execution that ends in an interrupted state (INPUT_REQUIRED, AUTH_REQUIRED) is unregistered as soon as the executor returns, and the task-store fallback from #277 turns only a terminal stored task into UnsupportedOperationError. So a stored task that waits for input falls through to TaskNotFoundError (-32001), although GetTask returns it.

§3.1.6: SubscribeToTask "can be used with any task that is not in a terminal state", and the stream MUST open with a Task carrying the current state. The official TCK builds on that: STREAM-SUB-001 (the first event is a Task), STREAM-SUB-002 (subscribe to an input-required task, send the follow-up message, the stream closes at the terminal state) and STREAM-ORDER-002..004 all create their task with tck-input-required. Against a2a-rs all five are skipped because the subscription errors.

Fix

ExecutionManager keeps, per task id, the subscriptions that found no registered execution (waiting). subscribe_to_task registers one under the registry lock before it reads the store, so an execution that registers from then on finds it: start subscribes every waiter to the new broadcast channel before the executor runs, so none of the turn's events is missed, and from there the stream is an ordinary resubscription (the task as the turn found it, the turn's events, closing where the execution ends). After the store read the errors are unchanged — a missing task is TaskNotFoundError, a terminal one UnsupportedOperationError — and an execution that attached meanwhile is served as is. Otherwise the stream opens with the stored Task and continues with the next execution's events once a follow-up message starts it.

A waiter is a Drop guard: a client that disconnects before the next turn removes its own entry, so nothing stays in the registry and no task is spawned for it. Lock order is the existing one (executions, then the new std::sync::Mutex around waiting, never held across an await). A subscription that is not polled while the next turn publishes more than EXECUTION_BUFFER_CAPACITY events lags and ends with the existing sanitized error, as a resubscription does.

§11.7 closes a turn's stream at an interrupted state. A subscription opened between turns has no turn to close with; following the next one is what STREAM-SUB-002 expects, and that stream still ends when the execution it attached to ends, interrupted or terminal, exactly as a resubscription to it would.

Tests

  • test_subscribe_to_a_task_awaiting_input_opens_with_the_stored_task — the regression; on main it fails with -32001.
  • test_subscriptions_to_a_task_awaiting_input_follow_its_next_turn — two subscribers, and the whole next turn runs before either stream is polled: both see Task INPUT_REQUIRED (stored), Task INPUT_REQUIRED (as the turn found it, with the follow-up in its history), COMPLETED, then close; a new subscription is then UnsupportedOperationError.
  • test_a_dropped_subscription_to_a_task_awaiting_input_unregisters; the unknown-task and terminal-task tests now also assert the registry is left empty.
  • just ci on Linux (stable 1.97.1): lockfile --locked, cargo hack check --all-targets --each-feature, cargo fmt --check, cargo hack clippy --all-targets --each-feature -- -D warnings, cargo test --workspace (doc tests included), copyright headers — all green.
  • Official TCK (a2aproject/a2a-tck d4d6fae, JSON-RPC + HTTP+JSON) against an agent built on the patched crates ([patch.crates-io] → this branch): MUST 69 passed / 4 failed / 20 skipped, before 64 / 4 / 25 — the same four failures as before (TCK-side, fix(a2a-server): SSE event ids, keep-alive, and a request body limit #202/fix(ci): exempt bot commits in bot-authored pull requests from DCO #231/chore: release #240), and STREAM-SUB-001, STREAM-SUB-002, STREAM-ORDER-002, STREAM-ORDER-003, STREAM-ORDER-004 pass instead of being skipped; STREAM-SUB-003/004 still pass.

Overlap with the A2A 1.1 stack

This touches ExecutionManager and subscribe_to_task, which #385–#392 rewrite (handler/live.rs, subscribe_stream). The stack tip (refs/pull/392/head) answers this case with the same -32001 — the regression test above fails on it the same way — so the fix is needed on either base. Happy to rebase onto the stack once it lands; the same idea fits its per-task change notifier and the subscribe_stream fallback.

🤖 Generated with Claude Code

@gulldan
gulldan requested a review from a team as a code owner October 29, 2026 13:56
gulldan and others added 7 commits October 31, 2026 13:20
The JSON-RPC transport never looked at the status of a response. An
error that something in front of the agent answers itself (an
authenticating proxy's 401, a gateway's 429 or 502) carries no
JSON-RPC envelope, so the caller got "failed to parse JSON-RPC
response" and could not tell a credential or rate-limit problem from
a broken server. A streaming call answered that way without a content
type was handed to the SSE parser and came back as an empty stream.

A body that is not a JSON-RPC envelope is now reported with its
status, as the REST transport already does ("HTTP 429 Too Many
Requests: <body>"); a JSON-RPC envelope is still decoded whatever the
status, and a non-success status is never treated as a stream.

Fixes: a2aproject#404

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: gulldan <5641375+gulldan@users.noreply.github.com>
The streaming test matched on the result and panicked in an `Ok` arm that a
passing run never reaches, so that line stayed uncovered. Taking the error
with `.err().expect()` keeps the assertion and needs no `Debug` on the stream.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: gulldan <5641375+gulldan@users.noreply.github.com>
google.protobuf.Value carries every number as a double, and
proto_value_to_json_value turned each one back into a JSON float, so
an integer in a data part or in metadata reached the executor, and a
client's decoded response, as 3600.0. serde then refuses it for an
integer type ("invalid type: floating point `3600.0`, expected u32").

A whole number within +-2^53, where a double is exact, is now given
back as an integer; other numbers stay floats. The round-trip tests
that used floats to work around this now use integers, and a new
protojson test decodes a JSON-RPC SendMessage whose data part holds an
integer and deserializes it into u32.

Fixes: a2aproject#399

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: gulldan <5641375+gulldan@users.noreply.github.com>
InMemoryTaskStore never removed a task, and TaskStore has no delete,
so a long-running agent kept every task it ever served: 20,000 tasks
with a 4 KiB artifact each held 192 MiB.

InMemoryTaskStore::with_terminal_task_limit(n) keeps at most n tasks
in a terminal state and purges the one that reached its terminal state
earliest. Tasks that are not terminal are never purged, and a purged
task reads as TaskNotFoundError, which A2A 3.3.2 allows for a task
that was completed and purged. new() keeps every task, as before. With
a limit of 1,000 the same run stays at about 20 MiB.

Fixes: a2aproject#401

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: gulldan <5641375+gulldan@users.noreply.github.com>
Only the JSON-RPC binding read A2A-Version: HTTP+JSON and gRPC served
a request with `A2A-Version: 99.0`, or with no version at all, as a
1.0 request, although 3.6.2 requires every interface to answer
VersionNotSupportedError and to read an empty value as 0.3.

The JSON-RPC check moves unchanged to middleware::check_a2a_version.
rest_router runs it ahead of every route and, when the header is
absent, reads the A2A-Version query parameter that 3.6.1 allows
instead; GrpcHandler runs it on the metadata of every method. Both
answer with the existing VERSION_NOT_SUPPORTED mapping (400 /
FAILED_PRECONDITION). Tests that call the transports directly now send
the version, as the JSON-RPC ones already do.

Fixes: a2aproject#410

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: gulldan <5641375+gulldan@users.noreply.github.com>
`subscribe_to_task` resubscribed only to a task whose execution was
still registered. An execution that ends in an interrupted state
(INPUT_REQUIRED, AUTH_REQUIRED) is unregistered as soon as the executor
returns, and the task-store fallback turned only a terminal stored task
into UnsupportedOperationError, so a stored task that waits for input
was reported as TaskNotFoundError. A2A §3.1.6 allows subscribing to
any non-terminal task, and the stream must open with the task as it
stands.

A subscription that finds no registered execution is now registered
with the execution manager, under its lock and before the store is
read, so an execution that registers from then on attaches it: `start`
subscribes every waiter before the executor runs, so none of that
turn's events is missed, and from there the stream is an ordinary
resubscription. The stream opens with the stored task and ends where
the execution it attached to ends. A missing task is still
TaskNotFoundError, a terminal one still UnsupportedOperationError, and
a waiter dropped first unregisters itself.

Tested with handler unit tests (a turn that asks for input and one that
completes; two subscribers polled only after the next turn ran; a
dropped subscription; the unchanged errors), `just ci` on Linux, and
the official TCK against an agent built on the patched crates, where
STREAM-SUB-001/002 and STREAM-ORDER-002..004 now pass instead of being
skipped.

Fixes: a2aproject#393

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: gulldan <5641375+gulldan@users.noreply.github.com>
@codecov

codecov Bot commented Oct 31, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 98.00885% with 9 lines in your changes missing coverage. Please review.

Files with missing lines Patch % Lines
a2a-server/src/handler.rs 95.52% 9 Missing ⚠️

📢 Thoughts on this report? Let us know!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

a2a-server: SubscribeToTask on an input-required task answers TaskNotFoundError

2 participants