diff --git a/.changeset/acp-route-a-claude-cli-bridge.md b/.changeset/acp-route-a-claude-cli-bridge.md index 27a3f3efbe..b3f60068bb 100644 --- a/.changeset/acp-route-a-claude-cli-bridge.md +++ b/.changeset/acp-route-a-claude-cli-bridge.md @@ -7,5 +7,6 @@ Route Fusion's Claude CLI path through the ACP bridge (`claude-code-cli-acp`) in - **U10** — forward `mcpServers` on ACP `session/new` through the runtime contract (`AgentRuntimeOptions.mcpServers` + the plugin's `newAcpSession`); defaults to `[]` so existing read-only ACP "ask" turns are unchanged. - **U11** — `streamViaAcp`: the `pi-claude-cli` provider can drive Claude through the bundled ACP bridge, returning the same `AssistantMessageEventStream` as the `-p` path. Dispatched only when `FUSION_CLAUDE_ACP=1` and a bridge path are present, so the live `-p` path is byte-for-byte untouched by default. Full-history prompting, schema-only MCP forwarding with break-early on pi-known tools, control-char/size sanitization, env allow-list, process-registry registration, and inactivity timeout. - **KTD10** — the ACP runtime plugin publishes its identity-pinned bundled bridge path on load so the kill-switch needs no manual path; it does not enable the transport. +- **OQ2** — opt-in connection reuse (`FUSION_CLAUDE_ACP_REUSE=1`, default OFF): a warm bridge connection + ACP session is kept across turns of one conversation (keyed by `sessionId`), so multi-turn lanes skip the cold bridge/`claude` spawn and `session/new` round-trip and send only the latest-turn delta (`buildResumePrompt`). A stable `router` indirection serves each turn's handlers; a warm-child death routes failure to the current owner turn (no 30-min inactivity hang), eviction is cache-identity-aware (a concurrent cold turn can't kill a newer entry's child), an empty resume cold-starts instead of issuing an empty prompt, and a per-turn token drops cross-turn stray updates. The idle reaper is `unref`'d. Default OFF → the cold path is functionally unchanged. The Claude-via-pi OAuth path is unchanged. Live verification confirmed the bridge gates tool execution behind `session/request_permission` (forwarded MCP tools and native tools do not execute when cancelled). Remaining for a follow-up: picker/auth/status surface (U12), workflow `model`-node verification (U13), and production rollout. diff --git a/packages/pi-claude-cli/src/__tests__/acp-driver.test.ts b/packages/pi-claude-cli/src/__tests__/acp-driver.test.ts index 2dedde00e4..ee9e1a854d 100644 --- a/packages/pi-claude-cli/src/__tests__/acp-driver.test.ts +++ b/packages/pi-claude-cli/src/__tests__/acp-driver.test.ts @@ -5,6 +5,9 @@ import { PassThrough } from "node:stream"; // Synthetic ACP session/update sequence the mocked prompt() will replay. let scriptedUpdates: Array> = []; let scriptedUsage: Record | undefined; +// When set, prompt() never resolves — simulates a turn waiting on the bridge so +// only an out-of-band event (child death / abort) can end it. +let scriptedHang = false; // Driver validates the bridge path with existsSync — make the fake path "exist". // writeFileSync/unlinkSync back the R17 auth-failure signal (spied). @@ -33,6 +36,7 @@ vi.mock("@agentclientprotocol/sdk", () => ({ this.initialize = vi.fn(async () => ({ protocolVersion: 1 })); this.newSession = vi.fn(async () => ({ sessionId: "s1" })); this.prompt = vi.fn(async () => { + if (scriptedHang) return new Promise(() => {}); // never resolves for (const u of scriptedUpdates) await handler.sessionUpdate({ update: u }); return { stopReason: "end_turn", usage: scriptedUsage }; }); @@ -54,6 +58,8 @@ vi.mock("@earendil-works/pi-ai", () => ({ calculateCost: vi.fn(), })); +import { spawn } from "node:child_process"; +import { ClientSideConnection } from "@agentclientprotocol/sdk"; import { streamViaAcp, buildBridgeEnv } from "../acp-driver.js"; const MODEL = { id: "claude-sonnet-4-5", name: "Claude Sonnet 4.5" } as never; @@ -66,7 +72,7 @@ function eventsOf(stream: { _events: Array> }) { const flush = () => new Promise((r) => setTimeout(r, 30)); describe("streamViaAcp — ACP→pi translation (U11)", () => { - beforeEach(() => { scriptedUpdates = []; scriptedUsage = undefined; }); + beforeEach(() => { scriptedUpdates = []; scriptedUsage = undefined; scriptedHang = false; }); it("feeds ACP token usage (incl. cache tokens) into the done message (item 2)", async () => { scriptedUsage = { inputTokens: 11, outputTokens: 22, cachedReadTokens: 5, cachedWriteTokens: 3 }; @@ -194,6 +200,123 @@ describe("streamViaAcp — ACP→pi translation (U11)", () => { }); }); +describe("connection reuse (item 1) — gated by FUSION_CLAUDE_ACP_REUSE", () => { + const savedReuse = process.env.FUSION_CLAUDE_ACP_REUSE; + beforeEach(() => { + scriptedUpdates = []; + scriptedUsage = undefined; + scriptedHang = false; + vi.mocked(spawn).mockClear(); + vi.mocked(ClientSideConnection).mockClear(); + }); + afterEach(() => { + if (savedReuse === undefined) delete process.env.FUSION_CLAUDE_ACP_REUSE; + else process.env.FUSION_CLAUDE_ACP_REUSE = savedReuse; + }); + + it("reuses one warm bridge connection across turns; turn 2 skips spawn + session/new", async () => { + process.env.FUSION_CLAUDE_ACP_REUSE = "1"; + const reuseOpts = { ...OPTS, sessionId: "conv-reuse-1" }; + + // Turn 1 (cold): needs >1 message so reuseKey activates and the connection caches. + const ctx1 = { messages: [{ role: "user", content: "hi" }, { role: "assistant", content: "hello" }] } as never; + scriptedUpdates = [{ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "turn one" } }]; + streamViaAcp(MODEL, ctx1, reuseOpts); + await flush(); + expect(vi.mocked(spawn)).toHaveBeenCalledTimes(1); + expect(vi.mocked(ClientSideConnection)).toHaveBeenCalledTimes(1); + + // Turn 2 (warm): same sessionId → no new spawn, no new connection. + const ctx2 = { messages: [...(ctx1 as unknown as { messages: unknown[] }).messages, { role: "user", content: "again" }] } as never; + scriptedUpdates = [{ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "turn two" } }]; + const s2 = streamViaAcp(MODEL, ctx2, reuseOpts) as unknown as { _events: Array> }; + await flush(); + expect(vi.mocked(spawn)).toHaveBeenCalledTimes(1); // no second spawn + expect(vi.mocked(ClientSideConnection)).toHaveBeenCalledTimes(1); // no second connection + + // The single warm connection's prompt() ran once per turn. + const conn = vi.mocked(ClientSideConnection).mock.instances[0] as unknown as { prompt: ReturnType; newSession: ReturnType }; + expect(conn.prompt).toHaveBeenCalledTimes(2); + expect(conn.newSession).toHaveBeenCalledTimes(1); // session/new only on the cold turn + const done = s2._events.find((e) => e.type === "done"); + expect(done!.reason).toBe("stop"); + + // Cleanup: evict the warm connection + clear its (unref'd) idle timer. + (vi.mocked(spawn).mock.results[0].value as EventEmitter).emit("close", 0); + }); + + it("fails a reuse turn FAST when the warm child dies mid-prompt (P0: no 30min hang)", async () => { + process.env.FUSION_CLAUDE_ACP_REUSE = "1"; + const reuseOpts = { ...OPTS, sessionId: "conv-death" }; + + // Turn 1 (cold) caches the warm connection. + const ctx1 = { messages: [{ role: "user", content: "hi" }, { role: "assistant", content: "hello" }] } as never; + scriptedUpdates = [{ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "one" } }]; + streamViaAcp(MODEL, ctx1, reuseOpts); + await flush(); + const child = vi.mocked(spawn).mock.results[0].value as EventEmitter; + + // Turn 2 (warm) hangs on prompt() — only the child-death path can end it. + scriptedHang = true; + const ctx2 = { messages: [...(ctx1 as unknown as { messages: unknown[] }).messages, { role: "user", content: "again" }] } as never; + const s2 = streamViaAcp(MODEL, ctx2, reuseOpts) as unknown as { _events: Array> }; + await flush(); + expect(s2._events.some((e) => e.type === "done")).toBe(false); // still waiting + + // The warm child dies. The cold turn's close handler routes failure to the + // CURRENT (reuse) turn via router.fail, so it ends immediately. + child.emit("close", 1); + await flush(); + const done = s2._events.find((e) => e.type === "done") as { reason?: string; message?: { content?: Array<{ text?: string }> } }; + expect(done).toBeDefined(); + expect(done!.reason).toBe("stop"); + expect(JSON.stringify(done!.message?.content)).toContain("Error"); + + // Cache was evicted: a subsequent turn cold-spawns a fresh bridge. + scriptedHang = false; + scriptedUpdates = [{ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "fresh" } }]; + const ctx3 = { messages: [...(ctx2 as unknown as { messages: unknown[] }).messages, { role: "assistant", content: "" }, { role: "user", content: "q3" }] } as never; + streamViaAcp(MODEL, ctx3, reuseOpts); + await flush(); + expect(vi.mocked(spawn)).toHaveBeenCalledTimes(2); // turn 1 + the post-death cold restart + }); + + it("cold-starts (no warm reuse) when the resume delta is empty (P1: no empty-prompt hang)", async () => { + process.env.FUSION_CLAUDE_ACP_REUSE = "1"; + const reuseOpts = { ...OPTS, sessionId: "conv-empty" }; + + // Turn 1 (cold) caches. + const ctx1 = { messages: [{ role: "user", content: "hi" }, { role: "assistant", content: "hello" }] } as never; + scriptedUpdates = [{ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "one" } }]; + streamViaAcp(MODEL, ctx1, reuseOpts); + await flush(); + expect(vi.mocked(spawn)).toHaveBeenCalledTimes(1); + + // Turn 2 whose context ends in an assistant message → buildResumePrompt is + // empty → must NOT take the warm path (would hang); cold-starts instead. + const ctx2 = { messages: [...(ctx1 as unknown as { messages: unknown[] }).messages, { role: "user", content: "x" }, { role: "assistant", content: "y" }] } as never; + scriptedUpdates = [{ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "two" } }]; + const s2 = streamViaAcp(MODEL, ctx2, reuseOpts) as unknown as { _events: Array> }; + await flush(); + expect(vi.mocked(spawn)).toHaveBeenCalledTimes(2); // empty resume → fresh spawn + const done = s2._events.find((e) => e.type === "done"); + expect(done!.reason).toBe("stop"); // produced a normal turn, did not hang + }); + + it("does NOT reuse when the flag is off (default): each turn spawns a fresh bridge", async () => { + delete process.env.FUSION_CLAUDE_ACP_REUSE; + const reuseOpts = { ...OPTS, sessionId: "conv-off" }; + const ctx = { messages: [{ role: "user", content: "hi" }, { role: "assistant", content: "x" }] } as never; + scriptedUpdates = [{ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "a" } }]; + streamViaAcp(MODEL, ctx, reuseOpts); + await flush(); + streamViaAcp(MODEL, ctx, reuseOpts); + await flush(); + expect(vi.mocked(spawn)).toHaveBeenCalledTimes(2); + expect(vi.mocked(ClientSideConnection)).toHaveBeenCalledTimes(2); + }); +}); + describe("buildBridgeEnv — R17 auth opt-in (item 3)", () => { const saved = { flag: process.env.FUSION_CLAUDE_ACP_FORWARD_AUTH, diff --git a/packages/pi-claude-cli/src/acp-driver.ts b/packages/pi-claude-cli/src/acp-driver.ts index daca18941d..d034475dda 100644 --- a/packages/pi-claude-cli/src/acp-driver.ts +++ b/packages/pi-claude-cli/src/acp-driver.ts @@ -45,7 +45,7 @@ import { } from "@agentclientprotocol/sdk"; import { AssistantMessageEventStream } from "@earendil-works/pi-ai"; import type { Api, Model, SimpleStreamOptions } from "@earendil-works/pi-ai"; -import { buildPrompt, buildSystemPrompt, type PiContext } from "./prompt-builder.js"; +import { buildPrompt, buildResumePrompt, buildSystemPrompt, type PiContext } from "./prompt-builder.js"; import { createEventBridge } from "./event-bridge.js"; import { registerProcess, captureStderr } from "./process-manager.js"; import { isPiKnownClaudeTool } from "./tool-mapping.js"; @@ -190,6 +190,61 @@ function toAcpPromptBlocks( return out; } +/** + * FNXC:ClaudeAcp 2026-06-15-14:10: + * Connection-reuse cache (item 1 / OQ2). Gated behind `FUSION_CLAUDE_ACP_REUSE` + * (default OFF). When on, a live bridge connection + ACP session is kept warm + * across turns of one conversation (keyed by the stable `options.sessionId`), so + * multi-turn lanes skip the cold bridge+claude spawn, the `session/new` + * round-trip, AND the full-history resend — sending only `buildResumePrompt` + * (delta) on reuse. A stable `router` indirection lets the long-lived connection + * handler serve each turn's fresh per-call state. Default OFF → the cold path + * below is functionally unchanged. + */ +const REUSE_IDLE_MS = 5 * 60_000; +interface AcpRouter { + onUpdate: ((p: { update?: Record } & Record) => Promise) | null; + onPermission: ((p: Record) => Promise<{ outcome: { outcome: "cancelled" } }>) | null; + // Liveness: invoked when the warm child dies so the turn CURRENTLY owning the + // connection fails fast instead of hanging until the inactivity timeout. The + // long-lived `child.on("close")` is bound to the cold turn's closure, so + // without this a reuse turn's death would never reach its own `failWith`. + // Repointed to each turn's `failWith`; nulled on release (idle → just evict). + fail: ((msg: string) => void) | null; +} +interface CachedAcpConn { + conn: ClientSideConnection; + child: ChildProcess; + acpSessionId: string; + cwd: string; + inUse: boolean; + router: AcpRouter; + idleTimer?: ReturnType; + // Monotonic id of the turn currently owning the connection. A stray + // session/update from a finished turn is dropped when it no longer matches. + activeTurn: number; +} +const acpSessionCache = new Map(); +let acpTurnCounter = 0; +function acpReuseEnabled(): boolean { + return process.env.FUSION_CLAUDE_ACP_REUSE === "1"; +} +/** + * Kill a cached connection's child and evict it — but only delete the map key + * if it STILL points at this exact entry. A concurrent cold turn may have + * replaced the entry under the same key; a stale close handler / idle timer + * must not evict (or kill the child of) that newer, live entry. The passed + * entry's own child is always killed (it is the dead/finished one). + */ +function evictCachedAcpConn(key: string, entry: CachedAcpConn): void { + if (acpSessionCache.get(key) === entry) acpSessionCache.delete(key); + if (entry.idleTimer) { clearTimeout(entry.idleTimer); entry.idleTimer = undefined; } + entry.router.onUpdate = null; + entry.router.onPermission = null; + entry.router.fail = null; + try { entry.child.kill("SIGKILL"); } catch { /* registry SIGKILL is authoritative */ } +} + /** * Stream a Claude response via the ACP bridge as an `AssistantMessageEventStream`. * Mirrors `streamViaCli`'s contract (start → deltas → done; break-early on tools). @@ -205,6 +260,12 @@ export function streamViaAcp( const bridge = createEventBridge(stream, model); (async () => { + const cwd = options.cwd ?? process.cwd(); + const reuseKey = + acpReuseEnabled() && options.sessionId && context.messages.length > 1 + ? options.sessionId + : undefined; + let child: ChildProcess | undefined; let getStderr: (() => string) | undefined; let ended = false; @@ -214,11 +275,44 @@ export function streamViaAcp( let sawToolCall = false; let inactivity: ReturnType | undefined; let onAbort: (() => void) | undefined; + // The cache entry this turn is bound to (set on reuse, or after a cold turn + // caches its connection). Identity-checked against the map before release. + let cacheEntry: CachedAcpConn | undefined; + // This turn's monotonic id, stamped onto the shared cache entry when the + // turn acquires it. Handlers drop updates once the entry moves to a newer + // turn (defends the warm connection against cross-turn content bleed). + let myTurn = 0; - const cleanup = () => { + // End the turn. `destroy` kills+evicts the connection; otherwise a cached + // connection is released (kept warm for the next turn) and a one-shot + // (non-reuse) connection is killed. + const endTurn = (destroy: boolean) => { if (inactivity) { clearTimeout(inactivity); inactivity = undefined; } if (onAbort && options.signal) options.signal.removeEventListener("abort", onAbort); - try { child?.kill("SIGKILL"); } catch { /* registry SIGKILL is authoritative */ } + const entry = cacheEntry; + const keepWarm = + !destroy && entry !== undefined && reuseKey !== undefined && + acpSessionCache.get(reuseKey) === entry; + if (keepWarm) { + // Release the warm connection: drop this turn's handlers (so a late + // update can't reach a finished turn or the liveness hook fire stale), + // mark idle, and arm an unref'd reaper bound to THIS entry. + entry!.router.onUpdate = null; + entry!.router.onPermission = null; + entry!.router.fail = null; + entry!.inUse = false; + if (entry!.idleTimer) clearTimeout(entry!.idleTimer); + const idle = setTimeout(() => evictCachedAcpConn(reuseKey!, entry!), REUSE_IDLE_MS); + idle.unref?.(); // a warm-connection idle timer must not keep the process alive + entry!.idleTimer = idle; + return; + } + if (entry !== undefined && reuseKey !== undefined) { + // Kills this turn's child; evicts the map key only if still current. + evictCachedAcpConn(reuseKey, entry); + } else { + try { child?.kill("SIGKILL"); } catch { /* registry SIGKILL is authoritative */ } + } }; const armInactivity = () => { if (inactivity) clearTimeout(inactivity); @@ -251,7 +345,7 @@ export function streamViaAcp( bridge.handleEvent({ type: "message_delta", delta: { stop_reason: effective === "tool_use" ? "tool_use" : "end_turn" } } as ClaudeApiEvent); stream.push({ type: "done", reason: effective === "tool_use" ? "toolUse" : "stop", message: bridge.getOutput() }); stream.end(); - cleanup(); + endTurn(false); // clean turn → keep a cached connection warm for next turn }; const failWith = (msg: string) => { @@ -268,7 +362,7 @@ export function streamViaAcp( }, }); stream.end(); - cleanup(); + endTurn(true); // failed turn → destroy the connection (never reuse a broken one) }; const openBlock = (kind: "text" | "thinking") => { @@ -291,9 +385,11 @@ export function streamViaAcp( finish("tool_use"); }; - const clientHandler = { - async sessionUpdate(params: { update?: Record } & Record) { + const handleUpdate = async (params: { update?: Record } & Record): Promise => { if (ended) return; + // The warm connection is shared across turns; ignore a stray update once + // the entry has been handed to a newer turn (cross-turn bleed guard). + if (cacheEntry && cacheEntry.activeTurn !== myTurn) return; armInactivity(); const u = (params.update ?? params) as Record; const kind = u.sessionUpdate as string; @@ -320,13 +416,14 @@ export function streamViaAcp( surfaceToolAndBreak(claudeName, (u.toolCallId as string) ?? `acp_${blockIndex + 1}`, u.rawInput ?? u.input); } } - }, - async requestPermission(params: Record) { + }; + + const handlePermission = async (params: Record): Promise<{ outcome: { outcome: "cancelled" } }> => { // A permission request means the bridge is about to EXECUTE a tool. For a // pi-known tool, surface it to pi and break early (pi executes it); deny // by default otherwise. We always return cancelled so the bridge never // executes Fusion's tools itself. - if (!ended) { + if (!ended && !(cacheEntry && cacheEntry.activeTurn !== myTurn)) { const tc = (params.toolCall ?? {}) as Record; const claudeName = ((tc._meta as { claudeCode?: { toolName?: string } } | undefined)?.claudeCode?.toolName) ?? (tc.title as string) ?? ""; if (isPiKnownClaudeTool(claudeName)) { @@ -334,19 +431,92 @@ export function streamViaAcp( } } return { outcome: { outcome: "cancelled" as const } }; - }, + }; + + // Usage emission (OQ3) — shared by the cold + reuse paths. Coerces the + // untrusted bridge usage payload to finite, non-negative numbers. + const emitUsage = (res: unknown): void => { + if (sawToolCall) return; + const u = (res as { usage?: Record }).usage; + if (!u) return; + const num = (x: unknown): number | undefined => + typeof x === "number" && Number.isFinite(x) && x >= 0 ? x : undefined; + bridge.handleEvent({ + type: "message_delta", + delta: {}, + usage: { + input_tokens: num(u.inputTokens), + output_tokens: num(u.outputTokens), + cache_read_input_tokens: num(u.cachedReadTokens), + cache_creation_input_tokens: num(u.cachedWriteTokens), + }, + } as ClaudeApiEvent); }; try { + const withTimeout = (p: Promise, label: string) => + Promise.race([p, new Promise((_, rej) => setTimeout(() => rej(new Error(`ACP ${label} timeout`)), INITIALIZE_TIMEOUT_MS))]); + + // ── Reuse path: a warm connection for this conversation exists ────────── + // Skip spawn + initialize + session/new, and send ONLY the latest-turn + // delta (`buildResumePrompt`) because the warm `claude` session already + // holds the prior turns server-side (sending full history would duplicate + // it). Gated by `reuseKey`, which is undefined unless reuse is enabled. + let warm = reuseKey ? acpSessionCache.get(reuseKey) : undefined; + // Never reuse a busy connection or one bound to a different cwd. + if (warm && (warm.inUse || warm.cwd !== cwd)) warm = undefined; + // A reuse turn sends only the delta; if there's nothing new to send, an + // empty prompt to the warm session could hang. Drop the warm connection + // and cold-start with full history instead. + let resumeBlocks: ReturnType | undefined; + if (warm && reuseKey) { + const resume = buildResumePrompt(context); + const resumeEmpty = typeof resume === "string" ? resume.trim() === "" : resume.length === 0; + if (resumeEmpty) { evictCachedAcpConn(reuseKey, warm); warm = undefined; } + else resumeBlocks = toAcpPromptBlocks(resume as string | Array>); + } + if (warm && reuseKey && resumeBlocks) { + cacheEntry = warm; + myTurn = ++acpTurnCounter; + warm.activeTurn = myTurn; + warm.inUse = true; + if (warm.idleTimer) { clearTimeout(warm.idleTimer); warm.idleTimer = undefined; } + warm.router.onUpdate = handleUpdate; + warm.router.onPermission = handlePermission; + warm.router.fail = failWith; // a warm-child death now fails THIS turn fast + child = warm.child; + onAbort = () => failWith("aborted"); + if (options.signal) options.signal.addEventListener("abort", onAbort, { once: true }); + armInactivity(); + + // ACP ContentBlock[] — text/image shapes match; cast through unknown. + const res = await warm.conn.prompt({ sessionId: warm.acpSessionId, prompt: resumeBlocks as unknown as Parameters[0]["prompt"] }); + if (ended) return; + emitUsage(res); + if (!sawToolCall) finish("stop"); + return; + } + + // ── Cold path: spawn the bridge and open a fresh ACP session ─────────── if (!isAbsolute(options.bridgePath) || !existsSync(options.bridgePath)) { failWith(`ACP bridge path invalid (must be an absolute, existing binary): ${options.bridgePath}`); return; } - child = spawn(options.bridgePath, [], { stdio: ["pipe", "pipe", "pipe"], cwd: options.cwd ?? process.cwd(), env: buildBridgeEnv(options.bridgeEnv) }); + child = spawn(options.bridgePath, [], { stdio: ["pipe", "pipe", "pipe"], cwd, env: buildBridgeEnv(options.bridgeEnv) }); registerProcess(child); getStderr = captureStderr(child); - child.on("error", (e) => failWith(`ACP bridge spawn failed: ${e.message}`)); - child.on("close", (code) => { if (!ended) failWith(`ACP bridge exited (code ${code ?? "?"})${getStderr ? `: ${getStderr().slice(-500)}` : ""}`); }); + // Stable router indirection: the long-lived connection + child handlers + // always dispatch to whichever turn currently owns the connection. On + // reuse we repoint `router.*` at the new turn; `router.fail` lets a + // warm-child death fail the CURRENT owner (not the cold turn it spawned). + const router: AcpRouter = { onUpdate: handleUpdate, onPermission: handlePermission, fail: failWith }; + child.on("error", (e) => router.fail?.(`ACP bridge spawn failed: ${e.message}`)); + child.on("close", (code) => { + const msg = `ACP bridge exited (code ${code ?? "?"})${getStderr ? `: ${getStderr().slice(-500)}` : ""}`; + const fail = router.fail; // capture before evict nulls it + if (reuseKey && cacheEntry) evictCachedAcpConn(reuseKey, cacheEntry); // a dead child can never be reused + fail?.(msg); // fail the owning turn (no-op if idle / already ended) + }); onAbort = () => failWith("aborted"); if (options.signal) options.signal.addEventListener("abort", onAbort, { once: true }); armInactivity(); @@ -355,10 +525,15 @@ export function streamViaAcp( Writable.toWeb(child.stdin!) as unknown as WritableStream, Readable.toWeb(child.stdout!) as unknown as ReadableStream, ); - const conn = new ClientSideConnection(() => clientHandler, acpStream); - - const withTimeout = (p: Promise, label: string) => - Promise.race([p, new Promise((_, rej) => setTimeout(() => rej(new Error(`ACP ${label} timeout`)), INITIALIZE_TIMEOUT_MS))]); + const conn = new ClientSideConnection( + () => ({ + sessionUpdate: (p) => router.onUpdate?.(p as Parameters>[0]) ?? Promise.resolve(), + requestPermission: (p) => + router.onPermission?.(p as Parameters>[0]) ?? + Promise.resolve({ outcome: { outcome: "cancelled" as const } }), + }), + acpStream, + ); const init = await withTimeout( conn.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: { fs: { readTextFile: false, writeTextFile: false } } }), @@ -367,10 +542,17 @@ export function streamViaAcp( if (ended) return; if (init.protocolVersion !== PROTOCOL_VERSION) { failWith(`incompatible ACP protocol ${init.protocolVersion}`); return; } - const opened = await withTimeout(conn.newSession({ cwd: options.cwd ?? process.cwd(), mcpServers: options.mcpServers ?? [] }), "newSession"); + const opened = await withTimeout(conn.newSession({ cwd, mcpServers: options.mcpServers ?? [] }), "newSession"); if (ended) return; - const cwd = options.cwd ?? process.cwd(); + // Cache the warm connection so the next turn of this conversation reuses + // it. Only when reuse is enabled (reuseKey set) and the child is live. + if (reuseKey) { + myTurn = ++acpTurnCounter; + cacheEntry = { conn, child, acpSessionId: opened.sessionId, cwd, inUse: true, router, activeTurn: myTurn }; + acpSessionCache.set(reuseKey, cacheEntry); + } + const systemPrompt = buildSystemPrompt(context, cwd); const blocks = [ ...(systemPrompt ? [{ type: "text" as const, text: `${systemPrompt}\n\n` }] : []), @@ -379,30 +561,12 @@ export function streamViaAcp( // ACP ContentBlock[] — text/image shapes match; cast through unknown. const res = await conn.prompt({ sessionId: opened.sessionId, prompt: blocks as unknown as Parameters[0]["prompt"] }); + if (ended) return; // Feed token usage (experimental ACP field) into the bridge BEFORE finish() // so it lands in the `done` message. Tool-use turns break early and never // resolve here, so they inherently report zero usage. Zero-when-absent safe. - if (!sawToolCall) { - const u = (res as { usage?: Record }).usage; - // The bridge is untrusted (see BRIDGE_ENV_ALLOWLIST): coerce its usage - // payload to finite, non-negative numbers only so a malformed value - // (string/NaN/negative) can't corrupt totalTokens / cost downstream. - const num = (x: unknown): number | undefined => - typeof x === "number" && Number.isFinite(x) && x >= 0 ? x : undefined; - if (u) { - bridge.handleEvent({ - type: "message_delta", - delta: {}, - usage: { - input_tokens: num(u.inputTokens), - output_tokens: num(u.outputTokens), - cache_read_input_tokens: num(u.cachedReadTokens), - cache_creation_input_tokens: num(u.cachedWriteTokens), - }, - } as ClaudeApiEvent); - } - finish("stop"); - } + emitUsage(res); + if (!sawToolCall) finish("stop"); } catch (err) { failWith(err instanceof Error ? err.message : String(err)); }