Skip to content

feat(session): add llm stream failure diagnostics #771

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Astro-Han merged 4 commits into from
May 19, 2026
Merged

feat(session): add llm stream failure diagnostics #771

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
dde73c0
feat(session): add llm stream failure diagnostics
Astro-Han May 19, 2026
8cd3c2b
fix(session): harden llm stream diagnostics sanitization
Astro-Han May 19, 2026
5e54139
fix(session): clarify stream diagnostic export boundaries
Astro-Han May 19, 2026
3250996
fix(session): thread abort provenance into stream diagnostics
Astro-Han May 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
163 changes: 160 additions & 3 deletions packages/opencode/src/session/export.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -294,9 +295,7 @@ export namespace Export {
const title_generations = collectTitleGenerations(node)
return {
...(last ? { loop: { last } } : {}),
...(llm_traces.length
? { llm_trace_schema_version: LLMTrace.SCHEMA_VERSION, llm_traces }
: {}),
...(llm_traces.length ? { llm_trace_schema_version: LLMTrace.SCHEMA_VERSION, llm_traces } : {}),
...(aborts.length ? { aborts } : {}),
...(title_generations.length ? { title_generations } : {}),
}
Expand Down Expand Up @@ -942,6 +941,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),
})),
Expand Down Expand Up @@ -993,9 +1000,159 @@ 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 timeline = isRecord(stream.timeline) ? stream.timeline : undefined
const durations = timeline && isRecord(timeline.durations_ms) ? timeline.durations_ms : undefined
const watchdog = isRecord(stream.watchdog) ? stream.watchdog : undefined
const attempt = isRecord(stream.attempt) ? stream.attempt : undefined
const abort = isRecord(stream.abort) ? stream.abort : undefined
const rawError = isRecord(stream.error) ? stream.error : undefined
const safeError = rawError
? compactObject({
...(typeof rawError.boundary === "string" ? { boundary: rawError.boundary } : {}),
...(typeof rawError.confidence === "string" ? { confidence: rawError.confidence } : {}),
...(Array.isArray(rawError.evidence)
? { evidence: rawError.evidence.filter((item): item is string => typeof item === "string") }
: {}),
...safeErrorFingerprint(rawError),
})
: undefined
const rawProvider = isRecord(stream.provider) ? stream.provider : undefined
const safeProvider = rawProvider ? safeProviderCorrelation(rawProvider) : undefined
return compactObject({
...(stream.schema_version === 2 ? { schema_version: 2 } : {}),
...(attempt
? {
attempt: compactObject({
...(typeof attempt.attempt_index === "number" ? { attempt_index: attempt.attempt_index } : {}),
...(typeof attempt.attempt_id === "string" ? { attempt_id: attempt.attempt_id } : {}),
...(typeof attempt.terminal_attempt === "boolean" ? { terminal_attempt: attempt.terminal_attempt } : {}),
...(attempt.note === "terminal_attempt_only" || attempt.note === "per_attempt_recorded"
? { note: attempt.note }
: {}),
}),
}
: {}),
...(stream.legacy_v1_counters === "terminal_attempt" || stream.legacy_v1_counters === "aggregate"
? { legacy_v1_counters: stream.legacy_v1_counters }
: {}),
...(timeline
? {
timeline: compactObject({
...(typeof timeline.collector_created_at === "number"
? { collector_created_at: timeline.collector_created_at }
: {}),
...(typeof timeline.sdk_stream_returned_at === "number"
? { sdk_stream_returned_at: timeline.sdk_stream_returned_at }
: {}),
...(typeof timeline.watchdog_armed_at === "number"
? { watchdog_armed_at: timeline.watchdog_armed_at }
: {}),
...(typeof timeline.first_event_at === "number" ? { first_event_at: timeline.first_event_at } : {}),
...(typeof timeline.first_provider_progress_at === "number"
? { first_provider_progress_at: timeline.first_provider_progress_at }
: {}),
...(typeof timeline.last_event_at === "number" ? { last_event_at: timeline.last_event_at } : {}),
...(typeof timeline.last_provider_progress_at === "number"
? { last_provider_progress_at: timeline.last_provider_progress_at }
: {}),
...(typeof timeline.completed_at === "number" ? { completed_at: timeline.completed_at } : {}),
...(typeof timeline.failed_at === "number" ? { failed_at: timeline.failed_at } : {}),
...(durations
? {
durations_ms: compactObject({
...(typeof durations.created_to_sdk_stream_returned === "number"
? { created_to_sdk_stream_returned: durations.created_to_sdk_stream_returned }
: {}),
...(typeof durations.watchdog_armed_to_first_event === "number"
? { watchdog_armed_to_first_event: durations.watchdog_armed_to_first_event }
: {}),
...(typeof durations.watchdog_armed_to_first_provider_progress === "number"
? {
watchdog_armed_to_first_provider_progress:
durations.watchdog_armed_to_first_provider_progress,
}
: {}),
...(typeof durations.first_to_last_provider_progress === "number"
? { first_to_last_provider_progress: durations.first_to_last_provider_progress }
: {}),
...(typeof durations.last_provider_progress_to_failure === "number"
? { last_provider_progress_to_failure: durations.last_provider_progress_to_failure }
: {}),
...(typeof durations.total === "number" ? { total: durations.total } : {}),
}),
}
: {}),
}),
}
: {}),
...(watchdog
? {
watchdog: compactObject({
...(typeof watchdog.connect_timeout_ms === "number"
? { connect_timeout_ms: watchdog.connect_timeout_ms }
: {}),
...(typeof watchdog.stream_timeout_ms === "number"
? { stream_timeout_ms: watchdog.stream_timeout_ms }
: {}),
...(typeof watchdog.provider_progressed === "boolean"
? { provider_progressed: watchdog.provider_progressed }
: {}),
...(watchdog.phase_at_end === "before_first_provider_progress" ||
watchdog.phase_at_end === "between_provider_events" ||
watchdog.phase_at_end === "completed" ||
watchdog.phase_at_end === "unknown"
? { phase_at_end: watchdog.phase_at_end }
: {}),
...(typeof watchdog.fired === "boolean" ? { fired: watchdog.fired } : {}),
...(watchdog.fired_phase === "connect" || watchdog.fired_phase === "silent_stream"
? { fired_phase: watchdog.fired_phase }
: {}),
}),
}
: {}),
...(abort
? {
abort: compactObject({
...(typeof abort.signal_aborted_at_error === "boolean"
? { signal_aborted_at_error: abort.signal_aborted_at_error }
: {}),
...(typeof abort.provenance_source === "string" ? { provenance_source: abort.provenance_source } : {}),
...(typeof abort.provenance_reason === "string" ? { provenance_reason: abort.provenance_reason } : {}),
...(abort.provenance_mode === "soft" || abort.provenance_mode === "hard"
? { provenance_mode: abort.provenance_mode }
: {}),
...(typeof abort.provenance_recorded_at === "number"
? { provenance_recorded_at: abort.provenance_recorded_at }
: {}),
...(typeof abort.provenance_missing === "boolean"
? { provenance_missing: abort.provenance_missing }
: {}),
}),
}
: {}),
...(safeError && Object.keys(safeError).length > 0 ? { error: safeError } : {}),
...(safeProvider ? { provider: safeProvider } : {}),
Comment thread
coderabbitai[bot] marked this conversation as resolved.
})
}

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,
Expand Down
4 changes: 4 additions & 0 deletions packages/opencode/src/session/llm-trace/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
143 changes: 142 additions & 1 deletion packages/opencode/src/session/llm-trace/recorder.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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) {
Expand All @@ -42,6 +46,113 @@ export function createRecorder(input: RecorderInput): Recorder {
finishReason = reason
tokens = nextTokens
},
beginStream(next) {
streamMonotonicStart = next.monotonicMs
stream = {
schema_version: 2,
legacy_v1_counters: "aggregate",
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
const boundary = localAbortBoundary(stream) ?? {
boundary: next.boundary,
confidence: next.confidence,
evidence: next.evidence,
}
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,
}
},
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),
}
},
recordAbortState(next) {
if (!stream) return
stream.abort = {
...(stream.abort ?? {}),
...(typeof next.signalAbortedAtError === "boolean"
? { signal_aborted_at_error: next.signalAbortedAtError }
: {}),
...(next.provenanceSource ? { provenance_source: next.provenanceSource } : {}),
...(next.provenanceReason ? { provenance_reason: next.provenanceReason } : {}),
...(next.provenanceMode ? { provenance_mode: next.provenanceMode } : {}),
...(typeof next.provenanceRecordedAt === "number" ? { provenance_recorded_at: next.provenanceRecordedAt } : {}),
...(typeof next.provenanceMissing === "boolean" ? { provenance_missing: next.provenanceMissing } : {}),
}
refreshLocalAbortBoundary(stream)
},
recordProviderErrorEvent(next) {
if (!stream) return
if (stream.error?.boundary === "watchdog" && stream.error.confidence === "high") return
const provider = safeProviderCorrelation(next.provider)
Comment thread
Astro-Han marked this conversation as resolved.
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
Expand Down Expand Up @@ -71,11 +182,39 @@ 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)
}

