fix(acp): address code-review findings (security, correctness, reliability)

Tier-2 code review fixes:
- P1 correctness: EventBridge per-turn state never reset — once the per-turn
  output cap tripped, all later turns were silently suppressed and tool/accum
  state bled across turns. Surface resetTurn() and call it per prompt turn.
- P1: plan_update read a non-existent .entries field (wrong SDK shape) and
  wiped the displayed plan — now a documented no-op (full 'plan' is source of
  truth).
- P1 security: write-path TOCTOU — open without O_TRUNC, re-validate realpath,
  then truncate, so an intermediate-symlink-swapped escaped target is never
  truncated before rejection.
- DoS: fs read stat-gates and bounded-reads oversized files instead of loading
  them fully before the ceiling.
- Security: stderr redaction now spans chunk boundaries; secret deny-list adds
  .git-credentials/*.p12/*.pfx/*.keystore/.pgpass/.htpasswd/etc.
- Reliability: cancelAcpSession bounded by a timeout so a blocked stdin can't
  delay the registry SIGKILL.
- Maintainability: drop dead ACP_NOT_IMPLEMENTED export; type agentCapabilities
  via the SDK AgentCapabilities; strengthen the S1 write-denial assertion.

+4 tests (181 total); typecheck + eslint clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

plugins/fusion-plugin-acp-runtime/src/__tests__/event-bridge.test.ts

@@ -1,6 +1,6 @@
 import { describe, it, expect, vi } from "vitest";
 import type { SessionUpdate } from "@agentclientprotocol/sdk";
-import { createEventBridge } from "../event-bridge.js";
+import { createEventBridge, PER_TURN_OUTPUT_CAP_CHARS } from "../event-bridge.js";
 import type { AcpCallbacks } from "../types.js";
 
 function makeCallbacks() {
@@ -208,6 +208,66 @@ describe("event bridge: plan (full replacement)", () => {
     expect(second).toContain("Step C");
     expect(second).not.toContain("Step A");
   });
+
+  it("plan_update does NOT wipe the prior plan (no-op; full plan stays source of truth) (FIX 2)", () => {
+    const { callbacks, onThinking } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+
+    bridge.handleSessionUpdate({
+      sessionUpdate: "plan",
+      entries: [{ content: "Step A", priority: "high", status: "pending" }],
+    } as SessionUpdate);
+    // The PlanUpdate variant carries `plan`, not a top-level `entries` array.
+    // The bridge must treat it as a no-op rather than firing an empty-plan
+    // onThinking that wipes the displayed plan.
+    bridge.handleSessionUpdate({
+      sessionUpdate: "plan_update",
+      plan: { type: "items", items: [] },
+    } as unknown as SessionUpdate);
+
+    // Only the `plan` event fired onThinking; plan_update fired nothing.
+    expect(onThinking).toHaveBeenCalledTimes(1);
+    expect(onThinking.mock.calls[0][0]).toContain("Step A");
+  });
+});
+
+describe("event bridge: per-turn reset (FIX 1)", () => {
+  it("resetTurn clears the output-cap latch so a later turn is not suppressed", () => {
+    const { callbacks, onText, onThinking } = makeCallbacks();
+    const bridge = createEventBridge(callbacks);
+
+    // Turn 1: flood past the per-turn cap so the latch trips and the truncation
+    // flag fires. One oversized chunk is bounded per-chunk, so send enough chunks
+    // to cross the cumulative cap.
+    const chunk = "y".repeat(50_000);
+    const chunksToTrip = Math.ceil(PER_TURN_OUTPUT_CAP_CHARS / chunk.length) + 1;
+    for (let i = 0; i < chunksToTrip; i++) {
+      bridge.handleSessionUpdate({
+        sessionUpdate: "agent_message_chunk",
+        content: { type: "text", text: chunk },
+      } as SessionUpdate);
+    }
+    // The cap fired exactly one truncation flag via onThinking.
+    expect(
+      onThinking.mock.calls.some((c) => /output truncated/i.test(String(c[0]))),
+    ).toBe(true);
+
+    // After the latch trips, further text on the SAME turn is suppressed.
+    const callsAfterTrip = onText.mock.calls.length;
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_message_chunk",
+      content: { type: "text", text: "suppressed" },
+    } as SessionUpdate);
+    expect(onText.mock.calls.length).toBe(callsAfterTrip);
+
+    // Turn 2: reset, then ordinary text must flow again (latch cleared).
+    bridge.reset();
+    bridge.handleSessionUpdate({
+      sessionUpdate: "agent_message_chunk",
+      content: { type: "text", text: "turn 2 output" },
+    } as SessionUpdate);
+    expect(onText).toHaveBeenLastCalledWith("turn 2 output");
+  });
 });
 
 describe("event bridge: tolerance", () => {

plugins/fusion-plugin-acp-runtime/src/__tests__/fs-capabilities.test.ts

@@ -89,6 +89,21 @@ describe("readTextFile", () => {
     expect(res.content.length).toBe(100);
   });
 
+  it("reads a file larger than the ceiling WITHOUT loading it fully (bounded read) (FIX 4)", async () => {
+    // Content far larger than the ceiling: a full readFile would load it all
+    // before truncation. The bounded-read path must cap memory + output.
+    const ceiling = 100;
+    const huge = "a".repeat(50_000); // 500x the ceiling
+    await writeFile(path.join(cwd, "huge.txt"), huge, "utf8");
+    const res = await reader({ readMaxBytes: ceiling })({
+      sessionId: "s",
+      path: "huge.txt",
+    } as never);
+    // Output is capped at the ceiling and equals the first `ceiling` bytes.
+    expect(res.content.length).toBe(ceiling);
+    expect(res.content).toBe("a".repeat(ceiling));
+  });
+
   it("rejects a lexical ../ escape", async () => {
     await expect(
       reader()({ sessionId: "s", path: "../../etc/passwd" } as never),
@@ -151,7 +166,7 @@ describe("writeTextFile", () => {
     // no approver the write must be denied, not silently written.
     await expect(
       writer(allowGate)({ sessionId: "s", path: "out2.txt", content: "x" } as never),
-    ).rejects.toThrow();
+    ).rejects.toBeInstanceOf(FsWriteDeniedError);
   });
 
   it("rejects an oversized write before touching the fs", async () => {

plugins/fusion-plugin-acp-runtime/src/__tests__/path-jail.test.ts

@@ -141,6 +141,15 @@ describe("deny-list predicates", () => {
       "id_rsa",
       "id_ed25519.pub",
       "credentials",
+      // FIX 6: expanded secret deny-list.
+      ".git-credentials",
+      "server.p12",
+      "cert.pfx",
+      "release.keystore",
+      "app.jks",
+      ".dockercfg",
+      ".pgpass",
+      ".htpasswd",
     ]) {
       expect(isSecretPath(path.join(cwd, f))).toBe(true);
     }

plugins/fusion-plugin-acp-runtime/src/__tests__/process-manager.test.ts

@@ -95,6 +95,24 @@ describe("captureStderr", () => {
     expect(out).toContain("Authorization:");
     expect(out).not.toContain("sk-live-SECRETSECRETSECRET123456");
   });
+
+  it("redacts a token split across two stderr writes (cross-chunk) (FIX 5)", async () => {
+    // The secret is emitted in two separate write() calls so it straddles two
+    // `data` chunks. Per-chunk redaction would leak it; cross-boundary redaction
+    // must catch it.
+    const child = track(
+      spawn(process.execPath, [
+        "-e",
+        "process.stderr.write('Authorization: Bearer sk-live-SPLIT');" +
+          "setTimeout(()=>process.stderr.write('TOKENTOKENTOKEN123456\\n'),20);",
+      ]),
+    );
+    const getStderr = captureStderr(child);
+    await waitForExit(child);
+    const out = getStderr();
+    expect(out).not.toContain("sk-live-SPLITTOKENTOKENTOKEN123456");
+    expect(out).toContain("[REDACTED]");
+  });
 });
 
 describe("process registry (KTD4)", () => {

plugins/fusion-plugin-acp-runtime/src/event-bridge.ts

@@ -250,8 +250,10 @@ export function createEventBridge(callbacks: AcpCallbacks): EventBridge {
           handlePlan(update.entries);
           break;
         case "plan_update":
-          // Treat an incremental plan op as a plan refresh for v1.
-          handlePlan((update as { entries?: PlanEntry[] }).entries);
+          // The (experimental) `PlanUpdate` variant carries a `plan` field, NOT a
+          // top-level `entries` array — so there is nothing here to map to our
+          // entries-based snapshot. v1 treats it as a NO-OP rather than wiping the
+          // prior plan: the full `plan` event remains the source of truth.
           break;
         case "plan_removed":
           // Clearing the plan: surface nothing.

plugins/fusion-plugin-acp-runtime/src/fs-capabilities.ts

@@ -144,7 +144,25 @@ export function createFsHandlers(opts: FsHandlerOptions): FsHandlers {
       // Atomic, symlink-safe open (TOCTOU defense), then read.
       const handle = await openWithinCwd(resolved, opts.cwd, fsConstants.O_RDONLY);
       try {
-        const content = await handle.readFile({ encoding: "utf8" });
+        const hasLimit =
+          typeof params.limit === "number" &&
+          Number.isFinite(params.limit) &&
+          params.limit > 0;
+        // DoS guard (FIX 4): a multi-GB file would OOM if we `readFile` the whole
+        // thing before `applyReadWindow` truncates. When the file exceeds the byte
+        // ceiling AND no bounding `limit` was supplied, read at most ceiling+1
+        // bytes so memory stays bounded; the +1 still lets applyReadWindow apply
+        // its truncation marker logic identically to a full read. A `limit` is
+        // line-bounded and read in full (matches prior behavior).
+        const stat = await handle.stat();
+        let content: string;
+        if (!hasLimit && stat.size > readMaxBytes) {
+          const buf = Buffer.alloc(readMaxBytes + 1);
+          const { bytesRead } = await handle.read(buf, 0, readMaxBytes + 1, 0);
+          content = buf.subarray(0, bytesRead).toString("utf8");
+        } else {
+          content = await handle.readFile({ encoding: "utf8" });
+        }
         return {
           content: applyReadWindow(content, params.line, params.limit, readMaxBytes),
         };
@@ -214,16 +232,24 @@ export function createFsHandlers(opts: FsHandlerOptions): FsHandlers {
       }
       // disposition === "allow" → proceed.
 
-      // Atomic, symlink-safe create/truncate within cwd. O_NOFOLLOW (in
-      // openWithinCwd) prevents following a swapped-in symlink on the final
-      // component (TOCTOU). O_CREAT|O_TRUNC|O_WRONLY for a normal write.
+      // Atomic, symlink-safe create within cwd. O_NOFOLLOW (in openWithinCwd)
+      // guards ONLY the FINAL component; an intermediate dir swapped to a symlink
+      // is still followed. We therefore must NOT pass O_TRUNC into open(): doing
+      // so would TRUNCATE an escaped target BEFORE openWithinCwd's post-open
+      // realpath re-validation gets to reject it (write-path TOCTOU, FIX 3).
+      // Instead open create+write WITHOUT truncate, let openWithinCwd run its
+      // re-validation, and ONLY truncate (via the fd) AFTER it has proven the
+      // opened inode is still inside the jail.
       const handle = await openWithinCwd(
         resolved,
         opts.cwd,
-        fsConstants.O_WRONLY | fsConstants.O_CREAT | fsConstants.O_TRUNC,
+        fsConstants.O_WRONLY | fsConstants.O_CREAT,
         0o644,
       );
       try {
+        // Truncate-AFTER-validate: openWithinCwd returned only because the
+        // re-validation passed, so it is now safe to empty the file and write.
+        await handle.truncate(0);
         await handle.writeFile(content, { encoding: "utf8" });
       } finally {
         await handle.close().catch(() => undefined);

plugins/fusion-plugin-acp-runtime/src/path-jail.ts

@@ -53,6 +53,13 @@ const SECRET_BASENAME_PATTERNS: RegExp[] = [
   /^\.netrc$/i,
   /^id_.+$/i, // id_rsa, id_ed25519, id_rsa.pub, ...
   /^credentials$/i,
+  /^\.git-credentials$/i, // git stored plaintext credentials
+  /\.p12$/i, // PKCS#12 keystore
+  /\.pfx$/i, // PKCS#12 keystore (Windows)
+  /\.(keystore|jks)$/i, // Java keystore
+  /^\.dockercfg$/i, // legacy docker registry auth
+  /^\.pgpass$/i, // PostgreSQL password file
+  /^\.htpasswd$/i, // Apache basic-auth credentials
 ];
 
 /**

plugins/fusion-plugin-acp-runtime/src/process-manager.ts

@@ -143,12 +143,18 @@ export function redactSecrets(text: string): string {
  * Returns a getter for the current (redacted) buffer contents.
  */
 export function captureStderr(child: ChildProcess): () => string {
-  let buffer = "";
+  // FIX 5: redacting each chunk in isolation leaks a secret that straddles a
+  // chunk boundary (the token is split across two `data` events so neither half
+  // matches a pattern). Accumulate the RAW bytes into a bounded buffer first,
+  // then redact across the whole (bounded) buffer after each append so a
+  // boundary-spanning secret is caught. The buffer stays bounded by the existing
+  // ceiling; the returned getter always reports the redacted view.
+  let raw = "";
   child.stderr?.on("data", (data: Buffer) => {
-    buffer += redactSecrets(data.toString());
-    if (buffer.length > STDERR_BUFFER_CEILING) {
-      buffer = buffer.slice(buffer.length - STDERR_BUFFER_CEILING);
+    raw += data.toString();
+    if (raw.length > STDERR_BUFFER_CEILING) {
+      raw = raw.slice(raw.length - STDERR_BUFFER_CEILING);
     }
   });
-  return () => buffer;
+  return () => redactSecrets(raw);
 }

plugins/fusion-plugin-acp-runtime/src/provider.ts

@@ -18,6 +18,7 @@ import {
   ndJsonStream,
   PROTOCOL_VERSION,
   type Agent,
+  type AgentCapabilities,
   type Client,
   type ContentBlock,
   type RequestPermissionResponse,
@@ -93,6 +94,13 @@ export interface BridgingClientHandler {
    * immediately (U5 cancel-drain — KTD4a). Idempotent.
    */
   cancelPending(): void;
+  /**
+   * Reset the event bridge's PER-TURN state (tool correlation, delta
+   * accumulators, cumulative-output counter, output-cap latch). MUST be called
+   * at the start of each prompt turn so a turn that trips the per-turn output cap
+   * does not silently suppress every subsequent turn (FIX 1).
+   */
+  resetTurn(): void;
 }
 
 /**
@@ -183,14 +191,14 @@ export function createBridgingClientHandler(
   if (fsHandlers.readTextFile) handler.readTextFile = fsHandlers.readTextFile;
   if (fsHandlers.writeTextFile) handler.writeTextFile = fsHandlers.writeTextFile;
 
-  return { handler, cancelPending };
+  return { handler, cancelPending, resetTurn: () => bridge.reset() };
 }
 
 export interface AcpConnection {
   /** Live ACP connection — later units drive session/new, prompt, cancel, load. */
   conn: ClientSideConnection;
   child: ChildProcess;
-  agentCapabilities?: unknown;
+  agentCapabilities?: AgentCapabilities;
   /** Auth methods the agent advertised; non-empty means auth is required. */
   authMethods: Array<{ id: string }>;
   /** Current redacted stderr buffer. */
@@ -326,14 +334,9 @@ export async function connect(opts: ConnectOptions): Promise<AcpConnection> {
 // adapter drives one shape (open → prompt → cancel/resume) without touching SDK
 // types directly. v1 always sends an empty `mcpServers` (KTD5).
 
-/** Narrow view of the agent capabilities we read for resume routing. */
-interface AgentCapabilitiesView {
-  loadSession?: boolean;
-}
-
 function readsLoadSession(connection: AcpConnection): boolean {
-  const caps = connection.agentCapabilities as AgentCapabilitiesView | undefined;
-  return caps?.loadSession === true;
+  // `agentCapabilities` is already typed as `AgentCapabilities | undefined`.
+  return connection.agentCapabilities?.loadSession === true;
 }
 
 export interface NewAcpSessionResult {
@@ -380,12 +383,25 @@ export async function promptAcpSession(
  * runs during teardown where the registry SIGKILL is the authoritative guarantee
  * (KTD4a).
  */
+/** Upper bound on how long `cancelAcpSession` waits on the cancel write (FIX 7). */
+const CANCEL_TIMEOUT_MS = 2_000;
+
 export async function cancelAcpSession(
   connection: AcpConnection,
   sessionId: string,
 ): Promise<void> {
+  // `conn.cancel` writes to the agent's stdin pipe; a dead or full pipe can
+  // back-pressure and stall teardown (the adapter awaits this BEFORE the
+  // authoritative registry SIGKILL). Bound it so the kill still runs promptly
+  // (FIX 7). Errors are swallowed — this is already best-effort.
   try {
-    await connection.conn.cancel({ sessionId });
+    await Promise.race([
+      connection.conn.cancel({ sessionId }),
+      new Promise<void>((resolve) => {
+        const timer = setTimeout(resolve, CANCEL_TIMEOUT_MS);
+        timer.unref?.();
+      }),
+    ]);
   } catch {
     // fire-and-forget; teardown's SIGKILL is authoritative
   }

plugins/fusion-plugin-acp-runtime/src/runtime-adapter.ts

@@ -26,12 +26,6 @@ import type {
   AcpSession,
 } from "./types.js";
 
-/**
- * Retained for back-compat: earlier units' tests imported this marker. The real
- * adapter no longer throws it; it remains exported so external references resolve.
- */
-export const ACP_NOT_IMPLEMENTED = "acp_not_implemented";
-
 export class AcpRuntimeAdapter implements AgentRuntime {
   readonly id = "acp";
   readonly name = "ACP Runtime";
@@ -61,7 +55,7 @@ export class AcpRuntimeAdapter implements AgentRuntime {
     // default OFF (KTD6) — and confined to the task cwd by the path jail. The
     // same toggles drive the advertised `fs` capability in connect() below, so
     // advertisement and registered handlers stay consistent.
-    const { handler: clientHandler, cancelPending } = createBridgingClientHandler(
+    const { handler: clientHandler, cancelPending, resetTurn } = createBridgingClientHandler(
       callbacks,
       options.actionGateContext,
       {
@@ -110,6 +104,10 @@ export class AcpRuntimeAdapter implements AgentRuntime {
       // Persist the per-run gate (KTD3) so U5/U7 can reach the live action gate.
       gate: options.actionGateContext,
       connection,
+      // Reset the event bridge's per-turn state at the start of each turn so a
+      // turn that trips the per-turn output cap can't latch and suppress every
+      // subsequent turn (FIX 1).
+      resetTurn,
       dispose: () => {
         if (disposed) return;
         disposed = true;
@@ -132,6 +130,11 @@ export class AcpRuntimeAdapter implements AgentRuntime {
     if (!acp.connection) {
       throw new Error("ACP session has no live connection (createSession not completed)");
     }
+    // Clear per-turn event-bridge state BEFORE driving the turn so tool
+    // correlation, delta accumulators, and the output-cap latch all start clean
+    // each turn (FIX 1). Without this, a turn that hit the per-turn output cap
+    // would silently suppress all later turns.
+    acp.resetTurn?.();
     const blocks = buildPromptBlocks(prompt);
     // Resolve when the SDK prompt promise resolves — it already drains all
     // session/update notifications for the turn before reporting the stopReason.

plugins/fusion-plugin-acp-runtime/src/types.ts

@@ -112,6 +112,12 @@ export interface AcpSession {
    * agent through it. Undefined only for the bare session shell used in tests.
    */
   connection?: AcpConnection;
+  /**
+   * Reset the event bridge's per-turn state (tool correlation, delta
+   * accumulators, output-cap latch). Called by `promptWithFallback` at the start
+   * of each turn (FIX 1). Undefined for the bare session shell used in tests.
+   */
+  resetTurn?: () => void;
   dispose(): void;
 }