From 8021ad22ebddf27d7142437951973986c029e122 Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Fri, 21 Aug 2026 09:56:33 -0700 Subject: [PATCH] Bracket agent-initiated ACP turns so unprompted work renders (#2122) When an ACP agent streams work with no session/prompt in flight (OMP's async-job delivery), the bridge forwarded the updates without opening a turn, so the runtime assembler demoted each one to a hidden thread-scoped provider/unhandled row and the user saw nothing. The bridge now folds agent-initiated work into its open-turn mirror (activePromptKind: "turn" | "compaction" | "agent" | null). A work-kind update arriving idle opens a turn; a 5 s quiet window ends it; the next turn/start settles it first; thread/stop interrupts it; an agent exit fails it through the settling error. Permission requests inside an agent turn are handled like prompted ones instead of auto-cancelled. Co-Authored-By: Claude --- .../provider-acp/src/bridge/bridge.test.ts | 193 ++++++++++++++++++ plugins/provider-acp/src/bridge/bridge.ts | 114 ++++++++++- .../src/bridge/fake-acp-agent.mjs | 74 +++++++ plugins/provider-acp/src/visibility.ts | 13 +- 4 files changed, 388 insertions(+), 6 deletions(-) diff --git a/plugins/provider-acp/src/bridge/bridge.test.ts b/plugins/provider-acp/src/bridge/bridge.test.ts index c3ad0e461e..bfe4c0e42e 100644 --- a/plugins/provider-acp/src/bridge/bridge.test.ts +++ b/plugins/provider-acp/src/bridge/bridge.test.ts @@ -2045,6 +2045,199 @@ describe("acp bridge", () => { startedProviderThreadIds.pop(); }); + describe("agent-initiated turns (#2122)", () => { + /** + * Runs one prompted turn whose agent then streams unprompted work, and + * waits until that work has fully arrived (the closing chunk is on the + * wire) so each test observes the agent turn at a known point. + */ + async function promptThenAwaitAgentWork( + variant: string, + args?: StartThreadArgs, + ): Promise<{ bbThreadId: string; providerThreadId: string }> { + const thread = await startThread(args); + const turnId = sendTurnRequest("turn/start", thread.providerThreadId, { + input: [ + { type: "text", text: `agent-initiated${variant}`, mentions: [] }, + ], + }); + await waitForResponse(turnId); + await waitForTurnCompleted(); + await waitFor( + () => + agentMessageTexts().some((text) => text.includes("the answer is 42.")) + ? true + : undefined, + "agent-initiated work to arrive", + ); + return thread; + } + + it("brackets unprompted agent work as a turn and ends it when the agent goes quiet", async () => { + await promptThenAwaitAgentWork(""); + + // The work is a real turn with real items, not hidden raw-event rows. + expect(threadEventsOfType("turn/started")).toHaveLength(2); + expect(threadEventsOfType("provider/unhandled")).toHaveLength(0); + expect(agentMessageTexts().join("")).toContain( + "agent-initiated:job bg_4 finished, the answer is 42.", + ); + const toolItems = threadEventsOfType("item/started").filter( + (event) => + (event.item as { type: string }).type === "toolCall" || + (event.item as { type: string }).type === "commandExecution" || + (event.item as { type: string }).type === "fileRead", + ); + expect(toolItems.length).toBeGreaterThan(0); + // The echoed job result stays noise: one accepted input (the user's), + // no phantom user row. + expect( + emittedDeltaKinds().filter((kind) => kind === "input.accepted"), + ).toHaveLength(1); + + // No end-of-turn signal exists; the quiet window closes it as completed. + const completed = await waitFor(() => { + const events = threadEventsOfType("turn/completed"); + return events.length === 2 ? events[1] : undefined; + }, "agent turn to close after the quiet window"); + expect(completed).toMatchObject({ status: "completed" }); + }, 20_000); + + it("does not open a turn for unprompted non-work updates", async () => { + const { providerThreadId } = await startThread(); + const turnId = sendTurnRequest("turn/start", providerThreadId, { + input: [{ type: "text", text: "agent-initiated:noise", mentions: [] }], + }); + await waitForResponse(turnId); + await waitForTurnCompleted(); + await waitFor( + () => + emittedDeltaKinds().includes("contextWindow") ? true : undefined, + "idle usage_update to be processed", + ); + + expect(threadEventsOfType("turn/started")).toHaveLength(1); + }); + + it("settles the agent turn before the next user turn opens", async () => { + const { providerThreadId } = await promptThenAwaitAgentWork(""); + expect(threadEventsOfType("turn/completed")).toHaveLength(1); + + const nextId = sendTurnRequest("turn/start", providerThreadId, { + input: [{ type: "text", text: "hello there", mentions: [] }], + }); + const response = await waitForResponse(nextId); + expect(response.error).toBeUndefined(); + await waitFor( + () => + threadEventsOfType("turn/completed").length === 3 ? true : undefined, + "all three turns to settle", + ); + + expect(threadEventsOfType("turn/started")).toHaveLength(3); + expect( + threadEventsOfType("turn/completed").map((event) => event.status), + ).toEqual(["completed", "completed", "completed"]); + expect(agentMessageTexts().at(-1)).toBe("echo:hello there"); + }); + + it("interrupts the agent turn on thread/stop", async () => { + const { providerThreadId } = await promptThenAwaitAgentWork(""); + + const stopId = sendRequest("thread/stop", { + threadId: bbThreadIdFor(providerThreadId), + providerThreadId, + intent: "interrupt", + activeTurnId: null, + }); + const stopResponse = await waitForResponse(stopId); + expect(stopResponse.result).toEqual({ ok: true }); + + const completed = threadEventsOfType("turn/completed"); + expect(completed).toHaveLength(2); + expect(completed[1]).toMatchObject({ status: "interrupted" }); + startedProviderThreadIds.pop(); + }); + + it("fails the agent turn when the agent process exits mid-turn", async () => { + const { bbThreadId } = await promptThenAwaitAgentWork(":exit"); + + const errors = await waitFor(() => { + const errorNotifications = notifications("error"); + return errorNotifications.length > 0 ? errorNotifications : undefined; + }, "agent exit error notification"); + expect(errors).toHaveLength(1); + expect(errors[0]?.params).toMatchObject({ threadId: bbThreadId }); + + // The turn reaches a terminal state instead of hanging "working". + const completed = threadEventsOfType("turn/completed"); + expect(completed).toHaveLength(2); + expect(completed[1]).toMatchObject({ status: "failed" }); + startedProviderThreadIds.pop(); + }); + + it("auto-allows a permission request inside an agent turn in full mode", async () => { + await promptThenAwaitAgentWork(":permission", { permissionMode: "full" }); + + expect( + output.messages.filter( + (message) => message.method === "interaction/request", + ), + ).toHaveLength(0); + expect(agentMessageTexts().join("")).toContain("permission:yes "); + }); + + it("forwards a permission request inside an agent turn in ask mode", async () => { + const { bbThreadId, providerThreadId } = await startThread({ + permissionMode: "accept-edits", + permissionEscalation: "ask", + }); + const turnId = sendTurnRequest("turn/start", providerThreadId, { + input: [ + { type: "text", text: "agent-initiated:permission", mentions: [] }, + ], + }); + await waitForResponse(turnId); + await waitForTurnCompleted(); + + const forwarded = await waitFor( + () => + output.messages.find( + (message) => + message.method === "interaction/request" && + message.id !== undefined, + ), + "forwarded permission request", + ); + expect(forwarded.params).toMatchObject({ + threadId: bbThreadId, + providerThreadId, + payload: { + kind: "approval", + subject: expect.objectContaining({ command: "rm -rf build" }), + }, + }); + handleLine( + JSON.stringify({ + jsonrpc: "2.0", + id: forwarded.id, + result: { decision: "deny" }, + }), + ); + + await waitFor( + () => + agentMessageTexts().some((text) => text.includes("the answer is 42.")) + ? true + : undefined, + "agent-initiated work to finish after the decision", + ); + expect(agentMessageTexts().join("")).toContain("permission:no "); + // The whole exchange lives in the one agent turn. + expect(threadEventsOfType("turn/started")).toHaveLength(2); + }); + }); + it("forks an advertised ACP session with the target cwd and MCP servers", async () => { const forkLog = join(workspaceDir, "fork-params.json"); const forkId = sendRequest("thread/fork", { diff --git a/plugins/provider-acp/src/bridge/bridge.ts b/plugins/provider-acp/src/bridge/bridge.ts index 77af5cda82..7544996625 100644 --- a/plugins/provider-acp/src/bridge/bridge.ts +++ b/plugins/provider-acp/src/bridge/bridge.ts @@ -75,6 +75,7 @@ import { resolveAcpPermissionDecision, } from "../interactions.js"; import { acpProfileFromLaunchSpec, type AcpAgentProfile } from "../profiles.js"; +import { isAgentWorkAcpUpdateKind } from "../visibility.js"; import { buildAcpModelListParams, buildAcpSessionParams, @@ -183,10 +184,14 @@ interface AcpThreadSession { policy: AcpSessionPolicy; pendingInstructions: string | undefined; /** - * Which agent prompt is in flight for this bb turn: an ordinary `"turn"`, - * the provider-local `"compaction"` maintenance prompt, or none. + * The bridge's mirror of the turn it has open for this bb thread: an + * ordinary `"turn"` (a session/prompt is in flight), the provider-local + * `"compaction"` maintenance prompt, an `"agent"`-initiated turn the bridge + * bracketed around unprompted agent work, or none. */ - activePromptKind: "turn" | "compaction" | null; + activePromptKind: "turn" | "compaction" | "agent" | null; + /** Re-armed per work update; closes an `"agent"` turn once the agent is quiet. */ + agentTurnQuietTimer: ReturnType | undefined; queuedInputs: AcpPendingTurnInput[]; /** True while a session/prompt request is outstanding. */ promptRequestPending: boolean; @@ -215,6 +220,14 @@ let dynamicToolBridgePromise: Promise | null = null; // this timeout forces disposal. Stop remains a best-effort success boundary. const THREAD_STOP_CANCEL_TIMEOUT_MS = 4_000; +// ACP has no end-of-turn signal for agent-initiated work (no session/prompt +// result arrives), so a still-open `"agent"` turn closes once the agent has +// been quiet this long. Kept short: while the turn is open the thread reads as +// working and the server holds the last streamed message until the turn +// flushes. A long pause (a slow agent-side tool) splits the work into two +// turns, which is the cheaper failure. +const AGENT_TURN_QUIET_WINDOW_MS = 5_000; + // --------------------------------------------------------------------------- // stdout helpers (bridge → runtime) // --------------------------------------------------------------------------- @@ -327,7 +340,7 @@ function emitSessionError(session: AcpThreadSession, message: string): void { // prompt in flight the error stays a runtime notification — a settling // error delta on an idle thread would surface a diagnostic for a turn bb // never accepted. `activePromptKind` mirrors the turn the bridge itself - // opened with `turn.open`. + // opened with `turn.open`, an agent-initiated one included. if (session.activePromptKind !== null) { emitForSession(session, "error", { threadId: session.bbThreadId, @@ -1348,10 +1361,14 @@ function handlePermissionRequest( return; } + // A permission request belongs to a turn the user can see: a prompted one + // or an agent-initiated one. Outside both (idle, or the compaction prompt) + // there is no turn to present it against. if ( session.stopping || session.cancelRequested || - session.activePromptKind !== "turn" + (session.activePromptKind !== "turn" && + session.activePromptKind !== "agent") ) { responder.result({ outcome: { outcome: "cancelled" } }); return; @@ -1654,17 +1671,25 @@ async function startAgentSession( onExit: (info) => { const wasCurrent = sessionsByBbThreadId.get(bbThreadId) === session; cancelPendingPermissions(session); + clearAgentTurnQuietTimer(session); removeSession(session); if (!wasCurrent || session.stopping) { return; } void releaseCursorMcpApproval(session); + // Emitted while `activePromptKind` still names the open turn so the + // error settles it. A prompted turn clears its own mirror when the + // rejected prompt unwinds; an agent-initiated turn has no prompt, so it + // is cleared here. emitSessionError( session, `ACP agent "${agentLabel}" exited unexpectedly` + `${info.code !== null ? ` (code ${info.code})` : ""}` + `${info.stderrTail ? `: ${info.stderrTail}` : ""}`, ); + if (session.activePromptKind === "agent") { + session.activePromptKind = null; + } }, }); session = { @@ -1680,6 +1705,7 @@ async function startAgentSession( }, pendingInstructions: params.instructions, activePromptKind: null, + agentTurnQuietTimer: undefined, queuedInputs: [], promptRequestPending: false, cancelRequested: false, @@ -1872,6 +1898,9 @@ async function stopSession(session: AcpThreadSession): Promise { "ACP session stopped before the steer was sent", ); cancelPendingPermissions(session); + // An interrupt stops agent-initiated work too. There is no prompt to + // cancel, so the turn settles as interrupted and the agent is reaped. + settleAgentTurn(session, "cancelled"); if (session.activePromptKind !== null && !session.connection.exited) { session.connection.notify("session/cancel", { @@ -1908,6 +1937,12 @@ async function releaseSession(session: AcpThreadSession): Promise { "ACP session released before the steer was sent", ); cancelPendingPermissions(session); + // Like a released prompt turn, a released agent turn detaches without a + // fabricated terminal state. + clearAgentTurnQuietTimer(session); + if (session.activePromptKind === "agent") { + session.activePromptKind = null; + } session.connection.kill(); removeSession(session); await releaseCursorMcpApproval(session); @@ -2142,6 +2177,64 @@ function startCompaction( }); } +// --------------------------------------------------------------------------- +// Agent-initiated turns +// --------------------------------------------------------------------------- + +/** + * Work the agent streams with no prompt in flight (OMP delivering an async + * job's result, for one) is a turn bb never asked for. ACP sends no bracket + * for it, so the bridge opens one itself — the sanctioned shape for + * provider-internal activity (provider-bridge-protocol.md, turn lifecycle + * rule 3) — and owns every exit path: a quiet window ends it, the next + * bb-initiated turn settles a still-open one first, `thread/stop` interrupts + * it, and an agent exit fails it through `emitSessionError`. Without the + * bracket the assembler demotes each update to a hidden thread-scoped + * `provider/unhandled` row and the user sees nothing (#2122). + */ +function openAgentTurn(session: AcpThreadSession): void { + session.activePromptKind = "agent"; + emitForSession(session, ACP_TURN_STARTED_METHOD, { + threadId: session.bbThreadId, + }); +} + +function clearAgentTurnQuietTimer(session: AcpThreadSession): void { + if (session.agentTurnQuietTimer !== undefined) { + clearTimeout(session.agentTurnQuietTimer); + session.agentTurnQuietTimer = undefined; + } +} + +function settleAgentTurn( + session: AcpThreadSession, + stopReason: z.infer, +): void { + if (session.activePromptKind !== "agent") { + return; + } + clearAgentTurnQuietTimer(session); + session.activePromptKind = null; + emitForSession(session, ACP_TURN_COMPLETED_METHOD, { + threadId: session.bbThreadId, + stopReason, + }); +} + +function armAgentTurnQuietTimer(session: AcpThreadSession): void { + clearAgentTurnQuietTimer(session); + session.agentTurnQuietTimer = setTimeout(() => { + session.agentTurnQuietTimer = undefined; + // A permission the user has not answered is not quiet: the agent is + // waiting on bb, and its answer belongs to this turn. + if (session.pendingPermissions.size > 0) { + armAgentTurnQuietTimer(session); + return; + } + settleAgentTurn(session, "end_turn"); + }, AGENT_TURN_QUIET_WINDOW_MS); +} + // --------------------------------------------------------------------------- // Agent inbound traffic // --------------------------------------------------------------------------- @@ -2200,6 +2293,14 @@ function handleAgentNotification( ) { return; } + if (isAgentWorkAcpUpdateKind(parsed.data.update.sessionUpdate)) { + if (session.activePromptKind === null) { + openAgentTurn(session); + } + if (session.activePromptKind === "agent") { + armAgentTurnQuietTimer(session); + } + } emitForSession(session, ACP_UPDATE_METHOD, { threadId: session.bbThreadId, update: parsed.data.update, @@ -2517,6 +2618,9 @@ async function handleRequest( sendError(request.id, -32000, "No active ACP session"); return; } + // User input ends agent-initiated work: that turn settles before the + // requested one opens, so each reaches exactly one terminal state. + settleAgentTurn(session, "end_turn"); if (session.activePromptKind !== null) { sendError(request.id, -32000, "A turn is already active"); return; diff --git a/plugins/provider-acp/src/bridge/fake-acp-agent.mjs b/plugins/provider-acp/src/bridge/fake-acp-agent.mjs index 70c7a40691..dbf357ddc5 100755 --- a/plugins/provider-acp/src/bridge/fake-acp-agent.mjs +++ b/plugins/provider-acp/src/bridge/fake-acp-agent.mjs @@ -247,6 +247,70 @@ function captureMcpServers(message) { : []; } +async function streamAgentInitiatedWork(variant) { + if (variant === "noise") { + notifyUpdate({ sessionUpdate: "available_commands_update", commands: [] }); + notifyUpdate({ + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "[bg_4 finished] exit 0" }, + }); + // A usage_update last: a positive, observable sign the idle traffic was + // processed even though none of it may open a turn. + notifyUpdate({ sessionUpdate: "usage_update", used: 1_000, size: 128_000 }); + return; + } + notifyUpdate({ + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "[bg_4 finished] exit 0" }, + }); + notifyUpdate(messageChunk("agent-initiated:job bg_4 finished, ")); + notifyUpdate({ + sessionUpdate: "tool_call", + toolCallId: "agent-initiated-tool-1", + title: "cat result.txt", + kind: "read", + status: "pending", + rawInput: { path: "result.txt" }, + }); + await sleep(30); + notifyUpdate({ + sessionUpdate: "tool_call_update", + toolCallId: "agent-initiated-tool-1", + status: "completed", + content: [{ type: "content", content: { type: "text", text: "42" } }], + }); + if (variant === "permission") { + let outcome = "cancelled"; + try { + const result = await requestClient("session/request_permission", { + sessionId: activeSessionId, + toolCall: { + toolCallId: "agent-initiated-tool-2", + title: "Run rm", + kind: "execute", + rawInput: { command: "rm -rf build" }, + }, + options: [ + { optionId: "yes", name: "Allow", kind: "allow_once" }, + { optionId: "no", name: "Deny", kind: "reject_once" }, + ], + }); + outcome = + result?.outcome?.outcome === "selected" + ? result.outcome.optionId + : "cancelled"; + } catch { + outcome = "error"; + } + notifyUpdate(messageChunk(`permission:${outcome} `)); + } + notifyUpdate(messageChunk("the answer is 42.")); + if (variant === "exit") { + await sleep(30); + process.exit(3); + } +} + async function handlePrompt(message) { activePromptId = message.id; const text = promptText(message.params?.prompt); @@ -355,6 +419,16 @@ async function handlePrompt(message) { } catch { notifyUpdate(messageChunk("write:denied")); } + } else if (text.includes("agent-initiated")) { + // OMP async-job delivery shape: once this prompt's result has gone out, + // the agent streams work with no session/prompt driving it — an echoed + // user_message_chunk (the injected job result), agent text, a tool call + // that completes, and a closing chunk. Variants ride the prompt text: + // agent-initiated:permission ask for permission mid-stream + // agent-initiated:exit exit(3) right after the stream + // agent-initiated:noise only non-work updates (must not open a turn) + const variant = text.match(/agent-initiated:(\w+)/)?.[1] ?? ""; + setTimeout(() => void streamAgentInitiatedWork(variant), 40); } else if (text.includes("hang")) { // Stay pending until the client sends session/cancel. return; diff --git a/plugins/provider-acp/src/visibility.ts b/plugins/provider-acp/src/visibility.ts index d6903766cc..054d33b4f5 100644 --- a/plugins/provider-acp/src/visibility.ts +++ b/plugins/provider-acp/src/visibility.ts @@ -23,15 +23,26 @@ const NORMALIZED_ACP_METHODS = new Set([ ACP_WARNING_METHOD, ]); -const NORMALIZED_ACP_UPDATE_KINDS = new Set([ +// Update kinds that carry agent work: streamed text, thoughts, tool calls, +// and plans. Arriving with no prompt in flight they are an agent-initiated +// turn (e.g. OMP's async-job delivery), which the bridge brackets itself. +const AGENT_WORK_ACP_UPDATE_KINDS = new Set([ "agent_message_chunk", "agent_thought_chunk", "tool_call", "tool_call_update", "plan", +]); + +const NORMALIZED_ACP_UPDATE_KINDS = new Set([ + ...AGENT_WORK_ACP_UPDATE_KINDS, "usage_update", ]); +export function isAgentWorkAcpUpdateKind(updateKind: string): boolean { + return AGENT_WORK_ACP_UPDATE_KINDS.has(updateKind); +} + // Update kinds the agent may legitimately send but BB intentionally does not // render: replayed history, agent-side mode/command/config/session metadata. const NOISE_ACP_UPDATE_KINDS = new Set([