function hasAbortProvenance(stream: StreamDiagnostics) {
return !!stream.abort?.provenance_source
}

function localAbortBoundary(stream: StreamDiagnostics) {
if (!stream.abort?.signal_aborted_at_error || !hasAbortProvenance(stream)) return undefined
return classifyBoundary({ abortSignalAborted: true, abortProvenancePresent: true, iteratorError: true })
}

function refreshLocalAbortBoundary(stream: StreamDiagnostics) {
if (stream.error?.boundary === "watchdog" && stream.error.confidence === "high") return
if (!stream.error || stream.error.boundary !== "unknown") return
const boundary = localAbortBoundary(stream)
if (!boundary) return
stream.error = {
...stream.error,
boundary: boundary.boundary,
confidence: boundary.confidence,
evidence: boundary.evidence,
}
}

export function storedPartCounts(parts: MessageV2.Part[]): StoredParts {
const counts: StoredParts = {
text: 0,
Expand Down Expand Up @@ -158,5 +297,7 @@ function tokenSummary(tokens: MessageV2.Assistant["tokens"]): Tokens {
}

function isEmptyCompletion(finishReason: string | undefined, stored: StoredParts) {
return finishReason === "stop" && stored.text === 0 && stored.reasoning === 0 && stored.tool === 0 && stored.file === 0
return (
finishReason === "stop" && stored.text === 0 && stored.reasoning === 0 && stored.tool === 0 && stored.file === 0
)
}
Loading