From b144e28a60487e322a9a57b3c83887fe471f6136 Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 16:15:35 +1000 Subject: [PATCH 01/14] chore(claude): bump claude-agent-sdk to ^0.3.235 for background_tasks_changed --- a2a-claude/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/a2a-claude/package.json b/a2a-claude/package.json index edc68c2..f4e0991 100644 --- a/a2a-claude/package.json +++ b/a2a-claude/package.json @@ -55,7 +55,7 @@ "dependencies": { "@a2a-js/sdk": "^1.0.0", "@a2a-wrapper/core": "2.0.0", - "@anthropic-ai/claude-agent-sdk": "0.3.202", + "@anthropic-ai/claude-agent-sdk": "^0.3.235", "express": "^4.18.2", "uuid": "^9.0.0" }, From cb5ec7e86bcfc20553bd95c34c6c0b003442784a Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 16:17:31 +1000 Subject: [PATCH 02/14] feat(claude): add BackgroundTaskTracker with replace semantics Tracks the live set of Claude SDK background tasks from system/background_tasks_changed, which the SDK documents as a level signal with replace (not merge) semantics. Pure and dependency-free; not wired in yet. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018tTTwsL7XNgwm4iXHmxqfb --- .../claude/__tests__/background-tasks.test.ts | 65 +++++++++++++++++ a2a-claude/src/claude/background-tasks.ts | 72 +++++++++++++++++++ 2 files changed, 137 insertions(+) create mode 100644 a2a-claude/src/claude/__tests__/background-tasks.test.ts create mode 100644 a2a-claude/src/claude/background-tasks.ts diff --git a/a2a-claude/src/claude/__tests__/background-tasks.test.ts b/a2a-claude/src/claude/__tests__/background-tasks.test.ts new file mode 100644 index 0000000..701c42b --- /dev/null +++ b/a2a-claude/src/claude/__tests__/background-tasks.test.ts @@ -0,0 +1,65 @@ +import { describe, it, expect } from "vitest"; +import { BackgroundTaskTracker } from "../background-tasks.js"; +import type { SDKMessageLike } from "../client-factory.js"; + +function changed(...ids: string[]): SDKMessageLike { + return { + type: "system", + subtype: "background_tasks_changed", + tasks: ids.map((id) => ({ task_id: id, task_type: "shell", description: `task ${id}` })), + }; +} + +describe("BackgroundTaskTracker", () => { + it("starts empty", () => { + expect(new BackgroundTaskTracker().size).toBe(0); + }); + + it("replaces the set on each payload rather than merging", () => { + const t = new BackgroundTaskTracker(); + t.observe(changed("a", "b")); + expect(t.snapshot().map((x) => x.taskId).sort()).toEqual(["a", "b"]); + + t.observe(changed("b")); + expect(t.snapshot().map((x) => x.taskId)).toEqual(["b"]); + + t.observe(changed()); + expect(t.size).toBe(0); + }); + + it("reports whether membership changed", () => { + const t = new BackgroundTaskTracker(); + expect(t.observe(changed("a"))).toBe(true); + expect(t.observe(changed("a"))).toBe(false); + expect(t.observe(changed("a", "b"))).toBe(true); + expect(t.observe(changed())).toBe(true); + }); + + it("ignores every other message type, including the edge bookends", () => { + const t = new BackgroundTaskTracker(); + t.observe(changed("a")); + expect(t.observe({ type: "system", subtype: "task_started", task_id: "z" })).toBe(false); + expect(t.observe({ type: "system", subtype: "task_notification", task_id: "a", status: "completed" })).toBe(false); + expect(t.observe({ type: "result", subtype: "success", result: "hi" })).toBe(false); + expect(t.snapshot().map((x) => x.taskId)).toEqual(["a"]); + }); + + it("carries type and description through for status metadata", () => { + const t = new BackgroundTaskTracker(); + t.observe(changed("a")); + expect(t.snapshot()[0]).toEqual({ taskId: "a", type: "shell", description: "task a" }); + }); + + it("tolerates malformed payloads", () => { + const t = new BackgroundTaskTracker(); + t.observe({ type: "system", subtype: "background_tasks_changed", tasks: "nonsense" }); + expect(t.size).toBe(0); + + t.observe({ + type: "system", + subtype: "background_tasks_changed", + tasks: [{ task_id: "a" }, { description: "no id" }, null], + }); + expect(t.snapshot()).toEqual([{ taskId: "a", type: "unknown", description: "" }]); + }); +}); diff --git a/a2a-claude/src/claude/background-tasks.ts b/a2a-claude/src/claude/background-tasks.ts new file mode 100644 index 0000000..4b47e2f --- /dev/null +++ b/a2a-claude/src/claude/background-tasks.ts @@ -0,0 +1,72 @@ +/** + * Background Task Tracker — the live set of Claude's in-flight background work. + * + * Consumes only `system/background_tasks_changed`, which the SDK documents as a + * *level* signal with replace semantics: every payload carries the full set, so + * consumers swap their set rather than pairing `task_started` / + * `task_notification` edges. A missed bookend therefore cannot wedge a stale + * "still running" indicator, and the SDK explicitly leaves the level's ordering + * relative to those bookends unspecified — which is why they are ignored here. + * + * The level is per-process and nothing is emitted at startup, so a tracker must + * begin empty and be discarded when the CLI process goes away. That is enforced + * structurally: the executor creates one tracker per query, and a query is one + * CLI process, so instance lifetime is process lifetime. + */ + +import type { SDKMessageLike } from "./client-factory.js"; + +/** One live background task, in the shape the A2A status metadata carries. */ +export interface BackgroundTaskInfo { + taskId: string; + type: string; + description: string; +} + +export class BackgroundTaskTracker { + private live = new Map(); + + /** + * Fold one SDK message into the set. + * + * @returns `true` when set membership changed, so callers can emit a sideband + * event only on real transitions. + */ + observe(msg: SDKMessageLike): boolean { + if (msg.type !== "system" || msg.subtype !== "background_tasks_changed") return false; + + const raw = Array.isArray(msg.tasks) ? (msg.tasks as unknown[]) : []; + const next = new Map(); + + for (const entry of raw) { + if (entry === null || typeof entry !== "object") continue; + const task = entry as Record; + const taskId = typeof task.task_id === "string" ? task.task_id : ""; + if (!taskId) continue; + next.set(taskId, { + taskId, + type: typeof task.task_type === "string" ? task.task_type : "unknown", + description: typeof task.description === "string" ? task.description : "", + }); + } + + // Equal size plus every `next` key present in `live` forces set equality + // for finite sets (a same-size subset is the whole set), so this pair of + // checks alone is sufficient to detect any membership change — no need to + // walk `live`'s keys too. + const changed = + next.size !== this.live.size || [...next.keys()].some((id) => !this.live.has(id)); + this.live = next; + return changed; + } + + /** How many background tasks are live right now. */ + get size(): number { + return this.live.size; + } + + /** The live set, for status-update metadata. */ + snapshot(): BackgroundTaskInfo[] { + return [...this.live.values()]; + } +} From ddb7721dd8c8a60a1bdc5537acdc53e852bfeb40 Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 16:20:03 +1000 Subject: [PATCH 03/14] feat(claude): support streaming-input queries in the client interface Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018tTTwsL7XNgwm4iXHmxqfb --- .../src/claude/__tests__/fake-client.ts | 49 +++++++++++++++++-- .../claude/__tests__/prompt-builder.test.ts | 34 +++++++++++++ a2a-claude/src/claude/client-factory.ts | 29 +++++++++-- a2a-claude/src/claude/prompt-builder.ts | 24 +++++++++ 4 files changed, 129 insertions(+), 7 deletions(-) create mode 100644 a2a-claude/src/claude/__tests__/prompt-builder.test.ts diff --git a/a2a-claude/src/claude/__tests__/fake-client.ts b/a2a-claude/src/claude/__tests__/fake-client.ts index 96a78ed..9e3a43b 100644 --- a/a2a-claude/src/claude/__tests__/fake-client.ts +++ b/a2a-claude/src/claude/__tests__/fake-client.ts @@ -2,11 +2,23 @@ * Test fakes for ClaudeClientLike / QueryLike. */ -import type { ClaudeClientLike, QueryLike, QueryOptionsLike, SDKMessageLike } from "../client-factory.js"; +import type { + ClaudeClientLike, + QueryLike, + QueryOptionsLike, + SDKMessageLike, + SDKUserMessageLike, +} from "../client-factory.js"; export interface FakeCall { - prompt: string; + prompt: string | AsyncIterable; + /** Text of the first input message, whichever prompt form was used. */ + promptText: string; options: QueryOptionsLike; + /** True once the executor closed its input stream. Always true for a string prompt. */ + inputClosed: boolean; + /** Messages the executor pushed into the input stream. */ + inputMessages: SDKUserMessageLike[]; } export interface FakeTurnScript { @@ -76,8 +88,37 @@ export class FakeClaudeClient implements ClaudeClientLike { this.scripts = scripts; } - runQuery(prompt: string, options: QueryOptionsLike): QueryLike { - this.calls.push({ prompt, options }); + runQuery( + prompt: string | AsyncIterable, + options: QueryOptionsLike, + ): QueryLike { + const call: FakeCall = { + prompt, + promptText: typeof prompt === "string" ? prompt : "", + options, + inputClosed: typeof prompt === "string", + inputMessages: [], + }; + this.calls.push(call); + + // Drain the input stream the way the real SDK does, so tests can assert the + // executor closed it. The generator parks after its first message, so this + // loop stays pending until the executor resolves its deferred. + if (typeof prompt !== "string") { + void (async () => { + try { + for await (const msg of prompt) { + call.inputMessages.push(msg); + if (call.promptText === "") call.promptText = msg.message.content; + } + } catch { + // A rejected input stream is not something the executor should do; + // swallow it so an unhandled rejection cannot fail an unrelated test. + } + call.inputClosed = true; + })(); + } + const script = this.scripts[Math.min(this.calls.length - 1, this.scripts.length - 1)]; const q = new FakeQuery(script, options.abortController?.signal); this.queries.push(q); diff --git a/a2a-claude/src/claude/__tests__/prompt-builder.test.ts b/a2a-claude/src/claude/__tests__/prompt-builder.test.ts new file mode 100644 index 0000000..0ca6a27 --- /dev/null +++ b/a2a-claude/src/claude/__tests__/prompt-builder.test.ts @@ -0,0 +1,34 @@ +import { describe, it, expect } from "vitest"; +import { promptStream } from "../prompt-builder.js"; +import { createDeferred } from "@a2a-wrapper/core"; + +describe("promptStream", () => { + it("yields exactly one user message carrying the prompt text", async () => { + const closed = createDeferred(); + const it0 = promptStream("do the thing", closed.promise)[Symbol.asyncIterator](); + + const first = await it0.next(); + expect(first.done).toBe(false); + expect(first.value).toEqual({ + type: "user", + parent_tool_use_id: null, + message: { role: "user", content: "do the thing" }, + }); + }); + + it("parks after the first message and only ends once closed", async () => { + const closed = createDeferred(); + const it0 = promptStream("hi", closed.promise)[Symbol.asyncIterator](); + await it0.next(); + + const pending = it0.next(); + const raced = await Promise.race([ + pending.then(() => "ended"), + new Promise((r) => setTimeout(() => r("still-open"), 20)), + ]); + expect(raced).toBe("still-open"); + + closed.resolve(); + expect((await pending).done).toBe(true); + }); +}); diff --git a/a2a-claude/src/claude/client-factory.ts b/a2a-claude/src/claude/client-factory.ts index bdebb9a..8faa003 100644 --- a/a2a-claude/src/claude/client-factory.ts +++ b/a2a-claude/src/claude/client-factory.ts @@ -18,6 +18,17 @@ export interface SDKMessageLike { [key: string]: unknown; } +/** + * The one input-message shape this wrapper sends. Narrower than the SDK's + * `SDKUserMessage` on purpose: `uuid` and `session_id` are optional there, and + * everything else on it is for replay/subagent traffic we never originate. + */ +export interface SDKUserMessageLike { + type: "user"; + parent_tool_use_id: string | null; + message: { role: "user"; content: string }; +} + export interface QueryLike extends AsyncIterable { interrupt(): Promise; } @@ -50,7 +61,10 @@ export interface QueryOptionsLike { } export interface ClaudeClientLike { - runQuery(prompt: string, options: QueryOptionsLike): QueryLike; + runQuery( + prompt: string | AsyncIterable, + options: QueryOptionsLike, + ): QueryLike; } // ─── Option Mapping ────────────────────────────────────────────────────────── @@ -165,8 +179,17 @@ export function buildQueryOptions( */ export function createClaudeClient(_config: Required): ClaudeClientLike { return { - runQuery(prompt: string, options: QueryOptionsLike): QueryLike { - return query({ prompt, options: options as unknown as Options }) as unknown as QueryLike; + runQuery( + prompt: string | AsyncIterable, + options: QueryOptionsLike, + ): QueryLike { + // A string prompt makes the SDK close the CLI's stdin on the first result + // (`isSingleUserTurn`), which ends the process before any background-task + // wake could fire. Streaming input is what keeps that window open. + return query({ + prompt: prompt as Parameters[0]["prompt"], + options: options as unknown as Options, + }) as unknown as QueryLike; }, }; } diff --git a/a2a-claude/src/claude/prompt-builder.ts b/a2a-claude/src/claude/prompt-builder.ts index e783627..b29ff60 100644 --- a/a2a-claude/src/claude/prompt-builder.ts +++ b/a2a-claude/src/claude/prompt-builder.ts @@ -5,6 +5,30 @@ * (see `packages/core/src/events/part-utils.ts`) so this wrapper's * import paths stay stable. Inbound `Part` parsing is an A2A protocol * concern and lives in core, not here. + * + * Also owns `promptStream`, the SDK input stream for one A2A Task. */ +import type { SDKUserMessageLike } from "./client-factory.js"; + export { extractUserText } from "@a2a-wrapper/core"; + +/** + * The SDK input stream for one A2A Task: the user's prompt, then a park. + * + * Passing an async iterable (rather than a string) is what stops the SDK + * closing the CLI's stdin on the first result, which is the only reason a + * second turn — and therefore a background-task report — can ever arrive. + * Resolving `closed` ends the stream, which ends the CLI's input, which lets + * the process exit and the message iterator complete. + * + * The caller MUST resolve `closed` on every exit path or the generator parks + * forever. + */ +export async function* promptStream( + text: string, + closed: Promise, +): AsyncGenerator { + yield { type: "user", parent_tool_use_id: null, message: { role: "user", content: text } }; + await closed; +} From 478f5e82024986dbfd18f7f57626e12f6ae2f1b5 Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 16:21:46 +1000 Subject: [PATCH 04/14] feat(claude): add holdTaskForBackgroundWork and emitBackgroundTaskEvents flags Add two new feature flags to control background task lifecycle behavior. Both default to true: holdTaskForBackgroundWork holds the A2A Task open while background work is in flight; emitBackgroundTaskEvents publishes background-task set changes as sideband events. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018tTTwsL7XNgwm4iXHmxqfb --- a2a-claude/schemas/agent-config.schema.json | 8 ++++++++ a2a-claude/src/config/defaults.ts | 2 ++ a2a-claude/src/config/types.ts | 9 +++++++++ 3 files changed, 19 insertions(+) diff --git a/a2a-claude/schemas/agent-config.schema.json b/a2a-claude/schemas/agent-config.schema.json index 5053627..8e742a8 100644 --- a/a2a-claude/schemas/agent-config.schema.json +++ b/a2a-claude/schemas/agent-config.schema.json @@ -408,6 +408,10 @@ "FeatureFlags": { "additionalProperties": false, "properties": { + "emitBackgroundTaskEvents": { + "description": "Publish background-task set changes as sideband events. Default: true.", + "type": "boolean" + }, "emitFileChangeEvents": { "description": "Publish file change metadata as sideband events. Default: true.", "type": "boolean" @@ -428,6 +432,10 @@ "description": "Publish tool_call_start/end sideband events. Default: true.", "type": "boolean" }, + "holdTaskForBackgroundWork": { + "description": "Hold the A2A Task open in `working` while Claude has background work in flight, completing it only once a turn ends with nothing left running. Default: true. Set false to restore the previous behaviour of completing the Task at the first SDK result.", + "type": "boolean" + }, "streamArtifactChunks": { "description": "Stream artifact chunks (A2A spec-correct) vs single buffered artifact. Default: false.", "type": "boolean" diff --git a/a2a-claude/src/config/defaults.ts b/a2a-claude/src/config/defaults.ts index 2cc28b5..9949a06 100644 --- a/a2a-claude/src/config/defaults.ts +++ b/a2a-claude/src/config/defaults.ts @@ -48,6 +48,8 @@ export const DEFAULTS: Required = { emitFileChangeEvents: true, emitTodoEvents: true, emitRateLimitEvents: true, + holdTaskForBackgroundWork: true, + emitBackgroundTaskEvents: true, }, timeouts: { prompt: 600_000, diff --git a/a2a-claude/src/config/types.ts b/a2a-claude/src/config/types.ts index 94c9ef6..22c8828 100644 --- a/a2a-claude/src/config/types.ts +++ b/a2a-claude/src/config/types.ts @@ -178,6 +178,15 @@ export interface FeatureFlags { emitTodoEvents?: boolean; /** Publish rate-limit status changes as sideband events. Default: true. */ emitRateLimitEvents?: boolean; + /** + * Hold the A2A Task open in `working` while Claude has background work in + * flight, completing it only once a turn ends with nothing left running. + * Default: true. Set false to restore the previous behaviour of completing + * the Task at the first SDK result. + */ + holdTaskForBackgroundWork?: boolean; + /** Publish background-task set changes as sideband events. Default: true. */ + emitBackgroundTaskEvents?: boolean; } // ─── Timeout Config ───────────────────────────────────────────────────────── From 8040a38b7193ac3bff077192a3f3a94777254e94 Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 16:25:01 +1000 Subject: [PATCH 05/14] feat(claude): make agent lifecycle bookends per-task, add background_tasks event agent_started and agent_finished must fire exactly once per A2A Task, not once per SDK turn, now that a Task can span several turns while background work is in flight. EventMapper tracks sawInit/emittedFinished instance state to suppress duplicates, and handleResult takes a held flag so the executor can defer agent_finished until the turn that actually ends the Task. Also adds handleBackgroundTasks(), gated by features.emitBackgroundTaskEvents, so an orchestrator can see why a Task is sitting in working. Requires adding "background_tasks" to core's EventType union, following the same precedent as the rate_limit event type. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018tTTwsL7XNgwm4iXHmxqfb --- .../src/claude/__tests__/event-mapper.test.ts | 35 +++++++++++++++++ a2a-claude/src/claude/event-mapper.ts | 38 ++++++++++++++++++- packages/core/src/events/transport.ts | 3 +- 3 files changed, 73 insertions(+), 3 deletions(-) diff --git a/a2a-claude/src/claude/__tests__/event-mapper.test.ts b/a2a-claude/src/claude/__tests__/event-mapper.test.ts index 3d63f8b..fcc75f6 100644 --- a/a2a-claude/src/claude/__tests__/event-mapper.test.ts +++ b/a2a-claude/src/claude/__tests__/event-mapper.test.ts @@ -268,3 +268,38 @@ describe("sanitizeMessage", () => { expect(out.length).toBeLessThanOrEqual(2000); }); }); + +describe("EventMapper across a held-open task", () => { + it("emits agent_started once even when init is re-emitted on wake", () => { + const { mapper, emitted } = makeMapper(); + + const init = { type: "system", subtype: "init", model: "claude-test" }; + mapper.handleMessage(init); + mapper.handleMessage(init); + mapper.handleMessage(init); + + expect(emitted.filter((e) => e.event === "agent_started")).toHaveLength(1); + }); + + it("suppresses agent_finished while the task is held, emitting once at the end", () => { + const { mapper, emitted } = makeMapper(); + + const result = { type: "result", subtype: "success", result: "x", usage: {}, total_cost_usd: 0, num_turns: 1 }; + mapper.handleResult(result, { held: true }); + mapper.handleResult(result, { held: true }); + mapper.handleResult(result, { held: false }); + + expect(emitted.filter((e) => e.event === "agent_finished")).toHaveLength(1); + }); + + it("emits background_tasks when the flag is on and not when it is off", () => { + const { mapper: onMapper, emitted: on } = makeMapper(); + onMapper.handleBackgroundTasks([{ taskId: "a", type: "shell", description: "build" }]); + expect(on.filter((e) => e.event === "background_tasks")).toHaveLength(1); + expect(on[0].data).toMatchObject({ backend: "claude", count: 1 }); + + const { mapper: offMapper, emitted: off } = makeMapper({ emitBackgroundTaskEvents: false }); + offMapper.handleBackgroundTasks([{ taskId: "a", type: "shell", description: "build" }]); + expect(off).toHaveLength(0); + }); +}); diff --git a/a2a-claude/src/claude/event-mapper.ts b/a2a-claude/src/claude/event-mapper.ts index 2d8fce4..47fc6c2 100644 --- a/a2a-claude/src/claude/event-mapper.ts +++ b/a2a-claude/src/claude/event-mapper.ts @@ -13,6 +13,7 @@ import type { AgentEventEmitter } from "@a2a-wrapper/core"; import type { AgentConfig } from "../config/types.js"; +import type { BackgroundTaskInfo } from "./background-tasks.js"; import type { SDKMessageLike } from "./client-factory.js"; import type { RateLimitVerdict } from "./rate-limit-tracker.js"; import { logger } from "../utils/logger.js"; @@ -80,6 +81,16 @@ export class EventMapper { private readonly emitter: AgentEventEmitter; private readonly config: Required; + /** + * A background-task wake re-emits `system/init` for the same session, so + * without this an A2A Task that spans several SDK turns would emit + * `agent_started` once per turn. Both bookends are per-A2A-Task, and this + * mapper is constructed per `execute()` call, so instance state is the + * right scope. + */ + private sawInit = false; + private emittedFinished = false; + constructor(emitter: AgentEventEmitter, config: Required) { this.emitter = emitter; this.config = config; @@ -98,7 +109,7 @@ export class EventMapper { if (msg.parent_tool_use_id == null) this.handleUser(msg); break; case "result": - this.handleResult(msg); + this.handleResult(msg, { held: false }); break; case "stream_event": break; // consumed by the executor for artifact deltas @@ -138,8 +149,23 @@ export class EventMapper { }); } + /** + * Emit the live background-task set. Called by the executor only when + * membership actually changed, so this is a transition, not a heartbeat. + */ + handleBackgroundTasks(tasks: BackgroundTaskInfo[]): void { + if (!this.config.features.emitBackgroundTaskEvents) return; + this.emitter.emit("background_tasks", { + backend: "claude", + count: tasks.length, + tasks, + }); + } + private handleSystem(msg: SDKMessageLike): void { if (msg.subtype === "init") { + if (this.sawInit) return; + this.sawInit = true; this.emitter.emit("agent_started", { backend: "claude", model: typeof msg.model === "string" ? msg.model : "", @@ -257,8 +283,16 @@ export class EventMapper { } } - private handleResult(msg: SDKMessageLike): void { + /** + * @param opts.held - True when the executor is keeping the A2A Task open + * because background work is still in flight. `agent_finished` is a + * per-A2A-Task bookend, not a per-SDK-turn one, so it is suppressed until + * the turn that actually ends the Task. + */ + handleResult(msg: SDKMessageLike, opts: { held: boolean } = { held: false }): void { if (msg.subtype === "success") { + if (opts.held || this.emittedFinished) return; + this.emittedFinished = true; this.emitter.emit("agent_finished", { backend: "claude", usage: sanitizeData(msg.usage) ?? null, diff --git a/packages/core/src/events/transport.ts b/packages/core/src/events/transport.ts index c289005..cc44c57 100644 --- a/packages/core/src/events/transport.ts +++ b/packages/core/src/events/transport.ts @@ -54,7 +54,8 @@ export type EventType = | "agent_finished" | "agent_error" | "context_window" - | "rate_limit"; + | "rate_limit" + | "background_tasks"; /** * A single agent event carrying structured trace data. From 723afce5eec602f9cb109bce2091afc2c44c9c2f Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 16:33:39 +1000 Subject: [PATCH 06/14] feat(claude): hold the A2A task open while background work is in flight Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018tTTwsL7XNgwm4iXHmxqfb --- .../executor-background-tasks.test.ts | 214 ++++++++++++++++++ a2a-claude/src/claude/executor.ts | 151 +++++++++--- 2 files changed, 338 insertions(+), 27 deletions(-) create mode 100644 a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts diff --git a/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts new file mode 100644 index 0000000..886ad4b --- /dev/null +++ b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts @@ -0,0 +1,214 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { ClaudeExecutor } from "../executor.js"; +import { FakeClaudeClient } from "./fake-client.js"; +import type { SDKMessageLike } from "../client-factory.js"; +import { DEFAULTS } from "../../config/defaults.js"; +import type { AgentConfig } from "../../config/types.js"; +import type { RequestContext, ExecutionEventBus } from "@a2a-js/sdk/server"; +import { TaskState } from "@a2a-js/sdk"; + +const STATE_NAME: Partial> = { + [TaskState.TASK_STATE_SUBMITTED]: "submitted", + [TaskState.TASK_STATE_WORKING]: "working", + [TaskState.TASK_STATE_COMPLETED]: "completed", + [TaskState.TASK_STATE_CANCELED]: "canceled", + [TaskState.TASK_STATE_FAILED]: "failed", +}; + +interface PublishedEvent { + kind?: string; + data?: { status?: { state?: TaskState; message?: unknown }; [k: string]: unknown }; + [k: string]: unknown; +} + +function makeBus() { + const events: PublishedEvent[] = []; + let finishedCount = 0; + const bus = { + publish: (e: PublishedEvent) => { events.push(e); }, + finished: () => { finishedCount++; }, + on: () => bus, off: () => bus, once: () => bus, removeAllListeners: () => bus, + } as unknown as ExecutionEventBus; + return { bus, events, finished: () => finishedCount }; +} + +function makeCtx(taskId: string, contextId: string, text = "do the thing"): RequestContext { + return { + taskId, contextId, task: undefined, + userMessage: { + messageId: "m1", contextId, taskId, role: 1, + parts: [{ content: { $case: "text", value: text }, metadata: undefined }], + metadata: undefined, extensions: [], referenceTaskIds: [], + }, + } as unknown as RequestContext; +} + +const states = (events: PublishedEvent[]): string[] => + events.filter((e) => e.kind === "statusUpdate") + .map((e) => STATE_NAME[e.data?.status?.state as TaskState] ?? ""); + +const artifacts = (events: PublishedEvent[]): PublishedEvent[] => + events.filter((e) => e.kind === "artifactUpdate"); + +const init = (sessionId: string): SDKMessageLike => + ({ type: "system", subtype: "init", session_id: sessionId, model: "claude-test" }); + +const bgChanged = (...ids: string[]): SDKMessageLike => + ({ + type: "system", subtype: "background_tasks_changed", + tasks: ids.map((id) => ({ task_id: id, task_type: "shell", description: `task ${id}` })), + }); + +const result = (text: string): SDKMessageLike => + ({ + type: "result", subtype: "success", result: text, + usage: { input_tokens: 1, output_tokens: 1 }, total_cost_usd: 0.01, num_turns: 1, + }); + +const errorResult = (subtype: string): SDKMessageLike => + ({ type: "result", subtype, errors: ["boom"], usage: {}, total_cost_usd: 0, num_turns: 1 }); + +let ws: string; +let config: Required; + +beforeEach(() => { + ws = mkdtempSync(join(tmpdir(), "a2a-claude-bg-")); + config = JSON.parse(JSON.stringify({ ...DEFAULTS, configDir: ws })) as Required; + config.claude.workingDirectory = ws; + config.events = { enabled: false } as Required["events"]; +}); + +afterEach(() => rmSync(ws, { recursive: true, force: true })); + +describe("held-open A2A task", () => { + it("stays working when a result arrives with background work in flight", async () => { + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("build started"), bgChanged(), result("build passed")], + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + expect(states(events)).toEqual(["submitted", "working", "working", "completed"]); + expect(finished()).toBe(1); + }); + + it("publishes one artifact per round", async () => { + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("build started"), bgChanged(), result("build passed")], + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + const texts = artifacts(events).map((a) => JSON.stringify(a)); + expect(texts).toHaveLength(2); + expect(texts[0]).toContain("build started"); + expect(texts[1]).toContain("build passed"); + }); + + it("carries the live set as status metadata on the held update", async () => { + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("waiting"), bgChanged(), result("done")], + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + // [0] submitted, [1] working "Processing request...", [2] the held update. + const statusUpdates = events.filter((e) => e.kind === "statusUpdate"); + expect(states(events)).toEqual(["submitted", "working", "working", "completed"]); + const held = statusUpdates[2]; + expect(JSON.stringify(held)).toContain("bg1"); + }); + + it("loops for as many rounds as the chain needs", async () => { + const client = new FakeClaudeClient([{ + messages: [ + init("s1"), + bgChanged("bg1"), result("stage 1 running"), + bgChanged(), bgChanged("bg2"), result("stage 2 running"), + bgChanged(), result("both done"), + ], + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + expect(states(events)).toEqual(["submitted", "working", "working", "working", "completed"]); + expect(artifacts(events)).toHaveLength(3); + expect(finished()).toBe(1); + }); + + it("completes at the first result when nothing is in flight", async () => { + const client = new FakeClaudeClient([{ messages: [init("s1"), result("hello world")] }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + expect(states(events)).toEqual(["submitted", "working", "completed"]); + expect(artifacts(events)).toHaveLength(1); + expect(finished()).toBe(1); + }); + + it("completes at the first result when the flag is off", async () => { + config.features.holdTaskForBackgroundWork = false; + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("build started"), bgChanged(), result("never read")], + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + expect(states(events)).toEqual(["submitted", "working", "completed"]); + expect(artifacts(events)).toHaveLength(1); + }); + + it("closes the input stream on the success path", async () => { + const client = new FakeClaudeClient([{ messages: [init("s1"), result("done")] }]); + const ex = new ClaudeExecutor(config, () => client); + + await ex.execute(makeCtx("t1", "ctx-1"), makeBus().bus); + await new Promise((r) => setTimeout(r, 10)); + + expect(client.calls[0].inputClosed).toBe(true); + expect(client.calls[0].promptText).toBe("do the thing"); + }); + + it("closes the input stream when a result errors", async () => { + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), errorResult("error_max_turns")], + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + await new Promise((r) => setTimeout(r, 10)); + + expect(states(events)).toEqual(["submitted", "working", "failed"]); + expect(finished()).toBe(1); + expect(client.calls[0].inputClosed).toBe(true); + }); + + it("falls back to completing when the iterator ends while still held", async () => { + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("still waiting")], + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + expect(states(events)).toEqual(["submitted", "working", "working", "completed"]); + expect(finished()).toBe(1); + }); +}); diff --git a/a2a-claude/src/claude/executor.ts b/a2a-claude/src/claude/executor.ts index a9bc92f..c57a744 100644 --- a/a2a-claude/src/claude/executor.ts +++ b/a2a-claude/src/claude/executor.ts @@ -28,11 +28,13 @@ import { import type { RateLimitSnapshot } from "./rate-limit-tracker.js"; import { validateMcpServers, toClaudeMcpEntry } from "./mcp-adapter.js"; import { CLAUDE_BACKEND_PATHS } from "./backend-paths.js"; -import { extractUserText } from "./prompt-builder.js"; +import { extractUserText, promptStream } from "./prompt-builder.js"; +import { BackgroundTaskTracker } from "./background-tasks.js"; import { resolveTransport, AgentEventEmitter, + createDeferred, materializeMemory, bootstrapSubAgents, publishTask, @@ -260,6 +262,10 @@ export class ClaudeExecutor implements AgentExecutor { // A prompt timeout of 0 (or any value <= 0) disables the bound entirely: // the turn runs until the SDK iterator completes. Without this guard // setTimeout would coerce such a delay to the next tick and abort at once. + // + // Note this timer is armed once, at turn start, and never re-armed — so + // for a held-open task it bounds the whole A2A Task including the idle + // gaps between SDK turns. See the README caveat. const promptTimeout = this.config.timeouts.prompt ?? 600_000; const timer = promptTimeout > 0 @@ -275,9 +281,28 @@ export class ClaudeExecutor implements AgentExecutor { let rateLimited: RateLimitSnapshot | null = null; let finalText = ""; let streamArtifactStarted = false; - const streamArtifactId = `response-${taskId}`; const streaming = this.config.features.streamArtifactChunks === true; + // One artifact per round, so a held-open task's later rounds cannot + // append onto an earlier round's artifact and make its lastChunk + // marker's fullText a lie. + let round = 1; + const streamArtifactId = (): string => `response-${taskId}-${round}`; + + // Terminality is decided inside the loop now, so both the catch and the + // post-loop block need to know whether it already happened. + let terminalPublished = false; + + // One tracker per query. A query is one CLI process, and the SDK's + // background-task level signal is per-process, so this scoping is what + // makes "reset to empty when the process restarts" structural. + const backgroundTasks = new BackgroundTaskTracker(); + const holdEnabled = this.config.features.holdTaskForBackgroundWork !== false; + + // Resolving this ends the SDK input stream, which lets the CLI exit. + // It MUST be resolved on every exit path — see the finally block. + const inputClosed = createDeferred(); + /** Single definition of the rate-limit ending, used by both paths. */ const endTurnRateLimited = (snapshot: RateLimitSnapshot): void => { // Tear down the subprocess — same break-then-abort teardown the @@ -289,10 +314,10 @@ export class ClaudeExecutor implements AgentExecutor { abortController.abort(); // Already-sent chunks would otherwise leave the client's artifact - // open forever. This closes the stream; finalText is intentionally - // "" here, since no success result arrives on this path. + // open forever. This closes the current round's stream; finalText is + // intentionally "" here, since no success result arrives on this path. if (streaming && streamArtifactStarted) { - publishLastChunkMarker(bus, taskId, contextId, streamArtifactId, finalText); + publishLastChunkMarker(bus, taskId, contextId, streamArtifactId(), finalText); } // Always terminal. The SDK cannot resume an interrupted turn — a @@ -307,6 +332,7 @@ export class ClaudeExecutor implements AgentExecutor { true, rateLimitMetadata(snapshot), ); + terminalPublished = true; bus.finished(); }; @@ -317,7 +343,10 @@ export class ClaudeExecutor implements AgentExecutor { resume: session.sessionId ?? undefined, abortController, }); - const q = this.client!.runQuery(promptText, options); + // Streaming input, not a string: a string prompt makes the SDK close + // the CLI's stdin on the first result, ending the process before any + // background-task wake could fire. + const q = this.client!.runQuery(promptStream(promptText, inputClosed.promise), options); this.sessionManager!.attachQuery(taskId, q); let resultError: string | null = null; @@ -331,6 +360,10 @@ export class ClaudeExecutor implements AgentExecutor { break; } + if (backgroundTasks.observe(msg)) { + mapper.handleBackgroundTasks(backgroundTasks.snapshot()); + } + if (msg.type === "system" && msg.subtype === "init" && session.sessionId === null) { if (typeof msg.session_id === "string") session.sessionId = msg.session_id; } @@ -346,25 +379,64 @@ export class ClaudeExecutor implements AgentExecutor { const delta = event?.delta as Record | undefined; if (event?.type === "content_block_delta" && delta?.type === "text_delta" && typeof delta.text === "string") { streamArtifactStarted = true; - publishStreamingChunk(bus, taskId, contextId, streamArtifactId, delta.text); + publishStreamingChunk(bus, taskId, contextId, streamArtifactId(), delta.text); } } - if (msg.type === "result") { - if (msg.subtype === "success" && typeof msg.result === "string") { - finalText = msg.result; - } else if (msg.subtype !== "success") { - const reasons: Record = { - error_max_turns: "Turn limit reached (max_turns).", - error_max_budget_usd: "Budget limit reached (max_budget_usd).", - error_during_execution: "Error during execution.", - error_max_structured_output_retries: "Structured output retries exhausted.", - }; - resultError = reasons[String(msg.subtype)] ?? `Execution failed (${String(msg.subtype)}).`; - } + if (msg.type !== "result") { + mapper.handleMessage(msg); + continue; + } + + // ── A result message: decide whether this ends the A2A Task ── + if (msg.subtype === "success" && typeof msg.result === "string") { + finalText = msg.result; + } else if (msg.subtype !== "success") { + const reasons: Record = { + error_max_turns: "Turn limit reached (max_turns).", + error_max_budget_usd: "Budget limit reached (max_budget_usd).", + error_during_execution: "Error during execution.", + error_max_structured_output_retries: "Structured output retries exhausted.", + }; + resultError = reasons[String(msg.subtype)] ?? `Execution failed (${String(msg.subtype)}).`; + } + + const holding = holdEnabled && resultError === null && backgroundTasks.size > 0; + mapper.handleResult(msg, { held: holding }); + + if (resultError !== null) { + // The CLI stays alive on streaming input, so an error result would + // hang the loop unless we close the stream ourselves. + inputClosed.resolve(); + break; + } + + // This round's output, closed either way so the next round starts a + // fresh artifact. + if (streaming && streamArtifactStarted) { + publishLastChunkMarker(bus, taskId, contextId, streamArtifactId(), finalText); + } else if (finalText) { + publishFinalArtifact(bus, taskId, contextId, finalText); + } + streamArtifactStarted = false; + + if (holding) { + publishStatus( + bus, taskId, contextId, "working", + finalText || undefined, + false, + { backgroundTasks: backgroundTasks.snapshot() }, + ); + finalText = ""; + round += 1; + continue; } - mapper.handleMessage(msg); + publishStatus(bus, taskId, contextId, "completed", undefined, true); + terminalPublished = true; + bus.finished(); + inputClosed.resolve(); + break; } if (rateLimited) { @@ -374,18 +446,28 @@ export class ClaudeExecutor implements AgentExecutor { if (resultError) { publishStatus(bus, taskId, contextId, "failed", sanitizeMessage(resultError), true); + terminalPublished = true; bus.finished(); return; } - if (streaming && streamArtifactStarted) { - publishLastChunkMarker(bus, taskId, contextId, streamArtifactId, finalText); - } else { - publishFinalArtifact(bus, taskId, contextId, finalText); + if (!terminalPublished) { + // The iterator ended while we were still holding — the CLI died, or + // it closed input on us. Complete with whatever the last round left + // rather than hanging until the prompt timeout. + log.info("SDK iterator ended while the task was still held open", { + taskId, + liveBackgroundTasks: backgroundTasks.size, + }); + if (streaming && streamArtifactStarted) { + publishLastChunkMarker(bus, taskId, contextId, streamArtifactId(), finalText); + } else if (finalText) { + publishFinalArtifact(bus, taskId, contextId, finalText); + } + publishStatus(bus, taskId, contextId, "completed", undefined, true); + terminalPublished = true; + bus.finished(); } - - publishStatus(bus, taskId, contextId, "completed", undefined, true); - bus.finished(); } catch (err) { // A detected rate limit outranks whatever the teardown threw: the // `break` above awaits iterator.return(), so a failing teardown would @@ -401,6 +483,17 @@ export class ClaudeExecutor implements AgentExecutor { return; } + // We break out of the loop after publishing a terminal event, which + // awaits iterator.return(); a throw from that teardown must not + // produce a second, contradictory terminal event. + if (terminalPublished) { + log.debug("Ignoring teardown error after the task was already terminal", { + taskId, + error: err instanceof Error ? err.message : String(err), + }); + return; + } + const isAbort = err instanceof Error && (err.name === "AbortError" || err.message.includes("abort") || err.message.includes("canceled")); @@ -420,6 +513,10 @@ export class ClaudeExecutor implements AgentExecutor { bus.finished(); } } finally { + // The input generator parks forever if this never resolves, keeping + // the CLI subprocess alive. Every exit path lands here; resolving an + // already-resolved deferred is a no-op. + inputClosed.resolve(); if (timer) clearTimeout(timer); this.sessionManager?.untrackExecution(taskId); } From 273455e4da66b091ba9f030735ffb5f13bffe8a0 Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 16:49:16 +1000 Subject: [PATCH 07/14] refactor(claude): address review of held-open task lifecycle Follow-up to 723afce: - Post-loop fallback no longer reports `completed` for a turn the prompt timer aborted. Guarded with `!timedOut`, plus an explicit timed-out branch so that path still emits a terminal event. - Explain why the rate-limit `break` need not pre-resolve the input deferred, unlike the other two breaks. No behaviour change. - Extract `publishRoundArtifact()` and `endTurnCompleted()` closures, removing the two byte-identical repeats in `turnFn`. - Add opt-in `hangUntilInputClosed` to the test fake so a query models a real subprocess that only exits once stdin closes, and cover the held-task-never-wakes path with it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018tTTwsL7XNgwm4iXHmxqfb --- .../executor-background-tasks.test.ts | 22 +++++++ .../src/claude/__tests__/fake-client.ts | 37 ++++++++++- a2a-claude/src/claude/executor.ts | 65 ++++++++++++++----- 3 files changed, 106 insertions(+), 18 deletions(-) diff --git a/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts index 886ad4b..64a9650 100644 --- a/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts +++ b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts @@ -199,6 +199,28 @@ describe("held-open A2A task", () => { expect(client.calls[0].inputClosed).toBe(true); }); + it("releases the input stream of a held task whose CLI never wakes", async () => { + // `hangUntilInputClosed` models a real subprocess: it outlives its scripted + // output and only exits once stdin closes. The executor holds this task + // open (bg1 never clears) and never gets another message, so the prompt + // timeout is the only way out — and the input stream must still be closed + // on the way, or the subprocess would be wedged for good. + config.timeouts.prompt = 50; + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("waiting")], + hangUntilInputClosed: true, + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + await new Promise((r) => setTimeout(r, 10)); + + expect(states(events)).toEqual(["submitted", "working", "working", "failed"]); + expect(finished()).toBe(1); + expect(client.calls[0].inputClosed).toBe(true); + }); + it("falls back to completing when the iterator ends while still held", async () => { const client = new FakeClaudeClient([{ messages: [init("s1"), bgChanged("bg1"), result("still waiting")], diff --git a/a2a-claude/src/claude/__tests__/fake-client.ts b/a2a-claude/src/claude/__tests__/fake-client.ts index 9e3a43b..aa50041 100644 --- a/a2a-claude/src/claude/__tests__/fake-client.ts +++ b/a2a-claude/src/claude/__tests__/fake-client.ts @@ -28,6 +28,15 @@ export interface FakeTurnScript { delayMs?: number; /** After yielding messages, hang until aborted (for cancel/timeout tests). */ hangAfter?: boolean; + /** + * Model a real CLI: after the scripted messages, stay alive until the + * executor closes its input stream, then end the iterator. + * + * Opt-in, because the default (end as soon as the script is exhausted) is + * what most tests want. With this set, an executor that never resolves its + * input deferred wedges exactly the way a real subprocess would. + */ + hangUntilInputClosed?: boolean; /** * Make `iterator.return()` reject — what a consumer's `break` hits when the * SDK's teardown fails. A string customizes the error message. @@ -43,7 +52,12 @@ function abortError(): Error { class FakeQuery implements QueryLike { public interrupted = false; - constructor(private script: FakeTurnScript, private signal?: AbortSignal) {} + constructor( + private script: FakeTurnScript, + private signal?: AbortSignal, + /** Settles when the executor closes its input stream. */ + private inputClosed: Promise = Promise.resolve(), + ) {} async interrupt(): Promise { this.interrupted = true; @@ -70,6 +84,15 @@ class FakeQuery implements QueryLike { if (this.signal?.aborted) throw abortError(); yield msg; } + if (this.script.hangUntilInputClosed) { + // A real CLI exits when its stdin closes, not when it runs out of things + // to say. Aborting still tears it down mid-wait. + await new Promise((resolve, reject) => { + if (this.signal?.aborted) return reject(abortError()); + this.signal?.addEventListener("abort", () => reject(abortError()), { once: true }); + void this.inputClosed.then(resolve); + }); + } if (this.script.hangAfter) { await new Promise((_, reject) => { if (this.signal?.aborted) return reject(abortError()); @@ -101,10 +124,17 @@ export class FakeClaudeClient implements ClaudeClientLike { }; this.calls.push(call); + // A string prompt is closed the moment it is handed over; a stream is not + // closed until the executor resolves its deferred. + let markInputClosed!: () => void; + const inputClosed = new Promise((r) => { markInputClosed = r; }); + // Drain the input stream the way the real SDK does, so tests can assert the // executor closed it. The generator parks after its first message, so this // loop stays pending until the executor resolves its deferred. - if (typeof prompt !== "string") { + if (typeof prompt === "string") { + markInputClosed(); + } else { void (async () => { try { for await (const msg of prompt) { @@ -116,11 +146,12 @@ export class FakeClaudeClient implements ClaudeClientLike { // swallow it so an unhandled rejection cannot fail an unrelated test. } call.inputClosed = true; + markInputClosed(); })(); } const script = this.scripts[Math.min(this.calls.length - 1, this.scripts.length - 1)]; - const q = new FakeQuery(script, options.abortController?.signal); + const q = new FakeQuery(script, options.abortController?.signal, inputClosed); this.queries.push(q); return q; } diff --git a/a2a-claude/src/claude/executor.ts b/a2a-claude/src/claude/executor.ts index c57a744..edf52c3 100644 --- a/a2a-claude/src/claude/executor.ts +++ b/a2a-claude/src/claude/executor.ts @@ -336,6 +336,32 @@ export class ClaudeExecutor implements AgentExecutor { bus.finished(); }; + /** + * Emit this round's output artifact. + * + * Reads `streamArtifactStarted`/`finalText`/`round` at call time, so the + * in-loop caller and the post-loop fallback stay in lockstep by + * construction rather than by two copies agreeing. + * + * Deliberately NOT reused by `endTurnRateLimited`: that path must close + * an already-open stream without ever publishing a buffered artifact, + * so it keeps its streaming-only variant. + */ + const publishRoundArtifact = (): void => { + if (streaming && streamArtifactStarted) { + publishLastChunkMarker(bus, taskId, contextId, streamArtifactId(), finalText); + } else if (finalText) { + publishFinalArtifact(bus, taskId, contextId, finalText); + } + }; + + /** Single definition of the successful ending, mirroring endTurnRateLimited. */ + const endTurnCompleted = (): void => { + publishStatus(bus, taskId, contextId, "completed", undefined, true); + terminalPublished = true; + bus.finished(); + }; + try { publishStatus(bus, taskId, contextId, "working", "Processing request..."); @@ -357,6 +383,13 @@ export class ClaudeExecutor implements AgentExecutor { if (verdict.kind !== "none") mapper.handleRateLimit(verdict); if (verdict.kind === "rejected") { rateLimited = verdict.snapshot; + // Unlike the error-result and completed breaks below, this one + // does not pre-resolve `inputClosed`, and does not need to: the + // SDK launches its input pump fire-and-forget and `Query.return()` + // closes the transport without awaiting the parked input + // generator, so `break` cannot block on it. The `finally` still + // resolves it, which is what actually releases the generator. + // `endTurnRateLimited` then aborts, tearing down the subprocess. break; } @@ -413,11 +446,7 @@ export class ClaudeExecutor implements AgentExecutor { // This round's output, closed either way so the next round starts a // fresh artifact. - if (streaming && streamArtifactStarted) { - publishLastChunkMarker(bus, taskId, contextId, streamArtifactId(), finalText); - } else if (finalText) { - publishFinalArtifact(bus, taskId, contextId, finalText); - } + publishRoundArtifact(); streamArtifactStarted = false; if (holding) { @@ -432,9 +461,7 @@ export class ClaudeExecutor implements AgentExecutor { continue; } - publishStatus(bus, taskId, contextId, "completed", undefined, true); - terminalPublished = true; - bus.finished(); + endTurnCompleted(); inputClosed.resolve(); break; } @@ -451,7 +478,7 @@ export class ClaudeExecutor implements AgentExecutor { return; } - if (!terminalPublished) { + if (!terminalPublished && !timedOut) { // The iterator ended while we were still holding — the CLI died, or // it closed input on us. Complete with whatever the last round left // rather than hanging until the prompt timeout. @@ -459,12 +486,20 @@ export class ClaudeExecutor implements AgentExecutor { taskId, liveBackgroundTasks: backgroundTasks.size, }); - if (streaming && streamArtifactStarted) { - publishLastChunkMarker(bus, taskId, contextId, streamArtifactId(), finalText); - } else if (finalText) { - publishFinalArtifact(bus, taskId, contextId, finalText); - } - publishStatus(bus, taskId, contextId, "completed", undefined, true); + publishRoundArtifact(); + endTurnCompleted(); + } + + if (!terminalPublished) { + // `timedOut` is the only way to reach here: the timer's abort ended + // the iterator cleanly instead of throwing, so the catch's timeout + // branch never ran. Reporting `completed` would claim a turn that + // was cut short actually finished, and reporting nothing would end + // the Task with no terminal event at all — so publish the same + // failure the catch's timeout branch would have. + const msg = `Prompt timed out after ${promptTimeout}ms.`; + log.error("Task execution timed out", { taskId }); + publishStatus(bus, taskId, contextId, "failed", msg, true); terminalPublished = true; bus.finished(); } From 033ef4bc92c412e9d992bafe55416ea419a148c7 Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 17:48:09 +1000 Subject: [PATCH 08/14] fix(claude): never publish a terminal event on top of a cancel The post-loop fallback was guarded only on `timedOut`, so an abort that ended the iterator cleanly rather than throwing fell through to `completed`. After `cancelTask` had already published `canceled`, that handed the client two terminal events that contradict each other. Guard on `abortController.signal.aborted` instead, which covers both abort routes: a timeout publishes `failed` exactly once, and a cancel publishes nothing from `turnFn` because `cancelTask` already did. Both clean-end branches are now covered rather than defensive. Adds an opt-in `endCleanlyOnAbort` script flag to the test fake so the race is reachable at all; without it the fake always rejects on abort and the branches could not be exercised. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018tTTwsL7XNgwm4iXHmxqfb --- .../executor-background-tasks.test.ts | 46 +++++++++++++++++ .../src/claude/__tests__/fake-client.ts | 51 +++++++++++++++---- a2a-claude/src/claude/executor.ts | 26 +++++++--- 3 files changed, 105 insertions(+), 18 deletions(-) diff --git a/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts index 64a9650..7077b8f 100644 --- a/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts +++ b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts @@ -221,6 +221,52 @@ describe("held-open A2A task", () => { expect(client.calls[0].inputClosed).toBe(true); }); + it("publishes no second terminal event when a cancel ends the iterator cleanly", async () => { + // `endCleanlyOnAbort` models the race where the subprocess closes its + // stream just as the abort lands: the iterator ends normally, so none of + // the executor's abort handling runs and the post-loop fallback is what + // has to notice. The task is held (bg1 never clears), so nothing terminal + // was published in the loop — exactly the state where a naive fallback + // would emit `completed` on top of `cancelTask`'s `canceled`. + // + // One bus, as in production, so a contradictory pair is visible. + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("waiting")], + hangAfter: true, + endCleanlyOnAbort: true, + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + const p = ex.execute(makeCtx("t1", "ctx-1"), bus); + await new Promise((r) => setTimeout(r, 20)); + await ex.cancelTask("t1", bus); + await p; + + expect(states(events)).toEqual(["submitted", "working", "working", "canceled"]); + expect(finished()).toBe(1); + }); + + it("reports a timeout when the timer's abort ends the iterator cleanly", async () => { + // Same clean-end race as the cancel test above, but reached via the prompt + // timer. The catch's timeout branch never runs, so the post-loop block is + // the only thing standing between this and a Task that either claims it + // `completed` or never terminates at all. + config.timeouts.prompt = 50; + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("waiting")], + hangAfter: true, + endCleanlyOnAbort: true, + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + expect(states(events)).toEqual(["submitted", "working", "working", "failed"]); + expect(finished()).toBe(1); + }); + it("falls back to completing when the iterator ends while still held", async () => { const client = new FakeClaudeClient([{ messages: [init("s1"), bgChanged("bg1"), result("still waiting")], diff --git a/a2a-claude/src/claude/__tests__/fake-client.ts b/a2a-claude/src/claude/__tests__/fake-client.ts index aa50041..d8da9eb 100644 --- a/a2a-claude/src/claude/__tests__/fake-client.ts +++ b/a2a-claude/src/claude/__tests__/fake-client.ts @@ -37,6 +37,16 @@ export interface FakeTurnScript { * input deferred wedges exactly the way a real subprocess would. */ hangUntilInputClosed?: boolean; + /** + * On abort, end the iterator cleanly instead of rejecting with an + * `AbortError`. + * + * Opt-in, because rejecting is what the SDK normally does and what every + * other script here relies on. This models the race where the subprocess + * happens to close its stream just as the abort lands, so the consumer sees + * a normal end-of-iteration and none of its abort handling runs. + */ + endCleanlyOnAbort?: boolean; /** * Make `iterator.return()` reject — what a consumer's `break` hits when the * SDK's teardown fails. A string customizes the error message. @@ -77,27 +87,46 @@ class FakeQuery implements QueryLike { } as AsyncIterator; } + /** + * True once aborted — throwing `AbortError` unless the script asked for a + * clean end, in which case the caller should `return`. + */ + private abortedNow(): boolean { + if (!this.signal?.aborted) return false; + if (this.script.endCleanlyOnAbort) return true; + throw abortError(); + } + + /** + * Park until `until` settles, or until abort. Returns "aborted" only when the + * script opted into a clean end; otherwise abort rejects, as the SDK does. + */ + private park(until?: Promise): Promise<"settled" | "aborted"> { + return new Promise<"settled" | "aborted">((resolve, reject) => { + const onAbort = (): void => { + if (this.script.endCleanlyOnAbort) resolve("aborted"); + else reject(abortError()); + }; + if (this.signal?.aborted) return onAbort(); + this.signal?.addEventListener("abort", onAbort, { once: true }); + if (until) void until.then(() => resolve("settled")); + }); + } + private async *generate(): AsyncGenerator { for (const msg of this.script.messages) { - if (this.signal?.aborted) throw abortError(); + if (this.abortedNow()) return; if (this.script.delayMs) await new Promise((r) => setTimeout(r, this.script.delayMs)); - if (this.signal?.aborted) throw abortError(); + if (this.abortedNow()) return; yield msg; } if (this.script.hangUntilInputClosed) { // A real CLI exits when its stdin closes, not when it runs out of things // to say. Aborting still tears it down mid-wait. - await new Promise((resolve, reject) => { - if (this.signal?.aborted) return reject(abortError()); - this.signal?.addEventListener("abort", () => reject(abortError()), { once: true }); - void this.inputClosed.then(resolve); - }); + if ((await this.park(this.inputClosed)) === "aborted") return; } if (this.script.hangAfter) { - await new Promise((_, reject) => { - if (this.signal?.aborted) return reject(abortError()); - this.signal?.addEventListener("abort", () => reject(abortError()), { once: true }); - }); + if ((await this.park()) === "aborted") return; } } } diff --git a/a2a-claude/src/claude/executor.ts b/a2a-claude/src/claude/executor.ts index edf52c3..0f5ff2d 100644 --- a/a2a-claude/src/claude/executor.ts +++ b/a2a-claude/src/claude/executor.ts @@ -478,7 +478,15 @@ export class ClaudeExecutor implements AgentExecutor { return; } - if (!terminalPublished && !timedOut) { + // An abort can end the iterator *cleanly* rather than throwing, in + // which case none of the catch's abort handling runs and we have to + // reproduce it here. Everything that aborts is a reason the turn did + // not finish on its own, so none of them may report `completed`. + // (A rate-limit abort also lands here in principle, but that path + // returned above.) + const aborted = abortController.signal.aborted; + + if (!terminalPublished && !aborted) { // The iterator ended while we were still holding — the CLI died, or // it closed input on us. Complete with whatever the last round left // rather than hanging until the prompt timeout. @@ -490,12 +498,11 @@ export class ClaudeExecutor implements AgentExecutor { endTurnCompleted(); } - if (!terminalPublished) { - // `timedOut` is the only way to reach here: the timer's abort ended - // the iterator cleanly instead of throwing, so the catch's timeout - // branch never ran. Reporting `completed` would claim a turn that - // was cut short actually finished, and reporting nothing would end - // the Task with no terminal event at all — so publish the same + if (!terminalPublished && aborted && timedOut) { + // The timer's abort ended the iterator cleanly, so the catch's + // timeout branch never ran. Reporting `completed` would claim a turn + // that was cut short actually finished, and reporting nothing would + // end the Task with no terminal event at all — so publish the same // failure the catch's timeout branch would have. const msg = `Prompt timed out after ${promptTimeout}ms.`; log.error("Task execution timed out", { taskId }); @@ -503,6 +510,11 @@ export class ClaudeExecutor implements AgentExecutor { terminalPublished = true; bus.finished(); } + + // The remaining case — aborted, not timed out — is cancellation, and + // it deliberately publishes nothing: `cancelTask` already published + // `canceled` and called `bus.finished()`. Emitting `completed` here + // would hand the client two terminal events that contradict. } catch (err) { // A detected rate limit outranks whatever the teardown threw: the // `break` above awaits iterator.return(), so a failing teardown would From e45b99362fb1b8d28910b1c24df59d8f6666304c Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 17:49:53 +1000 Subject: [PATCH 09/14] test(claude): cover per-round streaming artifact ids --- .../executor-background-tasks.test.ts | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts index 7077b8f..dbe272d 100644 --- a/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts +++ b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts @@ -279,4 +279,40 @@ describe("held-open A2A task", () => { expect(states(events)).toEqual(["submitted", "working", "working", "completed"]); expect(finished()).toBe(1); }); + + it("gives each round its own streaming artifact id and lastChunk marker", async () => { + config.features.streamArtifactChunks = true; + const delta = (text: string): SDKMessageLike => ({ + type: "stream_event", + parent_tool_use_id: null, + event: { type: "content_block_delta", delta: { type: "text_delta", text } }, + }); + + const client = new FakeClaudeClient([{ + messages: [ + init("s1"), + bgChanged("bg1"), delta("build "), delta("started"), result("build started"), + bgChanged(), delta("build passed"), result("build passed"), + ], + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + const ids = artifacts(events).map( + (a) => (a.data as { artifact?: { artifactId?: string } }).artifact?.artifactId, + ); + expect(ids).toEqual([ + "response-t1-1", "response-t1-1", "response-t1-1", // 2 chunks + marker + "response-t1-2", "response-t1-2", // 1 chunk + marker + ]); + + const markers = artifacts(events).filter( + (a) => (a.data as { lastChunk?: boolean }).lastChunk === true, + ); + expect(markers).toHaveLength(2); + expect(JSON.stringify(markers[0])).toContain("build started"); + expect(JSON.stringify(markers[1])).toContain("build passed"); + }); }); From 13286e66f3c8a3a6a82138237c6397c813ebedbb Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 17:52:01 +1000 Subject: [PATCH 10/14] test(claude): add background-task lifecycle smoke scripts Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018tTTwsL7XNgwm4iXHmxqfb --- scripts/background-tasks-smoke/README.md | 31 ++++++ .../background-tasks-smoke/spike-chain.mjs | 93 ++++++++++++++++ .../background-tasks-smoke/spike-single.mjs | 103 ++++++++++++++++++ 3 files changed, 227 insertions(+) create mode 100644 scripts/background-tasks-smoke/README.md create mode 100644 scripts/background-tasks-smoke/spike-chain.mjs create mode 100644 scripts/background-tasks-smoke/spike-single.mjs diff --git a/scripts/background-tasks-smoke/README.md b/scripts/background-tasks-smoke/README.md new file mode 100644 index 0000000..b482ec7 --- /dev/null +++ b/scripts/background-tasks-smoke/README.md @@ -0,0 +1,31 @@ +# Background-task lifecycle smoke tests + +Manual end-to-end checks that Claude's background-task wake actually fires in +headless SDK mode. Unit tests use scripted fakes; only these run the real CLI. + +**These spend real quota** — roughly a minute of model time each — and need an +authenticated `claude` on PATH. They are deliberately not wired into `npm test`. + +## Running + +```bash +cd scripts/background-tasks-smoke +npm install @anthropic-ai/claude-agent-sdk@^0.3.235 +node spike-single.mjs # one background task: does a second result arrive at all +node spike-chain.mjs # two-stage chain: does the hold loop across rounds +``` + +## What to look for + +`spike-single.mjs` should show `background_tasks_changed` with one id, then +`RESULT #1`, then `background_tasks_changed []`, then `RESULT #2` — two results +on one query, with no second user message pushed. That is the whole premise of +the feature: the CLI wakes itself when background work settles. + +`spike-chain.mjs` mirrors the executor's hold-vs-complete decision inline and +should end with `decisions=["HOLD","HOLD","COMPLETE"]`. + +Both should show `session_state_changed` **never firing**. It is documented in +the SDK as the "authoritative turn-over signal", but it is not carried by the +stream-json transport, which is why the executor counts the background-task +level set instead. If it ever starts firing, revisit that decision. diff --git a/scripts/background-tasks-smoke/spike-chain.mjs b/scripts/background-tasks-smoke/spike-chain.mjs new file mode 100644 index 0000000..a029241 --- /dev/null +++ b/scripts/background-tasks-smoke/spike-chain.mjs @@ -0,0 +1,93 @@ +// Smoke test: two-stage background task chain. +// Proves that the hold loop spans multiple rounds: at each result, the executor +// decides HOLD if background tasks are still live, COMPLETE only when the set is empty. +// Also validates that bg_changed for a newly started task lands before its result. +// COSTS REAL QUOTA: ~2 minutes of model time. + +import { query } from "@anthropic-ai/claude-agent-sdk"; + +const t0 = Date.now(); +const log = (...a) => console.log(`[${String(Date.now() - t0).padStart(6)}ms]`, ...a); + +const PROMPT = `You will run a two-stage pipeline using background tasks. + +STAGE 1: Run exactly this as a background Bash task (run_in_background: true): + sleep 20 && echo STAGE_ONE_DONE +Do NOT wait for it or poll it. Immediately end your turn saying stage 1 is running. + +STAGE 2: When you are notified that stage 1 finished, immediately start exactly this +as a background Bash task (run_in_background: true): + sleep 20 && echo STAGE_TWO_DONE +Again do NOT wait or poll. Immediately end your turn saying stage 2 is running. + +FINALLY: When you are notified that stage 2 finished, report both stages and stop.`; + +let closeInput; +const closed = new Promise((r) => { closeInput = r; }); + +async function* input() { + yield { type: "user", parent_tool_use_id: null, message: { role: "user", content: PROMPT } }; + await closed; +} + +const stopTimer = setTimeout(() => { log("### hard stop — closing input"); closeInput(); }, 180_000); + +// Mirror the proposed executor logic exactly: track the level set, and at each +// result decide hold-vs-complete from it. +// Expected: decisions=["HOLD","HOLD","COMPLETE"] +const live = new Set(); +let results = 0; +const decisions = []; + +const q = query({ + prompt: input(), + options: { + cwd: process.cwd(), + // Bypass permissions only here: this is a sandboxed smoke test that needs + // to run unattended. Do not copy this pattern into production code. + permissionMode: "bypassPermissions", + allowDangerouslySkipPermissions: true, + settingSources: [], + strictMcpConfig: true, + }, +}); + +const clip = (s, n = 90) => (typeof s === "string" ? s.replace(/\s+/g, " ").slice(0, n) : ""); + +try { + for await (const m of q) { + const key = m.type + (m.subtype ? `/${m.subtype}` : ""); + + if (key === "system/background_tasks_changed") { + live.clear(); + for (const t of m.tasks ?? []) live.add(t.task_id); + log(`>>> bg_changed`, JSON.stringify([...live])); + } else if (key === "system/session_state_changed") { + log(`>>> session_state_changed`, m.state); + } else if (key === "system/init") { + log(` init (session ${m.session_id})`); + } else if (key === "system/task_notification") { + log(` task_notification`, m.task_id, m.status); + } else if (m.type === "result") { + results += 1; + const decision = live.size > 0 ? `HOLD (waiting on ${[...live].join(",")})` : "COMPLETE"; + decisions.push(decision); + log(`### RESULT #${results} (${m.subtype}) -> ${decision}`); + log(` text: ${JSON.stringify(clip(m.result, 110))}`); + if (live.size === 0) { log("### set empty at result — closing input"); closeInput(); } + } else if (m.type === "assistant") { + for (const b of m.message?.content ?? []) { + if (b.type === "tool_use") log(` tool_use`, b.name, JSON.stringify(clip(JSON.stringify(b.input), 80))); + } + } + } + log("### iterator completed normally"); +} catch (err) { + log("### iterator threw:", err?.name, clip(err?.message, 150)); +} finally { + clearTimeout(stopTimer); + closeInput(); +} + +log(`### results=${results} decisions=${JSON.stringify(decisions)}`); +process.exit(0); diff --git a/scripts/background-tasks-smoke/spike-single.mjs b/scripts/background-tasks-smoke/spike-single.mjs new file mode 100644 index 0000000..130753e --- /dev/null +++ b/scripts/background-tasks-smoke/spike-single.mjs @@ -0,0 +1,103 @@ +// Smoke test: one background task lifecycle. +// Proves that background_tasks_changed fires when a task finishes, +// and that a second result arrives on the same query (no second user message). +// COSTS REAL QUOTA: ~1 minute of model time. + +import { query } from "@anthropic-ai/claude-agent-sdk"; + +const t0 = Date.now(); +const log = (...a) => console.log(`[${String(Date.now() - t0).padStart(6)}ms]`, ...a); + +const PROMPT = `Run exactly this command as a background task (Bash with run_in_background: true): + +sleep 40 && echo BUILD_FINISHED + +Do NOT wait for it and do NOT poll it. Immediately end your turn with a single short +sentence saying you started it and are waiting for it to finish.`; + +let closeInput; +const closed = new Promise((r) => { closeInput = r; }); + +async function* input() { + yield { + type: "user", + parent_tool_use_id: null, + message: { role: "user", content: PROMPT }, + }; + await closed; +} + +const HARD_STOP_MS = 150_000; +const stopTimer = setTimeout(() => { + log("### hard stop reached — closing input stream"); + closeInput(); +}, HARD_STOP_MS); + +let resultCount = 0; + +const q = query({ + prompt: input(), + options: { + cwd: process.cwd(), + // Bypass permissions only here: this is a sandboxed smoke test that needs + // to run unattended. Do not copy this pattern into production code. + permissionMode: "bypassPermissions", + allowDangerouslySkipPermissions: true, + settingSources: [], + strictMcpConfig: true, + }, +}); + +const clip = (s, n = 90) => (typeof s === "string" ? s.replace(/\s+/g, " ").slice(0, n) : ""); + +try { + for await (const m of q) { + const key = m.type + (m.subtype ? `/${m.subtype}` : ""); + + switch (key) { + case "system/background_tasks_changed": + log(`>>> ${key}`, JSON.stringify((m.tasks ?? []).map((t) => t.task_id))); + break; + case "system/session_state_changed": + log(`>>> ${key}`, m.state); + break; + case "system/task_started": + log(` ${key}`, m.task_id, clip(m.description, 50)); + break; + case "system/task_notification": + log(`>>> ${key}`, m.task_id, m.status, clip(m.summary, 60)); + break; + case "system/init": + log(` ${key}`, "session", m.session_id); + break; + case "stream_event": + break; + default: { + if (m.type === "result") { + resultCount += 1; + log(`### RESULT #${resultCount} (${m.subtype})`, JSON.stringify(clip(m.result, 120))); + // Decide nothing here — just observe whether more arrive. + } else if (m.type === "assistant") { + const blocks = m.message?.content ?? []; + for (const b of blocks) { + if (b.type === "text" && b.text?.trim()) log(` assistant.text`, JSON.stringify(clip(b.text))); + if (b.type === "tool_use") log(` assistant.tool_use`, b.name, JSON.stringify(clip(JSON.stringify(b.input), 100))); + } + } else if (m.type === "user") { + log(` user`, `origin=${m.origin?.kind ?? "-"}`, m.isSynthetic ? "(synthetic)" : ""); + } else { + log(` ${key}`); + } + } + } + } + log("### iterator completed normally"); +} catch (err) { + log("### iterator threw:", err?.name, clip(err?.message, 150)); +} finally { + clearTimeout(stopTimer); + closeInput(); +} + +log(`### done — total results seen: ${resultCount}`); +process.exit(0); From f0344216d09e1604610f54636fedb6fdd347dfea Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 17:55:34 +1000 Subject: [PATCH 11/14] docs(claude): document the held-open task lifecycle --- .changeset/hold-task-for-background-work.md | 48 +++++++++++++ a2a-claude/README.md | 76 +++++++++++++++++++-- 2 files changed, 120 insertions(+), 4 deletions(-) create mode 100644 .changeset/hold-task-for-background-work.md diff --git a/.changeset/hold-task-for-background-work.md b/.changeset/hold-task-for-background-work.md new file mode 100644 index 0000000..bd5a062 --- /dev/null +++ b/.changeset/hold-task-for-background-work.md @@ -0,0 +1,48 @@ +--- +"a2a-claude": minor +"@a2a-wrapper/core": minor +--- + +Hold the A2A Task open while Claude has background work in flight, instead of +completing it the moment the first SDK turn ends. + +An A2A Task reached a terminal state as soon as Claude's first turn ended — +even when that turn started a background process and said it was waiting on +the result. A2A gives an agent no way to open a new turn against a terminal +Task, so the eventual follow-up report had nowhere to land: the client had +already been told the Task was done. + +The Task now stays in `working` for as long as Claude reports background work +in flight. Each SDK turn ("round") publishes its own `response` artifact plus +a non-final `working` status update whose `metadata.backgroundTasks` lists +what's still running (`taskId`, `type`, `description`). The Task only +completes once a round ends with nothing left. Chains of any length work this +way — check the build, start the deploy, report the result — as rounds of one +Task rather than a string of separate ones. + +This forced a change to how queries are issued: with a plain string prompt, +the SDK closes the CLI subprocess's stdin on the first result and the process +exits, making a second turn impossible at any SDK version. Queries now use +streaming-input mode instead, keeping the subprocess alive across rounds. A +per-query `BackgroundTaskTracker` follows the SDK's `background_tasks_changed` +message — a level signal with replace semantics, not a pair of start/stop +edges — to know what's still running. This required bumping +`@anthropic-ai/claude-agent-sdk` from `0.3.202` to `^0.3.235`, the first +version to emit that message. + +Both behaviors are gated by feature flags, on by default: +`features.holdTaskForBackgroundWork` (default `true`; set `false` to restore +the previous completes-at-first-result behavior) and +`features.emitBackgroundTaskEvents` (default `true`), which publishes a +`background_tasks` sideband event each time the live set changes. + +Two wire-visible changes worth knowing about even if you don't use the new +flags: a success result with empty text no longer publishes an empty +`response` artifact (the old code published one unconditionally), and +`agent_started` / `agent_finished` sideband events are now emitted once per +A2A Task rather than once per SDK turn — the SDK re-emits `system/init` on +every background-task wake, so without this a held-open Task would have +emitted `agent_started` several times over. + +Adds a `background_tasks` sideband event type to `@a2a-wrapper/core`, gated by +`features.emitBackgroundTaskEvents`. diff --git a/a2a-claude/README.md b/a2a-claude/README.md index 66c6522..06754ee 100644 --- a/a2a-claude/README.md +++ b/a2a-claude/README.md @@ -13,7 +13,7 @@ Claude Code is Anthropic's production-grade software engineering agent. It handl **Features:** - Native [A2A v1.0](https://a2a-protocol.org) protocol, backward compatible with v0.3.x clients — Agent Card, JSON-RPC, REST, streaming -- Powered by `@anthropic-ai/claude-agent-sdk` (pinned `0.3.202`) — `claude-sonnet-5`, `claude-opus-4-8`, and any SDK-compatible model +- Powered by `@anthropic-ai/claude-agent-sdk` (`^0.3.235`) — `claude-sonnet-5`, `claude-opus-4-8`, and any SDK-compatible model - Permission-mode guardrails — headless-safe modes only, with an explicit opt-in for unrestricted access - MCP tool support — stdio and Streamable HTTP transports - Multi-turn context continuity — each A2A `contextId` maps to a persistent Claude session (resumed via the SDK's `resume` option) @@ -200,7 +200,9 @@ Two more things worth knowing: "emitToolEvents": true, "emitFileChangeEvents": true, "emitTodoEvents": true, - "emitRateLimitEvents": true + "emitRateLimitEvents": true, + "holdTaskForBackgroundWork": true, + "emitBackgroundTaskEvents": true }, "timeouts": { @@ -360,6 +362,71 @@ sideband events with `action: "retrying"` carrying the SDK's `attempt`, `maxRetries`, and `delayMs`. Set a lower `timeouts.prompt` if a retry storm burning the window matters more to you than the retries succeeding. +### Background tasks + +Claude can start work that outlives a single SDK turn — a background shell +process, a long-running build — and end its turn saying it's waiting on the +result. Left alone, an A2A Task has no way to represent that: the Task reaches +a terminal state the moment the turn ends, and A2A gives an agent no way to +open a new turn against a terminal Task, so the eventual follow-up report +would have nowhere to land. + +`features.holdTaskForBackgroundWork` (default `true`) keeps the Task in +`working` for as long as Claude reports background work in flight, instead of +completing it at the first SDK result. Each SDK turn — a "round" — publishes +its own `response` artifact plus a non-final `working` status update whose +`metadata.backgroundTasks` lists what's still running (`taskId`, `type`, and +`description`, mirrored from the SDK's own `background_tasks_changed` +message). The Task only reaches a terminal state once a round ends with that +set empty. A chain of any length works as rounds of one Task rather than a +string of separate ones — check the build, kick off a deploy, report the +result. + +Set `holdTaskForBackgroundWork: false` to restore the previous behavior: the +Task completes at the first SDK result regardless of what Claude reports is +still running. + +`features.emitBackgroundTaskEvents` (default `true`) publishes a +`background_tasks` sideband event each time the live set changes, carrying the +same `taskId`/`type`/`description` list plus a `count`. See +[Sideband Events](#sideband-events). + +#### Caveats + +Four things worth knowing before relying on this. + +**`claude.maxTurns` now spans rounds.** A held-open Task accumulates SDK turns +across every round it takes, so a chain that used to run as several Tasks +under several separate budgets is now one Task under one budget. A +`maxTurns` that was comfortable before can be exhausted mid-chain, ending the +Task with `error_max_turns`. + +**Further messages on the same `contextId` queue behind a held-open Task.** +Turns are serialized per context — see [Prompt timeout](#prompt-timeout) — +so a Task that's waiting on background work blocks every later message on +that context, the same as any other slow turn would. `cancelTask` +(`tasks/cancel`) is currently the only way to release the queue early, and +there's a sharp edge worth calling out: the remedy a user would naturally +reach for — sending another message on the same context — is exactly what's +blocked. + +**`timeouts.prompt` bounds the whole Task, not one round.** The timer is +armed once at turn start and is never re-armed, so it also covers the idle +gaps between rounds while Claude's background work runs elsewhere. If you run +with a non-zero prompt timeout, raise it: the ten-minute default is usually +too low for a chain that holds the Task open, and a Task that runs out the +budget ends `failed` even if every round up to that point succeeded. + +**With `timeouts.prompt: 0`, a held-open Task has no automatic release.** +This is the sharpest edge of the four, and it applies directly to any +deployment that disables the prompt timeout. If the background-task set +never empties — and the SDK's `background_tasks_changed` level is the only +settle signal available, with no wake-up turn guaranteed to ever follow — the +Task stays in `working` indefinitely and holds its context's queue open with +it. `cancelTask` is the only escape. This is a known limitation of the +current design; a non-timeout release mechanism is planned. Operators running +with the prompt timeout disabled should monitor for Tasks stuck in `working`. + ## Example Agents | Config | Port | Permission mode | Description | @@ -429,15 +496,16 @@ Sideband events are published through `AgentEventEmitter` for every Claude Agent | Event | Emitted when | Notes | |---|---|---| -| `agent_started` | SDK `system`/`init` message | Includes `backend: "claude"` and the resolved model | +| `agent_started` | SDK `system`/`init` message | Includes `backend: "claude"` and the resolved model; emitted once per A2A Task, even across a held-open Task's several rounds — the SDK re-emits `init` on every background-task wake, so this bookend is deduplicated per Task rather than per SDK turn | | `thinking` | Assistant `thinking` content block | Controlled by `features.emitThinkingEvents` | | `tool_call_start` / `tool_call_end` | Assistant `tool_use` block / matching `tool_result` | `toolKind` is `"shell"` (Bash), `"mcp"`, `"a2a_subagent"` (mcp server `a2a-subagents`), or `"builtin"`; controlled by `features.emitToolEvents` | | `decision` (`kind: "file_change"`) | `Edit` / `Write` / `NotebookEdit` tool call | Path and change kind only — never file contents; controlled by `features.emitFileChangeEvents` | | `decision` (`kind: "todo_list"`) | `TodoWrite` tool call | Controlled by `features.emitTodoEvents` | | `decision` (`kind: "permission_denied"`) | SDK `system`/`permission_denied` message | Tool name + sanitized message | -| `agent_finished` | SDK `result`/`success` message | Includes sanitized `usage`, `totalCostUsd`, `numTurns` | +| `agent_finished` | SDK `result`/`success` message | Includes sanitized `usage`, `totalCostUsd`, `numTurns`; emitted once per A2A Task, on the round that finally completes it — not on every intermediate round of a held-open Task | | `agent_error` | SDK `result` failure subtypes / `error` message | Sanitized error message; reason mapped from the SDK's failure subtype (e.g. max turns, max budget) | | `rate_limit` | SDK `rate_limit_event`, `system`/`api_retry` with `error: "rate_limit"`, or an assistant `rate_limit` error | `action` is `"ended_turn"` (rejection — the turn stops), `"retrying"` (SDK internal retry, with the `retry` counters), or `"warning"`; carries `status` plus `rateLimitType` / `resetsAt` / `utilization` when the SDK reports them. Controlled by `features.emitRateLimitEvents` | +| `background_tasks` | SDK `system`/`background_tasks_changed` message | Level signal with replace semantics — each event carries the full live set (`taskId` / `type` / `description`) plus `count`, and is only emitted when membership actually changes. See [Background tasks](#background-tasks). Controlled by `features.emitBackgroundTaskEvents` | ## Docker From 8813d576a5f240c8492fb0013060205572f7761d Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Wed, 19 Aug 2026 18:16:49 +1000 Subject: [PATCH 12/14] fix(claude): deliver background_tasks events, close the agent_finished bookend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Six findings from the final review of the held-open task lifecycle. `background_tasks` had no entry in core's `EVENT_TO_TRACE_KEY`, and `A2ATransport.send` silently drops any event type it cannot map — so on the shipped defaults (`events.enabled: true`, `events.transport: "a2a"`) the event this branch documents was undeliverable. Adds a `trace.background_tasks` mapping so the README's claim is true. The sideband event is not redundant with `metadata.backgroundTasks` on the held status update: membership can change several times within one round, and only the event carries those transitions. `rate_limit` and `context_window` have the same gap; that is pre-existing and left alone, but now documented rather than silent. The post-loop "iterator ended while still held" fallback published `completed` without ever emitting `agent_finished`: the last round's result had been consumed with `{ held: true }`, which suppresses the bookend, and nothing re-emitted it. A consumer pairing bookends leaked a span. The fallback now closes it via `EventMapper.emitFinishedBookend`, which shares the `emittedFinished` latch so no path can double-fire. Test hardening, each proven by mutation: - Both "closes the input stream" tests were vacuous — the `finally` resolves `inputClosed` on every exit path, so deleting both in-loop resolves left every test green. Adds a `returnAwaitsInputClosed` script option whose `iterator.return()` drains the input pump, which wedges a `break` that did not pre-close the stream. The success and error breaks are now pinned by it; the rate-limit break, which deliberately does not pre-close, is pinned by a mirror test that runs without the option so only the `finally` can satisfy it. All three mutations now fail a test, and each fails a different one. - Nothing pinned the catch's `terminalPublished` guard. Adds the teardown-throw -after-completion case that produced a contradictory `failed` without it. - Nothing pinned the `else if (finalText)` artifact guard. Adds the empty-success-result case. Docs: the prompt-timeout section said it bounds a single turn, which stopped being true when `holdTaskForBackgroundWork` started defaulting on; the background-tasks smoke README claimed a `decisions` array the script has never printed; and `handleMessage`'s `case "result"` now carries a note that it is a secondary entry point with no hold state, so nobody re-routes results through it and loses suppression. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018tTTwsL7XNgwm4iXHmxqfb --- .changeset/hold-task-for-background-work.md | 5 +- a2a-claude/README.md | 15 +- .../src/claude/__tests__/event-mapper.test.ts | 46 ++++++ .../executor-background-tasks.test.ts | 148 +++++++++++++++++- .../src/claude/__tests__/fake-client.ts | 31 +++- a2a-claude/src/claude/event-mapper.ts | 39 +++++ a2a-claude/src/claude/executor.ts | 9 ++ .../src/__tests__/events/transport.test.ts | 70 +++++++++ packages/core/src/events/transport.ts | 11 +- scripts/background-tasks-smoke/README.md | 12 +- .../background-tasks-smoke/spike-chain.mjs | 3 +- 11 files changed, 373 insertions(+), 16 deletions(-) create mode 100644 packages/core/src/__tests__/events/transport.test.ts diff --git a/.changeset/hold-task-for-background-work.md b/.changeset/hold-task-for-background-work.md index bd5a062..8f206e0 100644 --- a/.changeset/hold-task-for-background-work.md +++ b/.changeset/hold-task-for-background-work.md @@ -45,4 +45,7 @@ every background-task wake, so without this a held-open Task would have emitted `agent_started` several times over. Adds a `background_tasks` sideband event type to `@a2a-wrapper/core`, gated by -`features.emitBackgroundTaskEvents`. +`features.emitBackgroundTaskEvents`. On the default `a2a` event transport it +is published as a trace artifact named `trace.background_tasks`, alongside the +existing `trace.lifecycle` / `trace.mcp` / `trace.thinking` / `trace.decision` +keys. diff --git a/a2a-claude/README.md b/a2a-claude/README.md index 06754ee..009cea0 100644 --- a/a2a-claude/README.md +++ b/a2a-claude/README.md @@ -217,11 +217,13 @@ Two more things worth knowing: ### Prompt timeout -`timeouts.prompt` bounds a single turn, in milliseconds (default `600000`, ten minutes). When it elapses the turn is aborted and the task is published as `failed`. +`timeouts.prompt` bounds one A2A task from start to terminal state, in milliseconds (default `600000`, ten minutes). When it elapses the task is aborted and published as `failed`. -Set it to `0` — or any value `<= 0` — to disable the bound entirely and let a turn run until it completes. This is the right setting for agents whose turns legitimately run for hours. +The timer is armed once, when the task starts, and is never re-armed. With `features.holdTaskForBackgroundWork` on (the default) a task can span several SDK turns, and this one budget covers all of them — including the idle gaps while Claude's background work runs elsewhere. See [Background tasks → Caveats](#caveats) before choosing a value. -Disabling it has one consequence worth knowing: turns are serialized per context, so a turn that never finishes holds its context's queue indefinitely and every later turn on the same `contextId` blocks behind it. Cancelling the task (`tasks/cancel`) still aborts the running turn and is the escape hatch. +Set it to `0` — or any value `<= 0` — to disable the bound entirely and let a task run until it completes. This is the right setting for agents whose turns legitimately run for hours. + +Disabling it has one consequence worth knowing: turns are serialized per context, so a turn that never finishes holds its context's queue indefinitely and every later turn on the same `contextId` blocks behind it. Cancelling the task (`tasks/cancel`) still aborts the running turn and is the escape hatch. With background-task holding on, that escape hatch is the *only* release — see [Background tasks → Caveats](#caveats). ### Session lifetime @@ -505,7 +507,12 @@ Sideband events are published through `AgentEventEmitter` for every Claude Agent | `agent_finished` | SDK `result`/`success` message | Includes sanitized `usage`, `totalCostUsd`, `numTurns`; emitted once per A2A Task, on the round that finally completes it — not on every intermediate round of a held-open Task | | `agent_error` | SDK `result` failure subtypes / `error` message | Sanitized error message; reason mapped from the SDK's failure subtype (e.g. max turns, max budget) | | `rate_limit` | SDK `rate_limit_event`, `system`/`api_retry` with `error: "rate_limit"`, or an assistant `rate_limit` error | `action` is `"ended_turn"` (rejection — the turn stops), `"retrying"` (SDK internal retry, with the `retry` counters), or `"warning"`; carries `status` plus `rateLimitType` / `resetsAt` / `utilization` when the SDK reports them. Controlled by `features.emitRateLimitEvents` | -| `background_tasks` | SDK `system`/`background_tasks_changed` message | Level signal with replace semantics — each event carries the full live set (`taskId` / `type` / `description`) plus `count`, and is only emitted when membership actually changes. See [Background tasks](#background-tasks). Controlled by `features.emitBackgroundTaskEvents` | +| `background_tasks` | SDK `system`/`background_tasks_changed` message | Level signal with replace semantics — each event carries the full live set (`taskId` / `type` / `description`) plus `count`, and is only emitted when membership actually changes. On the default `a2a` transport it arrives as a `trace.background_tasks` artifact. See [Background tasks](#background-tasks). Controlled by `features.emitBackgroundTaskEvents` | + +> **Note:** `rate_limit` (and `context_window`) have no A2A trace-artifact +> mapping, so on the default `a2a` transport they are dropped rather than +> delivered. They are observable on the `http` transport or a custom one. This +> is a pre-existing gap, tracked separately. ## Docker diff --git a/a2a-claude/src/claude/__tests__/event-mapper.test.ts b/a2a-claude/src/claude/__tests__/event-mapper.test.ts index fcc75f6..fde7aa8 100644 --- a/a2a-claude/src/claude/__tests__/event-mapper.test.ts +++ b/a2a-claude/src/claude/__tests__/event-mapper.test.ts @@ -292,6 +292,52 @@ describe("EventMapper across a held-open task", () => { expect(emitted.filter((e) => e.event === "agent_finished")).toHaveLength(1); }); + it("emits agent_finished once even when two unheld results arrive", () => { + // The `emittedFinished` latch is what lets the executor's post-loop + // fallback re-emit the bookend without risking a double-fire on a normal + // path. Pin it directly rather than trusting the claim. + const { mapper, emitted } = makeMapper(); + + const result = { type: "result", subtype: "success", result: "x", usage: {}, total_cost_usd: 0, num_turns: 1 }; + mapper.handleResult(result, { held: false }); + mapper.handleResult(result, { held: false }); + + expect(emitted.filter((e) => e.event === "agent_finished")).toHaveLength(1); + }); + + it("closes the bookend from the fallback, and only once", () => { + const { mapper, emitted } = makeMapper(); + + const result = { type: "result", subtype: "success", result: "x", usage: {}, total_cost_usd: 0.5, num_turns: 3 }; + mapper.handleResult(result, { held: true }); + mapper.emitFinishedBookend(result); + mapper.emitFinishedBookend(result); + + const finished = emitted.filter((e) => e.event === "agent_finished"); + expect(finished).toHaveLength(1); + expect(finished[0].data).toMatchObject({ totalCostUsd: 0.5, numTurns: 3 }); + }); + + it("does not re-emit the bookend from the fallback after a normal emit", () => { + const { mapper, emitted } = makeMapper(); + + const result = { type: "result", subtype: "success", result: "x", usage: {}, total_cost_usd: 0, num_turns: 1 }; + mapper.handleResult(result, { held: false }); + mapper.emitFinishedBookend(result); + + expect(emitted.filter((e) => e.event === "agent_finished")).toHaveLength(1); + }); + + it("still closes the bookend when no result ever arrived", () => { + const { mapper, emitted } = makeMapper(); + + mapper.emitFinishedBookend(null); + + const finished = emitted.filter((e) => e.event === "agent_finished"); + expect(finished).toHaveLength(1); + expect(finished[0].data).toMatchObject({ usage: null, totalCostUsd: null, numTurns: null }); + }); + it("emits background_tasks when the flag is on and not when it is off", () => { const { mapper: onMapper, emitted: on } = makeMapper(); onMapper.handleBackgroundTasks([{ taskId: "a", type: "shell", description: "build" }]); diff --git a/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts index dbe272d..5ddb30a 100644 --- a/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts +++ b/a2a-claude/src/claude/__tests__/executor-background-tasks.test.ts @@ -9,6 +9,7 @@ import { DEFAULTS } from "../../config/defaults.js"; import type { AgentConfig } from "../../config/types.js"; import type { RequestContext, ExecutionEventBus } from "@a2a-js/sdk/server"; import { TaskState } from "@a2a-js/sdk"; +import type { AgentEvent } from "@a2a-wrapper/core"; const STATE_NAME: Partial> = { [TaskState.TASK_STATE_SUBMITTED]: "submitted", @@ -71,6 +72,35 @@ const result = (text: string): SDKMessageLike => const errorResult = (subtype: string): SDKMessageLike => ({ type: "result", subtype, errors: ["boom"], usage: {}, total_cost_usd: 0, num_turns: 1 }); +const rejected = (): SDKMessageLike => + ({ + type: "rate_limit_event", + rate_limit_info: { + status: "rejected", rateLimitType: "five_hour", + resetsAt: Date.now() + 3_600_000, utilization: 1, + }, + }); + +/** + * Fail fast rather than hanging to vitest's default timeout. + * + * The input-stream tests below run against a transport whose `return()` + * drains the input pump. An executor that stops closing its stream before a + * `break` wedges there permanently — a wedge is exactly the failure those + * tests exist to catch, so it needs a bounded, legible signal. + */ +async function settleWithin(p: Promise, ms: number, what: string): Promise { + let timer: ReturnType | undefined; + const guard = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${what} did not settle within ${ms}ms`)), ms); + }); + try { + return await Promise.race([p, guard]); + } finally { + if (timer) clearTimeout(timer); + } +} + let ws: string; let config: Required; @@ -173,20 +203,57 @@ describe("held-open A2A task", () => { expect(artifacts(events)).toHaveLength(1); }); - it("closes the input stream on the success path", async () => { - const client = new FakeClaudeClient([{ messages: [init("s1"), result("done")] }]); + // ── Closing the SDK input stream ─────────────────────────────────────── + // + // Three sites decide when the stream closes, and the asymmetry between them + // is deliberate: the success and error `break`s close it first, the + // rate-limit `break` deliberately does not and leans on the `finally`. + // Asserting `inputClosed === true` alone cannot tell them apart — the + // `finally` makes that true on every exit path — so the first two tests run + // against `returnAwaitsInputClosed`, a transport that wedges on `break` + // unless the stream was already closed, and the third runs without it so + // that only the `finally` can satisfy it. + + it("closes the input stream before breaking on the success path", async () => { + const client = new FakeClaudeClient([{ + messages: [init("s1"), result("done")], + returnAwaitsInputClosed: true, + }]); const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); - await ex.execute(makeCtx("t1", "ctx-1"), makeBus().bus); - await new Promise((r) => setTimeout(r, 10)); + await settleWithin(ex.execute(makeCtx("t1", "ctx-1"), bus), 500, "execute()"); + expect(states(events)).toEqual(["submitted", "working", "completed"]); + expect(finished()).toBe(1); expect(client.calls[0].inputClosed).toBe(true); expect(client.calls[0].promptText).toBe("do the thing"); }); - it("closes the input stream when a result errors", async () => { + it("closes the input stream before breaking when a result errors", async () => { const client = new FakeClaudeClient([{ messages: [init("s1"), bgChanged("bg1"), errorResult("error_max_turns")], + returnAwaitsInputClosed: true, + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await settleWithin(ex.execute(makeCtx("t1", "ctx-1"), bus), 500, "execute()"); + + expect(states(events)).toEqual(["submitted", "working", "failed"]); + expect(finished()).toBe(1); + expect(client.calls[0].inputClosed).toBe(true); + }); + + it("leaves the rate-limit break's input stream to the finally", async () => { + // The mirror image of the two above: this path breaks without closing the + // stream, so the `finally` is the only thing that ever releases the input + // generator. Deliberately runs against the ordinary non-draining + // transport, matching the shipped SDK, where `break` cannot block on the + // input pump. + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), rejected()], + hangAfter: true, }]); const ex = new ClaudeExecutor(config, () => client); const { bus, events, finished } = makeBus(); @@ -280,6 +347,77 @@ describe("held-open A2A task", () => { expect(finished()).toBe(1); }); + it("emits the agent_finished bookend when the iterator ends while still held", async () => { + // The last round's result was consumed with `{ held: true }`, which + // suppressed the bookend, and this path never sees another result. If the + // fallback does not close it, the Task publishes `completed` while the + // trace stream shows `agent_started` with nothing pairing it. + config.events = { enabled: true } as Required["events"]; + const captured: AgentEvent[] = []; + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("still waiting")], + }]); + const ex = new ClaudeExecutor(config, () => client); + ex.customTransport = async (e: AgentEvent) => { captured.push(e); }; + const { bus, events } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + expect(states(events)).toEqual(["submitted", "working", "working", "completed"]); + expect(captured.filter((e) => e.eventType === "agent_started")).toHaveLength(1); + const finishedEvents = captured.filter((e) => e.eventType === "agent_finished"); + expect(finishedEvents).toHaveLength(1); + expect(finishedEvents[0].data).toMatchObject({ totalCostUsd: 0.01, numTurns: 1 }); + }); + + it("emits the agent_finished bookend exactly once on the ordinary path", async () => { + // The fallback re-emit must not double-fire when a round already closed + // the bookend normally. + config.events = { enabled: true } as Required["events"]; + const captured: AgentEvent[] = []; + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("round 1"), bgChanged(), result("round 2")], + }]); + const ex = new ClaudeExecutor(config, () => client); + ex.customTransport = async (e: AgentEvent) => { captured.push(e); }; + + await ex.execute(makeCtx("t1", "ctx-1"), makeBus().bus); + + expect(captured.filter((e) => e.eventType === "agent_finished")).toHaveLength(1); + }); + + it("publishes no second terminal event when the teardown throws after completing", async () => { + // `break` awaits iterator.return(); a throw there lands in the catch after + // `completed` was already published and `bus.finished()` already called. + // Without the catch's `terminalPublished` guard the client gets a + // contradictory `failed` on top of it. + const client = new FakeClaudeClient([{ + messages: [init("s1"), bgChanged("bg1"), result("round 1"), bgChanged(), result("round 2")], + throwOnReturn: "transport closed unexpectedly", + }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + expect(states(events)).toEqual(["submitted", "working", "working", "completed"]); + expect(finished()).toBe(1); + }); + + it("publishes no artifact for a success result with empty text", async () => { + // An empty `response` artifact is noise on the wire, and a client that + // renders every artifact shows a blank message for it. + const client = new FakeClaudeClient([{ messages: [init("s1"), result("")] }]); + const ex = new ClaudeExecutor(config, () => client); + const { bus, events, finished } = makeBus(); + + await ex.execute(makeCtx("t1", "ctx-1"), bus); + + expect(states(events)).toEqual(["submitted", "working", "completed"]); + expect(artifacts(events)).toHaveLength(0); + expect(finished()).toBe(1); + }); + it("gives each round its own streaming artifact id and lastChunk marker", async () => { config.features.streamArtifactChunks = true; const delta = (text: string): SDKMessageLike => ({ diff --git a/a2a-claude/src/claude/__tests__/fake-client.ts b/a2a-claude/src/claude/__tests__/fake-client.ts index d8da9eb..c5c979f 100644 --- a/a2a-claude/src/claude/__tests__/fake-client.ts +++ b/a2a-claude/src/claude/__tests__/fake-client.ts @@ -52,6 +52,23 @@ export interface FakeTurnScript { * SDK's teardown fails. A string customizes the error message. */ throwOnReturn?: boolean | string; + /** + * Make `iterator.return()` block until the input stream is closed — a + * transport whose teardown drains its input pump before completing. + * + * `break`ing out of a `for await` awaits `iterator.return()`, so under this + * model a consumer that breaks *without* first closing its input stream + * deadlocks: its own `finally` — the thing that would close the stream — + * cannot run until the `break` completes. That is what pins the executor's + * pre-`break` `inputClosed.resolve()` calls, which are otherwise + * indistinguishable from the unconditional resolve in its `finally`. + * + * Opt-in, because the SDK shipped today does not do this: `Query.return()` + * awaits `cleanup()` (transport close plus a bounded wait for process exit) + * and `streamInput` is launched fire-and-forget, so a real `break` cannot + * block on the input pump. Existing scripts keep the non-blocking default. + */ + returnAwaitsInputClosed?: boolean; } function abortError(): Error { @@ -76,13 +93,21 @@ class FakeQuery implements QueryLike { [Symbol.asyncIterator](): AsyncIterator { const gen = this.generate(); const failure = this.script.throwOnReturn; - if (!failure) return gen; + const drains = this.script.returnAwaitsInputClosed === true; + if (!failure && !drains) return gen; return { next: () => gen.next(), throw: (e?: unknown) => gen.throw(e), return: async (value?: unknown) => { - await gen.return(value as never).catch(() => {}); - throw new Error(typeof failure === "string" ? failure : "iterator teardown failed"); + // Drain first, then fail: a teardown that hangs never gets as far as + // reporting an error, and the ordering matters for scripts that set + // both. + if (drains) await this.inputClosed; + const done = await gen.return(value as never).catch(() => ({ done: true, value: undefined })); + if (failure) { + throw new Error(typeof failure === "string" ? failure : "iterator teardown failed"); + } + return done; }, } as AsyncIterator; } diff --git a/a2a-claude/src/claude/event-mapper.ts b/a2a-claude/src/claude/event-mapper.ts index 47fc6c2..9d682ef 100644 --- a/a2a-claude/src/claude/event-mapper.ts +++ b/a2a-claude/src/claude/event-mapper.ts @@ -108,6 +108,14 @@ export class EventMapper { case "user": if (msg.parent_tool_use_id == null) this.handleUser(msg); break; + // Secondary entry point only. The executor routes non-result messages + // here and calls `handleResult` directly for results, because only it + // knows whether the A2A Task is being held open. This case therefore + // hardcodes `{ held: false }` and would emit the `agent_finished` + // bookend on an intermediate round — do not re-route the executor's + // results through `handleMessage`, or hold suppression is silently + // lost. Kept because other callers (and tests) hand whole message + // streams to `handleMessage`. case "result": this.handleResult(msg, { held: false }); break; @@ -315,4 +323,35 @@ export class EventMapper { ...(errs.length > 0 ? { errors: errs } : {}), }); } + + /** + * Close the `agent_finished` bookend on a path that completes the A2A Task + * without an unheld result to carry it. + * + * The executor's post-loop fallback is the case: the SDK iterator ended + * while the Task was still held, so the last result was already consumed + * with `{ held: true }` and its bookend suppressed. That round is the one + * that completes the Task, so the bookend belongs to it — otherwise the + * trace stream shows `agent_started` with nothing closing it and a consumer + * pairing bookends leaks a span. + * + * Pass the last result seen so its usage figures survive; pass `null` when + * the iterator ended before any result arrived. Idempotent: shares the + * `emittedFinished` latch with {@link handleResult}, so a Task whose + * bookend already went out emits nothing here. + */ + emitFinishedBookend(lastResult: SDKMessageLike | null): void { + if (this.emittedFinished) return; + if (lastResult && lastResult.subtype === "success") { + this.handleResult(lastResult, { held: false }); + return; + } + this.emittedFinished = true; + this.emitter.emit("agent_finished", { + backend: "claude", + usage: null, + totalCostUsd: null, + numTurns: null, + }); + } } diff --git a/a2a-claude/src/claude/executor.ts b/a2a-claude/src/claude/executor.ts index 0f5ff2d..3e17f92 100644 --- a/a2a-claude/src/claude/executor.ts +++ b/a2a-claude/src/claude/executor.ts @@ -376,6 +376,9 @@ export class ClaudeExecutor implements AgentExecutor { this.sessionManager!.attachQuery(taskId, q); let resultError: string | null = null; + // The last result the loop saw, kept only so the post-loop fallback + // can close the `agent_finished` bookend with real usage figures. + let lastResult: SDKMessageLike | null = null; const rateLimits = new RateLimitTracker(); for await (const msg of q as AsyncIterable) { @@ -434,6 +437,7 @@ export class ClaudeExecutor implements AgentExecutor { resultError = reasons[String(msg.subtype)] ?? `Execution failed (${String(msg.subtype)}).`; } + lastResult = msg; const holding = holdEnabled && resultError === null && backgroundTasks.size > 0; mapper.handleResult(msg, { held: holding }); @@ -495,6 +499,11 @@ export class ClaudeExecutor implements AgentExecutor { liveBackgroundTasks: backgroundTasks.size, }); publishRoundArtifact(); + // This is the round that completes the Task, so it owes the + // `agent_finished` bookend — the last result was consumed with + // `{ held: true }`, which suppressed it. The mapper's latch keeps + // it to one per Task, so a path that already emitted is a no-op. + mapper.emitFinishedBookend(lastResult); endTurnCompleted(); } diff --git a/packages/core/src/__tests__/events/transport.test.ts b/packages/core/src/__tests__/events/transport.test.ts new file mode 100644 index 0000000..e797a46 --- /dev/null +++ b/packages/core/src/__tests__/events/transport.test.ts @@ -0,0 +1,70 @@ +import { describe, it, expect } from "vitest"; +import { A2ATransport } from "../../events/transport.js"; +import type { AgentEvent, EventType } from "../../events/transport.js"; +import type { ExecutionEventBus } from "@a2a-js/sdk/server"; + +/** + * `A2ATransport.send` drops any event type missing from its trace-key map, so + * "the transport accepted the call" is not evidence a client ever saw it. + * These tests assert on what actually reached the bus. + */ + +function createMockBus() { + const events: any[] = []; + const bus = { publish(e: any) { events.push(e); } } as unknown as ExecutionEventBus; + return { bus, events }; +} + +function event(eventType: EventType, data: Record = {}): AgentEvent { + return { + eventId: "e1", + eventType, + agentId: "agent-1", + agentName: "Agent One", + traceId: "trace-1", + parentAgentId: null, + timestamp: "2026-01-01T00:00:00.000Z", + data, + }; +} + +describe("A2ATransport", () => { + it("publishes background_tasks as a trace artifact on the default transport", async () => { + const { bus, events } = createMockBus(); + const transport = new A2ATransport(bus, "task-1", "ctx-1"); + + await transport.send( + event("background_tasks", { + backend: "claude", + count: 1, + tasks: [{ taskId: "bg1", type: "shell", description: "npm test" }], + }), + ); + + expect(events).toHaveLength(1); + const artifact = events[0].data.artifact; + expect(artifact.name).toBe("trace.background_tasks"); + expect(artifact.metadata.traceType).toBe("trace.background_tasks"); + const part = artifact.parts[0].content; + expect(part.$case).toBe("data"); + expect(part.value).toMatchObject({ + agent_id: "agent-1", + backend: "claude", + count: 1, + }); + expect(JSON.stringify(part.value)).toContain("bg1"); + }); + + it("still drops an event type with no trace-key mapping", async () => { + const { bus, events } = createMockBus(); + const transport = new A2ATransport(bus, "task-1", "ctx-1"); + + // rate_limit and context_window remain unmapped — a pre-existing gap this + // test documents rather than fixes, so the drop is a deliberate state and + // not an accident nobody noticed. + await transport.send(event("rate_limit", { status: "rejected" })); + await transport.send(event("context_window", { used: 1 })); + + expect(events).toHaveLength(0); + }); +}); diff --git a/packages/core/src/events/transport.ts b/packages/core/src/events/transport.ts index cc44c57..568c11d 100644 --- a/packages/core/src/events/transport.ts +++ b/packages/core/src/events/transport.ts @@ -211,7 +211,15 @@ class FunctionTransport implements EventTransport { // ─── Constants ─────────────────────────────────────────────────────────────── -/** Maps EventType → A2A trace artifact key. */ +/** + * Maps EventType → A2A trace artifact key. + * + * An event type absent from this map is silently dropped by + * {@link A2ATransport} — so anything documented as reaching a client on the + * default transport MUST have an entry here. `rate_limit` and + * `context_window` are still missing; they are only observable on a non-A2A + * transport today. + */ const EVENT_TO_TRACE_KEY: Record = { tool_call_start: "trace.mcp.start", tool_call_end: "trace.mcp", @@ -220,6 +228,7 @@ const EVENT_TO_TRACE_KEY: Record = { agent_started: "trace.lifecycle", agent_finished: "trace.lifecycle", agent_error: "trace.lifecycle", + background_tasks: "trace.background_tasks", }; /** Maps lifecycle EventType → state string. */ diff --git a/scripts/background-tasks-smoke/README.md b/scripts/background-tasks-smoke/README.md index b482ec7..ab9f193 100644 --- a/scripts/background-tasks-smoke/README.md +++ b/scripts/background-tasks-smoke/README.md @@ -23,7 +23,17 @@ on one query, with no second user message pushed. That is the whole premise of the feature: the CLI wakes itself when background work settles. `spike-chain.mjs` mirrors the executor's hold-vs-complete decision inline and -should end with `decisions=["HOLD","HOLD","COMPLETE"]`. +prints one decision per result, each `HOLD (waiting on )` or +`COMPLETE`. A healthy chain run ends on `COMPLETE` with at least one `HOLD` +before it — something like: + +``` +### results=3 decisions=["HOLD (waiting on bg_01…)","HOLD (waiting on bg_02…)","COMPLETE"] +``` + +The task ids are generated per run, so match on the shape rather than the +exact string. What matters is that at least one result was held and the last +one was not. Both should show `session_state_changed` **never firing**. It is documented in the SDK as the "authoritative turn-over signal", but it is not carried by the diff --git a/scripts/background-tasks-smoke/spike-chain.mjs b/scripts/background-tasks-smoke/spike-chain.mjs index a029241..af68ad2 100644 --- a/scripts/background-tasks-smoke/spike-chain.mjs +++ b/scripts/background-tasks-smoke/spike-chain.mjs @@ -34,7 +34,8 @@ const stopTimer = setTimeout(() => { log("### hard stop — closing input"); clo // Mirror the proposed executor logic exactly: track the level set, and at each // result decide hold-vs-complete from it. -// Expected: decisions=["HOLD","HOLD","COMPLETE"] +// Expected: one decision per result, at least one `HOLD (waiting on )` +// followed by a final `COMPLETE`. const live = new Set(); let results = 0; const decisions = []; From 14f471c6e753abe1006a6aee287bcc29ccd1b4d6 Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Tue, 25 Aug 2026 16:58:24 +1000 Subject: [PATCH 13/14] chore(claude): pin claude-agent-sdk to 0.3.245, clarify the hold flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review fixes ahead of the upstream PR. Pin the SDK exactly rather than `^0.3.235`. It is the only exact-pinned dependency in the package and was deliberately so: this is a pre-1.0 SDK whose message shapes change between patch releases — `0.3.202` did not emit `background_tasks_changed` at all, which is the whole reason for the bump. `0.3.235` is the floor, but `0.3.245` is what the feature was actually exercised against, so that is what gets pinned. The lockfile still recorded `0.3.202`, so `npm ci` would have installed an SDK that cannot emit the message this feature reads. Regenerating it also picks up unrelated stale entries the committed lockfile carried from the A2A v1.0 upgrade (workspace versions, core 1.7.0 → 2.0.0, npm `libc` fields). Narrow the `holdTaskForBackgroundWork` wording. The flag governs the Task completion decision only; streaming input is unconditional, so "restore the previous behaviour" over-promised a revert the flag does not perform. Trim the changeset to what a changelog reader needs. --- .changeset/hold-task-for-background-work.md | 75 ++++++-------- a2a-claude/README.md | 13 ++- a2a-claude/package.json | 2 +- a2a-claude/schemas/agent-config.schema.json | 2 +- a2a-claude/src/config/types.ts | 8 +- package-lock.json | 106 +++++++++++--------- scripts/background-tasks-smoke/README.md | 2 +- 7 files changed, 109 insertions(+), 99 deletions(-) diff --git a/.changeset/hold-task-for-background-work.md b/.changeset/hold-task-for-background-work.md index 8f206e0..2380bb3 100644 --- a/.changeset/hold-task-for-background-work.md +++ b/.changeset/hold-task-for-background-work.md @@ -3,49 +3,38 @@ "@a2a-wrapper/core": minor --- -Hold the A2A Task open while Claude has background work in flight, instead of -completing it the moment the first SDK turn ends. +Hold the A2A Task open while Claude has background work in flight. -An A2A Task reached a terminal state as soon as Claude's first turn ended — -even when that turn started a background process and said it was waiting on -the result. A2A gives an agent no way to open a new turn against a terminal -Task, so the eventual follow-up report had nowhere to land: the client had -already been told the Task was done. +A Task used to reach a terminal state as soon as Claude's first turn ended — +even when that turn had just started a background process and said it was +waiting on the result. A2A gives an agent no way to open a new turn against a +terminal Task, so the follow-up report had nowhere to land. The Task now stays in `working` for as long as Claude reports background work -in flight. Each SDK turn ("round") publishes its own `response` artifact plus -a non-final `working` status update whose `metadata.backgroundTasks` lists -what's still running (`taskId`, `type`, `description`). The Task only -completes once a round ends with nothing left. Chains of any length work this -way — check the build, start the deploy, report the result — as rounds of one -Task rather than a string of separate ones. - -This forced a change to how queries are issued: with a plain string prompt, -the SDK closes the CLI subprocess's stdin on the first result and the process -exits, making a second turn impossible at any SDK version. Queries now use -streaming-input mode instead, keeping the subprocess alive across rounds. A -per-query `BackgroundTaskTracker` follows the SDK's `background_tasks_changed` -message — a level signal with replace semantics, not a pair of start/stop -edges — to know what's still running. This required bumping -`@anthropic-ai/claude-agent-sdk` from `0.3.202` to `^0.3.235`, the first -version to emit that message. - -Both behaviors are gated by feature flags, on by default: -`features.holdTaskForBackgroundWork` (default `true`; set `false` to restore -the previous completes-at-first-result behavior) and -`features.emitBackgroundTaskEvents` (default `true`), which publishes a -`background_tasks` sideband event each time the live set changes. - -Two wire-visible changes worth knowing about even if you don't use the new -flags: a success result with empty text no longer publishes an empty -`response` artifact (the old code published one unconditionally), and -`agent_started` / `agent_finished` sideband events are now emitted once per -A2A Task rather than once per SDK turn — the SDK re-emits `system/init` on -every background-task wake, so without this a held-open Task would have -emitted `agent_started` several times over. - -Adds a `background_tasks` sideband event type to `@a2a-wrapper/core`, gated by -`features.emitBackgroundTaskEvents`. On the default `a2a` event transport it -is published as a trace artifact named `trace.background_tasks`, alongside the -existing `trace.lifecycle` / `trace.mcp` / `trace.thinking` / `trace.decision` -keys. +running, and completes only once a turn ends with nothing left. Each turn +publishes its own `response` artifact and a non-final `working` status update +whose `metadata.backgroundTasks` lists what is still in flight. Chains of any +length work this way, as rounds of one Task rather than several Tasks. + +Controlled by `features.holdTaskForBackgroundWork` (default `true`; set +`false` for the old complete-at-first-result behavior) and +`features.emitBackgroundTaskEvents` (default `true`), which publishes a new +`background_tasks` sideband event — added to `@a2a-wrapper/core` — each time +the live set changes. + +Bumps `@anthropic-ai/claude-agent-sdk` from `0.3.202` to `0.3.245`. The +feature needs at least `0.3.235`, the first version to emit +`background_tasks_changed`. + +Three changes apply even with `holdTaskForBackgroundWork` off: + +- Queries now use streaming input rather than a string prompt. A string prompt + makes the SDK close the CLI subprocess's stdin on the first result, which + ends the process before a second round is possible. This is not switchable. +- `agent_started` / `agent_finished` are emitted once per A2A Task rather than + once per SDK turn. +- A success result with empty text no longer publishes an empty `response` + artifact. + +See the a2a-claude README for caveats, including how `claude.maxTurns` and +`timeouts.prompt` now span a held-open Task's rounds. diff --git a/a2a-claude/README.md b/a2a-claude/README.md index 009cea0..c218feb 100644 --- a/a2a-claude/README.md +++ b/a2a-claude/README.md @@ -13,7 +13,7 @@ Claude Code is Anthropic's production-grade software engineering agent. It handl **Features:** - Native [A2A v1.0](https://a2a-protocol.org) protocol, backward compatible with v0.3.x clients — Agent Card, JSON-RPC, REST, streaming -- Powered by `@anthropic-ai/claude-agent-sdk` (`^0.3.235`) — `claude-sonnet-5`, `claude-opus-4-8`, and any SDK-compatible model +- Powered by `@anthropic-ai/claude-agent-sdk` (pinned `0.3.245`) — `claude-sonnet-5`, `claude-opus-4-8`, and any SDK-compatible model - Permission-mode guardrails — headless-safe modes only, with an explicit opt-in for unrestricted access - MCP tool support — stdio and Streamable HTTP transports - Multi-turn context continuity — each A2A `contextId` maps to a persistent Claude session (resumed via the SDK's `resume` option) @@ -384,9 +384,14 @@ set empty. A chain of any length works as rounds of one Task rather than a string of separate ones — check the build, kick off a deploy, report the result. -Set `holdTaskForBackgroundWork: false` to restore the previous behavior: the -Task completes at the first SDK result regardless of what Claude reports is -still running. +Set `holdTaskForBackgroundWork: false` to complete the Task at the first SDK +result, as before, regardless of what Claude reports is still running. + +The flag governs that completion decision and nothing else. Queries are issued +in streaming-input mode either way: with a plain string prompt the SDK closes +the CLI subprocess's stdin on the first result and the process exits, so +streaming input is what makes a second round possible at all. There is no +setting that reverts it. `features.emitBackgroundTaskEvents` (default `true`) publishes a `background_tasks` sideband event each time the live set changes, carrying the diff --git a/a2a-claude/package.json b/a2a-claude/package.json index f4e0991..b9d754f 100644 --- a/a2a-claude/package.json +++ b/a2a-claude/package.json @@ -55,7 +55,7 @@ "dependencies": { "@a2a-js/sdk": "^1.0.0", "@a2a-wrapper/core": "2.0.0", - "@anthropic-ai/claude-agent-sdk": "^0.3.235", + "@anthropic-ai/claude-agent-sdk": "0.3.245", "express": "^4.18.2", "uuid": "^9.0.0" }, diff --git a/a2a-claude/schemas/agent-config.schema.json b/a2a-claude/schemas/agent-config.schema.json index 8e742a8..96f5160 100644 --- a/a2a-claude/schemas/agent-config.schema.json +++ b/a2a-claude/schemas/agent-config.schema.json @@ -433,7 +433,7 @@ "type": "boolean" }, "holdTaskForBackgroundWork": { - "description": "Hold the A2A Task open in `working` while Claude has background work in flight, completing it only once a turn ends with nothing left running. Default: true. Set false to restore the previous behaviour of completing the Task at the first SDK result.", + "description": "Hold the A2A Task open in `working` while Claude has background work in flight, completing it only once a turn ends with nothing left running. Default: true. Set false to complete the Task at the first SDK result, as before. Governs the completion decision only — queries use streaming input either way.", "type": "boolean" }, "streamArtifactChunks": { diff --git a/a2a-claude/src/config/types.ts b/a2a-claude/src/config/types.ts index 22c8828..20c1d1f 100644 --- a/a2a-claude/src/config/types.ts +++ b/a2a-claude/src/config/types.ts @@ -181,8 +181,12 @@ export interface FeatureFlags { /** * Hold the A2A Task open in `working` while Claude has background work in * flight, completing it only once a turn ends with nothing left running. - * Default: true. Set false to restore the previous behaviour of completing - * the Task at the first SDK result. + * Default: true. Set false to complete the Task at the first SDK result, as + * before. + * + * This governs the completion decision only. Queries are issued in + * streaming-input mode either way — that is what keeps the CLI subprocess + * alive past the first result, and it is not switchable. */ holdTaskForBackgroundWork?: boolean; /** Publish background-task set changes as sideband events. Default: true. */ diff --git a/package-lock.json b/package-lock.json index 1dd65c6..98c8127 100644 --- a/package-lock.json +++ b/package-lock.json @@ -20,11 +20,11 @@ } }, "a2a-antigravity": { - "version": "0.1.1", + "version": "0.2.0", "license": "MIT", "dependencies": { "@a2a-js/sdk": "^1.0.0", - "@a2a-wrapper/core": "1.7.0", + "@a2a-wrapper/core": "2.0.0", "express": "^4.18.2", "uuid": "^9.0.0" }, @@ -57,12 +57,12 @@ } }, "a2a-claude": { - "version": "0.2.0", + "version": "0.3.0", "license": "MIT", "dependencies": { "@a2a-js/sdk": "^1.0.0", - "@a2a-wrapper/core": "1.7.0", - "@anthropic-ai/claude-agent-sdk": "0.3.202", + "@a2a-wrapper/core": "2.0.0", + "@anthropic-ai/claude-agent-sdk": "0.3.245", "express": "^4.18.2", "uuid": "^9.0.0" }, @@ -92,11 +92,11 @@ } }, "a2a-codex": { - "version": "1.6.1", + "version": "1.7.0", "license": "MIT", "dependencies": { "@a2a-js/sdk": "^1.0.0", - "@a2a-wrapper/core": "1.7.0", + "@a2a-wrapper/core": "2.0.0", "@openai/codex-sdk": "^0.137.0", "express": "^4.18.2", "uuid": "^9.0.0" @@ -127,12 +127,12 @@ } }, "a2a-copilot": { - "version": "1.7.0", + "version": "1.8.0", "hasInstallScript": true, "license": "MIT", "dependencies": { "@a2a-js/sdk": "^1.0.0", - "@a2a-wrapper/core": "1.7.0", + "@a2a-wrapper/core": "2.0.0", "@github/copilot-sdk": "^1.0.0", "express": "^4.18.2", "uuid": "^9.0.0" @@ -166,11 +166,11 @@ } }, "a2a-opencode": { - "version": "1.6.1", + "version": "1.7.0", "license": "MIT", "dependencies": { "@a2a-js/sdk": "^1.0.0", - "@a2a-wrapper/core": "1.7.0", + "@a2a-wrapper/core": "2.0.0", "@opencode-ai/sdk": "^1.15.13", "express": "^4.18.2", "swagger-ui-express": "^5.0.1", @@ -252,22 +252,22 @@ "link": true }, "node_modules/@anthropic-ai/claude-agent-sdk": { - "version": "0.3.202", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk/-/claude-agent-sdk-0.3.202.tgz", - "integrity": "sha512-LnaLxDtsZP7J6g++xRSnnpTX7CHNe4v+cvBRIlD2ar+N+xi0aqY2YDaCsxPsl+haVUB9kqlUMd0zosmwsfTGjQ==", + "version": "0.3.245", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk/-/claude-agent-sdk-0.3.245.tgz", + "integrity": "sha512-b/SXCxBxZfN4ItHFDUS1uJ3xhI5fOSv3/VxyZvekYmlsbSwZi/75UqKhVlT7qbB1LDJOB48ZmQdCTxWhJWjObA==", "license": "SEE LICENSE IN README.md", "engines": { "node": ">=18.0.0" }, "optionalDependencies": { - "@anthropic-ai/claude-agent-sdk-darwin-arm64": "0.3.202", - "@anthropic-ai/claude-agent-sdk-darwin-x64": "0.3.202", - "@anthropic-ai/claude-agent-sdk-linux-arm64": "0.3.202", - "@anthropic-ai/claude-agent-sdk-linux-arm64-musl": "0.3.202", - "@anthropic-ai/claude-agent-sdk-linux-x64": "0.3.202", - "@anthropic-ai/claude-agent-sdk-linux-x64-musl": "0.3.202", - "@anthropic-ai/claude-agent-sdk-win32-arm64": "0.3.202", - "@anthropic-ai/claude-agent-sdk-win32-x64": "0.3.202" + "@anthropic-ai/claude-agent-sdk-darwin-arm64": "0.3.245", + "@anthropic-ai/claude-agent-sdk-darwin-x64": "0.3.245", + "@anthropic-ai/claude-agent-sdk-linux-arm64": "0.3.245", + "@anthropic-ai/claude-agent-sdk-linux-arm64-musl": "0.3.245", + "@anthropic-ai/claude-agent-sdk-linux-x64": "0.3.245", + "@anthropic-ai/claude-agent-sdk-linux-x64-musl": "0.3.245", + "@anthropic-ai/claude-agent-sdk-win32-arm64": "0.3.245", + "@anthropic-ai/claude-agent-sdk-win32-x64": "0.3.245" }, "peerDependencies": { "@anthropic-ai/sdk": ">=0.93.0", @@ -276,9 +276,9 @@ } }, "node_modules/@anthropic-ai/claude-agent-sdk-darwin-arm64": { - "version": "0.3.202", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-darwin-arm64/-/claude-agent-sdk-darwin-arm64-0.3.202.tgz", - "integrity": "sha512-ujR3zDthDPkZs+AxW95iHpqLT5cuwGImsS3mVxLt1DlDij4qeTnihLX8+EpQTK+oNW9jjvFA86yKwa84fa1KYA==", + "version": "0.3.245", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-darwin-arm64/-/claude-agent-sdk-darwin-arm64-0.3.245.tgz", + "integrity": "sha512-oH1R4yxVKR8oSYMqKHb5NaAPYq8+/enKR0qZKi+lKm6ru64onCmoujT3ilD9rz6TYALCYp2L8Jl2zwOerKSpug==", "cpu": [ "arm64" ], @@ -289,9 +289,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-darwin-x64": { - "version": "0.3.202", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-darwin-x64/-/claude-agent-sdk-darwin-x64-0.3.202.tgz", - "integrity": "sha512-s/RVSGgkVmIMfyt1ndR8braLLu82bARoijmt1kk8d4IptUZ0Sc+zNUWKoFXwR9XqDBu6rBbBF9RIzD02raT57w==", + "version": "0.3.245", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-darwin-x64/-/claude-agent-sdk-darwin-x64-0.3.245.tgz", + "integrity": "sha512-VtK8dfnF0GhVzJgVylZxPdGRZb21DhOpd04WWefKRfnWVbgjdOtEGpafyMi4ZaND/ftYpc2PB2P7IX8OkIj5iA==", "cpu": [ "x64" ], @@ -302,12 +302,15 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-linux-arm64": { - "version": "0.3.202", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-arm64/-/claude-agent-sdk-linux-arm64-0.3.202.tgz", - "integrity": "sha512-a4YtRkgGYt3ogePJDW8Ts6bNW690jb9LHyZaiWXsi+zT53xCNqJB2zKPyRc7hXWOqzIk4nCfwJpjmhLzMu3WIg==", + "version": "0.3.245", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-arm64/-/claude-agent-sdk-linux-arm64-0.3.245.tgz", + "integrity": "sha512-qIi1grLff5a3Z6K9dUsWKrFKcjUuOGErX89DHC9m0UBfGN6swOs6cFj9zvFYhJhLBJeZu0JKyz18BHSXfeu3PA==", "cpu": [ "arm64" ], + "libc": [ + "glibc" + ], "license": "SEE LICENSE IN LICENSE.md", "optional": true, "os": [ @@ -315,12 +318,15 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-linux-arm64-musl": { - "version": "0.3.202", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-arm64-musl/-/claude-agent-sdk-linux-arm64-musl-0.3.202.tgz", - "integrity": "sha512-abSb3Gah45kUNyOeKjmQ/dd1KZ4CaQz5JAr9YQxRDXoOwx8wJVx6huBIpDxjms9wyS9X5Rqxn0Lx7zFP+wV2zQ==", + "version": "0.3.245", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-arm64-musl/-/claude-agent-sdk-linux-arm64-musl-0.3.245.tgz", + "integrity": "sha512-81QcZcFL5YcLLdvw2AXBq8Bxs/Fcq4qkq5dqUkjEEIi6jI2+WTk5gCjLsUg9zVaHG1yoVN3HKBFJss+sI+evHA==", "cpu": [ "arm64" ], + "libc": [ + "musl" + ], "license": "SEE LICENSE IN LICENSE.md", "optional": true, "os": [ @@ -328,12 +334,15 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-linux-x64": { - "version": "0.3.202", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-x64/-/claude-agent-sdk-linux-x64-0.3.202.tgz", - "integrity": "sha512-XIvhdCWAAT4OdOA82fOJII+WH0Tf8pFckckEbJMMmOgQBKOnHT+609Pd3Ehw6zGcA9iFrhG5mY8Ncuckeo1aMw==", + "version": "0.3.245", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-x64/-/claude-agent-sdk-linux-x64-0.3.245.tgz", + "integrity": "sha512-fvPtGYI61pGRP2rmaYskyLE83PytLLMV/NzDunmDlIRvqkzgQrxDBeOistt6mc29uAlI70EbaKW4zsSft6IhBQ==", "cpu": [ "x64" ], + "libc": [ + "glibc" + ], "license": "SEE LICENSE IN LICENSE.md", "optional": true, "os": [ @@ -341,12 +350,15 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-linux-x64-musl": { - "version": "0.3.202", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-x64-musl/-/claude-agent-sdk-linux-x64-musl-0.3.202.tgz", - "integrity": "sha512-fze5nAQL1ErcMCQNB10ILaWdM0QbJSaTQzBz8NVAy0FGW8ZL0t4Wf/VgFkfzXbfkaxmPuM1C27Dn5HiU7UDEHQ==", + "version": "0.3.245", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-linux-x64-musl/-/claude-agent-sdk-linux-x64-musl-0.3.245.tgz", + "integrity": "sha512-3Uxl7YDnqpQHbSpEOYCPytcbcuo1PcdYHDXDpoSotBPlvFOJEiLGCGhwWfEbTO/9RGdp94Qm0T2gGtPNE6QpFg==", "cpu": [ "x64" ], + "libc": [ + "musl" + ], "license": "SEE LICENSE IN LICENSE.md", "optional": true, "os": [ @@ -354,9 +366,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-win32-arm64": { - "version": "0.3.202", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-win32-arm64/-/claude-agent-sdk-win32-arm64-0.3.202.tgz", - "integrity": "sha512-N1J0HRvC+8a69bqNY7+ENIYQzR0i7s+rOIGH5XtuLxvLqOnZO8LHxWEZOe8ezabGq5eZqphSCgL6vQnQQpNh+A==", + "version": "0.3.245", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-win32-arm64/-/claude-agent-sdk-win32-arm64-0.3.245.tgz", + "integrity": "sha512-N8JTt+DuX2xwbbnLMLDUgmnbtz3VKSGI07WV4WjCEBeb6Olt2KJHDksnJzknrbFBapJz915Vbv2fO+izJcRI1g==", "cpu": [ "arm64" ], @@ -367,9 +379,9 @@ ] }, "node_modules/@anthropic-ai/claude-agent-sdk-win32-x64": { - "version": "0.3.202", - "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-win32-x64/-/claude-agent-sdk-win32-x64-0.3.202.tgz", - "integrity": "sha512-ytLGEC1fjTSiVSoXukS+j9G+06Mi20NSzxxzlG6uE75SEB0+17tHdWUaHqd8PhH/6GPzcYx81czxWQl1MVbq4Q==", + "version": "0.3.245", + "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-agent-sdk-win32-x64/-/claude-agent-sdk-win32-x64-0.3.245.tgz", + "integrity": "sha512-C8PHrQBPgExO6sr5bwG+IW7cTR2KDmvAUj9By++u+IiTFEAtPCEjOQ3CFgfPc5wTuGD26SrzWUbdEom5TQp1Dg==", "cpu": [ "x64" ], @@ -6509,7 +6521,7 @@ }, "packages/core": { "name": "@a2a-wrapper/core", - "version": "1.7.0", + "version": "2.0.0", "license": "MIT", "devDependencies": { "@a2a-js/sdk": "^1.0.0", diff --git a/scripts/background-tasks-smoke/README.md b/scripts/background-tasks-smoke/README.md index ab9f193..d1db06e 100644 --- a/scripts/background-tasks-smoke/README.md +++ b/scripts/background-tasks-smoke/README.md @@ -10,7 +10,7 @@ authenticated `claude` on PATH. They are deliberately not wired into `npm test`. ```bash cd scripts/background-tasks-smoke -npm install @anthropic-ai/claude-agent-sdk@^0.3.235 +npm install @anthropic-ai/claude-agent-sdk@0.3.245 node spike-single.mjs # one background task: does a second result arrive at all node spike-chain.mjs # two-stage chain: does the hold loop across rounds ``` From fbddf5a69d38902b17c9d578d67c850037eefc7a Mon Sep 17 00:00:00 2001 From: Colin Harris Date: Tue, 25 Aug 2026 17:08:54 +1000 Subject: [PATCH 14/14] test(claude): drop bypassPermissions from the background-task spikes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The spikes ran an unrestricted Claude in `process.cwd()` — the repo working tree, per the README's own instructions — while a comment claimed they were sandboxed. Nothing sandboxed them. They need a shell and nothing else, so pre-approve exactly that: `permissionMode: "dontAsk"` with Bash/BashOutput/KillShell allowed and everything else denied, in a fresh mkdtemp directory. File writes are now impossible rather than merely unlikely, and the comment describes what the code does. Deliberately not `auto`, which the SDK also offers: it resolves permissions with a model classifier, which would make an unattended run that spends real quota non-deterministic and add classifier traffic to the very message stream these spikes exist to observe. `dontAsk` is also the mode this repo already endorses for headless operation in VALID_PERMISSION_MODES. Log `system/permission_denied` explicitly in both spikes. It is the new failure mode, and spike-chain's if/else chain drops unmatched messages silently — a denial there would have looked like a stalled run. Not executed as part of this change: these cost real quota. --- scripts/background-tasks-smoke/README.md | 6 +++++ .../background-tasks-smoke/spike-chain.mjs | 25 +++++++++++++++---- .../background-tasks-smoke/spike-single.mjs | 25 +++++++++++++++---- 3 files changed, 46 insertions(+), 10 deletions(-) diff --git a/scripts/background-tasks-smoke/README.md b/scripts/background-tasks-smoke/README.md index d1db06e..0a654a7 100644 --- a/scripts/background-tasks-smoke/README.md +++ b/scripts/background-tasks-smoke/README.md @@ -6,6 +6,12 @@ headless SDK mode. Unit tests use scripted fakes; only these run the real CLI. **These spend real quota** — roughly a minute of model time each — and need an authenticated `claude` on PATH. They are deliberately not wired into `npm test`. +Each runs in a fresh temp directory, never the repo, under +`permissionMode: "dontAsk"` with `Bash` / `BashOutput` / `KillShell` +pre-approved and everything else denied — so a spike cannot write to your +working tree. If a run logs `!!! permission_denied`, widen `allowedTools` +rather than switching to `bypassPermissions`. + ## Running ```bash diff --git a/scripts/background-tasks-smoke/spike-chain.mjs b/scripts/background-tasks-smoke/spike-chain.mjs index af68ad2..55deea6 100644 --- a/scripts/background-tasks-smoke/spike-chain.mjs +++ b/scripts/background-tasks-smoke/spike-chain.mjs @@ -4,6 +4,10 @@ // Also validates that bg_changed for a newly started task lands before its result. // COSTS REAL QUOTA: ~2 minutes of model time. +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + import { query } from "@anthropic-ai/claude-agent-sdk"; const t0 = Date.now(); @@ -43,11 +47,16 @@ const decisions = []; const q = query({ prompt: input(), options: { - cwd: process.cwd(), - // Bypass permissions only here: this is a sandboxed smoke test that needs - // to run unattended. Do not copy this pattern into production code. - permissionMode: "bypassPermissions", - allowDangerouslySkipPermissions: true, + // An empty temp dir, not the repo: this runs unattended, and nothing it + // does needs a working tree. + cwd: mkdtempSync(join(tmpdir(), "bg-smoke-")), + // `dontAsk` denies anything not pre-approved, so the shell family below is + // the whole of what this can do — no file writes, no network tools. Note + // this is deliberately not `bypassPermissions`, and not `auto` either: a + // model classifier would make an unattended quota-spending run + // non-deterministic. + permissionMode: "dontAsk", + allowedTools: ["Bash", "BashOutput", "KillShell"], settingSources: [], strictMcpConfig: true, }, @@ -69,6 +78,12 @@ try { log(` init (session ${m.session_id})`); } else if (key === "system/task_notification") { log(` task_notification`, m.task_id, m.status); + } else if (key === "system/permission_denied") { + // The run needs Bash and nothing else. If this fires, the pre-approved + // tool list above is too narrow — widen it rather than reaching for + // bypassPermissions. This branch matters: everything unmatched below is + // dropped silently, so without it a denial would look like a stalled run. + log(`!!! permission_denied`, m.tool_name, clip(m.message, 80)); } else if (m.type === "result") { results += 1; const decision = live.size > 0 ? `HOLD (waiting on ${[...live].join(",")})` : "COMPLETE"; diff --git a/scripts/background-tasks-smoke/spike-single.mjs b/scripts/background-tasks-smoke/spike-single.mjs index 130753e..f1bc258 100644 --- a/scripts/background-tasks-smoke/spike-single.mjs +++ b/scripts/background-tasks-smoke/spike-single.mjs @@ -3,6 +3,10 @@ // and that a second result arrives on the same query (no second user message). // COSTS REAL QUOTA: ~1 minute of model time. +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + import { query } from "@anthropic-ai/claude-agent-sdk"; const t0 = Date.now(); @@ -38,11 +42,16 @@ let resultCount = 0; const q = query({ prompt: input(), options: { - cwd: process.cwd(), - // Bypass permissions only here: this is a sandboxed smoke test that needs - // to run unattended. Do not copy this pattern into production code. - permissionMode: "bypassPermissions", - allowDangerouslySkipPermissions: true, + // An empty temp dir, not the repo: this runs unattended, and nothing it + // does needs a working tree. + cwd: mkdtempSync(join(tmpdir(), "bg-smoke-")), + // `dontAsk` denies anything not pre-approved, so the shell family below is + // the whole of what this can do — no file writes, no network tools. Note + // this is deliberately not `bypassPermissions`, and not `auto` either: a + // model classifier would make an unattended quota-spending run + // non-deterministic. + permissionMode: "dontAsk", + allowedTools: ["Bash", "BashOutput", "KillShell"], settingSources: [], strictMcpConfig: true, }, @@ -70,6 +79,12 @@ try { case "system/init": log(` ${key}`, "session", m.session_id); break; + // The run needs Bash and nothing else. If this fires, the pre-approved + // tool list above is too narrow — widen it rather than reaching for + // bypassPermissions. + case "system/permission_denied": + log(`!!! ${key}`, m.tool_name, clip(m.message, 80)); + break; case "stream_event": break; default: {