feat(session): add llm stream failure diagnostics

packages/opencode/src/session/export.ts

@@ -25,6 +25,7 @@ import { isRecord } from "@/util/record"
 import { Glob } from "@/util/glob"
 import { safeToolFailureMetadata } from "./tool-failure"
 import { LLMTrace } from "./llm-trace"
+import { safeErrorFingerprint, safeProviderCorrelation } from "./llm-trace/stream-diagnostics"
 
 export function getRuntimeNamespace(): "pawwork" | "opencode" {
   return Runtime.isPawWork() ? "pawwork" : "opencode"
@@ -942,6 +943,14 @@ export namespace Export {
                 structured:
                   msg.info.structured === undefined ? undefined : { redacted: `assistant-structured:${msg.info.id}` },
                 error: namedError("assistant-error", msg.info.id, msg.info.error),
+                diagnostics: !msg.info.diagnostics
+                  ? msg.info.diagnostics
+                  : {
+                      ...msg.info.diagnostics,
+                      llm_trace: msg.info.diagnostics.llm_trace
+                        ? sanitizeLLMTrace(msg.info.diagnostics.llm_trace)
+                        : undefined,
+                    },
               },
         parts: msg.parts.map(partFn),
       })),
@@ -993,9 +1002,40 @@ export namespace Export {
             ? undefined
             : redact("title-generation-error-message", String(index), trace.error_message),
       })),
+      llm_traces: diagnostics.llm_traces?.map(sanitizeLLMTrace),
+    }
+  }
+
+  function sanitizeLLMTrace(trace: LLMTrace.Summary): LLMTrace.Summary {
+    const stream = trace.stream as Record<string, unknown> | undefined
+    if (!stream) return trace
+    return {
+      ...trace,
+      stream: sanitizeStreamDiagnostics(stream),
+    } as LLMTrace.Summary
+  }
+
+  function sanitizeStreamDiagnostics(stream: Record<string, unknown>) {
+    const rawError = isRecord(stream.error) ? stream.error : undefined
+    const safeError = rawError
+      ? {
+          ...rawError,
+          ...safeErrorFingerprint(rawError),
+        }
+      : undefined
+    const rawProvider = isRecord(stream.provider) ? stream.provider : undefined
+    const safeProvider = rawProvider ? safeProviderCorrelation(rawProvider) : undefined
+    return {
+      ...stream,
+      ...(safeError ? { error: compactObject(safeError) } : {}),
+      ...(safeProvider ? { provider: safeProvider } : {}),
     }
   }
 
+  function compactObject<T extends Record<string, unknown>>(input: T): T {
+    return Object.fromEntries(Object.entries(input).filter(([, value]) => value !== undefined)) as T
+  }
+
   // Snapshot-level sanitize. Wraps sanitizeTree (the conversation tree) AND redacts top-level
   // runtime_context/diagnostic fields that may carry user-machine paths or raw tool args.
   // Other runtime_context fields (app_version, build_channel, locale, timezone, model_refs,

packages/opencode/src/session/llm-trace/index.ts

@@ -5,19 +5,23 @@ import {
 } from "./recorder"
 import { SCHEMA_VERSION as VERSION } from "./types"
 import * as Types from "./types"
