From b11cd8438995349731797851e2c6c670b8126cc2 Mon Sep 17 00:00:00 2001 From: chilung Date: Tue, 8 Sep 2026 19:52:05 +0000 Subject: [PATCH 1/3] test(stream): prove bare upstream error at EOF meters failed without streamAborted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add two groups of regression tests for the usage-marker parity criterion in devlog/_plan/260906_opaque_transport_finality/010_failure_boundaries.md: tee path (consumeForInspection / stream-aborted-marker.test.ts): - bare { type: "error" } SSE event followed by clean EOF → onTerminal("failed", 502) with no streamAborted on the persisted attempt (semantic failure, not a read reset) - read error following a bare error event → streamAborted stays true (onReadError path) eager path (relaySseEagerBounded / relay-eager.test.ts, 090-13): - bare error frame + clean upstream EOF → onSynthetic receives ("failed", "upstream_error") so core.ts onSynthetic can record a semantic failed status without streamAborted No source changes; all existing tests continue to pass. The implementation already handles these paths correctly in relay.ts and relay-eager.ts; these tests make the contract explicit and regression-proof. --- tests/server/relay-eager.test.ts | 31 +++++++++ tests/server/stream-aborted-marker.test.ts | 75 ++++++++++++++++++++++ 2 files changed, 106 insertions(+) diff --git a/tests/server/relay-eager.test.ts b/tests/server/relay-eager.test.ts index e7be8ba57c..09346a2672 100644 --- a/tests/server/relay-eager.test.ts +++ b/tests/server/relay-eager.test.ts @@ -1617,6 +1617,37 @@ 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).toContain("event: response.failed"); + expect(out).toContain("provider stream failed"); + expect(syntheticCalls).toEqual([["failed", "upstream_error"]]); + expect(rec090.dones).toBe(1); + }); }); describe("createSseInspector — extraction locks (h)", () => { diff --git a/tests/server/stream-aborted-marker.test.ts b/tests/server/stream-aborted-marker.test.ts index 5e142a3429..cce7be1faf 100644 --- a/tests/server/stream-aborted-marker.test.ts +++ b/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(); + 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({ + 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(); + 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({ + 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); + }); }); From e12dfb163012766814bc8264a26b049a1ca01861 Mon Sep 17 00:00:00 2001 From: chilung Date: Tue, 8 Sep 2026 19:59:21 +0000 Subject: [PATCH 2/3] fix(logs): persist transport finality evidence Keep transport phase and terminal source when request-log rows are written to usage.jsonl and rehydrated after restart, so synthetic relay failures remain distinguishable from upstream terminals. Closes #3657 --- src/server/request-log.ts | 6 ++++ src/usage/log.ts | 24 +++++++++++++++ tests/usage/request-log.test.ts | 53 +++++++++++++++++++++++++++++++++ 3 files changed, 83 insertions(+) diff --git a/src/server/request-log.ts b/src/server/request-log.ts index 6fc90aab74..823dad9483 100644 --- a/src/server/request-log.ts +++ b/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 } : {}), diff --git a/src/usage/log.ts b/src/usage/log.ts index fd8408cc19..2944c22f9a 100644 --- a/src/usage/log.ts +++ b/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); } +const KNOWN_TRANSPORT_PHASES = new Set>([ + "pre_headers", "mid_stream", "terminal_sse", +]); + +export function isKnownTransportPhase(value: unknown): value is NonNullable { + return typeof value === "string" && KNOWN_TRANSPORT_PHASES.has(value as NonNullable); +} + +const KNOWN_TERMINAL_SOURCES = new Set>([ + "upstream", "synthetic", +]); + +export function isKnownTerminalSource(value: unknown): value is NonNullable { + return typeof value === "string" && KNOWN_TERMINAL_SOURCES.has(value as NonNullable); +} + 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 } : {}), diff --git a/tests/usage/request-log.test.ts b/tests/usage/request-log.test.ts index 33fbd13b1f..6080e52ceb 100644 --- a/tests/usage/request-log.test.ts +++ b/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", From 6bd3274ab2134ceb42f34fd904820330a6b32f6d Mon Sep 17 00:00:00 2001 From: chilung Date: Tue, 8 Sep 2026 20:22:50 +0000 Subject: [PATCH 3/3] test(server): verify exactly one failed terminal on upstream error --- tests/server/relay-eager.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/server/relay-eager.test.ts b/tests/server/relay-eager.test.ts index 09346a2672..ad4d4e458e 100644 --- a/tests/server/relay-eager.test.ts +++ b/tests/server/relay-eager.test.ts @@ -1643,7 +1643,8 @@ describe("relaySseEagerBounded — error paths", () => { const out = await readAll(relayed); await settle(); - expect(out).toContain("event: response.failed"); + 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);