feat(acp): opt-in warm connection reuse across turns (OQ2)

Keep a warm bridge connection + ACP session across turns of one
conversation (gated by FUSION_CLAUDE_ACP_REUSE=1, default OFF), 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.

Addresses the adversarial review of the reuse path:
- P0: a warm-child death routes failure to the CURRENT owner turn via
  router.fail, so a reuse turn fails fast instead of hanging until the
  30-min inactivity timeout.
- P1: eviction is cache-identity-aware (evictCachedAcpConn only deletes
  the map key when it still points at the entry), so a concurrent cold
  turn / stale close handler / idle timer can't evict or kill a newer
  live entry's child.
- P1: an empty resume delta cold-starts instead of issuing an empty
  prompt that could hang.
- P2: a per-turn token drops cross-turn stray updates on the shared
  warm connection.
- The idle reaper is unref'd so it never pins the process.

Default OFF → the cold path is functionally unchanged (reviewer-verified).
Adds multi-turn tests: reuse skips spawn+session/new, flag-off spawns
fresh, fail-fast on warm-child death, empty-resume cold fallback.
346/346 pass, tsc clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.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.

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<Record<string, unknown>> = [];
 let scriptedUsage: Record<string, number> | 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<Record<string, unknown>> }) {
 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<Record<string, unknown>> };
+    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<typeof vi.fn>; newSession: ReturnType<typeof vi.fn> };
+    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<Record<string, unknown>> };
+    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<Record<string, unknown>> };
+    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,

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<string, unknown> } & Record<string, unknown>) => Promise<void>) | null;
+  onPermission: ((p: Record<string, unknown>) => 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<typeof setTimeout>;
+  // 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<string, CachedAcpConn>();
+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<typeof setTimeout> | 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<string, unknown> } & Record<string, unknown>) {
+    const handleUpdate = async (params: { update?: Record<string, unknown> } & Record<string, unknown>): Promise<void> => {
         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<string, unknown>;
         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<string, unknown>) {
+    };
+
+    const handlePermission = async (params: Record<string, unknown>): 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<string, unknown>;
           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<string, unknown> }).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 = <T>(p: Promise<T>, label: string) =>
+        Promise.race([p, new Promise<never>((_, 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<typeof toAcpPromptBlocks> | 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<Record<string, unknown>>);
+      }
+      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<typeof warm.conn.prompt>[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<Uint8Array>,
         Readable.toWeb(child.stdout!) as unknown as ReadableStream<Uint8Array>,
       );
-      const conn = new ClientSideConnection(() => clientHandler, acpStream);
-
-      const withTimeout = <T>(p: Promise<T>, label: string) =>
-        Promise.race([p, new Promise<never>((_, rej) => setTimeout(() => rej(new Error(`ACP ${label} timeout`)), INITIALIZE_TIMEOUT_MS))]);
+      const conn = new ClientSideConnection(
+        () => ({
+          sessionUpdate: (p) => router.onUpdate?.(p as Parameters<NonNullable<AcpRouter["onUpdate"]>>[0]) ?? Promise.resolve(),
+          requestPermission: (p) =>
+            router.onPermission?.(p as Parameters<NonNullable<AcpRouter["onPermission"]>>[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<typeof conn.prompt>[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<string, unknown> }).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));
     }