From 3ed33b4012c4bd465e18eada707c185dce287dff Mon Sep 17 00:00:00 2001 From: chilung Date: Sat, 22 Aug 2026 08:48:06 +0000 Subject: [PATCH 1/3] feat(usage): add durable stream timeline and failure attribution to request history (closes #1217) --- src/usage/log.ts | 92 +++++++++++++++++++++++++++++++++++++++++ tests/usage-log.test.ts | 43 +++++++++++++++++++ 2 files changed, 135 insertions(+) diff --git a/src/usage/log.ts b/src/usage/log.ts index 66654b8b74..fcab3509dc 100644 --- a/src/usage/log.ts +++ b/src/usage/log.ts @@ -13,6 +13,30 @@ import { CODEX_ACCOUNT_LOG_LABEL_RE } from "../codex/account-label"; export type UsageStatus = "reported" | "unreported" | "unsupported" | "estimated"; export type CodexUsageAccountLogLabel = "main" | `p${string}`; +/** + * Bounded stream timing breakdown in elapsed ms from request/attempt start (issue #1217). + * Best-effort correlation metrics for streaming observability. + */ +export interface StreamTimeline { + upstreamDispatchMs?: number; + upstreamHeadersMs?: number; + upstreamFirstByteMs?: number; + upstreamFirstSemanticOutputMs?: number; + downstreamFirstWriteMs?: number; + upstreamEndMs?: number; + downstreamEndMs?: number; +} + +export type FailureSide = "upstream" | "relay" | "downstream" | "client" | "local"; +export type FailureStage = + | "pre_dispatch" + | "upstream_wait_headers" + | "upstream_read" + | "relay_transform" + | "downstream_write" + | "client_cancel" + | "terminal_delivery"; + export function isCodexUsageAccountLogLabel(value: unknown): value is CodexUsageAccountLogLabel { return value === "main" || (typeof value === "string" && CODEX_ACCOUNT_LOG_LABEL_RE.test(value)); } @@ -65,6 +89,10 @@ export interface PersistedUsageAttempt { reasoningWireValue?: string | number | boolean; /** Adapter-produced tier fact for this physical attempt; absent on pre-B0 rows. */ tierOutcome?: AttemptTierOutcome; + /** Bounded streaming timeline for this attempt (issue #1217). */ + streamTimeline?: StreamTimeline; + failureSide?: FailureSide; + failureStage?: FailureStage; } export interface PersistedUsageEntry { @@ -118,6 +146,14 @@ 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; + /** + * Bounded streaming timeline and causal failure attribution (issue #1217). + */ + streamTimeline?: StreamTimeline; + failureSide?: FailureSide; + failureStage?: FailureStage; + transportPhase?: string; + terminalSource?: string; /** * Bounded route-decision trace (RI-01): why this provider/model/account was * selected. Additive field; old rows without it parse unchanged. Never @@ -238,6 +274,38 @@ const TIER_CONFIRMATIONS = new Set([ const FAST_DOWNGRADE_REASONS = new Set>([ "route-unsupported", "wire-unavailable", "response-declined", ]); +const KNOWN_FAILURE_SIDES = new Set([ + "upstream", "relay", "downstream", "client", "local", +]); +const KNOWN_FAILURE_STAGES = new Set([ + "pre_dispatch", + "upstream_wait_headers", + "upstream_read", + "relay_transform", + "downstream_write", + "client_cancel", + "terminal_delivery", +]); + +function normalizeStreamTimeline(raw: unknown): StreamTimeline | null { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; + const t = raw as Record; + const out: StreamTimeline = {}; + for (const key of [ + "upstreamDispatchMs", + "upstreamHeadersMs", + "upstreamFirstByteMs", + "upstreamFirstSemanticOutputMs", + "downstreamFirstWriteMs", + "upstreamEndMs", + "downstreamEndMs", + ] as const) { + if (key in t && isNonNegativeFiniteNumber(t[key])) { + out[key] = t[key] as number; + } + } + return Object.keys(out).length > 0 ? out : null; +} export function isLabRouteSubjectId(value: unknown): value is string { return typeof value === "string" && LAB_ROUTE_SUBJECT_ID_RE.test(value); @@ -392,6 +460,15 @@ function normalizeUsageAttempt(raw: unknown): PersistedUsageAttempt | null { : { reasoningWireValue: attempt.reasoningWireValue } : {}), ...(tierOutcome ? { tierOutcome } : {}), + ...(normalizeStreamTimeline(attempt.streamTimeline) + ? { streamTimeline: normalizeStreamTimeline(attempt.streamTimeline)! } + : {}), + ...(typeof attempt.failureSide === "string" && KNOWN_FAILURE_SIDES.has(attempt.failureSide as FailureSide) + ? { failureSide: attempt.failureSide as FailureSide } + : {}), + ...(typeof attempt.failureStage === "string" && KNOWN_FAILURE_STAGES.has(attempt.failureStage as FailureStage) + ? { failureStage: attempt.failureStage as FailureStage } + : {}), }; } @@ -504,6 +581,21 @@ function normalizeUsageEntry(entry: PersistedUsageEntry): PersistedUsageEntry { ...(entry.terminalStatus ? { terminalStatus: entry.terminalStatus } : {}), ...(entry.closeReason ? { closeReason: entry.closeReason } : {}), ...(entry.upstreamError ? { upstreamError: entry.upstreamError } : {}), + ...(normalizeStreamTimeline(entry.streamTimeline) + ? { streamTimeline: normalizeStreamTimeline(entry.streamTimeline)! } + : {}), + ...(typeof entry.failureSide === "string" && KNOWN_FAILURE_SIDES.has(entry.failureSide as FailureSide) + ? { failureSide: entry.failureSide as FailureSide } + : {}), + ...(typeof entry.failureStage === "string" && KNOWN_FAILURE_STAGES.has(entry.failureStage as FailureStage) + ? { failureStage: entry.failureStage as FailureStage } + : {}), + ...(typeof entry.transportPhase === "string" && entry.transportPhase + ? { transportPhase: capMetadataString(entry.transportPhase) } + : {}), + ...(typeof entry.terminalSource === "string" && entry.terminalSource + ? { terminalSource: capMetadataString(entry.terminalSource) } + : {}), ...(routeDecision ? { routeDecision } : {}), }; } diff --git a/tests/usage-log.test.ts b/tests/usage-log.test.ts index 8414287b01..c93b7eafee 100644 --- a/tests/usage-log.test.ts +++ b/tests/usage-log.test.ts @@ -882,4 +882,47 @@ describe("usage log", () => { expect(readRecentUsageEntries(1)).toEqual([]); }, STORE_BUDGET_MS); + + test("normalizes and preserves streamTimeline and failure attribution (#1217)", () => { + appendUsageEntry({ + requestId: "ocx-stream-timeline-test", + timestamp: Date.now(), + provider: "anthropic", + model: "claude-sonnet-5", + status: 502, + durationMs: 61342, + firstOutputMs: 9107, + usageStatus: "unreported", + streamTimeline: { + upstreamDispatchMs: 12, + upstreamHeadersMs: 4410, + upstreamFirstByteMs: 4421, + upstreamFirstSemanticOutputMs: 9107, + downstreamFirstWriteMs: 4423, + upstreamEndMs: 61340, + downstreamEndMs: 61342, + }, + failureSide: "upstream", + failureStage: "upstream_read", + transportPhase: "mid_stream", + terminalSource: "synthetic", + }); + + const entries = readRecentUsageEntries(10); + const row = entries.find(e => e.requestId === "ocx-stream-timeline-test"); + expect(row).toBeDefined(); + expect(row?.streamTimeline).toEqual({ + upstreamDispatchMs: 12, + upstreamHeadersMs: 4410, + upstreamFirstByteMs: 4421, + upstreamFirstSemanticOutputMs: 9107, + downstreamFirstWriteMs: 4423, + upstreamEndMs: 61340, + downstreamEndMs: 61342, + }); + expect(row?.failureSide).toBe("upstream"); + expect(row?.failureStage).toBe("upstream_read"); + expect(row?.transportPhase).toBe("mid_stream"); + expect(row?.terminalSource).toBe("synthetic"); + }); }); From 464a6d9c11b90070cdebffb9842430a6a116d99a Mon Sep 17 00:00:00 2001 From: chilung Date: Sat, 22 Aug 2026 10:39:48 +0000 Subject: [PATCH 2/3] fix(usage): clean up stream timeline normalization (#1217) --- src/usage/log.ts | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/src/usage/log.ts b/src/usage/log.ts index fcab3509dc..5ea616f614 100644 --- a/src/usage/log.ts +++ b/src/usage/log.ts @@ -460,9 +460,7 @@ function normalizeUsageAttempt(raw: unknown): PersistedUsageAttempt | null { : { reasoningWireValue: attempt.reasoningWireValue } : {}), ...(tierOutcome ? { tierOutcome } : {}), - ...(normalizeStreamTimeline(attempt.streamTimeline) - ? { streamTimeline: normalizeStreamTimeline(attempt.streamTimeline)! } - : {}), + ...(normalizeStreamTimeline(attempt.streamTimeline) ? { streamTimeline: normalizeStreamTimeline(attempt.streamTimeline) as StreamTimeline } : {}), ...(typeof attempt.failureSide === "string" && KNOWN_FAILURE_SIDES.has(attempt.failureSide as FailureSide) ? { failureSide: attempt.failureSide as FailureSide } : {}), @@ -581,9 +579,7 @@ function normalizeUsageEntry(entry: PersistedUsageEntry): PersistedUsageEntry { ...(entry.terminalStatus ? { terminalStatus: entry.terminalStatus } : {}), ...(entry.closeReason ? { closeReason: entry.closeReason } : {}), ...(entry.upstreamError ? { upstreamError: entry.upstreamError } : {}), - ...(normalizeStreamTimeline(entry.streamTimeline) - ? { streamTimeline: normalizeStreamTimeline(entry.streamTimeline)! } - : {}), + ...(normalizeStreamTimeline(entry.streamTimeline) ? { streamTimeline: normalizeStreamTimeline(entry.streamTimeline) as StreamTimeline } : {}), ...(typeof entry.failureSide === "string" && KNOWN_FAILURE_SIDES.has(entry.failureSide as FailureSide) ? { failureSide: entry.failureSide as FailureSide } : {}), From f50ad03b1147d610d12c652e1b6b8f7ff56a2203 Mon Sep 17 00:00:00 2001 From: chilung Date: Sat, 22 Aug 2026 11:44:08 +0000 Subject: [PATCH 3/3] fix(usage): enforce strict validation on transportPhase and terminalSource (#1217) --- src/usage/log.ts | 31 ++++++++++++++++++++++------ tests/usage-log.test.ts | 45 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 6 deletions(-) diff --git a/src/usage/log.ts b/src/usage/log.ts index 5ea616f614..cfb87c3ece 100644 --- a/src/usage/log.ts +++ b/src/usage/log.ts @@ -36,6 +36,8 @@ export type FailureStage = | "downstream_write" | "client_cancel" | "terminal_delivery"; +export type TransportPhase = "pre_headers" | "mid_stream" | "terminal_sse"; +export type TerminalSource = "upstream" | "synthetic"; export function isCodexUsageAccountLogLabel(value: unknown): value is CodexUsageAccountLogLabel { return value === "main" || (typeof value === "string" && CODEX_ACCOUNT_LOG_LABEL_RE.test(value)); @@ -93,6 +95,8 @@ export interface PersistedUsageAttempt { streamTimeline?: StreamTimeline; failureSide?: FailureSide; failureStage?: FailureStage; + transportPhase?: TransportPhase; + terminalSource?: TerminalSource; } export interface PersistedUsageEntry { @@ -152,8 +156,8 @@ export interface PersistedUsageEntry { streamTimeline?: StreamTimeline; failureSide?: FailureSide; failureStage?: FailureStage; - transportPhase?: string; - terminalSource?: string; + transportPhase?: TransportPhase; + terminalSource?: TerminalSource; /** * Bounded route-decision trace (RI-01): why this provider/model/account was * selected. Additive field; old rows without it parse unchanged. Never @@ -286,6 +290,15 @@ const KNOWN_FAILURE_STAGES = new Set([ "client_cancel", "terminal_delivery", ]); +const KNOWN_TRANSPORT_PHASES = new Set([ + "pre_headers", + "mid_stream", + "terminal_sse", +]); +const KNOWN_TERMINAL_SOURCES = new Set([ + "upstream", + "synthetic", +]); function normalizeStreamTimeline(raw: unknown): StreamTimeline | null { if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; @@ -467,6 +480,12 @@ function normalizeUsageAttempt(raw: unknown): PersistedUsageAttempt | null { ...(typeof attempt.failureStage === "string" && KNOWN_FAILURE_STAGES.has(attempt.failureStage as FailureStage) ? { failureStage: attempt.failureStage as FailureStage } : {}), + ...(typeof attempt.transportPhase === "string" && KNOWN_TRANSPORT_PHASES.has(attempt.transportPhase as TransportPhase) + ? { transportPhase: attempt.transportPhase as TransportPhase } + : {}), + ...(typeof attempt.terminalSource === "string" && KNOWN_TERMINAL_SOURCES.has(attempt.terminalSource as TerminalSource) + ? { terminalSource: attempt.terminalSource as TerminalSource } + : {}), }; } @@ -586,11 +605,11 @@ function normalizeUsageEntry(entry: PersistedUsageEntry): PersistedUsageEntry { ...(typeof entry.failureStage === "string" && KNOWN_FAILURE_STAGES.has(entry.failureStage as FailureStage) ? { failureStage: entry.failureStage as FailureStage } : {}), - ...(typeof entry.transportPhase === "string" && entry.transportPhase - ? { transportPhase: capMetadataString(entry.transportPhase) } + ...(typeof entry.transportPhase === "string" && KNOWN_TRANSPORT_PHASES.has(entry.transportPhase as TransportPhase) + ? { transportPhase: entry.transportPhase as TransportPhase } : {}), - ...(typeof entry.terminalSource === "string" && entry.terminalSource - ? { terminalSource: capMetadataString(entry.terminalSource) } + ...(typeof entry.terminalSource === "string" && KNOWN_TERMINAL_SOURCES.has(entry.terminalSource as TerminalSource) + ? { terminalSource: entry.terminalSource as TerminalSource } : {}), ...(routeDecision ? { routeDecision } : {}), }; diff --git a/tests/usage-log.test.ts b/tests/usage-log.test.ts index c93b7eafee..1982ef202e 100644 --- a/tests/usage-log.test.ts +++ b/tests/usage-log.test.ts @@ -925,4 +925,49 @@ describe("usage log", () => { expect(row?.transportPhase).toBe("mid_stream"); expect(row?.terminalSource).toBe("synthetic"); }); + + test("drops unknown or invalid transportPhase and terminalSource (#1217)", () => { + appendUsageEntry({ + requestId: "ocx-stream-invalid-attribution-test", + timestamp: Date.now(), + provider: "anthropic", + model: "claude-sonnet-5", + status: 502, + durationMs: 1000, + usageStatus: "unreported", + failureSide: "invalid_side" as unknown as any, + failureStage: "invalid_stage" as unknown as any, + transportPhase: "invalid_phase" as unknown as any, + terminalSource: "invalid_source" as unknown as any, + attempts: [ + { + ordinal: 1, + adapter: "anthropic", + sendCount: 1, + usageStatus: "unreported", + timestamp: Date.now(), + provider: "anthropic", + model: "claude-sonnet-5", + status: 502, + durationMs: 1000, + failureSide: "bogus_side" as unknown as any, + failureStage: "bogus_stage" as unknown as any, + transportPhase: "bogus_phase" as unknown as any, + terminalSource: "bogus_source" as unknown as any, + }, + ], + }); + + const entries = readRecentUsageEntries(10); + const row = entries.find(e => e.requestId === "ocx-stream-invalid-attribution-test"); + expect(row).toBeDefined(); + expect(row?.failureSide).toBeUndefined(); + expect(row?.failureStage).toBeUndefined(); + expect(row?.transportPhase).toBeUndefined(); + expect(row?.terminalSource).toBeUndefined(); + expect(row?.attempts?.[0].failureSide).toBeUndefined(); + expect(row?.attempts?.[0].failureStage).toBeUndefined(); + expect(row?.attempts?.[0].transportPhase).toBeUndefined(); + expect(row?.attempts?.[0].terminalSource).toBeUndefined(); + }); });