fix(server): finalize native-passthrough request log on client cancel (#44)

Native-passthrough SSE is inspected on a teed background stream; Codex disconnects
the instant it finishes reading, aborting that stream. consumeForInspection took
the cancelled path and dropped the /api/logs entry (early-abort returned before
onDone; mid-drain abort suppressed onTerminal). Add an onCancel finalizer fired on
both abort branches (idempotent via finalizeNativePassthroughLog's logged guard),
run onDone on early-abort, and stop reportNativeTerminal from downgrading a real
detected terminal to a cancel — so a turn logs completed/failed when it finished,
499 client_cancel only on a pure cancel. Exactly one entry per turn.

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

Author: lidge-jun

src/server.ts

@@ -444,15 +444,22 @@ async function handleResponses(
       linkAbortSignal(upstream, turnAc.signal);
       registerTurn(turnAc);
       if (recordTerminalOutcomes) {
+        // A real terminal was parsed from the (teed) inspection stream — record it as the outcome
+        // even if the client has already disconnected: the turn genuinely reached that terminal, so
+        // it must log as completed/failed, not be dropped or downgraded to a cancel (#44). A pure
+        // client-cancel (no terminal seen) is finalized separately via consumeForInspection's onCancel.
         const reportNativeTerminal = (status: ResponsesTerminalStatus) => {
-          if (options.abortSignal?.aborted) {
-            options.onNativePassthroughCancel?.();
-            return;
-          }
           terminalRecorder?.(status);
           options.onNativePassthroughTerminal?.(status);
         };
-        consumeForInspection(inspectBody, reportNativeTerminal, turnAc.signal, () => unregisterTurn(turnAc), logCtx);
+        consumeForInspection(
+          inspectBody,
+          reportNativeTerminal,
+          turnAc.signal,
+          () => unregisterTurn(turnAc),
+          logCtx,
+          () => options.onNativePassthroughCancel?.(),
+        );
       } else {
         consumeForResponseLogMetadata(inspectBody, logCtx, turnAc.signal, () => unregisterTurn(turnAc));
       }
@@ -1302,12 +1309,13 @@ export function relaySseWithHeartbeat(
  * Background-consume an SSE stream purely for terminal-outcome inspection (quota tracking).
  * Does not produce output; safe to ignore errors (the client-facing stream is separate).
  */
-function consumeForInspection(
+export function consumeForInspection(
   body: ReadableStream<Uint8Array>,
   onTerminal: (status: ResponsesTerminalStatus) => void,
   signal?: AbortSignal,
   onDone?: () => void,
   logCtx?: RequestLogContext,
+  onCancel?: () => void,
 ): void {
   const reader = body.getReader();
   const decoder = new TextDecoder();
@@ -1316,13 +1324,21 @@ function consumeForInspection(
   let cancelled = false;
   if (signal) {
     if (signal.aborted) {
+      // Aborted before we could read anything (Codex disconnects the instant it finishes reading).
+      // Finalize as a client-cancel and release the turn — the early return skips pump()'s finally,
+      // so onDone/onCancel must run here or the entry is silently dropped (#44).
       cancelled = true;
       reader.cancel(signal.reason).catch(() => {});
+      onCancel?.();
+      onDone?.();
       return;
     }
     signal.addEventListener("abort", () => {
+      // Mid-drain disconnect: record a client-cancel entry (idempotent downstream) instead of the
+      // suppressed onTerminal path. onDone still fires via pump()'s finally after the read rejects.
       cancelled = true;
       reader.cancel(signal.reason).catch(() => {});
+      onCancel?.();
     }, { once: true });
   }
   const pump = async () => {

tests/consume-for-inspection-cancel.test.ts

@@ -0,0 +1,51 @@
+import { describe, expect, test } from "bun:test";
+import { consumeForInspection } from "../src/server";
+
+// Regression for issue #44: native-passthrough turns are inspected on a teed background stream.
+// Codex disconnects the instant it finishes reading, so the inspection stream is frequently
+// aborted. The cancel path must finalize (onCancel) and release the turn (onDone) instead of
+// silently dropping the /api/logs entry.
+
+function pendingStream(): ReadableStream<Uint8Array> {
+  // A stream whose read never resolves on its own — only reader.cancel() (via abort) ends it.
+  return new ReadableStream<Uint8Array>({ start() {}, pull() { /* never enqueue/close */ } });
+}
+
+function closingStream(): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({ start(c) { c.close(); } });
+}
+
+const tick = () => new Promise(r => setTimeout(r, 5));
+
+describe("consumeForInspection cancel finalization (#44)", () => {
+  test("already-aborted signal → onCancel + onDone fire, onTerminal does not", () => {
+    const ac = new AbortController();
+    ac.abort();
+    let terminal = 0, cancel = 0, done = 0;
+    consumeForInspection(pendingStream(), () => terminal++, ac.signal, () => done++, undefined, () => cancel++);
+    expect(cancel).toBe(1);
+    expect(done).toBe(1);
+    expect(terminal).toBe(0);
+  });
+
+  test("mid-drain abort → onCancel + onDone fire, onTerminal suppressed", async () => {
+    const ac = new AbortController();
+    let terminal = 0, cancel = 0, done = 0;
+    consumeForInspection(pendingStream(), () => terminal++, ac.signal, () => done++, undefined, () => cancel++);
+    ac.abort();
+    await tick();
+    expect(cancel).toBe(1);
+    expect(done).toBe(1);
+    expect(terminal).toBe(0);
+  });
+
+  test("clean close without a terminal payload → onTerminal(incomplete), not a cancel", async () => {
+    let terminalStatus: string | null = null;
+    let cancel = 0, done = 0;
+    consumeForInspection(closingStream(), s => { terminalStatus = s; }, undefined, () => done++, undefined, () => cancel++);
+    await tick();
+    expect(terminalStatus).toBe("incomplete");
+    expect(cancel).toBe(0);
+    expect(done).toBe(1);
+  });
+});