From 3039a8b5e971c07401f3fb2ed8eb8d5c2c5dc99e Mon Sep 17 00:00:00 2001 From: James Long Date: Wed, 11 Nov 2026 15:04:39 -0600 Subject: [PATCH] fix(acp): stream the parent's reaction to background subagent reports (#63395) --- packages/cli/src/acp/translate.ts | 68 +++++++++++-------- packages/cli/src/acp/turn.ts | 62 ++++++++++++----- .../acp/background-turn.subprocess.test.ts | 48 +++++++++++++ packages/cli/test/acp/subprocess.ts | 2 + packages/cli/test/acp/turn.test.ts | 57 ++++++++++++++++ 5 files changed, 192 insertions(+), 45 deletions(-) create mode 100644 packages/cli/test/acp/background-turn.subprocess.test.ts diff --git a/packages/cli/src/acp/translate.ts b/packages/cli/src/acp/translate.ts index c6960b4cc29..f85be8cc561 100644 --- a/packages/cli/src/acp/translate.ts +++ b/packages/cli/src/acp/translate.ts @@ -25,7 +25,7 @@ export type Terminal = "succeeded" | "failed" | "interrupted" export type TurnContext = { readonly sessionID: Session.ID readonly cwd: string - readonly start: TurnStart + readonly start?: TurnStart readonly childUpdates: boolean readonly compaction: boolean } @@ -44,12 +44,14 @@ type RetryStatus = { readonly error: SessionError.Error } +type Child = ACPChild.Session & { readonly turnID?: string } + export type TurnState = { readonly started: boolean readonly tools: ReadonlyMap readonly retries: ReadonlyMap readonly compactions: ACPCompaction.Tracked - readonly children: ReadonlyMap + readonly children: ReadonlyMap readonly openChildren: ReadonlySet readonly asks: ReadonlySet readonly finish?: SessionMessage.Assistant["finish"] @@ -69,9 +71,9 @@ export type Output = readonly _tag: "PermissionAsk" readonly event: PermissionEvent readonly tool?: Tool - readonly child?: ACPChild.Session + readonly child?: Child } - | { readonly _tag: "FormAsk"; readonly form: FormEvent["data"]["form"]; readonly child?: ACPChild.Session } + | { readonly _tag: "FormAsk"; readonly form: FormEvent["data"]["form"]; readonly child?: Child } | { readonly _tag: "AskSettled"; readonly id: string } export type Folded = { @@ -95,20 +97,21 @@ export function fold(state: TurnState, event: OpenCodeEvent, ctx: TurnContext): const sessionID = sessionIDFromEvent(event) const child = sessionID ? state.children.get(sessionID) : undefined + const openChild = child && state.openChildren.has(child.id) ? child : undefined - if (event.type === "permission.asked" && (event.data.sessionID === ctx.sessionID || child)) { + if (event.type === "permission.asked" && (event.data.sessionID === ctx.sessionID || openChild)) { const tool = event.data.source?.id ? state.tools.get(toolKey(event.data.sessionID, event.data.source.id)) : undefined return { state: { ...state, asks: new Set(state.asks).add(event.data.id) }, - outputs: [{ _tag: "PermissionAsk", event, tool, child }], + outputs: [{ _tag: "PermissionAsk", event, tool, child: openChild }], } } - if (event.type === "form.created" && (event.data.form.sessionID === ctx.sessionID || child)) { + if (event.type === "form.created" && (event.data.form.sessionID === ctx.sessionID || openChild)) { return { state: { ...state, asks: new Set(state.asks).add(event.data.form.id) }, - outputs: [{ _tag: "FormAsk", form: event.data.form, child }], + outputs: [{ _tag: "FormAsk", form: event.data.form, child: openChild }], } } const settledID = @@ -123,15 +126,18 @@ export function fold(state: TurnState, event: OpenCodeEvent, ctx: TurnContext): return { state: { ...state, asks }, outputs: [{ _tag: "AskSettled", id: settledID }] } } if (!sessionID || (sessionID !== ctx.sessionID && !child)) return { state, outputs: [] } - if (event.type === "session.inbox.delivered" && event.data.inboxID === ctx.start.id) + if ( + sessionID === ctx.sessionID && + event.type === "session.inbox.delivered" && + (!ctx.start || event.data.inboxID === ctx.start.id) + ) return { state: { ...state, started: true }, outputs: [] } - if (!state.started) return { state, outputs: [] } + if (!child && !state.started) return { state, outputs: [] } return child ? childEvent(state, event, ctx, child) : rootEvent(state, event, ctx) } -export function fromTrackedChild(state: TurnState, event: OpenCodeEvent) { - const sessionID = event.type === "session.created" ? event.data.parentID : sessionIDFromEvent(event) - return sessionID !== undefined && state.children.has(sessionID) +export function projectChild(ctx: TurnContext, child: Child) { + return !ctx.childUpdates && ctx.start !== undefined && child.turnID === ctx.start.id } export function failure(state: TurnState) { @@ -202,7 +208,13 @@ function childCreated(state: TurnState, event: CreatedEvent, ctx: TurnContext): if (!parentID) return { state, outputs: [] } const parent = parentID === ctx.sessionID ? undefined : state.children.get(parentID) if (!parent && parentID !== ctx.sessionID) return { state, outputs: [] } - const child = { id: event.data.sessionID, parentID, depth: parent ? parent.depth + 1 : 1, title: event.data.title } + const child: Child = { + id: event.data.sessionID, + parentID, + depth: parent ? parent.depth + 1 : 1, + title: event.data.title, + turnID: parent ? parent.turnID : ctx.start?.id, + } return { state: { ...state, @@ -224,20 +236,27 @@ function rootEvent(state: TurnState, event: OpenCodeEvent, ctx: TurnContext): Fo return { state: { ...recorded, stepError: event.data.error }, outputs: [] } } case "session.execution.succeeded": - return { state, outputs: [], terminal: "succeeded" } + return { state: { ...state, started: Boolean(ctx.start) }, outputs: [], terminal: "succeeded" } case "session.execution.interrupted": - return { state, outputs: [], terminal: "interrupted" } + return { state: { ...state, started: Boolean(ctx.start) }, outputs: [], terminal: "interrupted" } case "session.execution.failed": - return { state: { ...state, executionError: event.data.error }, outputs: [], terminal: "failed" } + return { + state: { ...state, started: Boolean(ctx.start), executionError: event.data.error }, + outputs: [], + terminal: "failed", + } default: return sessionEvent(state, event, ctx, undefined) } } -function childEvent(state: TurnState, event: OpenCodeEvent, ctx: TurnContext, child: ACPChild.Session): Folded { +function childEvent(state: TurnState, event: OpenCodeEvent, ctx: TurnContext, child: Child): Folded { switch (event.type) { case "session.execution.started": - return { state, outputs: childStatus(ctx, child, { type: "status", status: "running" }) } + return { + state: { ...state, openChildren: new Set(state.openChildren).add(child.id) }, + outputs: childStatus(ctx, child, { type: "status", status: "running" }), + } case "session.execution.succeeded": return childEnded(state, ctx, child, { type: "status", status: "completed" }) case "session.execution.interrupted": @@ -249,12 +268,7 @@ function childEvent(state: TurnState, event: OpenCodeEvent, ctx: TurnContext, ch } } -function sessionEvent( - state: TurnState, - event: OpenCodeEvent, - ctx: TurnContext, - child: ACPChild.Session | undefined, -): Folded { +function sessionEvent(state: TurnState, event: OpenCodeEvent, ctx: TurnContext, child: Child | undefined): Folded { const sessionID = child?.id ?? ctx.sessionID const send = (update: SessionUpdate) => route(ctx, child, update) switch (event.type) { @@ -415,11 +429,11 @@ function newTool(sessionID: string, id: string, name = "tool"): Tool { return { sessionID, id, name, input: {}, metadata: {} } } -function route(ctx: TurnContext, child: ACPChild.Session | undefined, update: SessionUpdate): Output[] { +function route(ctx: TurnContext, child: Child | undefined, update: SessionUpdate): Output[] { if (!child) return [{ _tag: "SessionUpdate", update }] const projected = ACPChild.project(update, child) if (ctx.childUpdates) return childStatus(ctx, child, { type: "update", update: projected }) - return [{ _tag: "SessionUpdate", update: projected }] + return projectChild(ctx, child) ? [{ _tag: "SessionUpdate", update: projected }] : [] } function childStatus(ctx: TurnContext, child: ACPChild.Session, event: ACPChild.Event): Output[] { diff --git a/packages/cli/src/acp/turn.ts b/packages/cli/src/acp/turn.ts index f43d82eb226..7e5637b584a 100644 --- a/packages/cli/src/acp/turn.ts +++ b/packages/cli/src/acp/turn.ts @@ -21,7 +21,7 @@ import { findModel, type ACPCatalog } from "./catalog" import { ACPChild } from "./child" import { ACPClient } from "./client" import { currentModel } from "./config-option" -import type { ACPConnection } from "./connection" +import { ACPConnection } from "./connection" import { ACPElicitation } from "./elicitation" import { ACPError } from "./error" import { ACPPermission } from "./permission" @@ -60,7 +60,11 @@ type Turn = { readonly state: Ref.Ref readonly subscription: Subscription readonly cancelled: Deferred.Deferred - readonly background: boolean +} + +type Background = { + readonly state: Ref.Ref + readonly fiber?: Fiber.Fiber } export const make = Effect.fnUntraced(function* (input: { @@ -73,8 +77,24 @@ export const make = Effect.fnUntraced(function* (input: { const scope = yield* Effect.scope const drainTimeout = yield* CancelDrainTimeout const turns = yield* FiberMap.make() + const backgrounds = new WeakMap() - const open = Effect.fnUntraced(function* (ctx: ACPTranslate.TurnContext, state: Ref.Ref) { + const open = Effect.fnUntraced(function* ( + attached: Attached, + ctx: ACPTranslate.TurnContext, + state: Ref.Ref, + ) { + const previous = backgrounds.get(attached) + if (previous) { + backgrounds.delete(attached) + if (previous.fiber) yield* Fiber.interrupt(previous.fiber) + const inherited = yield* Ref.get(previous.state) + yield* Ref.set(state, { + ...ACPTranslate.initial, + children: inherited.children, + openChildren: inherited.openChildren, + }) + } // Parented to the service scope; the session scope may already be closed. const subscriptionScope = yield* Scope.fork(scope) const subscription: Subscription = { @@ -93,7 +113,7 @@ export const make = Effect.fnUntraced(function* (input: { Effect.forever, Effect.forkIn(subscriptionScope), ) - return { ctx, state, subscription, cancelled: yield* Deferred.make(), background: false } satisfies Turn + return { ctx, state, subscription, cancelled: yield* Deferred.make() } satisfies Turn }) const take = (subscription: Subscription) => @@ -141,7 +161,6 @@ export const make = Effect.fnUntraced(function* (input: { const interpret = (turn: Turn, output: ACPTranslate.Output) => { switch (output._tag) { case "SessionUpdate": - if (turn.background) return Effect.void return ACPPermission.withCompletedDiffs(output.update, output.diff, turn.ctx.cwd).pipe( Effect.flatMap((update) => input.connection.sessionUpdate({ sessionId: turn.ctx.sessionID, update })), ) @@ -185,7 +204,7 @@ export const make = Effect.fnUntraced(function* (input: { requestedSchema, clientSessionID: turn.ctx.sessionID, child: output.child, - toolCallSent: !turn.background && (!output.child || !turn.ctx.childUpdates), + toolCallSent: !output.child || ACPTranslate.projectChild(turn.ctx, output.child), } yield* queueAsk(turn, output.form.id, ACPElicitation.ask(elicitation), "cancel", (outcome) => ACPElicitation.respond(elicitation, outcome), @@ -203,10 +222,7 @@ export const make = Effect.fnUntraced(function* (input: { const advance = Effect.fnUntraced(function* (turn: Turn) { const event = yield* take(turn.subscription) const folded = yield* Ref.modify(turn.state, (current) => { - const next = - turn.background && !ACPTranslate.fromTrackedChild(current, event) - ? { state: current, outputs: [] } - : ACPTranslate.fold(current, event, turn.ctx) + const next = ACPTranslate.fold(current, event, turn.ctx) return [next, next.state] }) yield* Effect.forEach(folded.outputs, (output) => interpret(turn, output), { discard: true }) @@ -226,7 +242,7 @@ export const make = Effect.fnUntraced(function* (input: { const followChildren = Effect.fnUntraced(function* (turn: Turn) { while (true) { const folded = yield* advance(turn) - if (folded.state.openChildren.size === 0) return yield* asksSettled(turn.subscription) + if (folded.terminal && folded.state.openChildren.size === 0) return yield* asksSettled(turn.subscription) } }) @@ -312,13 +328,22 @@ export const make = Effect.fnUntraced(function* (input: { const close = Scope.close(turn.subscription.scope, Exit.void) if (Exit.isFailure(exit) && !Cause.hasInterruptsOnly(exit.cause)) return yield* close const state = yield* Ref.get(turn.state) - if (state.openChildren.size === 0) return yield* close - // Children outlive a cancelled turn, so their asks still reach the client. - const background = followChildren({ - ...turn, - state: yield* Ref.make(state), - cancelled: yield* Deferred.make(), - background: true, + if (state.openChildren.size === 0) { + if (state.children.size > 0) backgrounds.set(attached, { state: turn.state }) + return yield* close + } + const responded = yield* ACPConnection.Responded + const nextState = yield* Ref.make({ ...state, started: false }) + const cancelled = yield* Deferred.make() + // Children outlive a cancelled turn, so their asks and report-triggered turns still reach the client. + const background = Effect.withFiber((fiber) => { + backgrounds.set(attached, { state: nextState, fiber }) + return Effect.exit(responded).pipe( + Effect.andThen( + followChildren({ ...turn, ctx: { ...turn.ctx, start: undefined }, state: nextState, cancelled }), + ), + Effect.interruptible, + ) }).pipe(Effect.ignore, Effect.ensuring(close), Effect.withSpan("cli.acp.turn.background")) yield* input.sessions.fork(attached, background).pipe(Effect.catchTag("ACPSessionNotFoundError", () => close)) }) @@ -368,6 +393,7 @@ export const make = Effect.fnUntraced(function* (input: { const state = yield* Ref.make(ACPTranslate.initial) const exit = yield* Effect.acquireUseRelease( open( + attached, { sessionID: attached.id, cwd: attached.cwd, diff --git a/packages/cli/test/acp/background-turn.subprocess.test.ts b/packages/cli/test/acp/background-turn.subprocess.test.ts new file mode 100644 index 00000000000..5894004c437 --- /dev/null +++ b/packages/cli/test/acp/background-turn.subprocess.test.ts @@ -0,0 +1,48 @@ +import type { PromptResponse } from "@agentclientprotocol/sdk" +import { expect, test } from "bun:test" +import { Schema } from "effect" +import { createAcpFixture, expectOk, initialize, newSession, toolCallStream } from "./subprocess" + +const modelRequest = Schema.decodeUnknownSync(Schema.Struct({ messages: Schema.Array(Schema.Unknown) })) + +test("streams the parent's response to a subagent report after end_turn", async () => { + const child = Promise.withResolvers() + const resumed = Promise.withResolvers() + await using fixture = await createAcpFixture({ + respond(request) { + const last = JSON.stringify(modelRequest(request).messages.at(-1)) + if (last.includes("("session/prompt", { + sessionId, + prompt: [{ type: "text", text: "run the child" }], + }), + ) + expect(response.stopReason).toBe("end_turn") + child.resolve("child result") + await resumed.promise + const message = await acp.waitForMessage( + (message) => message.method === "session/update" && JSON.stringify(message.params).includes("parent reaction"), + "the resumed parent's agent_message_chunk", + ) + expect(message.params).toMatchObject({ + sessionId, + update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "parent reaction" } }, + }) +}, 60_000) diff --git a/packages/cli/test/acp/subprocess.ts b/packages/cli/test/acp/subprocess.ts index dee0ee22c6a..9ca913371f8 100644 --- a/packages/cli/test/acp/subprocess.ts +++ b/packages/cli/test/acp/subprocess.ts @@ -37,6 +37,7 @@ type AcpProcess = { readonly pid: number readonly exited: Promise readonly request: (method: string, params?: unknown) => Promise> + readonly waitForMessage: (predicate: Waiter["predicate"], description: string) => Promise readonly close: () => Promise readonly stderr: () => string readonly [Symbol.asyncDispose]: () => Promise @@ -294,6 +295,7 @@ function spawnAcp(input: { readonly env: Record }): pid: child.pid, exited: child.exited, request, + waitForMessage: (predicate, description) => take(predicate, 10_000, description), async close() { if (!inputClosed) { inputClosed = true diff --git a/packages/cli/test/acp/turn.test.ts b/packages/cli/test/acp/turn.test.ts index 54bfa5a8aa7..b56e3f3b494 100644 --- a/packages/cli/test/acp/turn.test.ts +++ b/packages/cli/test/acp/turn.test.ts @@ -136,6 +136,63 @@ describe("acp turn events over the wire", () => { }, ) + test("streams resumed parent turns after background children report across prompts", async () => { + await using acp = await startSession({ + permission: () => ({ outcome: { outcome: "selected", optionId: "once" } }), + onPrompt: ({ sessionID, id, text }) => { + if (text === "launch") + return turn( + sessionID, + id, + childCreated("ses_first", sessionID, "First"), + childCreated("ses_second", sessionID, "Second"), + ) + if (text === "followup") + return turn(sessionID, id, durableEvent("session.execution.started", { sessionID: "ses_first" })) + return turn( + sessionID, + id, + textDelta("ses_second", "msg_child_2", "older child during chat"), + textDelta(sessionID, "msg_chat", "chat reply"), + ) + }, + }) + const sessionID = acp.sessionId + + expect((await acp.prompt(sessionID, "launch")).stopReason).toBe("end_turn") + acp.server.send( + textDelta("ses_first", "msg_child", "child only"), + succeeded("ses_first"), + ...turn(sessionID, "msg_report_1", textDelta(sessionID, "msg_reaction_1", "first reaction")), + textDelta(sessionID, "msg_stale", "stale"), + ) + await acp.waitForUpdate(({ update }) => JSON.stringify(update).includes("first reaction")) + + expect((await acp.prompt(sessionID, "chat")).stopReason).toBe("end_turn") + acp.server.send( + permissionAsked("ses_second", "perm_second", { action: "read" }), + succeeded("ses_second"), + ...turn(sessionID, "msg_report_2", textDelta(sessionID, "msg_reaction_2", "second reaction")), + ) + await acp.waitForUpdate(({ update }) => JSON.stringify(update).includes("second reaction")) + await acp.until(() => acp.server.replies.length === 1, "second child permission reply") + + expect((await acp.prompt(sessionID, "followup")).stopReason).toBe("end_turn") + acp.server.send( + succeeded("ses_first"), + ...turn(sessionID, "msg_report_3", textDelta(sessionID, "msg_reaction_3", "followup reaction")), + ) + await acp.waitForUpdate(({ update }) => JSON.stringify(update).includes("followup reaction")) + + expect(acp.server.replies).toEqual([{ sessionID: "ses_second", requestID: "perm_second", decision: "once" }]) + expect(turnUpdates(acp.updates).map(({ update }) => update)).toMatchObject([ + { messageId: "msg_reaction_1", content: { text: "first reaction" } }, + { messageId: "msg_chat", content: { text: "chat reply" } }, + { messageId: "msg_reaction_2", content: { text: "second reaction" } }, + { messageId: "msg_reaction_3", content: { text: "followup reaction" } }, + ]) + }) + test("cancels unsupported session forms so execution can continue", async () => { await using acp = await startSession({ onPrompt: ({ sessionID, id }) => [