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

src/server/index.ts

@@ -1624,6 +1624,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "unknown",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
           inboundProtocol: "responses",
         };
@@ -1658,6 +1659,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;
@@ -1713,6 +1715,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 => {
@@ -1737,6 +1740,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "unknown",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
           inboundProtocol: "responses",
         };
@@ -1811,6 +1815,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "unknown",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
           inboundProtocol: "messages",
         };
@@ -1841,6 +1846,7 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server<W
         const logCtx: RequestLogContext = {
           model: "unknown",
           provider: "unknown",
+          requestStartedAt: start,
           ...admissionFields(admission),
           inboundProtocol: "chat",
         };
@@ -1872,6 +1878,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 => {
@@ -1909,6 +1916,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(sessionLaneIdFromRequest(req.headers));
@@ -2151,6 +2159,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

@@ -15,6 +15,7 @@ import {
   httpStatusForRequestLogTerminal,
   inspectResponseLogJson,
   inspectResponseLogSsePayloadParsed,
+  noteStreamTimelineEvent,
   recordFirstOutput,
   type RequestLogContext,
   type RequestLogEntry,
@@ -585,6 +586,7 @@ export function trackSseForRequestLog(
   onCancel: () => void,
   logCtx?: RequestLogContext,
   onFirstOutput?: () => void,
+  inspectedSource?: "upstream" | "relay",
 ): ReadableStream<Uint8Array> {
   const reader = body.getReader();
   let terminalReported = false;
@@ -601,6 +603,7 @@ export function trackSseForRequestLog(
     onTerminal: reportTerminal,
     logCtx,
     onFirstOutput,
+    inspectedSource,
   });
 
   return new ReadableStream<Uint8Array>({
@@ -645,7 +648,9 @@ export function responseWithDeferredRequestLog(
   start: number,
   logCtx: RequestLogContext,
   addLog: (entry: RequestLogEntry) => void = addRequestLog,
+  inspectedSource?: "upstream" | "relay",
 ): Response {
+  if (logCtx.requestStartedAt === undefined) logCtx.requestStartedAt = start;
   const contentType = response.headers.get("content-type")?.toLowerCase() ?? "";
   if (isUsageDebugEnabled() && !logCtx.usageDebugContentType && contentType) {
     logCtx.usageDebugContentType = contentType;
@@ -707,6 +712,7 @@ export function responseWithDeferredRequestLog(
     },
     logCtx,
     () => recordFirstOutput(logCtx, start),
+    inspectedSource,
   );
   return new Response(body, {
     status: response.status,
@@ -839,6 +845,11 @@ export type SseInspectorHandlers = {
    * between `response.created` and `response.completed`.
    */
   pinCompletedResponseIdToFirstSeen?: boolean;
+  /**
+   * Provenance of the stream being inspected.
+   * Raw upstream SSE defaults to "upstream"; adapter or bridge produced SSE uses "relay".
+   */
+  inspectedSource?: "upstream" | "relay";
 };
 
 type CompletedOutputItem = { item: unknown; sourceBytes: number };
@@ -1021,6 +1032,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 && firstOutputFromParsed(parsed)) {
+      noteStreamTimelineEvent(handlers.logCtx, "upstreamFirstSemanticOutputMs");
+    }
     const status = terminalStatusFromParsed(parsed);
     const policyTerminal = status === "failed"
       && isPolicyRewriteType(parsed)
@@ -1030,8 +1044,15 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector
       try {
         reported = true;
         if (handlers.logCtx) {
+          const source = handlers.inspectedSource ?? "upstream";
           handlers.logCtx.transportPhase = "terminal_sse";
-          handlers.logCtx.terminalSource = "upstream";
+          handlers.logCtx.terminalSource = source;
+          if (status === "failed") {
+            handlers.logCtx.failureSide = handlers.logCtx.failureSide
+              ?? (source === "relay" ? "relay" : "upstream");
+            handlers.logCtx.failureStage = handlers.logCtx.failureStage
+              ?? (source === "relay" ? "relay_transform" : "terminal_delivery");
+          }
         }
         handlers.onTerminal(status, policyTerminal ? 400 : undefined);
       } finally {
@@ -1158,7 +1179,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;
@@ -1184,6 +1209,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;
@@ -1349,6 +1375,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,
@@ -1367,7 +1396,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");
       }
     },
@@ -1379,6 +1413,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).
@@ -1399,6 +1435,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

@@ -24,13 +24,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 {
@@ -50,6 +56,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. */
@@ -134,8 +142,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;
 }
@@ -198,9 +209,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;
 }
@@ -312,6 +326,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 } : {}),
   };
 }
@@ -368,10 +387,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 {
@@ -430,6 +468,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 {
@@ -461,6 +504,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;
@@ -926,6 +993,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)
@@ -954,6 +1022,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,
@@ -1027,6 +1102,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()) {