diff --git a/README.md b/README.md index a2021abd..ad7a3c4e 100644 --- a/README.md +++ b/README.md @@ -37,14 +37,34 @@ npm install @agentclientprotocol/claude-agent-acp@preview ### Subagent sessions -Subagents are exposed only after bilateral capability negotiation. Until the released ACP SDKs -preserve the draft `clientCapabilities.subagents` field, a supporting client may advertise -`nativeSubagentSessions` in `_meta.jetbrains.air.capabilities`; the adapter mirrors the capability -in its initialize response. The canonical field remains supported and takes precedence once it is -available. Without either client signal, Agent/Task lifecycle keeps its legacy ordinary ACP -tool-call representation and child interactions stay on the root session. Clients that use the -historical `_meta["subagent-transcript"]` capability or `forwardSubagentText` session option retain -the flattened child transcript behavior. See [AIR extensions](docs/air-extensions.md#native-subagent-sessions). +A client that declares `clientCapabilities.subagents`, other than JetBrains AIR, gets each Agent or +Task subagent as its own child session, following the [subagents RFD](https://github.com/agentclientprotocol/agent-client-protocol/blob/main/docs/rfds/subagents.mdx): + +- `subagent_update` on the parent announces the child before any of its traffic, with the + parent's `title` and `description` for it, and reports its work state: `running`, and `idle` + when its work ends, with `end_turn` or `cancelled` when the SDK gives the reason. A failure is + `idle` with the `error` stop reason and the SDK's error text as its `error`. +- The prompt of the Agent or Task call that launches the child is a `session_message` in the + child's transcript. The tool call stays a tool call of the session that made it. +- `session/cancel` on the parent interrupts Claude Code, which stops the foreground children that + the turn waits on and, currently, background children too. Each stopped child reports `idle` + with `cancelled` from Claude Code's own events, and the open requests of the foreground ones are + withdrawn. The adapter ends no child at a cancel itself, so a child that keeps running keeps its + open requests. A child that still runs when Claude Code exits is `idle` with no stop reason. +- A child whose parent is not yet known when it ends is not exposed, since the RFD forbids + guessing its parent. +- A SendMessage that resumes a child is not shown yet: the child stays `idle`, and only the + permission and question requests of the resumed run appear in its session. +- A client cannot prompt, cancel, or otherwise change a child yet, and a replay does not restore + children yet. + +JetBrains AIR keeps the earlier draft of the RFD (`subagent_spawned`, `subagent_state_update`), +which it enables with `nativeSubagentSessions` in `_meta.jetbrains.air.capabilities`. See +[AIR extensions](docs/air-extensions.md#native-subagent-sessions). + +Without either signal, Agent and Task stay ordinary tool calls and child interactions stay on the +root session. Clients that use the historical `_meta["subagent-transcript"]` capability or the +`forwardSubagentText` session option retain the flattened child transcript. ## Contribution Policy diff --git a/docs/air-extensions.md b/docs/air-extensions.md index bf5ccedb..468b24e0 100644 --- a/docs/air-extensions.md +++ b/docs/air-extensions.md @@ -1200,8 +1200,9 @@ This section covers only the AIR bridge. - The canonical client field is `clientCapabilities.subagents: {}`. - Released ACP SDKs can strip that draft field. AIR can instead declare `nativeSubagentSessions` in `_meta.jetbrains.air.capabilities`. -- Either signal enables native subagent sessions. The canonical field takes precedence when it is available. -- The agent advertises `nativeSubagentSessions` to AIR, and `agentCapabilities.sessionCapabilities.subagents` to every client. +- Either signal enables native subagent sessions, and AIR gets this earlier draft of the RFD whichever it declares. +- The agent advertises `nativeSubagentSessions` and `agentCapabilities.sessionCapabilities.subagents` to AIR. +- A client that is not AIR and declares `clientCapabilities.subagents` gets the revised RFD instead (`subagent_update`, `session_message`), see the README. - With native sessions, the adapter sends `subagent_spawned` and `subagent_state_update`. The child output goes to the child session. The Agent or Task tool call is not the subagent card. - `subagent_spawned` can carry an optional `prompt`: the exact prompt of the child session. diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 4f74eef3..7e72ebc4 100644 --- a/src/acp-agent.ts +++ b/src/acp-agent.ts @@ -111,6 +111,7 @@ import { asSdkSessionNotification, clientSupportsSubagents, SubagentAwareSessionCapabilities, + subagentForm, type SubagentState, } from "./acp-subagents.js"; import { @@ -2804,6 +2805,8 @@ export class ClaudeAcpAgent { } } + // The subagents RFD has only the client capability. AIR's draft also had + // this agent capability, which AIR gets (`docs/air-extensions.md`). const sessionCapabilities: SubagentAwareSessionCapabilities = { additionalDirectories: {}, close: {}, @@ -2811,7 +2814,7 @@ export class ClaudeAcpAgent { fork: {}, list: {}, resume: {}, - subagents: {}, + ...(this.toolCallCapabilities.air.client ? { subagents: {} } : {}), }; return { @@ -4147,7 +4150,7 @@ export class ClaudeAcpAgent { * stamps from `parent_tool_use_id`, and never reach the top-level feed * as the turn's answer. */ const subagents = (session.nativeSubagentRuntime ??= new NativeSubagentRuntime( - clientSupportsSubagents(this.clientCapabilities), + subagentForm(this.clientCapabilities), params.sessionId, session, async (notification) => this.client.sessionUpdate(asSdkSessionNotification(notification)), @@ -5911,6 +5914,7 @@ export class ClaudeAcpAgent { message.status, sendUpdate, message.tool_use_id, + { reason: message.reason }, ); await asyncTasks.taskNotification({ task_id: message.task_id, @@ -5953,7 +5957,13 @@ export class ClaudeAcpAgent { message.patch.status === "failed" || message.patch.status === "killed" ) { - await subagents.finishTask(message.task_id, message.patch.status, sendUpdate); + await subagents.finishTask( + message.task_id, + message.patch.status, + sendUpdate, + undefined, + { error: message.patch.status === "failed" ? message.patch.error : undefined }, + ); const parentToolUseId = session.liveBackgroundTasks.get( message.task_id, )?.parentToolUseId; @@ -7495,13 +7505,18 @@ export class ClaudeAcpAgent { if (toolCallId === null || !session.emittedToolCalls.has(toolCallId)) { break; } - const subagentParentToolUseId = message.parent_tool_use_id - ? [...session.liveBackgroundTasks.values()].some( - (task) => task.isSubagent && task.parentToolUseId === message.parent_tool_use_id, - ) + // The beat is in the subagent whose Agent call is + // `parent_tool_use_id`, unless it reports on that Agent call + // itself: the Agent call's own heartbeat falls back to its id, and + // a tool call is never its own parent. + const subagentParentToolUseId = + message.parent_tool_use_id && + message.parent_tool_use_id !== toolCallId && + [...session.liveBackgroundTasks.values()].some( + (task) => task.isSubagent && task.parentToolUseId === message.parent_tool_use_id, + ) ? message.parent_tool_use_id - : undefined - : undefined; + : undefined; // `tool_name` names the tool that reports the beat. When the beat // falls back to the parent call, AIR gets the name of that call, // or no name. Every other client gets `tool_name`, like upstream. @@ -7746,9 +7761,18 @@ export class ClaudeAcpAgent { // their compaction is running. Close it before ANY cancelled prompt // resolves, not just the active-turn idle/backstop paths. await session.contextCompaction?.interrupt(); + // A cancel drops the state of each tool call (the session that a + // permission request emitted it in, the fields the client holds, the hook + // callbacks), even when publishing fails. AIR's draft ends every child + // here, so a later update of an ended child's tool call is then dropped + // instead of reaching that child's session. The RFD form keeps the state: + // a background subagent can keep running after a cancel + // (agentclientprotocol/agent-client-protocol#2308), and its tool call that + // a permission request emitted on the root before the child was announced + // must stay on the root, which the RFD requires. Each tool call drops its + // own state when it ends. try { - await session.nativeSubagentRuntime?.finishAll( - "cancelled", + await session.nativeSubagentRuntime?.parentCancelled( session.nativeSubagentDeliver ?? (async (notification) => this.client.sessionUpdate(asSdkSessionNotification(notification))), @@ -7759,9 +7783,11 @@ export class ClaudeAcpAgent { error, ); } finally { - session.eagerToolCallSessions?.clear(); - session.toolCallFields?.clear(); - clearHookCallbacks(params.sessionId); + if (session.nativeSubagentRuntime?.form !== "rfd") { + session.eagerToolCallSessions?.clear(); + session.toolCallFields?.clear(); + clearHookCallbacks(params.sessionId); + } } // The user messages this cancel abandons while Claude Code may still have // them queued: they are dropped from its queue before the interrupt. @@ -8301,7 +8327,10 @@ export class ClaudeAcpAgent { }) : undefined; let replayTurnId: string | undefined; - const nativeReplayEnabled = clientSupportsSubagents(this.clientCapabilities); + // Only AIR's draft replays child sessions so far. A client on the RFD gets + // no children on replay, which the RFD allows (it SHOULD replay them), until + // the replay keeps the children's live session ids. + const nativeReplayEnabled = subagentForm(this.clientCapabilities) === "air"; const replayCompactionUpdates = clientSupportsCompactionUpdates(this.clientCapabilities); const replayTerminalStates = new Map(); const replayChildren = new Map< diff --git a/src/acp-subagents.ts b/src/acp-subagents.ts index 055492c4..6ce78457 100644 --- a/src/acp-subagents.ts +++ b/src/acp-subagents.ts @@ -6,17 +6,33 @@ import type { import { AIR_NATIVE_SUBAGENT_SESSIONS_CAPABILITY, clientSupportsAirCapability, + isAirClient, } from "./air-extension.js"; export { AIR_NATIVE_SUBAGENT_SESSIONS_CAPABILITY } from "./air-extension.js"; /** - * Temporary typed surface for agentclientprotocol/agent-client-protocol#1992. + * How a client gets native subagent sessions. * - * The wire contract is already defined by the ACP draft, but the published - * TypeScript SDK does not contain it yet. Keep the compatibility boundary in - * this file so it can be replaced by SDK exports without changing lifecycle - * code when the draft ships. + * - `rfd`: the subagents RFD (agentclientprotocol/agent-client-protocol#1992, + * as revised): `subagent_update` and `session_message`, which the SDK types. + * A client that declares `clientCapabilities.subagents` and is not AIR gets + * it. (ACP v2 clients do not get subagents yet.) + * - `air`: the earlier draft of that RFD, which AIR implements: + * `subagent_spawned` and `subagent_state_update`, typed below. AIR gets it, + * whichever capability it declares (`docs/air-extensions.md`). + */ +export type SubagentForm = "rfd" | "air"; + +/** The form of native subagent sessions that the client gets, or none. */ +export function subagentForm(capabilities?: ClientCapabilities | null): SubagentForm | undefined { + if (!clientSupportsSubagents(capabilities)) return undefined; + return isAirClient(capabilities) ? "air" : "rfd"; +} + +/** + * The earlier draft of the subagents RFD, which AIR implements and the SDK + * does not type. */ export type SubagentSessionCapabilities = { cancel?: boolean; @@ -116,7 +132,7 @@ export function clientSupportsSubagents(capabilities?: ClientCapabilities | null return clientSupportsAirCapability(capabilities, AIR_NATIVE_SUBAGENT_SESSIONS_CAPABILITY); } -/** The only cast needed until the TypeScript SDK publishes PR #1992. */ +/** The cast that AIR's draft updates and the async task updates need. */ export function asSdkSessionNotification( notification: AcpSessionNotification, ): SessionNotification { diff --git a/src/native-subagents.ts b/src/native-subagents.ts index f302fd04..d51ba8d1 100644 --- a/src/native-subagents.ts +++ b/src/native-subagents.ts @@ -1,6 +1,12 @@ -import type { AcpSessionNotification, SubagentState } from "./acp-subagents.js"; +import { RequestError, type SessionNotification } from "@agentclientprotocol/sdk"; +import type { AcpSessionNotification, SubagentForm, SubagentState } from "./acp-subagents.js"; import { AIR_SUBAGENT_KEY, airExtensionMeta } from "./air-extension.js"; +/** The work state of a child, as the RFD's `subagent_update.state` carries it. */ +type WorkState = NonNullable< + Extract["state"] +>; + export type NativeSubagent = { sessionId: string; parentSessionId: string; @@ -9,9 +15,28 @@ export type NativeSubagent = { task: string; /** * The exact prompt of this generation, sent as `prompt` in - * `subagent_spawned`. It is absent when the adapter has no prompt. + * `subagent_spawned`, or as the child's first `session_message` in the RFD + * form. It is absent when the adapter has no prompt. */ prompt?: string; + /** The RFD form: the parent's label for the child, `subagent_update.title`. */ + title?: string; + /** The RFD form: the child's role, its `subagent_type`, for the title. */ + role?: string; + /** The RFD form: the child's purpose, `subagent_update.description`. */ + description?: string; + /** + * The RFD form: the work state the client holds. The child is reusable, so + * `idle` is not terminal. AIR's draft uses {@link terminalState} instead. + */ + workState?: WorkState["state"]; + /** + * The RFD form: the child's idle snapshot reports a failure without the + * SDK's error text, which a later event of the same failure may carry. + */ + missingError?: boolean; + /** The RFD form: the ids of the prompt messages in the child's transcript. */ + promptIds?: Set; announced?: boolean; terminalState?: SubagentState; /** Connection-local single-flight state; never serialized on the wire. */ @@ -29,6 +54,19 @@ export type NativeSubagentSession = { type Publish = (notification: AcpSessionNotification) => Promise; type Logger = { log(message: string): void }; +/** What the SDK says about the end of a task, beyond its status. */ +export type TaskEnd = { + /** The SDK's text for a failure. */ + error?: string; + /** The SDK's reason for an unusual end, such as `worker_restart`. */ + reason?: string; + /** + * The adapter knows that the work ended, but not why: the stream ended, or + * the conversation was reset, with the task still running. + */ + unexplained?: boolean; +}; + type TaskStarted = { taskId: string; toolUseId?: string | null; @@ -56,6 +94,8 @@ const MAX_CHILD_TOOL_CALLS = 2048; */ export class NativeSubagentRuntime { readonly enabled: boolean; + /** The form of the child sessions; see {@link SubagentForm}. */ + readonly form: SubagentForm | undefined; private readonly children: Map; private readonly taskByToolUse: Map; @@ -74,14 +114,16 @@ export class NativeSubagentRuntime { private readonly pending = new Map(); private pendingCount = 0; + /** `form` is undefined for a client without native subagent sessions. */ constructor( - enabled: boolean, + form: SubagentForm | undefined, private readonly rootSessionId: string, private readonly session: NativeSubagentSession, private readonly publish: Publish, private readonly logger: Logger, ) { - this.enabled = enabled; + this.form = form; + this.enabled = form !== undefined; this.children = session.nativeSubagentsByTaskId ??= new Map(); this.taskByToolUse = session.nativeSubagentTaskIdByToolUseId ??= new Map(); this.parentByToolUse = session.nativeSubagentParentByToolUseId ??= new Map(); @@ -116,28 +158,47 @@ export class NativeSubagentRuntime { mergeSubagentIdentity(this.identityByToolUse.get(toolCallId), identity), ); } + const parentChild = claudeMeta?.parentToolUseId + ? this.childByParentToolUse.get(claudeMeta.parentToolUseId) + : undefined; const parentSessionId = claudeMeta?.parentToolUseId - ? this.childByParentToolUse.get(claudeMeta.parentToolUseId)?.sessionId + ? parentChild?.sessionId : this.rootSessionId; - this.parentByToolUse.set(toolCallId, parentSessionId ?? this.rootSessionId); - + // The RFD form: a call made inside a child that is not announced yet + // establishes no parent, because the RFD forbids announcing its child + // under the root or before that child. The update is buffered below + // under the child's tool use, and routed again, with the child as the + // parent, once the child is announced. + const parentPending = + this.form === "rfd" && !!claudeMeta?.parentToolUseId && !parentChild?.announced; const child = this.childByParentToolUse.get(toolCallId); - if (child && !child.announced) { - child.parentSessionId = parentSessionId ?? this.rootSessionId; - applySubagentIdentity(child, this.identityByToolUse.get(toolCallId)); - await announceNativeSubagent(child, this.publish); - for (const pending of this.takePending(toolCallId)) await deliver(pending); - } - if (!child && isFailedToolCallUpdate(update)) { - this.takePending(toolCallId); - const initial = this.controlByToolUse.get(toolCallId); - this.cleanupControl(toolCallId); - if (forcedSessionId) { - return { ...notification, sessionId: forcedSessionId }; + if (!parentPending) { + this.parentByToolUse.set(toolCallId, parentSessionId ?? this.rootSessionId); + if (child && !child.announced) { + child.parentSessionId = parentSessionId ?? this.rootSessionId; + applySubagentIdentity(child, this.identityByToolUse.get(toolCallId)); + await this.announce(child); + for (const pending of this.takePending(toolCallId)) await deliver(pending); } - return failedControlFallback(initial, notification, parentSessionId ?? this.rootSessionId); } - return forcedSessionId ? { ...notification, sessionId: forcedSessionId } : null; + if (this.form === "air") { + if (!child && isFailedToolCallUpdate(update)) { + this.takePending(toolCallId); + const initial = this.controlByToolUse.get(toolCallId); + this.cleanupControl(toolCallId); + if (forcedSessionId) { + return { ...notification, sessionId: forcedSessionId }; + } + return failedControlFallback( + initial, + notification, + parentSessionId ?? this.rootSessionId, + ); + } + return forcedSessionId ? { ...notification, sessionId: forcedSessionId } : null; + } + // The RFD form: the Agent/Task tool call is the parent's record of the + // delegation, so it stays a tool call of the session that made it. } const toolCallId = @@ -180,6 +241,8 @@ export class NativeSubagentRuntime { return; } const previous = this.children.get(task.taskId); + // An RFD child never has a terminal state, so a resume does not change it + // yet (see `taskResumed`). if (previous && previous.terminalState === undefined) return; // A SendMessage resume reuses the finished generation's tool id, whose @@ -207,6 +270,11 @@ export class NativeSubagentRuntime { identity?.description ?? task.description, ), ...promptField(promptText(task.prompt) ?? identity?.prompt), + ...rfdLabels( + identity?.name, + identity?.subagentType ?? task.subagentType, + identity?.description ?? task.description, + ), }, !!knownParentSessionId || !task.toolUseId, deliver, @@ -219,6 +287,11 @@ export class NativeSubagentRuntime { * running `task_updated` patch or a SendMessage `resumedAgentId` is the * signal. A child that did not finish is not changed. The `prompt` is the * SendMessage text that resumed the child, when the adapter knows it. + * + * The RFD form does not show a resume yet: its children never have a + * terminal state, so a resumed child stays idle. The resumed run's output + * carries the id of the call that resumed it, which is not the child's, so + * it is buffered and then dropped. */ async taskResumed(taskId: string, deliver: Publish, prompt?: string): Promise { if (!this.enabled) return; @@ -242,27 +315,51 @@ export class NativeSubagentRuntime { ); } + /** + * Reports that the SDK ended the task's work. `end` has what the SDK says + * about it: its text for a failure, and its reason for an unusual end. + */ async finishTask( taskId: string, status: unknown, deliver: Publish, toolUseId?: string | null, + end?: TaskEnd, ): Promise { if (!this.enabled) return; const state = nativeSubagentState(status); const child = toolUseId ? this.childByParentToolUse.get(toolUseId) : this.children.get(taskId); if (child && toolUseId && this.taskByToolUse.get(toolUseId) !== taskId) return; - if (!state || !child || child.terminalState !== undefined) return; + if (!state || !child) return; + if (this.form === "rfd") { + if (!child.announced && !child.announcePromise) { + this.withdraw(taskId, child); + return; + } + } const existing = this.taskFinishPromises.get(taskId); - if (existing) return existing; + if (existing) { + await existing; + if (this.form === "rfd") await this.supplyError(child, state, end?.error); + return; + } + if (this.form === "air" ? child.terminalState !== undefined : child.workState === "idle") { + if (this.form === "rfd") await this.supplyError(child, state, end?.error); + return; + } const finish = Promise.resolve().then(async () => { try { - await announceNativeSubagent(child, this.publish); + await this.announce(child); if (child.parentToolUseId) { for (const pending of this.takePending(child.parentToolUseId)) await deliver(pending); } - await finishNativeSubagent(this.session, taskId, state, this.publish); + if (this.form === "rfd") { + await this.reportState(child, rfdEndState(state, end)); + child.missingError = state === "failed" && !end?.error && !end?.unexplained; + } else { + await finishNativeSubagent(this.session, taskId, state, this.publish); + } } finally { if (child.parentToolUseId) { this.cleanupControl(child.parentToolUseId); @@ -277,12 +374,20 @@ export class NativeSubagentRuntime { } } + /** + * Ends every child that still runs, because the stream ended or the + * conversation was reset. AIR's draft reports `state` for each. In the RFD + * form the cause of each child's end is unknown, so it is idle with no stop + * reason: a child that a cancel stopped already reported it from the SDK's + * events, and a background child that outlived the cancel was not cancelled. + */ async finishAll(state: SubagentState, deliver: Publish): Promise { const errors: unknown[] = []; + const end = this.form === "rfd" ? { unexplained: true } : undefined; try { for (const taskId of [...this.children.keys()].reverse()) { try { - await this.finishTask(taskId, state, deliver); + await this.finishTask(taskId, state, deliver, undefined, end); } catch (error) { errors.push(error); } @@ -298,20 +403,80 @@ export class NativeSubagentRuntime { if (errors.length > 1) throw new AggregateError(errors, "Failed to finish native subagents"); } + /** + * The parent session got `session/cancel`. AIR's draft ends every child as + * cancelled here. The RFD form reports nothing: acknowledging a cancel is no + * evidence that a child stopped. Claude Code's interrupt stops the + * foreground children that the turn waits on and, currently, background + * children too, and the SDK's task events report each as cancelled. A child + * that keeps running keeps its open requests + * (agentclientprotocol/agent-client-protocol#2308). + */ + async parentCancelled(deliver: Publish): Promise { + if (this.form === "air") await this.finishAll("cancelled", deliver); + } + discardPending(parentToolUseId: string): void { this.takePending(parentToolUseId); } /** - * Routes an update to the child session. An update of a finished child is - * dropped: it never goes to the root session. + * The RFD form: a child that ends before its parent is known stays + * unexposed, since the RFD forbids guessing its parent. Its buffered updates + * are dropped, and it is forgotten, so a late frame does not announce it. + * (Buffering until that frame would also meet the RFD, but the frame comes + * before the task in the SDK's order, so a child ending without it is rare.) + */ + private withdraw(taskId: string, child: NativeSubagent): void { + this.children.delete(taskId); + if (child.parentToolUseId) { + this.takePending(child.parentToolUseId); + this.cleanupControl(child.parentToolUseId); + this.childByParentToolUse.delete(child.parentToolUseId); + this.taskByToolUse.delete(child.parentToolUseId); + } + } + + /** + * The RFD form: a later event of a failure can carry the SDK's error text + * that the reported idle snapshot lacks. + */ + private async supplyError( + child: NativeSubagent, + state: SubagentState, + error: string | undefined, + ): Promise { + if (state !== "failed" || !error || !child.missingError) return; + await this.reportState(child, rfdEndState("failed", { error })); + } + + /** Announces `child` in the client's form. A runtime without one never does. */ + private async announce(child: NativeSubagent): Promise { + if (this.form) await announceNativeSubagent(child, this.publish, this.form); + } + + /** The RFD form: publishes the work state of `child` on its parent. */ + private async reportState(child: NativeSubagent, state: WorkState): Promise { + child.workState = state.state; + child.missingError = false; + await this.publish({ + sessionId: child.parentSessionId, + update: { sessionUpdate: "subagent_update", sessionId: child.sessionId, state }, + }); + } + + /** + * Routes an update to the child session. It never goes to the root session: + * in AIR's draft, an update of a finished child is dropped. */ private toChild( child: NativeSubagent, notification: AcpSessionNotification, toolCallId: string | undefined, ): AcpSessionNotification | null { - if (child.terminalState !== undefined || child.terminalPromise) { + // AIR's draft ends a child for good. In the RFD form an idle child is + // reusable, and its background work can still report. + if (this.form === "air" && (child.terminalState !== undefined || child.terminalPromise)) { this.logger.log( `Session ${this.rootSessionId}: ignoring late update for terminal subagent ${child.sessionId}`, ); @@ -327,10 +492,11 @@ export class NativeSubagentRuntime { * tool call when the work started. An async task can outlive that child: a * subagent can start a background command and end its turn. So the route * still sends `async_task_spawned`, `async_task_progress` and - * `async_task_state_update` to the child generation after it finished. The route drops every other update - * after the child finished. `undefined` means that the root session owns the - * tool call. `eagerSessionId` is the session where a permission request - * created the tool call before the stream routed it. + * `async_task_state_update` to the child generation after it finished. In + * AIR's draft, the route drops every other update after the child finished; + * in the RFD form, an idle child gets them all. `undefined` means that the + * root session owns the tool call. `eagerSessionId` is the session where a + * permission request created the tool call before the stream routed it. */ routeOfToolCall( toolCallId: string, @@ -438,7 +604,14 @@ export class NativeSubagentRuntime { previous: NativeSubagent | undefined, fields: Pick< NativeSubagent, - "parentSessionId" | "parentToolUseId" | "name" | "task" | "prompt" + | "parentSessionId" + | "parentToolUseId" + | "name" + | "task" + | "prompt" + | "title" + | "role" + | "description" >, announce: boolean, deliver: Publish, @@ -455,10 +628,20 @@ export class NativeSubagentRuntime { this.controlByToolUse.delete(toolUseId); } if (!announce) return; - await announceNativeSubagent(child, this.publish); + await this.announce(child); for (const pending of toolUseId ? this.takePending(toolUseId) : []) await deliver(pending); } + /** + * The ACP session id of a child: the SDK agent id, and in AIR's draft a + * generation suffix for each resume. The RFD needs it unique within the + * connection and never under two parents. Agent ids are random, and an + * agent keeps its parent, so only a fork could repeat one under another + * root. That holds today because the SDK's `forkSession` copies no + * `subagents/` history, so a fork cannot resume an agent of the original. + * If a fork ever can, the RFD requires remapping the copied ids, for + * example by deriving the child id from the root session id. + */ private nextChildSessionId(taskId: string, previous: NativeSubagent | undefined): string { if (!previous) { this.generationByTaskId.set(taskId, 1); @@ -470,13 +653,45 @@ export class NativeSubagentRuntime { } } +/** + * Announces the child on its parent, once, before any update of the child. + * The RFD form announces it as running and then sends the prompt as the + * first message of the child. The child gets no capabilities, so the client + * may not cancel or otherwise change it. + */ export async function announceNativeSubagent( child: NativeSubagent, publish: Publish, + form: SubagentForm, ): Promise { if (child.announced) return; if (child.announcePromise) return child.announcePromise; const announce = Promise.resolve().then(async () => { + if (form === "rfd") { + child.workState = "running"; + await publish({ + sessionId: child.parentSessionId, + update: { + sessionUpdate: "subagent_update", + sessionId: child.sessionId, + ...(child.title ? { title: child.title } : {}), + ...(child.description ? { description: child.description } : {}), + state: { state: "running" }, + }, + }); + child.announced = true; + // The immediate parent made the Agent/Task call that launched the child. + if (child.prompt) { + await publishPrompt( + child, + child.prompt, + child.parentToolUseId, + child.parentSessionId, + publish, + ); + } + return; + } await publish({ sessionId: child.parentSessionId, update: { @@ -508,7 +723,7 @@ export async function finishNativeSubagent( if (!child || child.terminalState !== undefined) return; if (child.terminalPromise) return child.terminalPromise; const finish = Promise.resolve().then(async () => { - await announceNativeSubagent(child, publish); + await announceNativeSubagent(child, publish, "air"); await publish({ sessionId: child.parentSessionId, update: { @@ -527,6 +742,93 @@ export async function finishNativeSubagent( } } +/** + * The RFD form: a message to the child, in the child's transcript (its + * incoming view). Its id is the id of the tool use that sent it, the + * Agent/Task call, so a prompt whose id is already in the transcript is the + * same delegation and is not sent again. `sender` is the + * session that made that call, when known. + */ +async function publishPrompt( + child: NativeSubagent, + prompt: string, + toolUseId: string | undefined, + sender: string | undefined, + publish: Publish, +): Promise { + child.promptIds ??= new Set(); + const messageId = toolUseId ?? `${child.sessionId}:prompt:${child.promptIds.size + 1}`; + if (child.promptIds.has(messageId)) return; + child.promptIds.add(messageId); + await publish({ + sessionId: child.sessionId, + update: { + sessionUpdate: "session_message", + messageId, + ...(sender ? { senderSessionId: sender } : {}), + recipientSessionId: child.sessionId, + content: [{ type: "text", text: prompt }], + }, + }); +} + +/** + * The RFD form: the work state at the end of a child's work. + * + * A failure is idle with the `error` stop reason and, when the SDK gives its + * text, the JSON-RPC `error`, as the RFD asks: a v1 child has no prompt + * response, so this is how a client learns that its work failed. The error is + * an internal error, as for a failed turn on the v2 surface. + * + * A task that a worker restart orphaned, or that was still running when the + * stream ended, is idle with no stop reason: its work ended, but the adapter + * did not see a cancellation or a failure, which the RFD forbids claiming. + */ +function rfdEndState(state: SubagentState, end: TaskEnd | undefined): WorkState { + if (end?.unexplained || end?.reason === "worker_restart") return { state: "idle" }; + switch (state) { + case "completed": + return { state: "idle", stopReason: "end_turn" }; + case "cancelled": + return { state: "idle", stopReason: "cancelled" }; + case "failed": + return { + state: "idle", + stopReason: "error", + ...(end?.error + ? { error: { code: RequestError.internalError().code, message: end.error } } + : {}), + }; + default: + // "disconnected": the child can no longer be observed. + return { state: "unknown" }; + } +} + +/** + * The RFD form: the parent's labels for a child. The title is the agent's + * name, with its role after it when both are known, else the role. The + * description is the task's short description, never the prompt. + */ +function rfdLabels( + name: unknown, + role: unknown, + description: unknown, +): { title?: string; role?: string; description?: string } { + const agentName = nonBlankString(name); + const agentRole = nonBlankString(role); + const title = + agentName && agentRole && agentName !== agentRole + ? `${agentName} (${agentRole})` + : (agentName ?? agentRole); + const text = nonBlankString(description); + return { + ...(title ? { title } : {}), + ...(agentRole ? { role: agentRole } : {}), + ...(text ? { description: text } : {}), + }; +} + /** * The agent id that a successful SendMessage result resumed. The SDK puts it * in `tool_use_result.resumedAgentId` when a finished agent runs again. @@ -746,6 +1048,14 @@ function applySubagentIdentity( child.task = subagentDescription(identity.prompt, identity.description); } child.prompt ??= identity.prompt; + const labels = rfdLabels( + identity.name, + identity.subagentType ?? child.role, + identity.description, + ); + child.title = labels.title ?? child.title; + child.role = labels.role ?? child.role; + child.description = labels.description ?? child.description; } /** The prompt text unchanged, or `undefined` when it is not a non-blank string. */ diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index c9400bca..75fad9c8 100644 --- a/src/tests/acp-agent.test.ts +++ b/src/tests/acp-agent.test.ts @@ -110,6 +110,7 @@ import type { BetaWebFetchToolResultBlockParam, BetaCodeExecutionToolResultBlockParam, } from "@anthropic-ai/sdk/resources/beta.mjs"; +import { validateRecorded } from "./acp-scenarios/schema.js"; /** The capabilities of an AIR client. Only AIR gets the AIR extensions of * `docs/air-extensions.md`. */ @@ -117,6 +118,13 @@ const AIR_CLIENT_CAPABILITIES = { _meta: { jetbrains: { air: { version: 1, capabilities: [] as string[] } } }, }; +/** AIR with native subagent sessions. AIR gets the earlier draft of the + * subagents RFD (`subagent_spawned`, `subagent_state_update`, a session per + * generation), which the tests that use this pin. */ +const AIR_NATIVE_SUBAGENTS = { + _meta: { jetbrains: { air: { version: 1, capabilities: ["nativeSubagentSessions"] } } }, +} as ClientCapabilities; + /** A `system`/init frame advertising the msg_lifecycle_v1 capability, so the * consumer latches `session.msgLifecycleV1` and cancel() routes orphan * accounting through `orphanCommands` (CLIs 2.1.206+). */ @@ -3509,7 +3517,7 @@ describe("subagent transcript replay", () => { await initializeClient( agent, capability === "native" - ? ({ subagents: {} } as ClientCapabilities) + ? AIR_NATIVE_SUBAGENTS : capability === "legacy" ? { _meta: { "subagent-transcript": true } } : {}, @@ -3645,7 +3653,7 @@ describe("subagent transcript replay", () => { } as unknown as AcpClient, { log: () => {}, error: () => {} }, ); - await initializeClient(agent, { subagents: {} } as ClientCapabilities); + await initializeClient(agent, AIR_NATIVE_SUBAGENTS); vi.mocked(getSessionInfo).mockResolvedValueOnce({ cwd: "/tmp/proj" } as any); vi.mocked(getSubagentMessages).mockClear(); await ( @@ -5563,6 +5571,228 @@ describe("subagent permission attribution (issue #851)", () => { expect(requests[0].sessionId).toBe("session-1"); }); + it("keeps a raced permission's tool call on the root after a cancel, while its background child runs on", async () => { + const updates: AcpSessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (notification: AcpSessionNotification) => { + updates.push(notification); + }, + requestPermission: async () => ({ + outcome: { outcome: "selected", optionId: "allow-once" }, + }), + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + await agent.initialize({ + protocolVersion: 1, + clientCapabilities: { subagents: {} } as ClientCapabilities, + }); + const permitted = Promise.withResolvers(); + const cancelled = Promise.withResolvers(); + injectGeneratorSession(agent, (input) => + (async function* () { + const { value: prompt } = await input[Symbol.asyncIterator]().next(); + yield userEcho(prompt); + // The child's task starts before the root's Agent frame, so the child + // is known but not announced when it asks for permission. + yield { + ...taskStarted("agent-42", "toolu_parent"), + subagent_type: "Explore", + is_backgrounded: true, + }; + await permitted.promise; + yield rootToolUse("toolu_parent", "Agent"); + await cancelled.promise; + // The child runs in the background, so it outlives the cancel. + yield { + type: "tool_progress", + tool_use_id: "toolu_sub", + tool_name: "Bash", + parent_tool_use_id: "toolu_parent", + elapsed_time_seconds: 3, + uuid: randomUUID(), + session_id: "test-session", + }; + yield successResult(); + })(), + ); + const turn = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "go" }], + }); + await vi.waitFor(() => + expect(agent.sessions["test-session"]?.nativeSubagentsByTaskId?.has("agent-42")).toBe(true), + ); + await agent.canUseTool("test-session")("Bash", { command: "ls" }, { + signal: new AbortController().signal, + suggestions: [], + toolUseID: "toolu_sub", + agentID: "agent-42", + } as any); + permitted.resolve(); + await vi.waitFor(() => + expect(updates.map(({ update }) => update.sessionUpdate)).toContain("subagent_update"), + ); + + await agent.cancel({ sessionId: "test-session" }); + cancelled.resolve(); + await expect(turn).resolves.toMatchObject({ stopReason: "cancelled" }); + + const sessionsOfToolCall = updates.flatMap(({ sessionId, update }) => + (update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update") && + update.toolCallId === "toolu_sub" + ? [`${update.sessionUpdate} ${update.status ?? ""} ${sessionId}`] + : [], + ); + expect(sessionsOfToolCall).toEqual([ + "tool_call pending test-session", + "tool_call_update in_progress test-session", + ]); + }); + + it("sends AIR no update of a child's permission tool call after the cancel ended the child", async () => { + const updates: AcpSessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (notification: AcpSessionNotification) => { + updates.push(notification); + }, + requestPermission: async () => ({ + outcome: { outcome: "selected", optionId: "allow-once" }, + }), + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + await agent.initialize({ protocolVersion: 1, clientCapabilities: AIR_NATIVE_SUBAGENTS }); + const cancelled = Promise.withResolvers(); + injectGeneratorSession(agent, (input) => + (async function* () { + const { value: prompt } = await input[Symbol.asyncIterator]().next(); + yield userEcho(prompt); + yield rootToolUse("toolu_parent", "Agent"); + yield { + ...taskStarted("agent-42", "toolu_parent"), + subagent_type: "Explore", + is_backgrounded: true, + }; + await cancelled.promise; + // The child runs in the background, so it outlives the cancel. + yield { + type: "tool_progress", + tool_use_id: "toolu_sub", + tool_name: "Bash", + parent_tool_use_id: "toolu_parent", + elapsed_time_seconds: 3, + uuid: randomUUID(), + session_id: "test-session", + }; + yield successResult(); + })(), + ); + const turn = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "go" }], + }); + await vi.waitFor(() => + expect(updates.map(({ update }) => update.sessionUpdate)).toContain("subagent_spawned"), + ); + await agent.canUseTool("test-session")("Bash", { command: "ls" }, { + signal: new AbortController().signal, + suggestions: [], + toolUseID: "toolu_sub", + agentID: "agent-42", + } as any); + + await agent.cancel({ sessionId: "test-session" }); + cancelled.resolve(); + await expect(turn).resolves.toMatchObject({ stopReason: "cancelled" }); + + // AIR's draft ends the child at the cancel, and an ended child gets no + // more updates. + const childEvents = updates.flatMap(({ sessionId, update }) => + update.sessionUpdate === "subagent_state_update" + ? [`subagent ${update.state}`] + : (update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update") && + update.toolCallId === "toolu_sub" + ? [`${update.sessionUpdate} ${update.status ?? ""} ${sessionId}`] + : [], + ); + expect(childEvents).toEqual(["tool_call pending agent-42", "subagent cancelled"]); + }); + + it("keeps the Agent call's own heartbeat and result on the session that made the call (the subagents RFD)", async () => { + const updates: AcpSessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (notification: AcpSessionNotification) => { + updates.push(notification); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + await initializeClient(agent, { subagents: {} } as ClientCapabilities); + injectGeneratorSession( + agent, + makeGenerator([ + rootToolUse("toolu_parent", "Agent", { + subagent_type: "Explore", + description: "Investigate", + prompt: "Find the parser", + }), + { ...taskStarted("agent-42", "toolu_parent"), subagent_type: "Explore" }, + // A heartbeat of the running Agent call: its own id is synthetic, and + // the SDK names the Agent call as `parent_tool_use_id`. + { + type: "tool_progress", + tool_use_id: "toolu_parent-heartbeat-1", + tool_name: "Agent", + parent_tool_use_id: "toolu_parent", + elapsed_time_seconds: 30, + uuid: randomUUID(), + session_id: "test-session", + }, + { + type: "system", + subtype: "task_notification", + task_id: "agent-42", + tool_use_id: "toolu_parent", + status: "completed", + output_file: "", + summary: "done", + uuid: randomUUID(), + session_id: "test-session", + }, + { + type: "user", + uuid: randomUUID(), + session_id: "test-session", + parent_tool_use_id: null, + message: { + role: "user", + content: [ + { type: "tool_result", tool_use_id: "toolu_parent", content: "It is in x.ts." }, + ], + }, + }, + successResult(), + ]), + ); + + await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "go" }] }); + + const agentCall = updates.flatMap(({ sessionId, update }) => + (update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update") && + update.toolCallId === "toolu_parent" + ? [{ sessionId, status: update.status }] + : [], + ); + expect(agentCall.map(({ status }) => status)).toEqual( + expect.arrayContaining(["in_progress", "completed"]), + ); + expect(new Set(agentCall.map(({ sessionId }) => sessionId))).toEqual(new Set(["test-session"])); + }); + it("routes a native child AskUserQuestion and its tool call to the child session", async () => { const { agent, updates, elicitations, session } = setup(); await agent.initialize({ @@ -5626,6 +5856,25 @@ describe("subagent permission attribution (issue #851)", () => { }; } + /** The call `toolUseId` of the tool `name` that the root session makes. */ + function rootToolUse(toolUseId: string, name: string, input: Record = {}) { + return { + type: "assistant" as const, + uuid: randomUUID(), + session_id: "test-session", + parent_tool_use_id: null, + message: { + id: `msg-${toolUseId}`, + model: "claude-sonnet-4-5", + role: "assistant" as const, + type: "message", + stop_reason: "tool_use", + content: [{ type: "tool_use" as const, id: toolUseId, name, input }], + usage: { input_tokens: 0, output_tokens: 0 }, + }, + }; + } + function successResult() { return { type: "result" as const, @@ -5995,9 +6244,7 @@ describe("subagent permission attribution (issue #851)", () => { ); await agent.initialize({ protocolVersion: 1, - clientCapabilities: { subagents: {} } as ClientCapabilities & { - subagents: Record; - }, + clientCapabilities: AIR_NATIVE_SUBAGENTS, }); injectGeneratorSession( agent, @@ -6041,6 +6288,312 @@ describe("subagent permission attribution (issue #851)", () => { ); }); + describe("the subagents RFD, for a client that declares subagents and is not AIR", () => { + // Each task_started names a subagent type: a task without one is not a subagent. + /** The updates of `frames`, for a client that declares `subagents`. */ + async function run(frames: unknown[]): Promise { + return (await runAgent(frames)).updates; + } + /** {@link run}, and the agent that ran the frames. */ + async function runAgent(frames: unknown[]) { + const updates: AcpSessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: AcpSessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + await agent.initialize({ + protocolVersion: 1, + clientCapabilities: { subagents: {} } as ClientCapabilities, + }); + // The root's Agent call comes first: the child's parent is the root. + injectGeneratorSession( + agent, + makeGenerator([rootToolUse("toolu_parent", "Agent"), ...frames, successResult()] as never), + ); + await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "go" }] }); + // Every update is a valid v1 session update, without an extension. + for (const payload of updates) { + expect(validateRecorded({ kind: "sessionUpdate", payload }, new Set())).toEqual([]); + } + return { agent, updates }; + } + const subagentTraffic = (updates: AcpSessionNotification[]) => + updates.filter(({ update }) => + [ + "subagent_update", + "session_message", + "subagent_spawned", + "subagent_state_update", + ].includes(update.sessionUpdate), + ); + const taskUpdated = (patch: Record) => ({ + type: "system", + subtype: "task_updated", + task_id: "agent-42", + patch, + uuid: randomUUID(), + session_id: "test-session", + }); + + it("announces the child as running, then sends it the prompt as a message", async () => { + const updates = await run([ + { ...taskStarted("agent-42", "toolu_parent"), subagent_type: "Explore", prompt: "Go" }, + taskUpdated({ status: "completed" }), + ]); + + expect(subagentTraffic(updates)).toEqual([ + { + sessionId: "test-session", + update: { + sessionUpdate: "subagent_update", + sessionId: "agent-42", + title: "Explore", + description: "Investigate", + state: { state: "running" }, + }, + }, + { + sessionId: "agent-42", + update: { + sessionUpdate: "session_message", + messageId: "toolu_parent", + senderSessionId: "test-session", + recipientSessionId: "agent-42", + content: [{ type: "text", text: "Go" }], + }, + }, + { + sessionId: "test-session", + update: { + sessionUpdate: "subagent_update", + sessionId: "agent-42", + state: { state: "idle", stopReason: "end_turn" }, + }, + }, + ]); + }); + + it("reports a failed child as idle, with the SDK's error", async () => { + const updates = await run([ + { ...taskStarted("agent-42", "toolu_parent"), subagent_type: "Explore" }, + taskUpdated({ status: "failed", error: "Context too small" }), + ]); + + expect(subagentTraffic(updates).at(-1)?.update).toEqual({ + sessionUpdate: "subagent_update", + sessionId: "agent-42", + state: { + state: "idle", + stopReason: "error", + error: { code: -32603, message: "Context too small" }, + }, + }); + }); + + it("reports a child that the SDK stopped as cancelled", async () => { + const updates = await run([ + { ...taskStarted("agent-42", "toolu_parent"), subagent_type: "Explore" }, + taskUpdated({ status: "killed" }), + ]); + + expect(subagentTraffic(updates).at(-1)?.update).toMatchObject({ + state: { state: "idle", stopReason: "cancelled" }, + }); + }); + + it("sends no update of the earlier draft that AIR implements", async () => { + const updates = await run([ + { ...taskStarted("agent-42", "toolu_parent"), subagent_type: "Explore" }, + taskUpdated({ status: "completed" }), + ]); + + expect(subagentTraffic(updates).length).toBeGreaterThan(0); + expect( + updates.filter(({ update }) => + ["subagent_spawned", "subagent_state_update"].includes(update.sessionUpdate), + ), + ).toEqual([]); + }); + + const taskNotification = ( + toolUseId: string, + status: string, + extra: Record = {}, + ) => ({ + type: "system", + subtype: "task_notification", + task_id: "agent-42", + tool_use_id: toolUseId, + status, + output_file: "", + summary: "done", + uuid: randomUUID(), + session_id: "test-session", + ...extra, + }); + + // agentclientprotocol/agent-client-protocol#2308: a cancel stops the + // descendant work that the cancelled work waits on, and leaves the rest + // to the Agent; the Client cleans up a descendant's requests only once it + // reports a cancelled idle. + it("at a root cancel, stops the foreground child and leaves the background child running", async () => { + const log: string[] = []; + const updates: AcpSessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: AcpSessionNotification) => { + updates.push(update); + if (update.update.sessionUpdate === "subagent_update") { + log.push(`${update.update.sessionId} ${JSON.stringify(update.update.state)}`); + } + }, + requestPermission: (params: { sessionId: string }, signal?: AbortSignal) => { + log.push(`request @${params.sessionId}`); + signal?.addEventListener("abort", () => log.push(`withdrawn @${params.sessionId}`)); + return new Promise(() => {}); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + await agent.initialize({ + protocolVersion: 1, + clientCapabilities: { subagents: {} } as ClientCapabilities, + }); + + const later = new Pushable(); + const input = new Pushable(); + let started!: () => void; + const ready = new Promise((resolve) => (started = resolve)); + const launch = (id: string, input: Record) => + rootToolUse(id, "Agent", { description: id, prompt: "Work", ...input }); + async function* stream() { + const { value } = await input[Symbol.asyncIterator]().next(); + yield userEcho(value); + yield launch("toolu_fg", { subagent_type: "Explore" }); + yield { ...taskStarted("agent-fg", "toolu_fg"), subagent_type: "Explore" }; + yield launch("toolu_bg", { subagent_type: "Explore", run_in_background: true }); + yield { + ...taskStarted("agent-bg", "toolu_bg"), + subagent_type: "Explore", + is_backgrounded: true, + }; + started(); + yield* later; + } + const query = wrapQuery(stream()); + // The CLI's interrupt (read in the bundled CLI): the foreground child's + // abort controller is linked to the turn's, so its request is withdrawn + // and its run reports killed, then stopped, then the turn goes idle. The + // background child's is not linked, so nothing happens to it. + const foreground = new AbortController(); + query.interrupt = vi.fn(async () => { + foreground.abort(); + setTimeout(() => { + later.push({ ...taskUpdated({ status: "killed" }), task_id: "agent-fg" }); + later.push({ ...taskNotification("toolu_fg", "stopped"), task_id: "agent-fg" }); + later.push({ + type: "system", + subtype: "session_state_changed", + state: "idle", + uuid: randomUUID(), + session_id: "test-session", + }); + }, 5); + }); + agent.sessions["test-session"] = mockSessionState({ query, input }); + + const prompt = agent + .prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "go" }] }) + .then((response) => { + log.push(`prompt ${response.stopReason}`); + return response; + }); + await ready; + await new Promise((resolve) => setTimeout(resolve, 20)); + const canUseTool = agent.canUseTool("test-session"); + const ask = (agentID: string, signal: AbortSignal) => + canUseTool("Bash", { command: "make" }, { + signal, + suggestions: [], + toolUseID: `${agentID}-bash`, + agentID, + } as never).catch(() => {}); + void ask("agent-fg", foreground.signal); + void ask("agent-bg", new AbortController().signal); + await new Promise((resolve) => setTimeout(resolve, 20)); + + await agent.cancel({ sessionId: "test-session" }); + await prompt; + const atResponse = [...log]; + // The CLI process exits while the background child still runs. + later.end(); + await new Promise((resolve) => setTimeout(resolve, 20)); + + const before = (prefix: string) => atResponse.filter((line) => line.startsWith(prefix)); + // The foreground child: withdrawn, then reported cancelled before the + // cancelled prompt resolves. + expect(atResponse).toContain("withdrawn @agent-fg"); + expect(before("agent-fg ").at(-1)).toBe('agent-fg {"state":"idle","stopReason":"cancelled"}'); + expect(atResponse.at(-1)).toBe("prompt cancelled"); + // The background child keeps running, with its request still open. + expect(atResponse).not.toContain("withdrawn @agent-bg"); + expect(before("agent-bg ")).toEqual(['agent-bg {"state":"running"}']); + // When the stream ends, its end has no reason the adapter saw. + expect(log.filter((line) => line.startsWith("agent-bg ")).at(-1)).toBe( + 'agent-bg {"state":"idle"}', + ); + for (const payload of updates) { + expect(validateRecorded({ kind: "sessionUpdate", payload }, new Set())).toEqual([]); + } + }); + + it("adds the SDK's error to a failure that a notification reported first", async () => { + const updates = await run([ + { ...taskStarted("agent-42", "toolu_parent"), subagent_type: "Explore" }, + taskNotification("toolu_parent", "failed"), + taskUpdated({ status: "failed", error: "Context too small" }), + ]); + + expect( + subagentTraffic(updates) + .slice(-2) + .map(({ update }) => update), + ).toEqual([ + { + sessionUpdate: "subagent_update", + sessionId: "agent-42", + state: { state: "idle", stopReason: "error" }, + }, + { + sessionUpdate: "subagent_update", + sessionId: "agent-42", + state: { + state: "idle", + stopReason: "error", + error: { code: -32603, message: "Context too small" }, + }, + }, + ]); + }); + + it("reports a child that a worker restart orphaned as idle, not cancelled", async () => { + const updates = await run([ + { ...taskStarted("agent-42", "toolu_parent"), subagent_type: "Explore" }, + taskNotification("toolu_parent", "stopped", { reason: "worker_restart" }), + ]); + + expect(subagentTraffic(updates).at(-1)?.update).toEqual({ + sessionUpdate: "subagent_update", + sessionId: "agent-42", + state: { state: "idle" }, + }); + }); + }); + it("buffers child output until spawn and drops duplicates and late updates", async () => { const updates: AcpSessionNotification[] = []; const agent = new ClaudeAcpAgent( @@ -6053,9 +6606,7 @@ describe("subagent permission attribution (issue #851)", () => { ); await agent.initialize({ protocolVersion: 1, - clientCapabilities: { subagents: {} } as ClientCapabilities & { - subagents: Record; - }, + clientCapabilities: AIR_NATIVE_SUBAGENTS, }); const childMessage = (text: string) => ({ type: "stream_event" as const, @@ -6175,9 +6726,7 @@ describe("subagent permission attribution (issue #851)", () => { ); await agent.initialize({ protocolVersion: 1, - clientCapabilities: { subagents: {} } as ClientCapabilities & { - subagents: Record; - }, + clientCapabilities: AIR_NATIVE_SUBAGENTS, }); injectGeneratorSession( agent, @@ -6250,9 +6799,7 @@ describe("subagent permission attribution (issue #851)", () => { ); await agent.initialize({ protocolVersion: 1, - clientCapabilities: { subagents: {} } as ClientCapabilities & { - subagents: Record; - }, + clientCapabilities: AIR_NATIVE_SUBAGENTS, }); const agentTool = (id: string, parent_tool_use_id: string | null) => ({ type: "assistant" as const, @@ -6349,9 +6896,7 @@ describe("subagent permission attribution (issue #851)", () => { ); await agent.initialize({ protocolVersion: 1, - clientCapabilities: { subagents: {} } as ClientCapabilities & { - subagents: Record; - }, + clientCapabilities: AIR_NATIVE_SUBAGENTS, }); injectGeneratorSession( agent, @@ -6383,9 +6928,7 @@ describe("subagent permission attribution (issue #851)", () => { ); await agent.initialize({ protocolVersion: 1, - clientCapabilities: { subagents: {} } as ClientCapabilities & { - subagents: Record; - }, + clientCapabilities: AIR_NATIVE_SUBAGENTS, }); injectGeneratorSession( agent, @@ -6425,9 +6968,7 @@ describe("subagent permission attribution (issue #851)", () => { ); await agent.initialize({ protocolVersion: 1, - clientCapabilities: { subagents: {} } as ClientCapabilities & { - subagents: Record; - }, + clientCapabilities: AIR_NATIVE_SUBAGENTS, }); let childStarted!: () => void; const started = new Promise((resolve) => (childStarted = resolve)); @@ -6501,6 +7042,8 @@ describe("subagent permission attribution (issue #851)", () => { agent, makeGenerator([ ...childMessages, + // The spawn arrives after the child's output, which waits for it. + rootToolUse("toolu_parent", "Agent"), { ...taskStarted("agent-42", "toolu_parent"), subagent_type: "Explore" }, { type: "system", @@ -7403,7 +7946,7 @@ describe("native subagent eager tool ownership", () => { nativeSubagentParentByToolUseId: new Map(), }; const runtime = new NativeSubagentRuntime( - true, + "air", "root", session, async (notification) => { @@ -7463,7 +8006,7 @@ describe("native subagent eager tool ownership", () => { nativeSubagentTaskIdByToolUseId: new Map([["toolu_agent", "agent-1"]]), nativeSubagentParentByToolUseId: new Map(), }; - const runtime = new NativeSubagentRuntime(true, "root", session, async () => {}, { + const runtime = new NativeSubagentRuntime("air", "root", session, async () => {}, { log: () => {}, }); const terminal = { @@ -10561,22 +11104,19 @@ describe("logout", () => { expect(response.agentCapabilities?.auth?.logout).toEqual({}); }); - it("advertises the agent subagent capability independently of client negotiation", async () => { + it("advertises the agent subagent capability of AIR's draft to AIR alone", async () => { const agent = createMockAgent(); - const unsupported = await agent.initialize({ protocolVersion: 1, clientCapabilities: {} }); - const supported = await agent.initialize({ - protocolVersion: 1, - clientCapabilities: { subagents: {} } as ClientCapabilities & { - subagents: Record; - }, - }); + const advertised = async (clientCapabilities: ClientCapabilities) => + ( + (await agent.initialize({ protocolVersion: 1, clientCapabilities })).agentCapabilities + ?.sessionCapabilities as { subagents?: unknown } + ).subagents; - expect( - (unsupported.agentCapabilities?.sessionCapabilities as { subagents?: unknown }).subagents, - ).toEqual({}); - expect( - (supported.agentCapabilities?.sessionCapabilities as { subagents?: unknown }).subagents, - ).toEqual({}); + // The subagents RFD has only the client capability. + expect(await advertised({})).toBeUndefined(); + expect(await advertised({ subagents: {} } as ClientCapabilities)).toBeUndefined(); + expect(await advertised(AIR_CLIENT_CAPABILITIES)).toEqual({}); + expect(await advertised(AIR_NATIVE_SUBAGENTS)).toEqual({}); }); it("negotiates subagents through AIR metadata when the SDK strips the draft field", async () => { @@ -15813,19 +16353,19 @@ describe("post-error recovery", () => { error: (...args) => errors.push(args), }); async function* emptyQuery() {} + const parentCancelled = vi.fn(async () => { + throw new Error("client disconnected"); + }); const session = mockSessionState({ query: wrapQuery(emptyQuery()), - nativeSubagentRuntime: { - finishAll: vi.fn(async () => { - throw new Error("client disconnected"); - }), - } as unknown as NativeSubagentRuntime, + nativeSubagentRuntime: { form: "air", parentCancelled } as unknown as NativeSubagentRuntime, eagerToolCallSessions: new Map([["tool", "test-session"]]), }); agent.sessions["test-session"] = session; await expect(agent.cancel({ sessionId: "test-session" })).resolves.toBeUndefined(); + expect(parentCancelled).toHaveBeenCalledOnce(); expect(session.query.interrupt).toHaveBeenCalledOnce(); expect(session.eagerToolCallSessions).toEqual(new Map()); expect(errors.flat().join(" ")).toContain("failed to publish cancelled subagent state"); @@ -19160,9 +19700,7 @@ describe("deferred settlement for live background subagents (issues #864/#866)", if (options.subagents) { await agent.initialize({ protocolVersion: 1, - clientCapabilities: { subagents: {} } as ClientCapabilities & { - subagents: Record; - }, + clientCapabilities: AIR_NATIVE_SUBAGENTS, }); } const first = await agent.prompt({ diff --git a/src/tests/acp-scenarios/compare.ts b/src/tests/acp-scenarios/compare.ts index 69f3eeea..5d11ea27 100644 --- a/src/tests/acp-scenarios/compare.ts +++ b/src/tests/acp-scenarios/compare.ts @@ -19,6 +19,9 @@ * - An `available_commands_update` also lists the `mcp` command of the * adapter (see {@link ADAPTER_COMMANDS}), when origin/main did not list * `mcp`. The adapter replaces the text of `/mcp` for every client. + * - The `initialize` response does not advertise + * `agentCapabilities.sessionCapabilities.subagents`: the subagents RFD has + * only the client capability, and only AIR's draft keeps the agent one. * - A `terminal_exit` has a `null` exit code where origin/main sent a code * that the tool result does not say (see `CommandOutput.exitCode`): 1 for a * failure that names no code or an interrupted command, 0 for a @@ -198,13 +201,29 @@ function isMeta(key: string): boolean { return key.startsWith("_meta."); } +/** `record` without the agent subagent capability, when it is an `initialize` response. */ +function withoutAgentSubagentCapability(record: Recorded): Recorded { + if (record.kind !== "initialize") return record; + const payload = record.payload as Json; + const agent = payload.agentCapabilities as Json | undefined; + const session = agent?.sessionCapabilities as Json | undefined; + if (!session || !("subagents" in session)) return record; + return { + ...record, + payload: { + ...payload, + agentCapabilities: { ...agent, sessionCapabilities: without(session, "subagents") }, + }, + } as Recorded; +} + /** * Returns the differences between the baseline and the current traffic that * the compatibility rule does not allow. An empty list means compatible. */ export function compareWithBaseline(baseline: Recorded[], current: Recorded[]): string[] { const expected = baseline - .map((record) => withoutAirOnlyKeys(record) as Recorded) + .map((record) => withoutAgentSubagentCapability(withoutAirOnlyKeys(record) as Recorded)) .filter((record) => { const update = updateOf(record); return !(update?.sessionUpdate === "session_info_update" && Object.keys(update).length === 1); diff --git a/src/tests/native-subagents.test.ts b/src/tests/native-subagents.test.ts index cd59c0b8..35b49a69 100644 --- a/src/tests/native-subagents.test.ts +++ b/src/tests/native-subagents.test.ts @@ -68,8 +68,8 @@ describe("NativeSubagentRuntime lifecycle", () => { await release.promise; }); - const first = announceNativeSubagent(value, publish); - const second = announceNativeSubagent(value, publish); + const first = announceNativeSubagent(value, publish, "air"); + const second = announceNativeSubagent(value, publish, "air"); await Promise.resolve(); expect(publish).toHaveBeenCalledTimes(1); @@ -101,7 +101,7 @@ describe("NativeSubagentRuntime lifecycle", () => { }); it("falls back to one ordinary failed tool call when no child ever starts", async () => { - const runtime = new NativeSubagentRuntime(true, "root", {}, async () => {}, { log: () => {} }); + const runtime = new NativeSubagentRuntime("air", "root", {}, async () => {}, { log: () => {} }); await expect( runtime.route(control("tool_call", "pending"), async () => {}), @@ -124,7 +124,7 @@ describe("NativeSubagentRuntime lifecycle", () => { }); it("creates a schema-valid non-native failed tool call without a cached initial call", async () => { - const runtime = new NativeSubagentRuntime(true, "root", {}, async () => {}, { log: () => {} }); + const runtime = new NativeSubagentRuntime("air", "root", {}, async () => {}, { log: () => {} }); const fallback = await runtime.route( { @@ -166,7 +166,7 @@ describe("NativeSubagentRuntime lifecycle", () => { const session = sessionWith(outer); const published: AcpSessionNotification[] = []; const runtime = new NativeSubagentRuntime( - true, + "air", "root", session, async (notification) => { @@ -211,7 +211,7 @@ describe("NativeSubagentRuntime lifecycle", () => { const published: AcpSessionNotification[] = []; const delivered: AcpSessionNotification[] = []; const runtime = new NativeSubagentRuntime( - true, + "air", "root", {}, async (notification) => { @@ -258,7 +258,7 @@ describe("NativeSubagentRuntime lifecycle", () => { it("creates a distinct ACP lifecycle when the SDK restarts the same task id", async () => { const published: AcpSessionNotification[] = []; const runtime = new NativeSubagentRuntime( - true, + "air", "root", {}, async (notification) => { @@ -332,7 +332,7 @@ describe("NativeSubagentRuntime lifecycle", () => { it("announces resumed subagent generation immediately when SendMessage restarts the task (#1158)", async () => { const published: AcpSessionNotification[] = []; const runtime = new NativeSubagentRuntime( - true, + "air", "root", {}, async (notification) => { @@ -393,7 +393,7 @@ describe("NativeSubagentRuntime lifecycle", () => { it("announces a resumed nested subagent under a live ancestor when its parent has finished", async () => { const published: AcpSessionNotification[] = []; const runtime = new NativeSubagentRuntime( - true, + "air", "root", {}, async (notification) => { @@ -439,7 +439,7 @@ describe("NativeSubagentRuntime lifecycle", () => { ): Promise> { const published: AcpSessionNotification[] = []; const runtime = new NativeSubagentRuntime( - true, + "air", "root", {}, async (notification) => { @@ -507,7 +507,7 @@ describe("NativeSubagentRuntime lifecycle", () => { const published: AcpSessionNotification[] = []; const logged: string[] = []; const runtime = new NativeSubagentRuntime( - true, + "air", "root", {}, async (notification) => { @@ -625,7 +625,7 @@ describe("NativeSubagentRuntime lifecycle", () => { it("keeps the work of a child tool call in the generation that started it", async () => { const published: AcpSessionNotification[] = []; const runtime = new NativeSubagentRuntime( - true, + "air", "root", {}, async (notification) => { @@ -729,3 +729,293 @@ describe("NativeSubagentRuntime lifecycle", () => { }); }); }); + +/** + * The subagents RFD (agentclientprotocol/agent-client-protocol#1992, as + * revised), for a client that declares `subagents` and is not AIR. + */ +describe("NativeSubagentRuntime in the RFD form", () => { + const launch = ( + toolCallId: string, + input: Record, + parentToolUseId?: string, + ): AcpSessionNotification => + ({ + sessionId: "root", + update: { + sessionUpdate: "tool_call", + toolCallId, + title: "Agent", + status: "pending", + rawInput: input, + _meta: { + claudeCode: { toolName: "Agent", ...(parentToolUseId ? { parentToolUseId } : {}) }, + }, + }, + }) as AcpSessionNotification; + const output = (parentToolUseId: string): AcpSessionNotification => + ({ + sessionId: "root", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "working" }, + _meta: { claudeCode: { parentToolUseId } }, + }, + }) as AcpSessionNotification; + /** A runtime whose child `worker-1` was launched by `launch-1` and runs. */ + async function runningWorker(input: Record = {}) { + const published: AcpSessionNotification[] = []; + const runtime = new NativeSubagentRuntime( + "rfd", + "root", + {}, + async (notification) => { + published.push(notification); + }, + { log: () => {} }, + ); + const control = await runtime.route( + launch("launch-1", { description: "Fix the build", prompt: "Make it pass", ...input }), + async () => {}, + ); + await runtime.taskStarted( + { taskId: "worker-1", toolUseId: "launch-1", subagentType: "Explore" }, + async () => {}, + ); + return { runtime, published, control }; + } + const states = (published: AcpSessionNotification[]) => + published.flatMap(({ update }) => + update.sessionUpdate === "subagent_update" && update.state ? [update.state] : [], + ); + + it("keeps the Agent tool call as a tool call of the session that made it", async () => { + const { runtime, control } = await runningWorker(); + + expect(control).toMatchObject({ sessionId: "root", update: { toolCallId: "launch-1" } }); + // A nested launch is a tool call of the child that made it. + await expect( + runtime.route(launch("launch-2", { description: "Dig" }, "launch-1"), async () => {}), + ).resolves.toMatchObject({ sessionId: "worker-1", update: { toolCallId: "launch-2" } }); + }); + + it("announces the child with the parent's labels, then sends it the prompt", async () => { + const { published } = await runningWorker({ name: "builder" }); + + expect(published).toEqual([ + { + sessionId: "root", + update: { + sessionUpdate: "subagent_update", + sessionId: "worker-1", + title: "builder (Explore)", + description: "Fix the build", + state: { state: "running" }, + }, + }, + { + sessionId: "worker-1", + update: { + sessionUpdate: "session_message", + messageId: "launch-1", + senderSessionId: "root", + recipientSessionId: "worker-1", + content: [{ type: "text", text: "Make it pass" }], + }, + }, + ]); + }); + + it("titles a child without a name by its role", async () => { + const { published } = await runningWorker(); + + expect(published[0]?.update).toMatchObject({ title: "Explore", description: "Fix the build" }); + }); + + it("does not show a resume yet: the idle child keeps its one session", async () => { + const { runtime, published } = await runningWorker(); + await runtime.finishTask("worker-1", "completed", async () => {}); + + // An idle child is reusable, so its updates still go to it. + await expect(runtime.route(output("launch-1"), async () => {})).resolves.toMatchObject({ + sessionId: "worker-1", + }); + // The SDK runs a resumed agent again under the id of the call that + // resumed it. The child stays idle, and no second session opens. + await runtime.taskStarted( + { taskId: "worker-1", toolUseId: "send-1", subagentType: "Explore", prompt: "Again" }, + async () => {}, + ); + await runtime.taskResumed("worker-1", async () => {}, "Again"); + await expect(runtime.route(output("send-1"), async () => {})).resolves.toBeNull(); + + expect(states(published)).toEqual([ + { state: "running" }, + { state: "idle", stopReason: "end_turn" }, + ]); + expect( + published.some(({ update }) => "sessionId" in update && update.sessionId !== "worker-1"), + ).toBe(false); + }); + + it("keeps a nested child that ends before its parent is known unexposed", async () => { + const { runtime, published } = await runningWorker(); + // worker-2's spawning frame has not arrived, so its parent is unknown. + await runtime.taskStarted( + { taskId: "worker-2", toolUseId: "launch-2", subagentType: "Explore" }, + async () => {}, + ); + await expect(runtime.route(output("launch-2"), async () => {})).resolves.toBeNull(); + // A resume signal does not give the unexposed child another call. + await runtime.taskStarted({ taskId: "worker-2", toolUseId: "send-9" }, async () => {}); + await runtime.finishTask("worker-2", "completed", async () => {}); + // The frame arrives late: it is the tool call of worker-1 that made it. + await expect( + runtime.route(launch("launch-2", { description: "Dig" }, "launch-1"), async () => {}), + ).resolves.toMatchObject({ sessionId: "worker-1" }); + + expect( + published.some( + ({ update }) => + (update.sessionUpdate === "subagent_update" && update.sessionId === "worker-2") || + (update.sessionUpdate === "session_message" && update.recipientSessionId === "worker-2"), + ), + ).toBe(false); + expect(published.some(({ sessionId }) => sessionId === "worker-2")).toBe(false); + }); + + describe("a nested child whose own parent is not announced yet", () => { + /** A runtime, the updates it sends in order, and the deliver function of + * the agent, which routes a buffered update again (as `sendUpdate` does). */ + function setup() { + const sent: AcpSessionNotification[] = []; + const runtime: NativeSubagentRuntime = new NativeSubagentRuntime( + "rfd", + "root", + {}, + async (notification) => { + sent.push(notification); + }, + { log: () => {} }, + ); + const deliver = async (notification: AcpSessionNotification): Promise => { + const routed = await runtime.route(notification, deliver); + if (routed) sent.push(routed); + }; + return { runtime, sent, deliver }; + } + /** The announcements, as [parent, child], and the sessions that got + * traffic before each was announced. */ + function check(sent: AcpSessionNotification[]) { + const announced = new Set(["root"]); + const early: string[] = []; + const announcements: [string, string][] = []; + for (const { sessionId, update } of sent) { + if (!announced.has(sessionId)) early.push(sessionId); + if (update.sessionUpdate === "subagent_update" && !announced.has(update.sessionId)) { + announced.add(update.sessionId); + announcements.push([sessionId, update.sessionId]); + } + } + return { announcements, early }; + } + + it("announces it after that parent, when the parent's task started before its launch frame", async () => { + const { runtime, sent, deliver } = setup(); + // worker-1 is known, but its launch frame has not come: not announced. + await runtime.taskStarted( + { taskId: "worker-1", toolUseId: "launch-1", subagentType: "Explore" }, + deliver, + ); + await deliver(launch("launch-2", { description: "Dig" }, "launch-1")); + await runtime.taskStarted( + { taskId: "worker-2", toolUseId: "launch-2", subagentType: "Explore" }, + deliver, + ); + await deliver(launch("launch-1", { description: "Fix the build" })); + + expect(check(sent)).toEqual({ + announcements: [ + ["root", "worker-1"], + ["worker-1", "worker-2"], + ], + early: [], + }); + }); + + it("never announces it under the root, when its parent is not known yet", async () => { + const { runtime, sent, deliver } = setup(); + await deliver(launch("launch-2", { description: "Dig" }, "launch-1")); + await runtime.taskStarted( + { taskId: "worker-2", toolUseId: "launch-2", subagentType: "Explore" }, + deliver, + ); + await deliver(launch("launch-1", { description: "Fix the build" })); + await runtime.taskStarted( + { taskId: "worker-1", toolUseId: "launch-1", subagentType: "Explore" }, + deliver, + ); + + expect(check(sent)).toEqual({ + announcements: [ + ["root", "worker-1"], + ["worker-1", "worker-2"], + ], + early: [], + }); + }); + }); + + it("keeps the role in the title when a late frame names the child", async () => { + const published: AcpSessionNotification[] = []; + const runtime = new NativeSubagentRuntime( + "rfd", + "root", + {}, + async (notification) => { + published.push(notification); + }, + { log: () => {} }, + ); + await runtime.taskStarted( + { taskId: "worker-1", toolUseId: "launch-1", subagentType: "Explore" }, + async () => {}, + ); + await runtime.route(launch("launch-1", { name: "builder" }), async () => {}); + + expect(published[0]?.update).toMatchObject({ title: "builder (Explore)" }); + }); + + it("reports nothing when the parent is cancelled, until the SDK ends the child", async () => { + const { runtime, published } = await runningWorker(); + + await runtime.parentCancelled(async () => {}); + expect(states(published)).toEqual([{ state: "running" }]); + await runtime.finishTask("worker-1", "killed", async () => {}); + expect(states(published).at(-1)).toEqual({ state: "idle", stopReason: "cancelled" }); + }); + + it("ends the children that still run with the stream as idle, with no reason it did not see", async () => { + // After a cancel, a child that still runs is a background child that + // outlived it: the stream's end is no evidence that it was cancelled. + const cancelled = await runningWorker(); + await cancelled.runtime.finishAll("cancelled", async () => {}); + expect(states(cancelled.published).at(-1)).toEqual({ state: "idle" }); + + const failed = await runningWorker(); + await failed.runtime.finishAll("failed", async () => {}); + expect(states(failed.published).at(-1)).toEqual({ state: "idle" }); + // A later error event does not turn that end into a failure. + await failed.runtime.finishTask("worker-1", "failed", async () => {}, undefined, { + error: "boom", + }); + expect(states(failed.published)).toHaveLength(2); + }); + + it("reports a child that can no longer be observed as unknown", async () => { + const { runtime, published } = await runningWorker(); + + await runtime.finishTask("worker-1", "disconnected", async () => {}); + expect(states(published).at(-1)).toEqual({ state: "unknown" }); + }); +});