mirrors/opencode
1
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-11-15 12:41:37 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(acp): stream the parent's reaction to background subagent reports (#63395)

Browse Source
This commit is contained in:
James Long
2026-11-11 15:04:39 -06:00
committed by GitHub
Unverified
parent 0bd8cee7c2
commit 3039a8b5e9
5 changed files with 192 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/cli/src/acp/translate.ts
+41 -27
View File
@@ -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<string, Tool>
readonly retries: ReadonlyMap<string, RetryStatus>
readonly compactions: ACPCompaction.Tracked
readonly children: ReadonlyMap<string, ACPChild.Session>
readonly children: ReadonlyMap<string, Child>
readonly openChildren: ReadonlySet<string>
readonly asks: ReadonlySet<string>
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[] {
packages/cli/src/acp/turn.ts
+44 -18
View File
@@ -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<ACPTranslate.TurnState>
readonly subscription: Subscription
readonly cancelled: Deferred.Deferred<void>
readonly background: boolean
}
type Background = {
readonly state: Ref.Ref<ACPTranslate.TurnState>
readonly fiber?: Fiber.Fiber<unknown, unknown>
}
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<string, PromptResponse, ACPError.Failure>()
const backgrounds = new WeakMap<Attached, Background>()
const open = Effect.fnUntraced(function* (ctx: ACPTranslate.TurnContext, state: Ref.Ref<ACPTranslate.TurnState>) {
const open = Effect.fnUntraced(function* (
attached: Attached,
ctx: ACPTranslate.TurnContext,
state: Ref.Ref<ACPTranslate.TurnState>,
) {
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<void>(), background: false } satisfies Turn
return { ctx, state, subscription, cancelled: yield* Deferred.make<void>() } 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<void>(),
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<void>()
// 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,
packages/cli/test/acp/background-turn.subprocess.test.ts
+48
View File
@@ -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<string>()
const resumed = Promise.withResolvers<void>()
await using fixture = await createAcpFixture({
respond(request) {
const last = JSON.stringify(modelRequest(request).messages.at(-1))
if (last.includes("<subagent") && last.includes("child result")) {
resumed.resolve()
return "parent reaction"
}
if (last.includes("run the child"))
return toolCallStream("call_child", "subagent", {
agent: "general",
description: "Background task",
prompt: "work in child session",
})
if (last.includes("work in child session")) return child.promise
return "launched"
},
})
const acp = fixture.spawn()
await initialize(acp)
const { sessionId } = await newSession(acp, fixture.home)
const response = expectOk(
await acp.request<PromptResponse>("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)
packages/cli/test/acp/subprocess.ts
+2
View File
@@ -37,6 +37,7 @@ type AcpProcess = {
readonly pid: number
readonly exited: Promise<number>
readonly request: <T>(method: string, params?: unknown) => Promise<JsonRpcResponse<T>>
readonly waitForMessage: (predicate: Waiter["predicate"], description: string) => Promise<JsonRpcMessage>
readonly close: () => Promise<number>
readonly stderr: () => string
readonly [Symbol.asyncDispose]: () => Promise<void>
@@ -294,6 +295,7 @@ function spawnAcp(input: { readonly env: Record<string, string | undefined> }):
pid: child.pid,
exited: child.exited,
request,
waitForMessage: (predicate, description) => take(predicate, 10_000, description),
async close() {
if (!inputClosed) {
inputClosed = true
packages/cli/test/acp/turn.test.ts
+57
View File
@@ -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 }) => [