diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 9ab7092b..10dfd99e 100644 --- a/src/acp-agent.ts +++ b/src/acp-agent.ts @@ -2715,6 +2715,11 @@ export class ClaudeAcpAgent { * express: see {@link ToolCallClientCapabilities.v2}. */ private readonly v2: boolean; + /** + * The subagent replay of each session's last load, until + * {@link publishCurrentSubagentStates} sends its children's current states. + */ + private readonly replayedSubagents = new Map(); /** See the `foreground` constructor option. */ private readonly foregroundEvents?: (sessionId: string) => ForegroundEvents; /** @@ -9680,6 +9685,7 @@ export class ClaudeAcpAgent { for (const message of messages) await replayMessage(message); if (rfdReplay && !pending?.stopped) { await rfdReplay.finish(clientSupportsNotices(this.clientCapabilities), sessionId); + this.replayedSubagents.set(sessionId, rfdReplay); } if (nativeReplayEnabled && !pending?.stopped) { @@ -10695,16 +10701,41 @@ export class ClaudeAcpAgent { /** * The work that follows a session setup response: the session's commands, - * and MCP server authentication. A v2 client gets the commands in the setup - * response itself (`src/v2/agent.ts`), so it gets no update with them here. + * the current states of the subagents a load replayed, and MCP server + * authentication. A v2 client gets the commands in the setup response itself, + * and the subagent states from the v2 surface once it has sent its own + * response (`src/v2/agent.ts`), so it gets neither here. */ private afterSetupResponse(sessionId: string, mcpServers: McpServer[]): void { setTimeout(() => { - if (!this.v2) this.sendAvailableCommandsUpdate(sessionId); + if (!this.v2) { + this.sendAvailableCommandsUpdate(sessionId); + void this.publishCurrentSubagentStates(sessionId); + } startMcpAuthentication(this, sessionId, mcpServers); }, 0); } + /** + * Sends the current state of each subagent that the last load of + * `sessionId` replayed (see {@link RfdSubagentReplay.currentStates}). Run it + * only after the load response: before it, the client treats subagent states + * as history. + */ + async publishCurrentSubagentStates(sessionId: string): Promise { + const replay = this.replayedSubagents.get(sessionId); + if (!replay) return; + this.replayedSubagents.delete(sessionId); + const runtime = this.sessions[sessionId]?.nativeSubagentRuntime; + try { + for (const notification of replay.currentStates((child) => runtime?.liveWorkState(child))) { + await this.client.sessionUpdate(asSdkSessionNotification(notification)); + } + } catch (error) { + this.logger.error(`Session ${sessionId}: failed to send current subagent states:`, error); + } + } + private async updateConfigOption( sessionId: string, configId: string, diff --git a/src/native-subagents.ts b/src/native-subagents.ts index a4b56d0d..342a0e7e 100644 --- a/src/native-subagents.ts +++ b/src/native-subagents.ts @@ -227,6 +227,16 @@ export class NativeSubagentRuntime { return routed; } + /** + * The RFD form: the work state that the client holds for the announced child + * `sessionId` of this live session, or undefined when no such child runs here. + */ + liveWorkState(sessionId: string): WorkState["state"] | undefined { + if (this.form !== "rfd") return undefined; + const child = this.children.get(sessionId); + return child?.announced ? child.workState : undefined; + } + /** Report the model of a child response, not the parent's model or a launch-time alias. */ async reportModel( parentToolUseId: string | null, diff --git a/src/subagent-replay.ts b/src/subagent-replay.ts index 22539abd..1fe08bf0 100644 --- a/src/subagent-replay.ts +++ b/src/subagent-replay.ts @@ -6,7 +6,8 @@ * `subagentHistory`). The replay announces each child right after that call, * under the session id it had live, the SDK agent id, and sends the child's * history to the child session. Every child's end goes out before the replay - * ends, since no subagent from before the reload still runs. + * ends. These states are history to the client, so each child's current + * state goes out again after the load response (`currentStates`). * * As live, the parent is not shown the Agent or Task call of a child that the * replay announces: the launch's prompt is the parent's outgoing message, and @@ -247,19 +248,19 @@ export class RfdSubagentReplay { } /** - * Sends the end of every announced child, and a notice when subagent history - * is missing. The end is the state of the child's last task notification, - * else `cancelled` for a child the user stopped, else the result of its - * foreground launch, else `idle` with no stop reason. + * Sends the end of every announced child (see `endState`), and a notice when + * subagent history is missing. */ async finish(supportsNotices: boolean, rootSessionId: string): Promise { for (const child of [...this.children.values()].reverse()) { if (!child.announced) continue; - const end = child.notified ?? (child.record.stoppedByUser ? "cancelled" : child.launchEnd); - const state: WorkState = end ? rfdEndState(end, undefined) : { state: "idle" }; await this.publish({ sessionId: child.parentSessionId, - update: { sessionUpdate: "subagent_update", sessionId: child.sessionId, state }, + update: { + sessionUpdate: "subagent_update", + sessionId: child.sessionId, + state: endState(child), + }, }); } const missing = this.omitted.size + this.withoutHistory; @@ -278,6 +279,38 @@ export class RfdSubagentReplay { } } + /** + * The current state of each announced child, to send after the load + * response. The RFD makes that response the freshness boundary: a client + * treats the replayed states before it as history, so they show no current + * activity and allow no cancel. + * + * `live` gives the work state of a child that still runs in this process, as + * when a live session is loaded again. Such a child reports that state and + * can be cancelled, as live. Any other child cannot be running, because the + * process that ran it is gone, so its replayed end is its current state. + */ + currentStates( + live: (sessionId: string) => WorkState["state"] | undefined, + ): AcpSessionNotification[] { + return [...this.children.values()] + .filter((child) => child.announced) + .map((child) => { + const liveState = live(child.sessionId); + const state: WorkState = + liveState === undefined || liveState === "idle" ? endState(child) : { state: liveState }; + return { + sessionId: child.parentSessionId, + update: { + sessionUpdate: "subagent_update", + sessionId: child.sessionId, + ...(liveState !== undefined ? { capabilities: { cancel: {} } } : {}), + state, + }, + }; + }); + } + private registerLaunch( toolUseId: string, input: Record, @@ -376,6 +409,16 @@ export function replaySubagentTerminalState( : "failed"; } +/** + * How `child`'s work ended: the state of its last task notification, else + * `cancelled` for a child the user stopped, else the result of its foreground + * launch, else `idle` with no stop reason. + */ +function endState(child: ReplayChild): WorkState { + const end = child.notified ?? (child.record.stoppedByUser ? "cancelled" : child.launchEnd); + return end ? rfdEndState(end, undefined) : { state: "idle" }; +} + function replayContentText(value: unknown, seen = new Set()): string { if (typeof value === "string") return value; if (typeof value !== "object" || value === null || seen.has(value)) return ""; diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index b3250ec7..4d5bc618 100644 --- a/src/tests/acp-agent.test.ts +++ b/src/tests/acp-agent.test.ts @@ -3808,6 +3808,7 @@ describe("subagent transcript replay", () => { root: object[], children: Record, capabilities: ClientCapabilities = { subagents: {}, session: { notices: {} } }, + afterResponse?: { live: (childSessionId: string) => string | undefined }, ): Promise { const configDir = await mkdtemp(path.join(os.tmpdir(), "claude-acp-rfd-replay-")); const originalConfigDir = process.env.CLAUDE_CONFIG_DIR; @@ -3848,12 +3849,25 @@ describe("subagent transcript replay", () => { await ( agent as unknown as { replaySessionHistory(sessionId: string): Promise } ).replaySessionHistory(sessionId); - return updates.flatMap(({ sessionId: target, update }) => { + const responseAt = updates.length; + if (afterResponse) { + // Stands in for the live session's runtime, which knows its running children. + (agent.sessions as Record)[sessionId] = { + nativeSubagentRuntime: { liveWorkState: afterResponse.live }, + }; + await agent.publishCurrentSubagentStates(sessionId); + } + return updates.flatMap(({ sessionId: target, update }, index) => { const on = target === sessionId ? "root" : target; const u = update as any; + const response = afterResponse && index === responseAt ? ["-- response --"] : []; switch (update.sessionUpdate) { case "subagent_update": - return [`${on} subagent_update ${u.sessionId} ${JSON.stringify(u.state)}`]; + return [ + ...response, + `${on} subagent_update ${u.sessionId} ${JSON.stringify(u.state)}` + + (u.capabilities ? ` capabilities=${JSON.stringify(u.capabilities)}` : ""), + ]; case "session_message": { const references = [ ...(u.senderMessageId ? [`senderMessageId=${u.senderMessageId}`] : []), @@ -4057,6 +4071,35 @@ describe("subagent transcript replay", () => { ]); }); + it("sends each child's current state after the load response", async () => { + const launch = (id: string, name: string) => + toolUse(id, "Agent", { name, subagent_type: "Explore", prompt: `Run ${name}` }); + const child = (name: string, launchId: string) => ({ + meta: { agentType: "Explore", name, toolUseId: launchId, requestShape: "foreground" }, + records: [user(`${name}1`, `Run ${name}`)], + }); + const trace = await replayRfd( + [ + user("u1", "Go"), + assistant("a1", [launch("toolu_live", "live"), launch("toolu_gone", "gone")]), + user("u2", [toolResult("toolu_live", "Live."), toolResult("toolu_gone", "Gone.")]), + ], + { aaa111: child("live", "toolu_live"), bbb222: child("gone", "toolu_gone") }, + undefined, + // The session is loaded again in the process that still runs `aaa111`. + { live: (childSessionId) => (childSessionId === "aaa111" ? "running" : undefined) }, + ); + + expect(trace.slice(trace.indexOf("-- response --"))).toEqual([ + "-- response --", + // A child that still runs reports its live state, and can be cancelled. + 'root subagent_update aaa111 {"state":"running"} capabilities={"cancel":{}}', + // Any other child's process is gone: its replayed end is its current + // state, and nothing can cancel it. + 'root subagent_update bbb222 {"state":"idle","stopReason":"end_turn"}', + ]); + }); + it("replays a child's hand-back as its message to the parent, instead of the launch's result", async () => { const trace = await replayRfd( [ diff --git a/src/tests/acp-v2.test.ts b/src/tests/acp-v2.test.ts index e5ec5493..a80aeb88 100644 --- a/src/tests/acp-v2.test.ts +++ b/src/tests/acp-v2.test.ts @@ -1116,14 +1116,27 @@ describe("ACP v2 prompts", () => { cwd, replayFrom: { type: "start" }, }); + client.sessionUpdates.push({ + sessionId: "response", + update: { sessionUpdate: "response" }, + } as never); + await vi.waitFor(() => + expect( + client.sessionUpdates.filter( + ({ sessionId, update }) => + sessionId === "abc123" && update.sessionUpdate === "state_update", + ), + ).toHaveLength(2), + ); await client.authUpdate(1); }); const child = client.sessionUpdates.flatMap(({ sessionId, update }) => - sessionId === "abc123" ? [update] : [], + sessionId === "abc123" || sessionId === "response" ? [update] : [], ); // The child's past `running` does not replay; its last state goes once, - // after the replay. + // after the replay. That state is history to the client, so the child's + // current state follows the response: its process is gone, so it is idle. expect(child).toEqual([ expect.objectContaining({ sessionUpdate: "session_message", content: text("Look around") }), // The client may still hold the child's message, so the replay clears it. @@ -1134,7 +1147,16 @@ describe("ACP v2 prompts", () => { content: text("Child report.")[0], }), { sessionUpdate: "state_update", state: "idle", stopReason: "end_turn" }, + { sessionUpdate: "response" }, + { sessionUpdate: "state_update", state: "idle", stopReason: "end_turn" }, ]); + // The current state on the parent grants nothing: the child cannot run again here. + expect( + client.sessionUpdates.filter( + ({ update }) => + update.sessionUpdate === "subagent_update" && (update as any).capabilities != null, + ), + ).toEqual([]); }); it("ends a resume with replay with one end event for a client with loadReplayEnd", async () => { diff --git a/src/v2/agent.ts b/src/v2/agent.ts index 9d1101be..d4b09a1b 100644 --- a/src/v2/agent.ts +++ b/src/v2/agent.ts @@ -95,10 +95,11 @@ export function v2AgentApp( sessionId: params.sessionId, update: stateSnapshot(agent.foregroundState(params.sessionId)), }); - return { - ...v2ResumeSessionResponse(response), - ...(await setupCommands(agent, params.sessionId, logger)), - }; + const commands = await setupCommands(agent, params.sessionId, logger); + // The response ends a replay. The subagents RFD treats the replayed + // subagent states before it as history, so their current states follow it. + setTimeout(() => void agent.publishCurrentSubagentStates(params.sessionId), 0); + return { ...v2ResumeSessionResponse(response), ...commands }; }) .onRequest(v2.methods.agent.session.fork, ({ params }) => agent.forking(params.sessionId, async () => {