+import { classifyBoundary as classify, safeProviderCorrelation as safeCorrelation } from "./stream-diagnostics"
 
 export namespace LLMTrace {
   export const SCHEMA_VERSION = VERSION
   export const Summary = Types.Summary
   export const createRecorder = createTraceRecorder
   export const requestSummary = summarizeRequest
   export const storedPartCounts = countStoredParts
+  export const classifyBoundary = classify
+  export const safeProviderCorrelation = safeCorrelation
 
   export type RequestSummary = Types.RequestSummary
   export type StreamEvents = Types.StreamEvents
   export type StoredParts = Types.StoredParts
   export type Tokens = Types.Tokens
   export type Flags = Types.Flags
+  export type StreamDiagnostics = Types.StreamDiagnostics
   export type Summary = Types.Summary
   export type RequestSummaryInput = Types.RequestSummaryInput
   export type RecorderInput = Types.RecorderInput

packages/opencode/src/session/llm-trace/recorder.ts

@@ -11,6 +11,8 @@ import type {
   Tokens,
 } from "./types"
 import { SCHEMA_VERSION } from "./types"
+import type { StreamDiagnostics } from "./types"
+import { classifyBoundary, safeErrorFingerprint, safeProviderCorrelation } from "./stream-diagnostics"
 
 export function requestSummary(input: RequestSummaryInput): RequestSummary {
   const options = safeOptions(input.options)
@@ -30,6 +32,8 @@ export function createRecorder(input: RecorderInput): Recorder {
   let request: RequestSummary | undefined
   let finishReason: string | undefined
   let tokens: MessageV2.Assistant["tokens"] | undefined
+  let stream: StreamDiagnostics | undefined
+  let streamMonotonicStart: number | undefined
 
   return {
     request(summary) {
@@ -42,6 +46,92 @@ export function createRecorder(input: RecorderInput): Recorder {
       finishReason = reason
       tokens = nextTokens
     },
+    beginStream(next) {
+      streamMonotonicStart = next.monotonicMs
+      stream = {
+        schema_version: 2,
+        legacy_v1_counters: "terminal_attempt",
+        timeline: {
+          collector_created_at: next.collectorCreatedAt,
+        },
+        watchdog: {
+          connect_timeout_ms: next.connectTimeoutMs,
+          stream_timeout_ms: next.streamTimeoutMs,
+          provider_progressed: false,
+          phase_at_end: "before_first_provider_progress",
+          fired: false,
+        },
+      }
+    },
+    recordStreamFailure(next) {
+      if (!stream) return
+      if (stream.error?.boundary === "watchdog" && stream.error.confidence === "high") return
+      stream.timeline.failed_at = next.failedAt
+      stream.timeline.durations_ms = {
+        ...(stream.timeline.durations_ms ?? {}),
+        total: durationSince(streamMonotonicStart, next.monotonicMs),
+      }
+      stream.error = {
+        ...safeErrorFingerprint(next.error),
+        boundary: next.boundary,
+        confidence: next.confidence,
+        evidence: next.evidence,
+      }
+    },
+    recordProviderProgress(next) {
+      if (!stream) return
+      stream.watchdog.provider_progressed = true
+      stream.watchdog.phase_at_end = "between_provider_events"
+      if (stream.timeline.first_provider_progress_at === undefined) {
+        stream.timeline.first_provider_progress_at = next.eventAt
+        stream.timeline.durations_ms = {
+          ...(stream.timeline.durations_ms ?? {}),
+          watchdog_armed_to_first_provider_progress: durationSince(streamMonotonicStart, next.monotonicMs),
+        }
+      }
+      stream.timeline.last_provider_progress_at = next.eventAt
+    },
+    recordWatchdogFired(next) {
+      if (!stream) return
+      stream.watchdog.fired = true
+      stream.watchdog.fired_phase = next.phase
+      if (next.phase === "connect") stream.watchdog.phase_at_end = "before_first_provider_progress"
+    },
+    recordStreamCompleted(next) {
+      if (!stream) return
+      stream.timeline.completed_at = next.completedAt
+      stream.watchdog.phase_at_end = "completed"
+      stream.timeline.durations_ms = {
+        ...(stream.timeline.durations_ms ?? {}),
+        total: durationSince(streamMonotonicStart, next.monotonicMs),
+      }
+    },
+    recordProviderErrorEvent(next) {
+      if (!stream) return
+      const provider = safeProviderCorrelation(next.provider)
+      stream.provider = provider
+      const boundary = classifyBoundary({
+        providerErrorEvent: true,
+        iteratorError: true,
+        requestIdPresent: provider?.request_id !== undefined || provider?.response_id !== undefined,
+        providerCorrelationUnavailable: provider?.unavailable_reason !== undefined,
+      })
+      stream.timeline.failed_at = next.failedAt
+      stream.timeline.durations_ms = {
+        ...(stream.timeline.durations_ms ?? {}),
+        total: durationSince(streamMonotonicStart, next.monotonicMs),
+      }
+      stream.error = {
+        ...safeErrorFingerprint(next.error),
+        boundary: boundary.boundary,
+        confidence: boundary.confidence,
+        evidence: boundary.evidence,
+      }
+    },
+    recordProviderCorrelation(input) {
+      if (!stream) return
+      stream.provider = safeProviderCorrelation(input)
+    },
     finalize(final: FinalizeInput) {
       const finalFinishReason = final.finishReason ?? finishReason
       const finalTokens = final.tokens ?? tokens
@@ -71,11 +161,17 @@ export function createRecorder(input: RecorderInput): Recorder {
         flags,
         created_at: input.createdAt,
         completed_at: final.completedAt,
+        ...(stream ? { stream } : {}),
       }
     },
   }
 }
 
+function durationSince(start: number | undefined, end: number) {
+  if (start === undefined) return undefined
+  return Math.max(0, end - start)
+}
+
 export function storedPartCounts(parts: MessageV2.Part[]): StoredParts {
   const counts: StoredParts = {
     text: 0,

packages/opencode/src/session/llm-trace/stream-diagnostics.ts

@@ -0,0 +1,159 @@
+import { isRecord } from "@/util/record"
+
+export type StreamBoundary = "watchdog" | "local_abort" | "sdk_transport" | "provider_stream" | "unknown"
+export type StreamConfidence = "low" | "medium" | "high"
+export type StreamEvidence =
+  | "watchdog_fired"
+  | "watchdog_error"
+  | "abort_signal_aborted"
+  | "abort_provenance_present"
+  | "abort_provenance_missing"
+  | "provider_error_event"
+  | "iterator_error"
+  | "provider_progress_seen"
+  | "request_id_present"
+  | "provider_correlation_unavailable"
+
+export type BoundaryInput = {
+  watchdogFired?: boolean
+  watchdogError?: boolean
+  abortSignalAborted?: boolean
+  abortProvenancePresent?: boolean
+  providerErrorEvent?: boolean
+  iteratorError?: boolean
+  providerProgressSeen?: boolean
+  requestIdPresent?: boolean
+  providerCorrelationUnavailable?: boolean
+}
+
+export type BoundaryResult = {
+  boundary: StreamBoundary
+  confidence: StreamConfidence
+  evidence: StreamEvidence[]
+}
+
+export type ProviderCorrelation = {
+  request_id?: string
+  response_id?: string
+  status_code?: number
+  safe_headers?: Record<string, string>
+  unavailable_reason?: string
+}
+
+const SAFE_HEADER_NAMES = new Set(["x-request-id", "request-id", "x-correlation-id", "x-trace-id", "traceparent"])
+
+export function classifyBoundary(input: BoundaryInput): BoundaryResult {
+  const evidence: StreamEvidence[] = []
+  if (input.watchdogFired) evidence.push("watchdog_fired")
+  if (input.watchdogError) evidence.push("watchdog_error")
+  if (input.abortSignalAborted) evidence.push("abort_signal_aborted")
+  if (input.abortProvenancePresent === true) evidence.push("abort_provenance_present")
+  if (input.abortProvenancePresent === false) evidence.push("abort_provenance_missing")
+  if (input.providerErrorEvent) evidence.push("provider_error_event")
+  if (input.iteratorError) evidence.push("iterator_error")
+  if (input.providerProgressSeen) evidence.push("provider_progress_seen")
+  if (input.requestIdPresent) evidence.push("request_id_present")
+  if (input.providerCorrelationUnavailable) evidence.push("provider_correlation_unavailable")
+
+  if (input.watchdogFired && input.watchdogError) return { boundary: "watchdog", confidence: "high", evidence }
+  if (input.abortSignalAborted && input.abortProvenancePresent) {
+    return { boundary: "local_abort", confidence: "high", evidence }
+  }
+  if (input.providerErrorEvent && input.requestIdPresent) {
+    return { boundary: "provider_stream", confidence: "high", evidence }
+  }
+  if (input.providerErrorEvent) return { boundary: "provider_stream", confidence: "medium", evidence }
+  if (input.abortSignalAborted && input.iteratorError) return { boundary: "unknown", confidence: "low", evidence }
+  if (input.providerProgressSeen && input.iteratorError)
+    return { boundary: "sdk_transport", confidence: "low", evidence }
+  if (input.iteratorError) return { boundary: "unknown", confidence: "low", evidence }
+  return { boundary: "unknown", confidence: "low", evidence }
+}
+
+export function safeProviderCorrelation(input: unknown): ProviderCorrelation | undefined {
+  if (!isRecord(input)) return { unavailable_reason: "provider_correlation_unavailable" }
+  const requestID = safeIdentifier(input.request_id ?? input.requestId ?? input["x-request-id"])
+  const responseID = safeIdentifier(input.response_id ?? input.responseId)
+  const statusCode = typeof input.status_code === "number" ? input.status_code : number(input.statusCode)
+  const safeHeaders = safeHeaderMap(input.headers ?? input.safe_headers)
+  const result: ProviderCorrelation = {
+    ...(requestID ? { request_id: requestID } : {}),
+    ...(responseID ? { response_id: responseID } : {}),
+    ...(statusCode !== undefined ? { status_code: statusCode } : {}),
+    ...(safeHeaders ? { safe_headers: safeHeaders } : {}),
+  }
+  return Object.keys(result).length ? result : { unavailable_reason: "provider_correlation_unavailable" }
+}
+
+export function safeErrorFingerprint(error: unknown) {
+  const record = isRecord(error) ? error : undefined
+  const cause = record && isRecord(record.cause) ? record.cause : undefined
+  return {
+    constructor_name: safeLowCardinality(record?.constructor?.name ?? constructorName(error)),
+    name: safeLowCardinality(record?.name),
+    message: safeMessage(record?.message ?? (typeof error === "string" ? error : undefined)),
+    code: safeLowCardinality(record?.code),
+    cause_constructor_name: safeLowCardinality(cause?.constructor?.name),
+    cause_name: safeLowCardinality(cause?.name),
+    cause_message: safeMessage(cause?.message),
+    cause_code: safeLowCardinality(cause?.code),
+    stack_hint: safeStackHint(record?.stack),
+  }
+}
+
+function safeHeaderMap(value: unknown) {
+  if (!isRecord(value)) return undefined
+  const result: Record<string, string> = {}
+  for (const [rawName, rawValue] of Object.entries(value)) {
+    const name = rawName.toLowerCase()
+    if (!SAFE_HEADER_NAMES.has(name)) continue
+    const safeValue = safeIdentifier(rawValue)
+    if (!safeValue) continue
+    result[name] = safeValue
+  }
+  return Object.keys(result).length ? result : undefined
+}
+
+function safeIdentifier(value: unknown) {
+  if (typeof value !== "string") return undefined
+  const trimmed = value.trim()
+  if (!trimmed || trimmed.length > 128) return undefined
+  if (/https?:\/\//i.test(trimmed)) return undefined
+  if (/bearer|cookie|token|secret|sk-/i.test(trimmed)) return undefined
+  if (!/^[a-zA-Z0-9_.:\/-]+$/.test(trimmed)) return undefined
+  return trimmed
+}
+
+function safeLowCardinality(value: unknown) {
+  if (typeof value !== "string") return undefined
+  const trimmed = value.trim()
+  if (!trimmed || trimmed.length > 80) return undefined
+  if (!/^[a-zA-Z0-9_.:-]+$/.test(trimmed)) return undefined
+  return trimmed
+}
+
+function safeMessage(value: unknown) {
+  if (typeof value !== "string") return undefined
+  let result = value.slice(0, 1024)
+  result = result.replace(/https?:\/\/\S+/gi, "[redacted:url]")
+  result = result.replace(/\bsk-[a-zA-Z0-9_-]+\b/g, "[redacted:secret]")
+  result = result.replace(/Bearer\s+[a-zA-Z0-9._~+/-]+/gi, "Bearer [redacted:secret]")
+  result = result.replace(/\/Users\/[^\s)]+/g, "[redacted:path]")
+  result = result.replace(/\\Users\\[^\s)]+/g, "[redacted:path]")
+  return result.slice(0, 1024)
+}
+
+function safeStackHint(value: unknown) {
+  if (typeof value !== "string") return undefined
+  const line = value.split("\n").find((entry) => entry.includes(" at "))
+  if (!line) return undefined
+  return safeMessage(line)?.slice(0, 160)
+}
+
+function constructorName(value: unknown) {
+  return isRecord(value) ? value.constructor?.name : undefined
+}
+
+function number(value: unknown) {
+  return typeof value === "number" && Number.isFinite(value) ? value : undefined
+}