fix(logs): persist transport finality evidence

src/server/request-log.ts

@@ -22,6 +22,8 @@ import {
   appendUsageEntry,
   isKnownAdmissionKind,
   isKnownInboundProtocol,
+  isKnownTerminalSource,
+  isKnownTransportPhase,
   isKnownUsageSurface,
   isCodexUsageAccountLogLabel,
   isValidReasoningWireValue,
@@ -320,6 +322,8 @@ export function requestLogEntryFromPersistedUsage(entry: PersistedUsageEntry): R
     ...(entry.usage ? { usage: entry.usage } : {}),
     ...(entry.totalTokens !== undefined ? { totalTokens: entry.totalTokens } : {}),
     ...(entry.attempts !== undefined ? { attempts: entry.attempts } : {}),
+    ...(isKnownTransportPhase(entry.transportPhase) ? { transportPhase: entry.transportPhase } : {}),
+    ...(isKnownTerminalSource(entry.terminalSource) ? { terminalSource: entry.terminalSource } : {}),
     ...(routeDecision ? { routeDecision } : {}),
     ...(claudeCompatibility ? { claudeCompatibility } : {}),
   };
@@ -441,6 +445,8 @@ export function addRequestLog(entry: RequestLogEntry) {
       ...(entry.usage ? { usage: entry.usage } : {}),
       ...(entry.totalTokens !== undefined ? { totalTokens: entry.totalTokens } : {}),
       ...(entry.attempts !== undefined ? { attempts: entry.attempts } : {}),
+      ...(isKnownTransportPhase(entry.transportPhase) ? { transportPhase: entry.transportPhase } : {}),
+      ...(isKnownTerminalSource(entry.terminalSource) ? { terminalSource: entry.terminalSource } : {}),
       ...failureDiagnostics,
       ...(entry.routeDecision ? { routeDecision: entry.routeDecision } : {}),
       ...(entry.claudeCompatibility ? { claudeCompatibility: entry.claudeCompatibility } : {}),

src/usage/log.ts

@@ -172,6 +172,10 @@ export interface PersistedUsageEntry {
   closeReason?: "terminal" | "client_cancel" | "non_stream" | "body_stall" | "body_overflow";
   /** Already redacted + capped at capture (request-log.ts redactSecretString().slice(0,500)). */
   upstreamError?: string;
+  /** Where the terminal/failure was observed; absent on historic rows. */
+  transportPhase?: "pre_headers" | "mid_stream" | "terminal_sse";
+  /** Whether the terminal came from upstream or a proxy-generated tail. */
+  terminalSource?: "upstream" | "synthetic";
   /**
    * Bounded route-decision trace (RI-01): why this provider/model/account was
    * selected. Additive field; old rows without it parse unchanged. Never
@@ -217,6 +221,22 @@ export function isKnownInboundProtocol(value: unknown): value is NonNullable<Per
   return typeof value === "string" && KNOWN_INBOUND_PROTOCOLS.has(value as NonNullable<PersistedUsageEntry["inboundProtocol"]>);
 }
 
+const KNOWN_TRANSPORT_PHASES = new Set<NonNullable<PersistedUsageEntry["transportPhase"]>>([
+  "pre_headers", "mid_stream", "terminal_sse",
+]);
+
+export function isKnownTransportPhase(value: unknown): value is NonNullable<PersistedUsageEntry["transportPhase"]> {
+  return typeof value === "string" && KNOWN_TRANSPORT_PHASES.has(value as NonNullable<PersistedUsageEntry["transportPhase"]>);
+}
+
+const KNOWN_TERMINAL_SOURCES = new Set<NonNullable<PersistedUsageEntry["terminalSource"]>>([
+  "upstream", "synthetic",
+]);
+
+export function isKnownTerminalSource(value: unknown): value is NonNullable<PersistedUsageEntry["terminalSource"]> {
+  return typeof value === "string" && KNOWN_TERMINAL_SOURCES.has(value as NonNullable<PersistedUsageEntry["terminalSource"]>);
+}
+
 export function usageLogPath(configDir?: string): string {
   return join(configDir ?? getConfigDir(), "usage.jsonl");
 }
@@ -511,6 +531,8 @@ function normalizeUsageEntry(entry: PersistedUsageEntry): PersistedUsageEntry {
   const responseServiceTier = sanitizeLogMetadataString(entry.responseServiceTier);
   const shadowCallRewrittenFrom = sanitizeLogMetadataString(entry.shadowCallRewrittenFrom);
   const claudeCompatibility = normalizeClaudeCompatibilityUsageLog(entry.claudeCompatibility);
+  const transportPhase = isKnownTransportPhase(entry.transportPhase) ? entry.transportPhase : undefined;
+  const terminalSource = isKnownTerminalSource(entry.terminalSource) ? entry.terminalSource : undefined;
   const routeDecision = entry.routeDecision
     ? normalizeRouteDecisionTrace(entry.routeDecision)
     : undefined;
@@ -579,6 +601,8 @@ function normalizeUsageEntry(entry: PersistedUsageEntry): PersistedUsageEntry {
     ...(entry.usage ? { usage: normalizeUsageValue(entry.usage) } : {}),
     ...(typeof entry.totalTokens === "number" ? { totalTokens: entry.totalTokens } : {}),
     ...(Array.isArray(entry.attempts) ? { attempts } : {}),
+    ...(transportPhase ? { transportPhase } : {}),
+    ...(terminalSource ? { terminalSource } : {}),
     ...(entry.errorCode ? { errorCode: entry.errorCode } : {}),
     ...(entry.terminalStatus ? { terminalStatus: entry.terminalStatus } : {}),
     ...(entry.closeReason ? { closeReason: entry.closeReason } : {}),

tests/server/relay-eager.test.ts

@@ -1617,6 +1617,38 @@ describe("relaySseEagerBounded — error paths", () => {
     expect(rec.synthetics).toEqual([]);
     expect(rec.dones).toBe(1);
   });
+
+  test("(090-13) bare upstream error event at clean EOF passes upstream_error reason to onSynthetic", async () => {
+    // A { type: "error" } bare error SSE frame arrives, then the upstream closes cleanly.
+    // The relay emits an upstreamErrorTailFrame rather than an adapterEofIncompleteFrame.
+    // onSynthetic must receive reason="upstream_error" so callers can distinguish a semantic
+    // upstream failure from a plain body-read reset (which carries no reason argument).
+    const up = controlledUpstream();
+    const syntheticCalls: Array<[string, string | undefined]> = [];
+    const rec090 = { dones: 0 };
+    const inspector090 = createSseInspector({});
+    const hooks090: EagerRelayHooks = {
+      inspectChunk: c => inspector090.feed(c),
+      finishInspection: () => inspector090.finish(),
+      disposeInspection: () => inspector090.dispose(),
+      sawTerminal: () => inspector090.reported(),
+      onSynthetic: (kind, reason) => syntheticCalls.push([kind, reason]),
+      onClientCancel: () => {},
+      onDone: () => { rec090.dones += 1; },
+    };
+    const relayed = relaySseEagerBounded(up.stream, new AbortController(), hooks090);
+    const bareErrorPayload = JSON.stringify({ type: "error", message: "provider stream failed" });
+    up.push(sse(bareErrorPayload));
+    up.close();
+    const out = await readAll(relayed);
+    await settle();
+
+    expect(out.match(/event: response\.failed/g)?.length).toBe(1);
+    expect(out).not.toContain("response.incomplete");
+    expect(out).toContain("provider stream failed");
+    expect(syntheticCalls).toEqual([["failed", "upstream_error"]]);
+    expect(rec090.dones).toBe(1);
+  });
 });
 
 describe("createSseInspector — extraction locks (h)", () => {

tests/server/stream-aborted-marker.test.ts

@@ -205,4 +205,79 @@ describe("streamAborted marker (codex-router #139)", () => {
     expect(row?.status).toBe(499);
     expect(row?.attempts?.[0]?.streamAborted).toBeUndefined();
   });
+
+  test("bare upstream error event at clean EOF meters as 502 without streamAborted", async () => {
+    const { logCtx, attempt } = makeLogCtx();
+    const terminalReported = Promise.withResolvers<void>();
+    const terminals: Array<[string, number | undefined]> = [];
+    // Stream sends a bare { type: "error" } SSE event then closes cleanly (no read error).
+    // The onCleanEof path in consumeForInspection detects the witnessed bare error and
+    // reports failed -- but a semantic EOF is not a body-read reset, so streamAborted is absent.
+    const barePayload = JSON.stringify({ type: "error", message: "provider failed cleanly" });
+    const body = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        controller.enqueue(encoder.encode("data: " + barePayload + "\n\n"));
+        controller.close();
+      },
+    });
+    consumeForInspection(
+      body,
+      (status, httpStatusOverride) => {
+        terminals.push([status, httpStatusOverride]);
+        terminalReported.resolve();
+      },
+      undefined,
+      () => {},
+      logCtx,
+    );
+    await terminalReported.promise;
+    expect(terminals).toEqual([["failed", 502]]);
+    expect(attempt.streamAborted).toBeUndefined();
+
+    addFinalRequestLog(
+      "ocx-bare-error-eof",
+      Date.now(),
+      logCtx,
+      httpStatusForRequestLogTerminal("failed", logCtx),
+      { terminalStatus: "failed", closeReason: "terminal" },
+      addRequestLog,
+    );
+    const [row] = readUsageEntries();
+    expect(row?.status).toBe(502);
+    expect(row?.attempts?.[0]?.status).toBe(502);
+    expect(row?.attempts?.[0]?.streamAborted).toBeUndefined();
+  });
+
+  test("read error after a bare upstream error event carries streamAborted", async () => {
+    const { logCtx, attempt } = makeLogCtx();
+    const terminalReported = Promise.withResolvers<void>();
+    const terminals: Array<[string, number | undefined]> = [];
+    // A bare error event arrives, then the body-read itself fails (socket reset).
+    // The read error takes the onReadError path and sets streamAborted.
+    const barePayload = JSON.stringify({ type: "error", message: "pre-reset error" });
+    let reads = 0;
+    const body = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        reads += 1;
+        if (reads === 1) {
+          controller.enqueue(encoder.encode("data: " + barePayload + "\n\n"));
+        } else {
+          controller.error(new Error("socket reset after error event"));
+        }
+      },
+    });
+    consumeForInspection(
+      body,
+      (status, httpStatusOverride) => {
+        terminals.push([status, httpStatusOverride]);
+        terminalReported.resolve();
+      },
+      undefined,
+      () => {},
+      logCtx,
+    );
+    await terminalReported.promise;
+    expect(terminals).toEqual([["failed", 502]]);
+    expect(attempt.streamAborted).toBe(true);
+  });
 });

tests/usage/request-log.test.ts

@@ -32,6 +32,7 @@ import { bridgeToResponsesSSE } from "../../src/bridge";
 import type { AdapterEvent, OcxConfig, OcxUsage } from "../../src/types";
 import {
   appendUsageEntry,
+  normalizeUsageEntryForTest,
   readUsageEntries,
   resetUsageReadCacheForTests,
   type PersistedUsageEntry,
@@ -447,6 +448,31 @@ describe("request log metadata", () => {
     }
   });
 
+  test("persists transport finality evidence from the final request log", () => {
+    const home = mkdtempSync(join(tmpdir(), "ocx-finality-usage-"));
+    const previousHome = process.env.OPENCODEX_HOME;
+    process.env.OPENCODEX_HOME = home;
+    try {
+      clearRequestLogsForTests();
+      resetUsageReadCacheForTests();
+      addFinalRequestLog("ocx-finality-persist", 1, {
+        model: "gpt-6-astra",
+        provider: "openai",
+        transportPhase: "mid_stream",
+        terminalSource: "synthetic",
+        upstreamError: "synthetic terminal",
+      }, 502, { terminalStatus: "failed", closeReason: "terminal" });
+      expect(getRequestLogEntries()[0]).toMatchObject({ transportPhase: "mid_stream", terminalSource: "synthetic" });
+      expect(readUsageEntries()[0]).toMatchObject({ transportPhase: "mid_stream", terminalSource: "synthetic" });
+    } finally {
+      clearRequestLogsForTests();
+      resetUsageReadCacheForTests();
+      if (previousHome === undefined) delete process.env.OPENCODEX_HOME;
+      else process.env.OPENCODEX_HOME = previousHome;
+      removeTreeWithRetry(home);
+    }
+  });
+
   // The value is caller-controlled, so proving it lands is only half the contract: the
   // persistence path must also be the SANITIZED one. A test that only ever writes a safe
   // short slug passes identically whether `sanitizeLogMetadataString` is applied or not.
@@ -1760,6 +1786,33 @@ describe("request log metadata", () => {
 });
 
 describe("request log restart hydrate", () => {
+  test("persists and rehydrates transport finality evidence", () => {
+    const persisted = {
+      requestId: "ocx-finality-evidence",
+      timestamp: 1_800_000_000_000,
+      provider: "openai",
+      model: "gpt-6-astra",
+      status: 502,
+      durationMs: 42,
+      usageStatus: "unreported",
+      errorCode: "upstream_server_error",
+      terminalStatus: "failed",
+      closeReason: "terminal",
+      upstreamError: "upstream failed",
+      transportPhase: "mid_stream",
+      terminalSource: "synthetic",
+    } as PersistedUsageEntry;
+
+    expect(normalizeUsageEntryForTest(persisted)).toMatchObject({
+      transportPhase: "mid_stream",
+      terminalSource: "synthetic",
+    });
+    expect(requestLogEntryFromPersistedUsage(persisted)).toMatchObject({
+      transportPhase: "mid_stream",
+      terminalSource: "synthetic",
+    });
+  });
+
   test("projects persisted usage rows into /api/logs entries", () => {
     const persisted: PersistedUsageEntry = {
       requestId: "ocx-revive",