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/server/relay-eager.test.ts b/tests/server/relay-eager.test.ts index e7be8ba57c..ad4d4e458e 100644 --- a/tests/server/relay-eager.test.ts +++ b/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)", () => { 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); + }); }); 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",