Preserve ACP bridge terminal evidence before disconnect (#15478)

Drain terminal frames through backpressure and record bounded terminal evidence without waiting for log persistence. Guard later events and input failures against writes after socket end.

Validated with 217 focused tests, full typecheck/build, independent review and green CI with Greptile 5/5.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/acp-run-lifecycle.md

@@ -58,6 +58,15 @@ removes it if startup fails. The child's environment values remain unchanged.
 If the launch shell fails before the wrapper emits a protocol event, the run
 log retains the shell's stderr alongside the exit code.
 
+The process-session bridge also writes one bounded terminal diagnostic to the
+run's stderr. It distinguishes a remote event, input delivery failure, output
+poll failure, output stream failure, and unexpected authenticated proxy closure.
+Exit codes and signals are validated; unavailable values are `unknown`. The line
+contains no raw error message, command, path, or environment value. It records
+the first observed terminal event, not proof of remote shutdown. An error frame
+drains to the local proxy before the connection closes, and a failed or stalled
+diagnostic write does not delay failure delivery or teardown.
+
 The settlement sequence is the one live cleanup owner for every settled path. It
 claims the ledger once, makes the pure reuse decision, then runs the ordered
 steps:

packages/adapter-utils/src/execution-target-sandbox.test.ts

@@ -1107,6 +1107,7 @@ describe("sandbox adapter execution targets", () => {
     // ACP keeps stdin open while it waits for a handshake. A remote child can
     // exit before replying; that must close the proxy and fail the handshake.
     const output = "final output\n".repeat(16_384);
+    const logs: string[] = [];
     const bridge = await startAdapterExecutionTargetProcessSessionBridge({
       runId: "run-open-stdin",
       target: {
@@ -1124,11 +1125,16 @@ describe("sandbox adapter execution targets", () => {
       env: {},
       timeoutSec: 10,
       streamOutputViaSession,
+      onLog: async (stream, chunk) => { if (stream === "stderr") logs.push(chunk); },
     });
     expect(bridge).not.toBeNull();
     try {
       const result = await runProxyWithInput(bridge!.agentCommand, "initialize\n", true);
       expect(result.code).toBe(exitCode ?? 1);
+      const terminal = logs.filter((line) => line.includes("ACP process session terminal:"));
+      expect(terminal).toEqual([
+        `[paperclip] ACP process session terminal: source=remote_event outcome=${exitCode === null ? "error" : "exit"} exitCode=${exitCode ?? "unknown"} signal=unknown.\n`,
+      ]);
       if (exitCode === null) {
         expect(result.stdout).toBe("");
         expect(result.stderr).toContain("ENOENT");
@@ -1141,6 +1147,194 @@ describe("sandbox adapter execution targets", () => {
     }
   }, 15_000);
 
+  it("flushes remote errors under backpressure and retains only bounded terminal evidence", async () => {
+    const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-terminal-flush-"));
+    cleanupDirs.push(rootDir);
+    const delegate = createLocalSandboxRunner();
+    const logs: string[] = [];
+    let emit!: (frame: string) => Promise<void>;
+    let complete!: () => void;
+    let failInput!: (error: Error) => void;
+    let inputStarted = false;
+    const inputFailure = new Promise<never>((_resolve, reject) => { failInput = reject; });
+    void inputFailure.catch(() => {});
+    const commandComplete = new Promise<void>((resolve) => { complete = resolve; });
+    const frame = {
+      type: "error", message: "synthetic-private-detail".repeat(200_000),
+      code: 1e100, signal: "untrusted-private-signal",
+    };
+    const bridge = await startAdapterExecutionTargetProcessSessionBridge({
+      runId: "terminal-flush", adapterKey: "acpx", runtimeRootDir: rootDir,
+      target: {
+        kind: "remote", transport: "sandbox", remoteCwd: rootDir,
+        runner: { execute: async (input) => {
+          if (!input.useSession) {
+            const script = input.args?.[1] ?? "";
+            if (script.startsWith("mkdir -p") && script.includes("/stdin/000000000001.json")) {
+              inputStarted = true;
+              return inputFailure;
+            }
+            return delegate.execute(input);
+          }
+          emit = async (text) => { await input.onLog?.("stdout", text); };
+          await commandComplete;
+          return { exitCode: 1, stdout: "", stderr: "", timedOut: false, signal: null, pid: null, startedAt: null };
+        } },
+      },
+      command: "cat", args: [], cwd: rootDir, env: {}, streamOutputViaSession: true,
+      onLog: async (stream, chunk) => { if (stream === "stderr") logs.push(chunk); },
+    });
+    let peer: net.Socket | undefined;
+    try {
+      const source = await readFile(bridge!.agentCommand, "utf8");
+      const port = Number(/port: (\d+)/.exec(source)![1]);
+      const token = JSON.parse(/const token = (".*?");/.exec(source)![1]) as string;
+      peer = net.createConnection({ host: "127.0.0.1", port });
+      peer.on("error", () => {});
+      peer.setEncoding("utf8");
+      let output = "";
+      peer.on("data", (chunk) => { output += chunk; });
+      const closed = new Promise<void>((resolve) => peer!.once("close", resolve));
+      await new Promise<void>((resolve) => peer!.once("connect", resolve));
+      peer.write(JSON.stringify({ token, type: "hello" }) + "\n");
+      // Confirm authentication by receiving a marker before applying backpressure.
+      await emit(JSON.stringify({ type: "data", stream: "stdout", data: "" }) + "\n");
+      await waitForCondition(() => output.length > 0, "Missing authenticated marker.");
+      output = "";
+      peer.write(JSON.stringify({ token, type: "stdin", data: "aW5wdXQ=" }) + "\n");
+      await waitForCondition(() => inputStarted, "Input write did not start.");
+      peer.pause();
+      // The real wrapper emits error then exit when a child fails to spawn.
+      // A later frame must not write after end() and abort the pending error.
+      await emit([
+        frame,
+        { type: "exit", code: 1 },
+        { type: "data", stream: "stdout", data: "bGF0ZQ==" },
+        { type: "shutdownAck" },
+      ].map((event) => JSON.stringify(event) + "\n").join(""));
+      // The reverse race matters too: an already-accepted input write can
+      // reject after the remote error started draining. Its end(data) must
+      // not abort or replace the first terminal frame.
+      failInput(new Error("synthetic-private-input-detail"));
+      await waitForCondition(
+        () => logs.some((line) => line.includes("ACP process session input delivery failed.")),
+        "Input failure was not handled.",
+      );
+      peer.resume();
+      await closed;
+      expect(JSON.parse(output)).toEqual(frame);
+      expect(logs.filter((line) => line.includes("ACP process session terminal:"))).toEqual([
+        "[paperclip] ACP process session terminal: source=remote_event outcome=error exitCode=unknown signal=unknown.\n",
+      ]);
+    } finally {
+      failInput(new Error("test cleanup"));
+      complete();
+      peer?.destroy();
+      await bridge?.stop();
+    }
+  }, 15_000);
+
+  it("reports a streamed transport failure even when diagnostic persistence stalls", async () => {
+    const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-terminal-stream-"));
+    cleanupDirs.push(rootDir);
+    const delegate = createLocalSandboxRunner();
+    const logs: string[] = [];
+    let releaseLog!: () => void;
+    const stalledLog = new Promise<void>((resolve) => { releaseLog = resolve; });
+    const bridge = await startAdapterExecutionTargetProcessSessionBridge({
+      runId: "terminal-stream", adapterKey: "acpx", runtimeRootDir: rootDir,
+      target: {
+        kind: "remote", transport: "sandbox", remoteCwd: rootDir,
+        runner: { execute: async (input) => {
+          if (input.useSession) throw new Error("synthetic-private-provider-detail");
+          return delegate.execute(input);
+        } },
+      },
+      command: "cat", args: [], cwd: rootDir, env: {}, streamOutputViaSession: true,
+      onLog: async (stream, chunk) => {
+        if (stream === "stderr") { logs.push(chunk); await stalledLog; }
+      },
+    });
+    try {
+      const result = await runProxyWithInput(bridge!.agentCommand, "", true);
+      expect(result.code).toBe(1);
+      expect(logs).toEqual([
+        "[paperclip] ACP process session terminal: source=output_stream outcome=error exitCode=unknown signal=unknown.\n",
+      ]);
+    } finally {
+      releaseLog();
+      await bridge?.stop();
+    }
+  }, 10_000);
+
+  it("closes a failed output poll without waiting for the run log", async () => {
+    const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-terminal-poll-"));
+    cleanupDirs.push(rootDir);
+    const delegate = createLocalSandboxRunner();
+    const logs: string[] = [];
+    let releaseLog!: () => void;
+    const stalledLog = new Promise<void>((resolve) => { releaseLog = resolve; });
+    const bridge = await startAdapterExecutionTargetProcessSessionBridge({
+      runId: "terminal-poll", adapterKey: "acpx", runtimeRootDir: rootDir,
+      target: {
+        kind: "remote", transport: "sandbox", remoteCwd: rootDir,
+        runner: { execute: async (input) => {
+          const script = input.args?.[1] ?? "";
+          if (script.includes("/events'/*.json")) throw new Error("synthetic-private-provider-detail");
+          return delegate.execute(input);
+        } },
+      },
+      command: "cat", args: [], cwd: rootDir, env: {},
+      onLog: async (stream, chunk) => {
+        if (stream === "stderr") { logs.push(chunk); await stalledLog; }
+      },
+    });
+    try {
+      const result = await runProxyWithInput(bridge!.agentCommand, "", true);
+      expect(result.code).toBe(1);
+      expect(logs).toEqual([
+        "[paperclip] ACP process session terminal: source=output_poll outcome=error exitCode=unknown signal=unknown.\n",
+      ]);
+    } finally {
+      releaseLog();
+      await bridge?.stop();
+    }
+  }, 10_000);
+
+  it("records an unexpected authenticated proxy close once and ignores unauthenticated peers", async () => {
+    const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-terminal-proxy-"));
+    cleanupDirs.push(rootDir);
+    const logs: string[] = [];
+    const bridge = await startAdapterExecutionTargetProcessSessionBridge({
+      runId: "terminal-proxy", adapterKey: "acpx", runtimeRootDir: rootDir,
+      target: { kind: "remote", transport: "sandbox", remoteCwd: rootDir, runner: createLocalSandboxRunner() },
+      command: "cat", args: [], cwd: rootDir, env: {},
+      onLog: async (stream, chunk) => { if (stream === "stderr") logs.push(chunk); },
+    });
+    try {
+      const source = await readFile(bridge!.agentCommand, "utf8");
+      const port = Number(/port: (\d+)/.exec(source)![1]);
+      const token = JSON.parse(/const token = (".*?");/.exec(source)![1]) as string;
+      for (const authenticated of [false, true]) {
+        const peer = net.createConnection({ host: "127.0.0.1", port });
+        peer.on("error", () => {});
+        const closed = new Promise<void>((resolve) => peer.once("close", resolve));
+        await new Promise<void>((resolve) => peer.once("connect", resolve));
+        if (authenticated) peer.end(JSON.stringify({ token, type: "hello" }) + "\n");
+        else peer.end();
+        await closed;
+        if (!authenticated) expect(logs).toEqual([]);
+      }
+      await waitForCondition(() => logs.length === 1, "Missing proxy-close diagnostic.");
+      await bridge!.stop();
+      expect(logs.filter((line) => line.includes("ACP process session terminal:"))).toEqual([
+        "[paperclip] ACP process session terminal: source=proxy_close outcome=error exitCode=unknown signal=unknown.\n",
+      ]);
+    } finally {
+      await bridge?.stop();
+    }
+  }, 10_000);
+
   it("buffers sandbox process session output until the local proxy connects", async () => {
     const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-process-session-buffer-"));
     cleanupDirs.push(rootDir);

packages/adapter-utils/src/execution-target.ts

@@ -2165,6 +2165,26 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
     }
   };
   let pollTimer: NodeJS.Timeout | null = null;
+  let terminalEvidenceRecorded = false;
+  const recordTerminalEvidence = (
+    source: "remote_event" | "input_delivery" | "output_poll" | "output_stream" | "proxy_close",
+    event: { type?: string; code?: number | null; signal?: string | null },
+  ) => {
+    if (terminalEvidenceRecorded) return;
+    terminalEvidenceRecorded = true;
+    // Remote frames are untrusted. Keep the retained diagnostic bounded and
+    // independent of messages, commands, paths, payloads, and provider errors.
+    const outcome = event.type === "exit" ? "exit" : "error";
+    const code = typeof event.code === "number" && Number.isInteger(event.code)
+      && event.code >= 0 && event.code <= 255 ? event.code : "unknown";
+    const signal = typeof event.signal === "string" && [
+      "SIGHUP", "SIGINT", "SIGQUIT", "SIGILL", "SIGABRT", "SIGFPE", "SIGKILL",
+      "SIGSEGV", "SIGPIPE", "SIGALRM", "SIGTERM", "SIGBUS", "SIGXCPU", "SIGXFSZ",
+    ].includes(event.signal) ? event.signal : "unknown";
+    logFailureWithoutWaiting(
+      `[paperclip] ACP process session terminal: source=${source} outcome=${outcome} exitCode=${code} signal=${signal}.\n`,
+    );
+  };
   const pendingRemoteEvents: Array<{
     type?: string;
     stream?: "stdout" | "stderr";
@@ -2191,22 +2211,30 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
 
   const writeRemoteEventToSocket = (event: (typeof pendingRemoteEvents)[number]) => {
     if (!socket) return false;
+    // `stopping` can already be set by a terminal frame buffered before auth.
+    // Use the socket's outbound state instead: once end() has been called,
+    // another write would destroy it and truncate the frame still draining.
+    // This also covers input-delivery failure, which ends the socket directly.
+    if (socket.writableEnded || socket.destroyed) return true;
     socket.write(jsonLine(event));
-    if (event.type === "exit") {
+    if (event.type === "exit" || event.type === "error") {
       stopping = true;
+      // Flush the terminal frame and all earlier output before closing. An
+      // immediate destroy can discard the cause and leave only connection_close.
       socket.end();
-    } else if (event.type === "error") {
-      stopping = true;
-      socket.destroy();
     }
     return true;
   };
 
-  const deliverRemoteEvent = (event: (typeof pendingRemoteEvents)[number]) => {
+  const deliverRemoteEvent = (
+    event: (typeof pendingRemoteEvents)[number],
+    source: "remote_event" | "output_poll" | "output_stream" = "remote_event",
+  ) => {
     if (event.type === "shutdownAck") {
       signalShutdownAcknowledged();
       return;
     }
+    if (event.type === "exit" || event.type === "error") recordTerminalEvidence(source, event);
     if (socket) {
       writeRemoteEventToSocket(event);
       return;
@@ -2248,6 +2276,7 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
     nextSocket.on("close", () => {
       clearTimeout(authTimer);
       liveSockets.delete(nextSocket);
+      if (authenticated && !stopping) recordTerminalEvidence("proxy_close", { type: "error" });
     });
     nextSocket.on("data", (chunk) => {
       connectionBuffer += chunk;
@@ -2304,11 +2333,17 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
             } catch {
               stdinDeliveryFailed = true;
               stopping = true;
+              recordTerminalEvidence("input_delivery", { type: "error" });
               const message = "ACP process session input delivery failed.";
               // Flush the diagnostic before closing; destroy() can discard it
               // and leave only ACP's generic connection_close error. Do not
               // expose provider error text, which may contain a command payload.
-              nextSocket.end(jsonLine({ type: "error", message }));
+              // A remote terminal frame may already be draining while this
+              // accepted stdin write fails. end(data) would write after end
+              // and discard that first failure just like socket.write(data).
+              if (!nextSocket.writableEnded && !nextSocket.destroyed) {
+                nextSocket.end(jsonLine({ type: "error", message }));
+              }
               // stop() awaits this input chain before sending shutdown. Run-log
               // persistence must not hold teardown open when it stalls or fails.
               logFailureWithoutWaiting(`[paperclip] ${message}\n`);
@@ -2340,8 +2375,8 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
       }
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
-      await onLog("stderr", `[paperclip] ACP process session bridge poll failed: ${message}\n`);
-      deliverRemoteEvent({ type: "error", message });
+      // A stuck log sink must not strand the adapter waiting for this failure.
+      deliverRemoteEvent({ type: "error", message }, "output_poll");
       return;
     } finally {
       if (!stopping) {
@@ -2467,14 +2502,15 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
             deliverRemoteEvent({
               type: "exit",
               code: typeof result.exitCode === "number" ? result.exitCode : null,
-            });
+              signal: result.signal,
+            }, "output_stream");
           }
         } catch (error) {
           if (!stopping) {
             deliverRemoteEvent({
               type: "error",
               message: error instanceof Error ? error.message : String(error),
-            });
+            }, "output_stream");
           }
         }
       })();