diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 9fd9e2b3..ba54f3c7 100644 --- a/src/acp-agent.ts +++ b/src/acp-agent.ts @@ -188,6 +188,7 @@ import { readSubagentMessages, subagentHistory, subagentLaunchIds, + subagentLaunchedBy, subagentLineage, } from "./subagent-history.js"; import { @@ -4329,6 +4330,7 @@ export class ClaudeAcpAgent { async (notification) => this.client.sessionUpdate(asSdkSessionNotification(notification)), this.logger, (agentId) => subagentLineage(params.sessionId, agentId, this.logger), + (launchId) => subagentLaunchedBy(params.sessionId, launchId), )); // Adapter-composed advisories (hook feedback, model fallbacks) are live // events, not something the model said. Clients on the notice contract get diff --git a/src/native-subagents.ts b/src/native-subagents.ts index 2662c33c..4ec7c281 100644 --- a/src/native-subagents.ts +++ b/src/native-subagents.ts @@ -108,6 +108,12 @@ export type HistoricalSubagent = { */ export type SubagentLineage = (agentId: string) => Promise; +/** + * The agent that the Agent or Task call `launchId` launched, from the session's + * subagent history; undefined when the history has no such call. + */ +export type LaunchedAgent = (launchId: string) => Promise; + const MAX_PENDING_PARENTS = 64; const MAX_PENDING_UPDATES = 256; const MAX_PENDING_UPDATES_PER_PARENT = 32; @@ -136,17 +142,30 @@ export class NativeSubagentRuntime { */ private readonly childByToolCall = new Map(); private readonly taskFinishPromises = new Map>(); - /** The RFD form: the session that made each SendMessage call. */ + /** The session that made each SendMessage call. */ private readonly senderByToolUse = new Map(); private readonly generationByTaskId = new Map(); private readonly pending = new Map(); private pendingCount = 0; + /** The AIR form: agents being restored from the history, by agent id. */ + private readonly agentRestores = new Map>(); + /** The AIR form: launch ids being resolved to an agent, by launch id. */ + private readonly launchRestores = new Map>(); + /** The AIR form: launch ids that the history has no restorable agent for. */ + private readonly unresolvedLaunches = new Set(); + /** The AIR form: agents that the history could not place, so it is read once. */ + private readonly unrestorableAgents = new Set(); + /** The latest generation number given to a restored agent. */ + private lastRestoredGeneration = 0; + /** Bumped when the restore state is dropped; a restore that began before it is void. */ + private restoreEpoch = 0; /** The RFD form: how many times the parent was cancelled (see {@link awaitingUser}). */ private parentCancels = 0; /** * `form` is undefined for a client without native subagent sessions. The RFD - * form restores an agent from before a load or resume with `lineage`. + * form restores an agent from before a load or resume with `lineage`. The AIR + * form does too, and `launchedAgent` finds the agent that a launch call made. */ constructor( form: SubagentForm | undefined, @@ -155,6 +174,7 @@ export class NativeSubagentRuntime { private readonly publish: Publish, private readonly logger: Logger, private readonly lineage?: SubagentLineage, + private readonly launchedAgent?: LaunchedAgent, ) { this.form = form; this.enabled = form !== undefined; @@ -174,7 +194,7 @@ export class NativeSubagentRuntime { forcedSessionId?: string, ): Promise { const routed = await this.routeUpdate(notification, deliver, forcedSessionId); - if (routed && this.form === "rfd") this.rememberSendMessage(routed); + if (routed) this.rememberSendMessage(routed); return routed; } @@ -266,7 +286,13 @@ export class NativeSubagentRuntime { if (this.enabled && claudeMeta?.parentToolUseId) { const child = this.childByParentToolUse.get(claudeMeta.parentToolUseId); if (!child || !child.announced) { - this.buffer(claudeMeta.parentToolUseId, notification); + const launchId = claudeMeta.parentToolUseId; + this.buffer(launchId, notification); + // An id that this process never saw is the launch call of an agent + // from before a load or resume; a new Agent call is known by now. + if (!child && this.form === "air" && this.isUnseenLaunch(launchId)) { + await this.restoreLaunch(launchId, deliver); + } return null; } return this.toChild(child, notification, toolCallId); @@ -294,6 +320,24 @@ export class NativeSubagentRuntime { ); return; } + // The SDK starts an agent from before a restart under the id of the + // SendMessage call that resumed it, as in the RFD form (see `restore`). + // Its output comes under the id of the call that launched it. + if ( + this.form === "air" && + task.subagentType && + task.toolUseId && + this.senderByToolUse.has(task.toolUseId) && + !this.children.has(task.taskId) + ) { + await this.restoreAgent(task.taskId, deliver, promptText(task.prompt)); + const restored = this.children.get(task.taskId); + if (restored) { + this.taskByToolUse.set(task.toolUseId, task.taskId); + this.childByParentToolUse.set(task.toolUseId, restored); + return; + } + } if (!task.subagentType) { if (task.toolUseId) { this.takePending(task.toolUseId); @@ -350,11 +394,17 @@ export class NativeSubagentRuntime { * In the RFD form, the running patch only reports that the child runs: its * prompt is a guess (the latest SendMessage to the agent), and the * `task_started` or the SendMessage result carries the delegation. + * + * In the AIR form, an agent that the runtime does not know is restored from + * the session's history (see {@link restoreAgent}). */ async taskResumed(taskId: string, deliver: Publish, prompt?: string): Promise { if (!this.enabled) return; const previous = this.children.get(taskId); - if (!previous) return; + if (!previous) { + if (this.form === "air") await this.restoreAgent(taskId, deliver, prompt); + return; + } const finishing = this.taskFinishPromises.get(taskId) ?? previous.terminalPromise; if (finishing) await finishing.catch(() => {}); if (this.form === "rfd") { @@ -508,6 +558,7 @@ export class NativeSubagentRuntime { this.identityByToolUse.clear(); this.controlByToolUse.clear(); this.parentByToolUse.clear(); + this.forgetRestores(); } if (errors.length === 1) throw errors[0]; if (errors.length > 1) throw new AggregateError(errors, "Failed to finish native subagents"); @@ -662,6 +713,115 @@ export class NativeSubagentRuntime { return this.children.get(agentId); } + /** + * The AIR form: whether no call of this process made `launchId`. A new Agent + * call has a control update before its `task_started`, and a launch that was + * restored or looked up is registered or remembered. + */ + private isUnseenLaunch(launchId: string): boolean { + return ( + !!this.launchedAgent && !this.unresolvedLaunches.has(launchId) && this.isUnseenCall(launchId) + ); + } + + /** Whether this runtime has no record of the tool call `toolUseId`. */ + private isUnseenCall(toolUseId: string): boolean { + return ( + !this.taskByToolUse.has(toolUseId) && + !this.childByParentToolUse.has(toolUseId) && + !this.controlByToolUse.has(toolUseId) && + !this.parentByToolUse.has(toolUseId) && + !this.identityByToolUse.has(toolUseId) + ); + } + + /** + * The AIR form: restores the agent that the call `launchId` launched, once. + * Concurrent callers wait for the same lookup. Without an agent, the updates + * buffered under `launchId` stay buffered and `launchId` is not looked up again. + */ + private restoreLaunch(launchId: string, deliver: Publish): Promise { + const existing = this.launchRestores.get(launchId); + if (existing) return existing; + const epoch = this.restoreEpoch; + const restoring = (async () => { + const agentId = await this.launchedAgent?.(launchId).catch((error: unknown) => { + this.logger.log( + `Session ${this.rootSessionId}: failed to read the history of launch ${launchId}: ${error}`, + ); + return undefined; + }); + if (epoch !== this.restoreEpoch) return; + if (agentId) await this.restoreAgent(agentId, deliver); + if (epoch !== this.restoreEpoch) return; + const child = this.childByParentToolUse.get(launchId); + if (!child) { + this.unresolvedLaunches.add(launchId); + return; + } + // Updates that arrived while the generation was announced. + if (child.announced) for (const pending of this.takePending(launchId)) await deliver(pending); + })().finally(() => { + if (this.launchRestores.get(launchId) === restoring) this.launchRestores.delete(launchId); + }); + this.launchRestores.set(launchId, restoring); + return restoring; + } + + /** + * The AIR form: opens a generation for the agent `agentId` that the runtime + * does not know, because it ran before the session was loaded or resumed. + * The session's subagent history (see {@link SubagentLineage}) gives the + * call that launched it: Claude Code sends the resumed run's messages under + * that call's id, so it is the generation's parent tool use, and the updates + * buffered under it are delivered. The parent is the agent that made the call + * if it is live, else the root. An agent that the history cannot place is + * not restored. Concurrent signals for one agent open one generation. + */ + private restoreAgent(agentId: string, deliver: Publish, prompt?: string): Promise { + if (!this.lineage || this.children.has(agentId) || this.unrestorableAgents.has(agentId)) { + return Promise.resolve(); + } + const existing = this.agentRestores.get(agentId); + if (existing) return existing; + const lineage = this.lineage; + const epoch = this.restoreEpoch; + const restoring = (async () => { + const entries = await lineage(agentId).catch((error: unknown) => { + this.logger.log( + `Session ${this.rootSessionId}: failed to read the history of subagent ${agentId}: ${error}`, + ); + return []; + }); + if (epoch !== this.restoreEpoch || this.children.has(agentId)) return; + const entry = entries.at(-1); + if (!entry || entry.agentId !== agentId) { + this.unrestorableAgents.add(agentId); + return; + } + const parent = entry.parentAgentId ? this.children.get(entry.parentAgentId) : undefined; + await this.openGeneration( + agentId, + undefined, + { + parentSessionId: + parent && this.isLiveSession(parent.sessionId) ? parent.sessionId : this.rootSessionId, + parentToolUseId: entry.launchId, + name: subagentDisplayName(entry.name, entry.description, entry.agentType, agentId), + task: subagentDescription(undefined, entry.description), + ...promptField(promptText(prompt)), + }, + true, + deliver, + true, + ); + })().finally(() => { + if (this.agentRestores.get(agentId) === restoring) this.agentRestores.delete(agentId); + }); + this.agentRestores.set(agentId, restoring); + return restoring; + } + /** * The RFD form: the child runs again, and keeps its session. `toolUseId` is * the call that delegated, usually a SendMessage: the SDK runs the resumed @@ -862,6 +1022,16 @@ export class NativeSubagentRuntime { this.generationByTaskId.clear(); this.pending.clear(); this.pendingCount = 0; + this.forgetRestores(); + } + + /** Lets a later resume signal restore an agent again, and look up a launch again. */ + private forgetRestores(): void { + this.agentRestores.clear(); + this.launchRestores.clear(); + this.unresolvedLaunches.clear(); + this.unrestorableAgents.clear(); + this.restoreEpoch++; } private takePending(parentToolUseId: string): AcpSessionNotification[] { @@ -932,9 +1102,10 @@ export class NativeSubagentRuntime { >, announce: boolean, deliver: Publish, + restored = false, ): Promise { const child: NativeSubagent = { - sessionId: this.nextChildSessionId(taskId, previous), + sessionId: this.nextChildSessionId(taskId, previous, restored), ...fields, }; const toolUseId = child.parentToolUseId; @@ -958,8 +1129,25 @@ export class NativeSubagentRuntime { * `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. + * + * An agent restored from the history (`restored`) is not on its first + * generation: before the restart the client already stored ids such as + * `taskId` and `taskId:generation:2`. Its generation is the current time in + * milliseconds, so it is `taskId:generation:1760000000000`, and each later + * resume in this process adds one. */ - private nextChildSessionId(taskId: string, previous: NativeSubagent | undefined): string { + private nextChildSessionId( + taskId: string, + previous: NativeSubagent | undefined, + restored = false, + ): string { + if (restored) { + // A clear() and a new restore within one millisecond must not repeat an id. + const generation = Math.max(Date.now(), this.lastRestoredGeneration + 1); + this.lastRestoredGeneration = generation; + this.generationByTaskId.set(taskId, generation); + return `${taskId}:generation:${generation}`; + } if (!previous) { this.generationByTaskId.set(taskId, 1); return taskId; diff --git a/src/subagent-history.ts b/src/subagent-history.ts index 56aee841..759b1dde 100644 --- a/src/subagent-history.ts +++ b/src/subagent-history.ts @@ -1,11 +1,12 @@ import { getSessionInfo, - getSessionMessages, getSubagentMessages, type SessionMessage, } from "@anthropic-ai/claude-agent-sdk"; +import { createReadStream } from "node:fs"; import { readdir, readFile } from "node:fs/promises"; import path from "node:path"; +import readline from "node:readline"; import { isNativeSubagentControlTool, type HistoricalSubagent } from "./native-subagents.js"; import { findTranscript } from "./resumed-session.js"; @@ -48,7 +49,7 @@ export async function subagentHistory(sessionId: string): Promise(); const transcript = await findTranscript(sessionId); if (!transcript) return { agents }; - const directory = path.join(path.dirname(transcript), sessionId, "subagents"); + const directory = subagentsDirectory(transcript, sessionId); let names: string[]; try { names = await readdir(directory); @@ -86,6 +87,22 @@ export async function subagentHistory(sessionId: string): Promise { + return (await subagentHistory(sessionId)).agents.get(launchId)?.agentId; +} + /** * The agent `agentId` and its ancestors, root first, from the subagent history * of the session `sessionId`. The runtime restores an agent from it when Claude @@ -109,15 +126,11 @@ export async function subagentLineage( } if (!launchOf.has(agentId)) return []; - // Without the project directory, as the replay reads it (`readResumedSession`). - const rootLaunches = new Set( - launchIdsIn( - await getSessionMessages(sessionId).catch((error: unknown) => { - logger.error(`Failed to read the transcript of session ${sessionId}:`, error); - return []; - }), - ), - ); + // The SDK's readers return only the active chain, which starts after the + // last compaction, so the launch calls come from the transcript files. + const transcript = await findTranscript(sessionId); + if (!transcript) return []; + const rootLaunches = await readLaunchIdsFromTranscript(transcript, null, logger); const agentLaunches = new Map>(); /** The agent whose transcript has the launch `launchId`, `null` for the root's. */ const launcherOf = async (launchId: string): Promise => { @@ -125,8 +138,10 @@ export async function subagentLineage( for (const candidate of launchOf.keys()) { let launches = agentLaunches.get(candidate); if (!launches) { - launches = new Set( - launchIdsIn(await readSubagentMessages(sessionId, candidate, history, logger)), + launches = await readLaunchIdsFromTranscript( + path.join(subagentsDirectory(transcript, sessionId), `agent-${candidate}.jsonl`), + candidate, + logger, ); agentLaunches.set(candidate, launches); } @@ -170,10 +185,52 @@ export function subagentLaunchIds(content: unknown): string[] { ); } -function launchIdsIn(messages: readonly SessionMessage[]): string[] { - return messages.flatMap((message) => - subagentLaunchIds((message as { message?: { content?: unknown } }).message?.content), - ); +/** + * The ids of the launch calls in the transcript `file`, every record of it, + * including those before a compaction. `owner` is the agent the transcript + * belongs to, null for the session's. The session's transcript can hold + * sidechain records, which are an agent's, not the root's; an agent's + * transcript keeps only the records of that agent. A file that cannot be read + * has no launches, so the agent is not placed under the root. + */ +async function readLaunchIdsFromTranscript( + file: string, + owner: string | null, + logger: { error(...args: unknown[]): void }, +): Promise> { + const launches = new Set(); + const stream = createReadStream(file, { encoding: "utf8" }); + try { + for await (const line of readline.createInterface({ input: stream, crlfDelay: Infinity })) { + if (!line.includes('"tool_use"')) continue; + let record: { + type?: unknown; + isSidechain?: unknown; + agentId?: unknown; + message?: { content?: unknown }; + } | null; + try { + record = JSON.parse(line); + } catch { + continue; + } + if (record?.type !== "assistant") continue; + if ( + owner === null + ? record.isSidechain === true + : record.agentId !== undefined && record.agentId !== owner + ) { + continue; + } + for (const id of subagentLaunchIds(record.message?.content)) launches.add(id); + } + } catch (error) { + logger.error(`Failed to read the transcript ${file}:`, error); + return new Set(); + } finally { + stream.destroy(); + } + return launches; } /** diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index b5c4e7d9..776845db 100644 --- a/src/tests/acp-agent.test.ts +++ b/src/tests/acp-agent.test.ts @@ -7506,22 +7506,8 @@ describe("subagent permission attribution (issue #851)", () => { ].join("\n") + "\n", ); // The SDK reads no transcript for a session id that is not a UUID, as - // the test session's is, so it answers with the transcript above. + // the test session's is, so it has no session info. vi.mocked(getSessionInfo).mockResolvedValueOnce({ cwd: "/tmp/proj" } as any); - vi.mocked(getSessionMessages).mockResolvedValueOnce([ - { - type: "assistant", - uuid: "a1", - session_id: "test-session", - parent_tool_use_id: null, - message: { - role: "assistant", - content: [ - { type: "tool_use", id: "toolu_old", name: "Agent", input: { prompt: "Go" } }, - ], - }, - }, - ] as Awaited>); await writeFile( path.join(directory, "test-session", "subagents", "agent-agent-7.meta.json"), JSON.stringify({ @@ -8400,6 +8386,167 @@ describe("subagent permission attribution (issue #851)", () => { }); }); + /** After an adapter restart the child's messages carry the old launch id; SendMessage resumes it. */ + describe("an agent from before an adapter restart, in the AIR form", () => { + /** + * Writes the history on disk, as the restarted adapter finds it: the root + * launched agent-7 with `toolu_old` before a compaction, which drops that + * call from the SDK's active chain. Runs `messages` through a new agent + * and returns the lifecycle and output the client saw. + */ + async function restarted(messages: unknown[]) { + const configDir = await mkdtemp(path.join(os.tmpdir(), "claude-acp-air-restore-")); + const originalConfigDir = process.env.CLAUDE_CONFIG_DIR; + process.env.CLAUDE_CONFIG_DIR = configDir; + try { + const directory = path.join(configDir, "projects", "-tmp-proj"); + await mkdir(path.join(directory, "test-session", "subagents"), { recursive: true }); + const record = (uuid: string, parentUuid: string | null, extra: object) => + JSON.stringify({ + sessionId: "test-session", + cwd: "/tmp/proj", + version: "2.0.0", + userType: "external", + isSidechain: false, + timestamp: "2026-10-13T00:00:00.000Z", + uuid, + parentUuid, + ...extra, + }); + await writeFile( + path.join(directory, "test-session.jsonl"), + [ + record("u1", null, { type: "user", message: { role: "user", content: "Go" } }), + record("a1", "u1", { + type: "assistant", + message: { + id: "m-a1", + role: "assistant", + model: "claude-opus-5", + content: [ + { type: "tool_use", id: "toolu_old", name: "Agent", input: { prompt: "Go" } }, + ], + }, + }), + record("b1", null, { + type: "system", + subtype: "compact_boundary", + content: "Conversation compacted", + }), + record("u2", "b1", { type: "user", message: { role: "user", content: "More" } }), + ].join("\n") + "\n", + ); + // The test session id is no UUID, so the SDK finds no session info. + vi.mocked(getSessionInfo).mockResolvedValue({ cwd: "/tmp/proj" } as any); + await writeFile( + path.join(directory, "test-session", "subagents", "agent-agent-7.meta.json"), + JSON.stringify({ + agentType: "Explore", + description: "Echo a word", + name: "echoer", + toolUseId: "toolu_old", + }), + ); + + 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: AIR_NATIVE_SUBAGENTS, + }); + injectGeneratorSession(agent, makeGenerator(messages)); + await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "go" }] }); + + return updates.flatMap(({ sessionId, update }) => + update.sessionUpdate === "subagent_spawned" + ? [[sessionId, update.sessionUpdate, update.subagentSessionId]] + : update.sessionUpdate === "agent_message_chunk" + ? [[sessionId, update.sessionUpdate, (update.content as { text: string }).text]] + : update.sessionUpdate === "subagent_state_update" + ? [[sessionId, update.sessionUpdate, update.subagentSessionId]] + : [], + ); + } finally { + vi.mocked(getSessionInfo).mockReset(); + if (originalConfigDir === undefined) delete process.env.CLAUDE_CONFIG_DIR; + else process.env.CLAUDE_CONFIG_DIR = originalConfigDir; + await rm(configDir, { recursive: true, force: true }); + } + } + const childMessage = (text: string) => ({ + type: "stream_event" as const, + parent_tool_use_id: "toolu_old", + uuid: randomUUID(), + session_id: "test-session", + event: { + type: "content_block_delta" as const, + index: 0, + delta: { type: "text_delta" as const, text }, + }, + }); + const resumed = { + type: "user", + parent_tool_use_id: null, + uuid: randomUUID(), + session_id: "test-session", + tool_use_result: { success: true, resumedAgentId: "agent-7" }, + message: { + role: "user", + content: [{ type: "tool_result", tool_use_id: "toolu_send", content: "Resuming" }], + }, + }; + const childId = expect.stringMatching(/^agent-7:generation:\d+$/); + + /** SendMessage resumes the agent; its launch is before a compaction; a repeated resume signal adds nothing. */ + it("restores it on a SendMessage resume and routes its run to one generation", async () => { + const seen = await restarted([ + rootToolUse("toolu_send", "SendMessage", { to: "agent-7", message: "Say PING" }), + { ...taskStarted("agent-7", "toolu_send"), subagent_type: "Explore", prompt: "Say PING" }, + childMessage("PING"), + resumed, + childMessage("PONG"), + { ...resumed, uuid: randomUUID() }, + { + type: "system", + subtype: "task_notification", + task_id: "agent-7", + tool_use_id: "toolu_send", + status: "completed", + output_file: "", + summary: "done", + uuid: randomUUID(), + session_id: "test-session", + }, + successResult(), + ]); + + expect(seen).toEqual([ + ["test-session", "subagent_spawned", childId], + [childId, "agent_message_chunk", "PING"], + [childId, "agent_message_chunk", "PONG"], + ["test-session", "subagent_state_update", childId], + ]); + }); + + /** Only a message under the old launch id arrives, with no resume signal. */ + it("restores it from a message under the old launch id alone", async () => { + const seen = await restarted([childMessage("PING"), childMessage("PONG"), successResult()]); + + expect(seen).toEqual([ + ["test-session", "subagent_spawned", childId], + [childId, "agent_message_chunk", "PING"], + [childId, "agent_message_chunk", "PONG"], + ]); + }); + }); + it("uses the immediate child as parent for a nested subagent", async () => { const updates: AcpSessionNotification[] = []; const agent = new ClaudeAcpAgent( diff --git a/src/tests/native-subagents.test.ts b/src/tests/native-subagents.test.ts index 49fd0adb..b9a562d1 100644 --- a/src/tests/native-subagents.test.ts +++ b/src/tests/native-subagents.test.ts @@ -1651,3 +1651,337 @@ describe("NativeSubagentRuntime in the RFD form", () => { ); }); }); + +describe("NativeSubagentRuntime in the AIR form, for an agent from before a load or resume", () => { + const worker: HistoricalSubagent = { + agentId: "worker-1", + launchId: "launch-1", + name: "worker", + agentType: "Explore", + description: "Inspect", + }; + const coordinator: HistoricalSubagent = { + agentId: "coordinator-1", + launchId: "launch-c", + name: "coordinator", + agentType: "general-purpose", + description: "Coordinate", + }; + const NOW = 1_760_000_000_000; + + const output = (parentToolUseId: string, text = "working"): AcpSessionNotification => + ({ + sessionId: "root", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text }, + _meta: { claudeCode: { parentToolUseId } }, + }, + }) as AcpSessionNotification; + const agentCall = (toolCallId: string): AcpSessionNotification => { + const launch = control("tool_call", "pending"); + return { ...launch, update: { ...launch.update, toolCallId } } as AcpSessionNotification; + }; + const spawned = (published: AcpSessionNotification[]) => + published.flatMap(({ sessionId, update }) => + update.sessionUpdate === "subagent_spawned" + ? [`${sessionId} -> ${update.subagentSessionId}`] + : [], + ); + + /** A runtime that knows no child, with the history's lineages and launches. */ + function reloaded( + lineages: Record, + { lookUpLaunches = true } = {}, + ) { + const published: AcpSessionNotification[] = []; + const log = vi.fn(); + const lineage = vi.fn(async (agentId: string) => lineages[agentId] ?? []); + const launchedAgent = vi.fn( + async (launchId: string) => + Object.values(lineages) + .flat() + .find((entry) => entry.launchId === launchId)?.agentId, + ); + const runtime = new NativeSubagentRuntime( + "air", + "root", + {}, + async (notification) => { + published.push(notification); + }, + { log }, + lineage, + lookUpLaunches ? launchedAgent : undefined, + ); + return { runtime, published, lineage, launchedAgent, log }; + } + + /** Runs `test` with the clock at {@link NOW}. */ + async function atNow(test: () => Promise) { + vi.useFakeTimers(); + vi.setSystemTime(NOW); + try { + await test(); + } finally { + vi.useRealTimers(); + } + } + + /** Output waits under the launch id; a resume signal restores the agent and a later resume numbers on from the first id. */ + it("restores the agent on a resume signal, delivers its buffered output, and numbers later generations from it", async () => { + await atNow(async () => { + const { runtime, published } = reloaded({ "worker-1": [worker] }, { lookUpLaunches: false }); + const delivered: AcpSessionNotification[] = []; + const deliver = async (n: AcpSessionNotification) => void delivered.push(n); + + await expect(runtime.route(output("launch-1", "early"), deliver)).resolves.toBeNull(); + await runtime.taskResumed("worker-1", deliver, "Say PING"); + + expect(spawned(published)).toEqual([`root -> worker-1:generation:${NOW}`]); + expect(published[0]?.update).toMatchObject({ + sessionUpdate: "subagent_spawned", + subagentSessionId: `worker-1:generation:${NOW}`, + name: "worker", + task: "Inspect", + prompt: "Say PING", + }); + expect(delivered).toHaveLength(1); + await expect(runtime.route(output("launch-1"), deliver)).resolves.toMatchObject({ + sessionId: `worker-1:generation:${NOW}`, + }); + + await runtime.finishTask("worker-1", "completed", deliver); + await runtime.taskResumed("worker-1", deliver); + expect(spawned(published)).toEqual([ + `root -> worker-1:generation:${NOW}`, + `root -> worker-1:generation:${NOW + 1}`, + ]); + }); + }); + + /** A message under a launch id that the history knows restores the agent with no resume signal. */ + it("restores the agent from a message under an unknown parent_tool_use_id alone, and routes it", async () => { + const { runtime, published, launchedAgent } = reloaded({ "worker-1": [worker] }); + const delivered: AcpSessionNotification[] = []; + const deliver = async (n: AcpSessionNotification) => void delivered.push(n); + + await expect(runtime.route(output("launch-1"), deliver)).resolves.toBeNull(); + + expect(launchedAgent).toHaveBeenCalledWith("launch-1"); + expect(spawned(published)).toHaveLength(1); + expect(delivered).toHaveLength(1); + await expect(runtime.route(output("launch-1"), deliver)).resolves.toMatchObject({ + sessionId: expect.stringMatching(/^worker-1:generation:\d+$/), + }); + }); + + /** An Agent call seen in this process waits for its task_started and is never looked up. */ + it("buffers a new Agent call of this process until its task_started, with no history lookup", async () => { + const { runtime, published, launchedAgent } = reloaded({}); + const delivered: AcpSessionNotification[] = []; + const deliver = async (n: AcpSessionNotification) => void delivered.push(n); + + await runtime.route(agentCall("launch-2"), deliver); + await expect(runtime.route(output("launch-2"), deliver)).resolves.toBeNull(); + expect(launchedAgent).not.toHaveBeenCalled(); + expect(published).toEqual([]); + + await runtime.taskStarted( + { taskId: "fresh-1", toolUseId: "launch-2", subagentType: "Explore" }, + deliver, + ); + expect(spawned(published)).toEqual(["root -> fresh-1"]); + expect(delivered).toHaveLength(1); + }); + + /** A launch id that resolves to no agent is looked up once. */ + it("looks up a launch id that the history does not know only once", async () => { + const { runtime, published, launchedAgent } = reloaded({}); + + await runtime.route(output("launch-9"), async () => {}); + await runtime.route(output("launch-9"), async () => {}); + + expect(launchedAgent).toHaveBeenCalledTimes(1); + expect(published).toEqual([]); + }); + + /** An agent launched by a live restored agent opens under it. */ + it("opens a nested agent under its parent while the parent is live", async () => { + const { runtime, published } = reloaded({ + "coordinator-1": [coordinator], + "worker-1": [coordinator, { ...worker, parentAgentId: "coordinator-1" }], + }); + await runtime.taskResumed("coordinator-1", async () => {}); + await runtime.taskResumed("worker-1", async () => {}); + + const parentId = (published[0]?.update as { subagentSessionId: string }).subagentSessionId; + expect(published[1]?.sessionId).toBe(parentId); + expect(spawned(published)).toEqual([ + `root -> ${parentId}`, + expect.stringMatching(/^coordinator-1:generation:\d+ -> worker-1:generation:\d+$/), + ]); + }); + + /** An agent whose parent is not live opens under the root. */ + it("opens a nested agent under the root when its parent is not live", async () => { + const { runtime, published } = reloaded({ + "worker-1": [coordinator, { ...worker, parentAgentId: "coordinator-1" }], + }); + await runtime.taskResumed("worker-1", async () => {}); + + expect(published[0]?.sessionId).toBe("root"); + }); + + /** An agent that the history cannot place is not announced. */ + it("opens nothing when the history cannot place the agent", async () => { + const { runtime, published, lineage } = reloaded({}); + + await runtime.taskResumed("worker-1", async () => {}); + + expect(lineage).toHaveBeenCalledWith("worker-1"); + expect(published).toEqual([]); + }); + + /** After clear(), a resume signal restores the agent again. */ + it("restores the agent again after the runtime was cleared", async () => { + const { runtime, published } = reloaded({ "worker-1": [worker] }); + await runtime.taskResumed("worker-1", async () => {}); + runtime.clear(); + await runtime.taskResumed("worker-1", async () => {}); + + const ids = spawned(published); + expect(ids).toHaveLength(2); + expect(new Set(ids).size).toBe(2); + }); + + /** After every child ended, a launch id that resolved to nothing is looked up again. */ + it("looks up an unknown launch again after the runtime ended every child", async () => { + const { runtime, launchedAgent } = reloaded({}); + await runtime.route(output("launch-9"), async () => {}); + await runtime.finishAll("cancelled", async () => {}); + await runtime.route(output("launch-9"), async () => {}); + + expect(launchedAgent).toHaveBeenCalledTimes(2); + }); + + /** Three signals for one agent in flight together open a single generation. */ + it("opens one generation for a message, a resumedAgentId and a task_updated that arrive together", async () => { + const { runtime, published, lineage } = reloaded({ "worker-1": [worker] }); + const delivered: AcpSessionNotification[] = []; + const deliver = async (n: AcpSessionNotification) => void delivered.push(n); + + await Promise.all([ + runtime.route(output("launch-1"), deliver), + runtime.sendMessageResumed("worker-1", deliver, "Say PING", "send-1"), + runtime.taskResumed("worker-1", deliver), + ]); + + expect(spawned(published)).toHaveLength(1); + expect(lineage).toHaveBeenCalledTimes(1); + expect(delivered).toHaveLength(1); + }); + + /** After a lineage that places no agent, every later message skips the history. */ + it("reads the history once for a launch whose agent cannot be placed", async () => { + const published: AcpSessionNotification[] = []; + const lineage = vi.fn(async () => []); + const launchedAgent = vi.fn(async () => "worker-1"); + const runtime = new NativeSubagentRuntime( + "air", + "root", + {}, + async (notification) => void published.push(notification), + { log: vi.fn() }, + lineage, + launchedAgent, + ); + + for (let i = 0; i < 5; i++) await runtime.route(output("launch-1"), async () => {}); + + expect(launchedAgent).toHaveBeenCalledTimes(1); + expect(lineage).toHaveBeenCalledTimes(1); + expect(published).toEqual([]); + }); + + /** A clear() while the lineage is read leaves the stale restore announcing nothing. */ + it("announces nothing for a restore that a clear() overtook", async () => { + const release = deferred(); + const published: AcpSessionNotification[] = []; + const runtime = new NativeSubagentRuntime( + "air", + "root", + {}, + async (notification) => void published.push(notification), + { log: vi.fn() }, + async () => { + await release.promise; + return [worker]; + }, + ); + + const resumed = runtime.taskResumed("worker-1", async () => {}); + runtime.clear(); + release.resolve(); + await resumed; + + expect(published).toEqual([]); + }); + + const sendMessage = (toolCallId: string): AcpSessionNotification => + ({ + sessionId: "root", + update: { + sessionUpdate: "tool_call", + toolCallId, + title: "SendMessage", + status: "pending", + _meta: { claudeCode: { toolName: "SendMessage" } }, + }, + }) as AcpSessionNotification; + + /** A SendMessage task_started restores the agent, and its id and the launch id both route. */ + it("restores the agent from a task_started under a SendMessage id", async () => { + const { runtime, published } = reloaded({ "worker-1": [worker] }, { lookUpLaunches: false }); + const deliver = async () => {}; + + await runtime.route(sendMessage("send-1"), deliver); + await runtime.taskStarted( + { taskId: "worker-1", toolUseId: "send-1", subagentType: "Explore", prompt: "Say PING" }, + deliver, + ); + await expect(runtime.route(output("launch-1"), deliver)).resolves.toMatchObject({ + sessionId: expect.stringMatching(/^worker-1:generation:\d+$/), + }); + await runtime.finishTask("worker-1", "completed", deliver, "send-1"); + + expect(spawned(published)).toHaveLength(1); + expect(published[0]?.update).toMatchObject({ prompt: "Say PING" }); + expect(published.at(-1)?.update.sessionUpdate).toBe("subagent_state_update"); + }); + + /** A task_started without a subagent type, such as a background shell, reads no history. */ + it("does not read the history for a task_started without a subagent type", async () => { + const { runtime, lineage } = reloaded({ "worker-1": [worker] }); + + await runtime.route(sendMessage("send-1"), async () => {}); + await runtime.taskStarted({ taskId: "shell-1", toolUseId: "send-1" }, async () => {}); + + expect(lineage).not.toHaveBeenCalled(); + }); + + /** A new Agent whose task_started precedes its tool call frame opens under its bare id and waits. */ + it("opens a new Agent whose task_started comes before its frame as the bare id, unannounced", async () => { + const { runtime, published, lineage } = reloaded({ "worker-1": [worker] }); + + await runtime.taskStarted( + { taskId: "worker-1", toolUseId: "launch-1", subagentType: "Explore" }, + async () => {}, + ); + + expect(lineage).not.toHaveBeenCalled(); + expect(published).toEqual([]); + await runtime.route(agentCall("launch-1"), async () => {}); + expect(spawned(published)).toEqual(["root -> worker-1"]); + }); +}); diff --git a/src/tests/subagent-history.test.ts b/src/tests/subagent-history.test.ts index 14e74893..273ee42e 100644 --- a/src/tests/subagent-history.test.ts +++ b/src/tests/subagent-history.test.ts @@ -70,7 +70,7 @@ describe("subagentLineage", () => { let parentUuid: string | null = null; return records .map((record) => { - const linked = { ...record, parentUuid }; + const linked = "parentUuid" in record ? record : { ...record, parentUuid }; parentUuid = (record as { uuid: string }).uuid; return JSON.stringify(linked); }) @@ -131,6 +131,66 @@ describe("subagentLineage", () => { ]); }); + const boundary = (uuid: string) => ({ + ...base, + type: "system", + subtype: "compact_boundary", + uuid, + parentUuid: null, + content: "Conversation compacted", + }); + + /** The root's launch comes before a compaction, which drops it from the SDK's active chain. */ + it("places an agent whose launch the root made before a compaction", async () => { + await writeSession( + [user("u1", "Go"), launch("a1", "launch-echoer"), boundary("b1"), user("u2", "More")], + { echoer1: { meta: meta("launch-echoer", "echoer"), records: [user("e1", "Work")] } }, + ); + + const lineage = await subagentLineage(sessionId, "echoer1", logger); + + expect(lineage.map(({ agentId, launchId }) => ({ agentId, launchId }))).toEqual([ + { agentId: "echoer1", launchId: "launch-echoer" }, + ]); + }); + + /** An agent's own launch of a nested agent comes before a compaction in that agent's transcript. */ + it("places a nested agent whose launch its parent agent made before a compaction", async () => { + await writeSession([user("u1", "Go"), launch("a1", "launch-coordinator")], { + coordinator1: { + meta: meta("launch-coordinator", "coordinator"), + records: [ + user("c1", "Work"), + launch("c2", "launch-worker"), + boundary("c3"), + user("c4", "More"), + ], + }, + worker1: { meta: meta("launch-worker", "worker"), records: [user("w1", "Work")] }, + }); + + const lineage = await subagentLineage(sessionId, "worker1", logger); + + expect( + lineage.map(({ agentId, launchId, parentAgentId }) => ({ agentId, launchId, parentAgentId })), + ).toEqual([ + { agentId: "coordinator1", launchId: "launch-coordinator", parentAgentId: undefined }, + { agentId: "worker1", launchId: "launch-worker", parentAgentId: "coordinator1" }, + ]); + }); + + /** A sidechain record in the root file is an agent's launch, so it does not make the root the parent. */ + it("does not take a launch in a sidechain record of the root file for the root's", async () => { + await writeSession( + [user("u1", "Go"), { ...launch("a1", "launch-worker"), isSidechain: true }], + { + worker1: { meta: meta("launch-worker", "worker"), records: [user("w1", "Work")] }, + }, + ); + + await expect(subagentLineage(sessionId, "worker1", logger)).resolves.toEqual([]); + }); + it("is empty for an agent whose launch no transcript has, since its parent is unknown", async () => { await writeSession([user("u1", "Go")], { lost1: { meta: meta("launch-lost", "lost"), records: [user("l1", "Work")] },