feat(usage): add durable stream timeline and failure attribution to request history

src/server/index.ts

@@ -1217,6 +1217,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "unknown",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
           inboundProtocol: "responses",
         };
@@ -1251,6 +1252,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "image_gen",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
         };
         const endpoint = url.pathname.endsWith("/edits") ? "edits" as const : "generations" as const;
@@ -1306,6 +1308,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "web_search",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
         };
         return runAdmittedHttpTurn(req, policy, async turnAdmissionLease => {
@@ -1330,6 +1333,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "unknown",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
           inboundProtocol: "responses",
         };
@@ -1401,6 +1405,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "unknown",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
           inboundProtocol: "messages",
         };
@@ -1431,6 +1436,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "unknown",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
           inboundProtocol: "chat",
         };
@@ -1462,6 +1468,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "gpt-live",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
         };
         return runAdmittedHttpTurn(req, policy, async turnAdmissionLease => {
@@ -1499,6 +1506,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "gpt-live",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
         };
         const turnAdmissionLease = tryAdmitTurn();
@@ -1679,6 +1687,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
           const logCtx: RequestLogContext = {
             model: "unknown",
             provider: "unknown",
+            requestStartedAt: start,
             ...(wsAdmission ? admissionFields(wsAdmission) : {}),
             inboundProtocol: "responses",
           };

src/server/relay.ts

@@ -7,6 +7,7 @@ import {
   httpStatusForRequestLogTerminal,
   inspectResponseLogJson,
   inspectResponseLogSsePayloadParsed,
+  noteStreamTimelineEvent,
   recordFirstOutput,
   type RequestLogContext,
   type RequestLogEntry,
@@ -424,6 +425,7 @@ export function responseWithDeferredRequestLog(
   logCtx: RequestLogContext,
   addLog: (entry: RequestLogEntry) => void = addRequestLog,
 ): Response {
+  if (logCtx.requestStartedAt === undefined) logCtx.requestStartedAt = start;
   const contentType = response.headers.get("content-type")?.toLowerCase() ?? "";
   if (isUsageDebugEnabled() && !logCtx.usageDebugContentType && contentType) {
     logCtx.usageDebugContentType = contentType;
@@ -799,6 +801,9 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
       try { handlers.onParsedPayload(parsed); } catch { /* inspection must never throw into the pump */ }
     }
     reportFirstOutput.parsed(parsed);
+    if (handlers.logCtx && handlers.logCtx.firstOutputMs !== undefined) {
+      noteStreamTimelineEvent(handlers.logCtx, "upstreamFirstSemanticOutputMs");
+    }
     const status = terminalStatusFromParsed(parsed);
     if (status) sawTerminal = true;
     if (!reported && handlers.onTerminal && status) {
@@ -807,6 +812,10 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
         if (handlers.logCtx) {
           handlers.logCtx.transportPhase = "terminal_sse";
           handlers.logCtx.terminalSource = "upstream";
+          if (status === "failed") {
+            handlers.logCtx.failureSide = "upstream";
+            handlers.logCtx.failureStage = "terminal_delivery";
+          }
         }
         handlers.onTerminal(status);
       } finally {
@@ -933,7 +942,11 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
 
   return {
     feed(chunk) {
-      if (!disposed) scanChunk(chunk);
+      if (disposed) return;
+      if (chunk.byteLength > 0 && handlers.logCtx) {
+        noteStreamTimelineEvent(handlers.logCtx, "upstreamFirstByteMs");
+      }
+      scanChunk(chunk);
     },
     finish() {
       if (disposed) return;
@@ -959,6 +972,7 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
 export type InspectionDrainBounds = { ms: number; bytes: number };
 
 export type InspectionConsumerOptions = {
+  requestStartedAt?: number;
   clientGoneSignal?: AbortSignal;
   drainBounds?: Partial<InspectionDrainBounds>;
   upstream?: AbortController;
@@ -1117,6 +1131,9 @@ export function consumeForInspection(
   onFirstOutput?: () => void,
   options?: InspectionConsumerOptions,
 ): void {
+  if (logCtx && options?.requestStartedAt !== undefined && logCtx.requestStartedAt === undefined) {
+    logCtx.requestStartedAt = options.requestStartedAt;
+  }
   const reader = body.getReader();
   const inspector = (options?.inspectorFactory ?? createSseInspector)({
     onTerminal,
@@ -1135,7 +1152,12 @@ export function consumeForInspection(
     onCancel,
     onCleanEof: () => {
       if (!inspector.reported()) {
-        if (logCtx) logCtx.terminalSource = "synthetic";
+        if (logCtx) {
+          logCtx.transportPhase = "mid_stream";
+          logCtx.terminalSource = "synthetic";
+          logCtx.failureSide = "upstream";
+          logCtx.failureStage = "upstream_read";
+        }
         onTerminal("incomplete");
       }
     },
@@ -1147,6 +1169,8 @@ export function consumeForInspection(
         if (logCtx) {
           logCtx.transportPhase = "mid_stream";
           logCtx.terminalSource = "synthetic";
+          logCtx.failureSide = "upstream";
+          logCtx.failureStage = "upstream_read";
           // A truncated 200 body must not meter as a success the client never
           // received; the router's equivalent turn carries 502 + streamAborted
           // (codex-router #139).
@@ -1167,6 +1191,9 @@ export function consumeForResponseLogMetadata(
   onFirstOutput?: () => void,
   options?: InspectionConsumerOptions,
 ): void {
+  if (options?.requestStartedAt !== undefined && logCtx.requestStartedAt === undefined) {
+    logCtx.requestStartedAt = options.requestStartedAt;
+  }
   const reader = body.getReader();
   // No onTerminal → the inspector's `reported` gate stays permanently false,
   // reproducing this consumer's unconditional logCtx inspection.

src/server/request-log.ts

@@ -21,13 +21,19 @@ import {
   isKnownUsageSurface,
   isCodexUsageAccountLogLabel,
   isValidReasoningWireValue,
+  normalizeStreamDiagnostics,
   readRecentUsageEntries,
   usageForFinalLog,
   usageStatusForFinalLog,
   usageTotalTokens,
   type AttemptRecoveryKind,
+  type FailureSide,
+  type FailureStage,
   type PersistedUsageAttempt,
   type PersistedUsageEntry,
+  type StreamTimeline,
+  type TerminalSource,
+  type TransportPhase,
   type UsageStatus,
 } from "../usage/log";
 import {
@@ -47,6 +53,8 @@ import { modelRecordValue } from "../reasoning-effort";
 export interface RequestLogContext {
   model: string;
   provider: string;
+  /** Internal request start timestamp in wall-clock ms. */
+  requestStartedAt?: number;
   /** TTFT: ms from request start to the first non-empty model output delta (WP4, devlog 040). */
   firstOutputMs?: number;
   /** Best-effort chat/session correlation for Logs grouping (#330). Opaque; omit when unknown. */
@@ -116,8 +124,11 @@ export interface RequestLogContext {
   /** Structured reason from `response.incomplete`; internal-only input to log classification. */
   terminalIncompleteReason?: string;
   affinity?: "reused" | "new_bind" | "rebound" | "cleared";
-  transportPhase?: "pre_headers" | "mid_stream" | "terminal_sse";
-  terminalSource?: "upstream" | "synthetic";
+  transportPhase?: TransportPhase;
+  terminalSource?: TerminalSource;
+  streamTimeline?: StreamTimeline;
+  failureSide?: FailureSide;
+  failureStage?: FailureStage;
   /** Bounded route-decision trace (RI-01); never contains secrets. */
   routeDecision?: RouteDecisionTraceV1;
 }
@@ -173,9 +184,12 @@ export interface RequestLogEntry {
   /** Codex pool affinity decision for this request (diagnostics for #186). */
   affinity?: "reused" | "new_bind" | "rebound" | "cleared";
   /** Where the upstream terminal/failure was observed. */
-  transportPhase?: "pre_headers" | "mid_stream" | "terminal_sse";
+  transportPhase?: TransportPhase;
   /** Whether the terminal came from a real upstream SSE event or a proxy synthetic tail. */
-  terminalSource?: "upstream" | "synthetic";
+  terminalSource?: TerminalSource;
+  streamTimeline?: StreamTimeline;
+  failureSide?: FailureSide;
+  failureStage?: FailureStage;
   /** Bounded route-decision trace (RI-01); never contains secrets. */
   routeDecision?: RouteDecisionTraceV1;
 }
@@ -287,6 +301,11 @@ export function requestLogEntryFromPersistedUsage(entry: PersistedUsageEntry): R
     ...(entry.usage ? { usage: entry.usage } : {}),
     ...(entry.totalTokens !== undefined ? { totalTokens: entry.totalTokens } : {}),
     ...(entry.attempts !== undefined ? { attempts: entry.attempts } : {}),
+    ...(entry.streamTimeline ? { streamTimeline: entry.streamTimeline } : {}),
+    ...(entry.failureSide ? { failureSide: entry.failureSide } : {}),
+    ...(entry.failureStage ? { failureStage: entry.failureStage } : {}),
+    ...(entry.transportPhase ? { transportPhase: entry.transportPhase } : {}),
+    ...(entry.terminalSource ? { terminalSource: entry.terminalSource } : {}),
     ...(routeDecision ? { routeDecision } : {}),
   };
 }
@@ -343,10 +362,29 @@ export function addRequestLog(entry: RequestLogEntry) {
   // line-oriented viewer — while `usage.jsonl` looked clean, which is the worst shape for a
   // sanitization bug because the safe surface is the one you check.
   const shadowCallRewrittenFrom = sanitizeLogMetadataString(entry.shadowCallRewrittenFrom);
-  const retained: RequestLogEntry = shadowCallRewrittenFrom === entry.shadowCallRewrittenFrom
-    ? entry
-    : { ...entry, ...(shadowCallRewrittenFrom ? { shadowCallRewrittenFrom } : {}) };
-  if (!shadowCallRewrittenFrom && retained !== entry) delete retained.shadowCallRewrittenFrom;
+  const diagnostics = normalizeStreamDiagnostics(entry);
+  const attempts = entry.attempts?.map(attempt => {
+    const normalized = { ...attempt };
+    const attemptDiagnostics = normalizeStreamDiagnostics(attempt);
+    if (!attemptDiagnostics.streamTimeline) delete normalized.streamTimeline;
+    if (!attemptDiagnostics.failureSide) delete normalized.failureSide;
+    if (!attemptDiagnostics.failureStage) delete normalized.failureStage;
+    if (!attemptDiagnostics.transportPhase) delete normalized.transportPhase;
+    if (!attemptDiagnostics.terminalSource) delete normalized.terminalSource;
+    return { ...normalized, ...attemptDiagnostics };
+  });
+  const retained: RequestLogEntry = {
+    ...entry,
+    ...(shadowCallRewrittenFrom ? { shadowCallRewrittenFrom } : {}),
+    ...(attempts ? { attempts } : {}),
+    ...diagnostics,
+  };
+  if (!shadowCallRewrittenFrom) delete retained.shadowCallRewrittenFrom;
+  if (!diagnostics.streamTimeline) delete retained.streamTimeline;
+  if (!diagnostics.failureSide) delete retained.failureSide;
+  if (!diagnostics.failureStage) delete retained.failureStage;
+  if (!diagnostics.transportPhase) delete retained.transportPhase;
+  if (!diagnostics.terminalSource) delete retained.terminalSource;
   entry = retained;
   retainRequestLogEntry(entry);
   try {
@@ -404,6 +442,11 @@ export function addRequestLog(entry: RequestLogEntry) {
       ...(entry.totalTokens !== undefined ? { totalTokens: entry.totalTokens } : {}),
       ...(entry.attempts !== undefined ? { attempts: entry.attempts } : {}),
       ...failureDiagnostics,
+      ...(entry.streamTimeline ? { streamTimeline: entry.streamTimeline } : {}),
+      ...(entry.failureSide ? { failureSide: entry.failureSide } : {}),
+      ...(entry.failureStage ? { failureStage: entry.failureStage } : {}),
+      ...(entry.transportPhase ? { transportPhase: entry.transportPhase } : {}),
+      ...(entry.terminalSource ? { terminalSource: entry.terminalSource } : {}),
       ...(entry.routeDecision ? { routeDecision: entry.routeDecision } : {}),
     });
   } catch {
@@ -436,6 +479,30 @@ export function recordFirstOutput(
   }
 }
 
+export function noteStreamTimelineEvent(
+  logCtx: RequestLogContext | undefined,
+  event: keyof StreamTimeline,
+  requestStartedAt?: number,
+  now = Date.now(),
+): void {
+  if (!logCtx) return;
+  if (requestStartedAt && !logCtx.requestStartedAt) {
+    logCtx.requestStartedAt = requestStartedAt;
+  }
+  if (!logCtx.streamTimeline) logCtx.streamTimeline = {};
+  const origin = requestStartedAt ?? logCtx.requestStartedAt ?? logCtx.activeAttemptStartedAt;
+  if (origin !== undefined && logCtx.streamTimeline[event] === undefined) {
+    logCtx.streamTimeline[event] = Math.max(0, now - origin);
+  }
+  if (logCtx.activeAttempt) {
+    if (!logCtx.activeAttempt.streamTimeline) logCtx.activeAttempt.streamTimeline = {};
+    const attemptOrigin = logCtx.activeAttemptStartedAt ?? requestStartedAt ?? logCtx.requestStartedAt;
+    if (attemptOrigin !== undefined && logCtx.activeAttempt.streamTimeline[event] === undefined) {
+      logCtx.activeAttempt.streamTimeline[event] = Math.max(0, now - attemptOrigin);
+    }
+  }
+}
+
 /** Snapshot target-specific requested effort even for runTurn adapters with no AdapterRequest. */
 export function recordAttemptRequestedEffort(logCtx: RequestLogContext): void {
   const attempt = logCtx.activeAttempt;
@@ -885,6 +952,7 @@ export function addFinalRequestLog(
   meta?: Pick<RequestLogEntry, "terminalStatus" | "closeReason">,
   addLog: (entry: RequestLogEntry) => void = addRequestLog,
 ): void {
+  if (!logCtx.requestStartedAt) logCtx.requestStartedAt = start;
   // Mid-stream web-search aborts used to emit response.failed and land as 502/upstream_server_error.
   // Prefer the client-close classification whenever the captured reason says so.
   const effectiveStatus = status >= 500 && logCtx.upstreamError && isClientClosedMessage(logCtx.upstreamError)
@@ -911,6 +979,13 @@ export function addFinalRequestLog(
     // semantic code on both so detailed attempt telemetry cannot regress to a generic status code.
     if (errorCode) logCtx.activeAttempt.errorCode = errorCode;
     else delete logCtx.activeAttempt.errorCode;
+    if (logCtx.streamTimeline && !logCtx.activeAttempt.streamTimeline) {
+      logCtx.activeAttempt.streamTimeline = { ...logCtx.streamTimeline };
+    }
+    if (logCtx.failureSide) logCtx.activeAttempt.failureSide = logCtx.failureSide;
+    if (logCtx.failureStage) logCtx.activeAttempt.failureStage = logCtx.failureStage;
+    if (logCtx.transportPhase) logCtx.activeAttempt.transportPhase = logCtx.transportPhase;
+    if (logCtx.terminalSource) logCtx.activeAttempt.terminalSource = logCtx.terminalSource;
   }
   const existing = finalizedUsage(
     logCtx.providerAdapter ?? logCtx.provider,
@@ -979,6 +1054,9 @@ export function addFinalRequestLog(
     ...(logCtx.affinity ? { affinity: logCtx.affinity } : {}),
     ...(logCtx.transportPhase ? { transportPhase: logCtx.transportPhase } : {}),
     ...(logCtx.terminalSource ? { terminalSource: logCtx.terminalSource } : {}),
+    ...(logCtx.streamTimeline ? { streamTimeline: logCtx.streamTimeline } : {}),
+    ...(logCtx.failureSide ? { failureSide: logCtx.failureSide } : {}),
+    ...(logCtx.failureStage ? { failureStage: logCtx.failureStage } : {}),
     ...(logCtx.routeDecision ? { routeDecision: logCtx.routeDecision } : {}),
   });
   if (isUsageDebugEnabled()) {

src/server/responses/core.ts

@@ -1837,6 +1837,7 @@ export async function handleComboResponses(
     const childLog: RequestLogContext = {
       model: pick.target.model,
       provider: pick.target.provider,
+      ...(logCtx.requestStartedAt !== undefined ? { requestStartedAt: logCtx.requestStartedAt } : {}),
       ...(logCtx.conversationId ? { conversationId: logCtx.conversationId } : {}),
       ...(logCtx.surface ? { surface: logCtx.surface } : {}),
     };
@@ -3835,6 +3836,7 @@ async function handleResponsesInner(
       linkAbortSignal(upstream, turnAc.signal);
       registerTurn(turnAc, options.turnAdmissionLease);
       const inspectionConsumerOptions = {
+        requestStartedAt: logCtx.requestStartedAt,
         clientGoneSignal: clientGone.signal,
         drainBounds: { ms: 15_000, bytes: 32 * 1024 * 1024 },
         upstream,