diff --git a/packages/opencode/src/session/export.ts b/packages/opencode/src/session/export.ts index 0cf88c2f7..5fce63858 100644 --- a/packages/opencode/src/session/export.ts +++ b/packages/opencode/src/session/export.ts @@ -358,8 +358,8 @@ export namespace Export { const walk = (t: Tree) => { for (const message of t.messages ?? []) { if (message.info.role !== "assistant") continue - const incident = message.info.diagnostics?.run_observability?.incident - if (incident) incidents.push(incident) + const summary = message.info.diagnostics?.run_observability + if (summary) incidents.push(...runIncidentsFromSummary(summary)) } for (const child of t.children ?? []) walk(child) } @@ -1036,8 +1036,11 @@ export namespace Export { function sanitizeDiagnostics(diagnostics: Snapshot["diagnostics"]): Snapshot["diagnostics"] { const last = diagnostics.loop?.last const sanitizedIncidents = ( - diagnostics.run_incidents ?? diagnostics.run_observability?.flatMap((summary) => summary.incident ?? []) + diagnostics.run_incidents ?? diagnostics.run_observability?.flatMap(runIncidentsFromSummary) )?.map(RunIncident.sanitize) + const hasRunIncident = + diagnostics.run_incidents?.length || + diagnostics.run_observability?.some((summary) => summary.incident || summary.recovered_incidents?.length) const sanitizedChains = sanitizedIncidents?.map(RunIncident.toExportChain) ?? diagnostics.incident_chains?.map((chain) => ({ @@ -1088,10 +1091,7 @@ export namespace Export { llm_traces: diagnostics.llm_traces?.map(sanitizeLLMTrace), run_observability: diagnostics.run_observability?.map(sanitizeRunObservability), run_incident_schema_version: - diagnostics.run_incident_schema_version ?? - (diagnostics.run_incidents?.length || diagnostics.run_observability?.some((summary) => summary.incident) - ? RunIncident.SCHEMA_VERSION - : undefined), + diagnostics.run_incident_schema_version ?? (hasRunIncident ? RunIncident.SCHEMA_VERSION : undefined), run_incidents: sanitizedIncidents, incident_chains: sanitizedChains, } @@ -1101,10 +1101,15 @@ export namespace Export { return { ...summary, incident: summary.incident ? RunIncident.sanitize(summary.incident) : undefined, + recovered_incidents: summary.recovered_incidents?.map(RunIncident.sanitize), error: summary.error, } } + function runIncidentsFromSummary(summary: RunObservability.Summary) { + return [...(summary.recovered_incidents ?? []), ...(summary.incident ? [summary.incident] : [])] + } + function sanitizeLLMTrace(trace: LLMTrace.Summary): LLMTrace.Summary { const stream = trace.stream as Record | undefined if (!stream) return trace diff --git a/packages/opencode/src/session/llm.ts b/packages/opencode/src/session/llm.ts index 1561c1f11..90b3c96f9 100644 --- a/packages/opencode/src/session/llm.ts +++ b/packages/opencode/src/session/llm.ts @@ -669,7 +669,7 @@ export const defaultLayer: Layer.Layer = Layer.suspend(() ), ) -function resolveTools(input: Pick) { +export function resolveTools(input: Pick) { const disabled = Permission.disabled( Object.keys(input.tools), Permission.merge(input.agent.permission, input.permission ?? []), diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index d32a65e94..264dc0b20 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -30,6 +30,8 @@ import { currentLifecycleCloseAction, lifecycleCloseActionMeta } from "./lifecyc const log = Log.create({ service: "session.processor" }) const TOOL_CLEANUP_TIMEOUT_MS = 1_000 +const SAFE_RECOVERY_AUTO_RETRY_BACKOFF_MS = 1_000 +const LOCAL_LIFECYCLE_CLOSE_INTERRUPTION_MESSAGE = "The run was interrupted by a local lifecycle close." export type Result = "compact" | "stop" | "continue" @@ -118,6 +120,55 @@ const TOOL_INTERRUPTION_ERRORS: Record = { tool_input_generation: "Tool call generation interrupted before the tool ran.", } +function watchdogPhase(error: unknown): "connect" | "silent_stream" | "unknown" | undefined { + const message = errorMessage(error).toLowerCase() + if (!message.includes("llm stream connection timed out")) return undefined + if (message.includes("without provider progress")) return "connect" + return "silent_stream" +} + +function sideEffectBoundarySnapshot(tools: LLM.StreamInput["tools"]): RunObservability.SideEffectBoundarySnapshot { + const names = Object.keys(tools ?? {}) + const effects = names.map((name) => RunObservability.toolEffect(name)) + const unknownCount = effects.filter((effect) => effect.kind === "unknown").length + const unclassifiedCount = effects.filter((effect) => !effect.complete).length + const incomplete = unknownCount > 0 || unclassifiedCount > 0 + return { + exposed_tool_count: names.length, + unknown_tool_count: unknownCount, + unclassified_effect_count: unclassifiedCount, + provider_executed_capability_present: false, + external_boundary_present: false, + proof_result: incomplete ? "incomplete" : "complete", + proof_reason: incomplete ? (unknownCount > 0 ? "unknown_tool_boundary" : "unclassified_effect") : "all_boundaries_classified", + } +} + +function recoveryInterruptionMessage(recovery: NonNullable["recovery"] | undefined) { + switch (recovery?.reason) { + case "visible_output_without_tool_execution": + return "The response was interrupted after output started. PawWork did not automatically retry to avoid duplicate text." + case "partial_tool_input_without_execution": + return "The connection broke while PawWork was preparing a tool call. The tool did not run." + case "tool_call_materialized_without_execution": + return "A tool call was prepared before the interruption. Recovery needs confirmation before continuing." + case "tool_execution_started": + return "The connection was interrupted after tool execution started. PawWork did not automatically retry." + case "unsafe_side_effect_started": + return "The connection was interrupted after a side effect may have started. PawWork did not automatically retry." + case "side_effect_facts_incomplete": + return "The connection was interrupted, and PawWork could not prove whether external side effects were possible." + case "local_lifecycle_close": + return LOCAL_LIFECYCLE_CLOSE_INTERRUPTION_MESSAGE + case "user_cancel": + return "The run was cancelled by the user." + case "no_visible_output_or_tool_execution": + return "The provider connection was interrupted before PawWork produced output or ran tools." + default: + return undefined + } +} + type PendingLoopAction = { loopAction: "block" | "stop" tool: string @@ -1073,10 +1124,11 @@ export const layer: Layer.Layer< const halt = Effect.fn("SessionProcessor.halt")(function* ( e: unknown, attemptID: RunObservability.AttemptID | undefined = ctx.currentAttemptID, + options?: { recordFailure?: boolean; interruptionMessage?: string }, ) { slog.error("process", { error: errorMessage(e), stack: e instanceof Error ? e.stack : undefined }) ctx.streamError = true - if (attemptID) { + if (attemptID && options?.recordFailure !== false) { ctx.runTrace.recordTransportFailure({ attemptID, at: Date.now(), @@ -1103,6 +1155,9 @@ export const layer: Layer.Layer< } const error = parse(e) + if (options?.interruptionMessage && isRecord(error.data)) { + error.data = { ...error.data, message: options.interruptionMessage } + } if (MessageV2.ContextOverflowError.isInstance(error)) { ctx.needsCompaction = true yield* bus.publish(Session.Event.Error, { sessionID: ctx.sessionID, error }) @@ -1116,95 +1171,190 @@ export const layer: Layer.Layer< yield* status.set(ctx.sessionID, { type: "idle" }) }) + const recordProcessInterrupt = Effect.fn("SessionProcessor.recordProcessInterrupt")(function* ( + attemptID: RunObservability.AttemptID | undefined, + ) { + aborted = true + const lifecycleAction = currentLifecycleCloseAction(ctx.directory) + ctx.runTrace.recordScopeClosed({ + at: Date.now(), + monotonicMs: performance.now(), + source: "session.processor.onInterrupt", + reason: "aborted", + propagationPoint: "session.processor.process.onInterrupt", + ...(lifecycleAction ? lifecycleCloseActionMeta(lifecycleAction) : {}), + }) + ctx.trace.recordAbortState({ + provenanceSource: "session.processor.onInterrupt", + provenanceReason: "aborted", + provenanceMode: "hard", + provenanceRecordedAt: Date.now(), + }) + if (!ctx.assistantMessage.error) { + yield* halt(new DOMException("Aborted", "AbortError"), attemptID) + } + }) + const process = Effect.fn("SessionProcessor.process")(function* (streamInput: LLM.StreamInput) { slog.info("process") ctx.needsCompaction = false ctx.shouldBreak = (yield* config.get()).experimental?.continue_loop_on_deny !== true let processAttemptID: RunObservability.AttemptID | undefined + let automaticStreamRetriesUsed = 0 + + const retryStillAllowed = Effect.fn("SessionProcessor.retryStillAllowed")(function* (stage: string) { + const lifecycleAction = currentLifecycleCloseAction(ctx.directory) + if (!lifecycleAction) return { allowed: true as const } + ctx.runTrace.recordScopeClosed({ + at: Date.now(), + monotonicMs: performance.now(), + source: `session.processor.safe_recovery.${stage}`, + reason: "lifecycle_close_before_auto_retry", + propagationPoint: "session.processor.safe_recovery", + ...lifecycleCloseActionMeta(lifecycleAction), + }) + return { + allowed: false as const, + interruptionMessage: LOCAL_LIFECYCLE_CLOSE_INTERRUPTION_MESSAGE, + } + }) + + const retrySignalFor = (error: unknown) => { + const phase = watchdogPhase(error) + if (phase) { + return { + retryable: true, + message: "Connection timed out", + watchdog: { phase }, + } + } + const parsed = parse(error) + const classification = SessionRetry.classifyRetry(parsed) + if (!classification) return { retryable: false } + if (SessionRetry.retryAction(classification) === "stop") { + ctx.terminalClassification = classification + return { retryable: false } + } + return { retryable: true, message: classification.raw } + } + + const runAttempt = Effect.fn("SessionProcessor.runAttempt")(function* () { + ctx.currentText = undefined + ctx.reasoningMap = {} + ctx.attemptCount++ + const activeTools = LLM.resolveTools(streamInput) + const attempt = ctx.runTrace.beginAttempt({ + attemptIndex: ctx.attemptCount, + at: Date.now(), + monotonicMs: performance.now(), + }) + ctx.currentAttemptID = attempt.attemptID + processAttemptID = attempt.attemptID + ctx.runTrace.recordSideEffectBoundarySnapshot({ + attemptID: attempt.attemptID, + at: Date.now(), + monotonicMs: performance.now(), + snapshot: sideEffectBoundarySnapshot(activeTools), + }) + let stream: Stream.Stream + try { + stream = llm.stream({ + ...ProviderTransform.streamTimeouts(streamInput.model), + ...streamInput, + tools: activeTools, + trace: ctx.trace, + }) + } catch (error) { + ctx.runTrace.recordSetupFailure({ at: Date.now(), monotonicMs: performance.now(), error }) + throw error + } + + yield* stream.pipe( + Stream.tap((event) => handleEvent(event, attempt.attemptID)), + // Stop draining the stream as soon as the loop gate fires a synthetic stop + // (ctx.blocked) so any trailing model text after the synthetic stop tool-error + // is dropped — the turn ends with the rendered Chinese summary alone. + Stream.takeUntil(() => ctx.needsCompaction || ctx.blocked), + Stream.runDrain, + ) + }) return yield* Effect.gen(function* () { - yield* Effect.gen(function* () { - ctx.currentText = undefined - ctx.reasoningMap = {} - ctx.attemptCount++ - const attempt = ctx.runTrace.beginAttempt({ - attemptIndex: ctx.attemptCount, + while (true) { + const result = yield* runAttempt().pipe( + Effect.onInterrupt(() => recordProcessInterrupt(processAttemptID)), + Effect.catchCauseIf( + (cause) => !Cause.hasInterruptsOnly(cause), + (cause) => Effect.fail(Cause.squash(cause)), + ), + Effect.catch((error: unknown) => Effect.succeed({ ok: false as const, error })), + ) + if (result === undefined) break + if (result.ok !== false) break + + const attemptID = processAttemptID + const retrySignal = retrySignalFor(result.error) + const decision = ctx.runTrace.recordAttemptFailureAndDeriveRecovery({ + attemptID, at: Date.now(), monotonicMs: performance.now(), + error: result.error, + evidence: retrySignal.watchdog ? ["watchdog_fired", "iterator_error"] : ["iterator_error"], + watchdog: retrySignal.watchdog, }) - ctx.currentAttemptID = attempt.attemptID - processAttemptID = attempt.attemptID - let stream: Stream.Stream - try { - stream = llm.stream({ - ...ProviderTransform.streamTimeouts(streamInput.model), - ...streamInput, - trace: ctx.trace, - }) - } catch (error) { - ctx.runTrace.recordSetupFailure({ at: Date.now(), monotonicMs: performance.now(), error }) - throw error - } - yield* stream.pipe( - Stream.tap((event) => handleEvent(event, attempt.attemptID)), - // Stop draining the stream as soon as the loop gate fires a synthetic stop - // (ctx.blocked) so any trailing model text after the synthetic stop tool-error - // is dropped — the turn ends with the rendered Chinese summary alone. - Stream.takeUntil(() => ctx.needsCompaction || ctx.blocked), - Stream.runDrain, - ) - }).pipe( - Effect.onInterrupt(() => - Effect.gen(function* () { - aborted = true - const lifecycleAction = currentLifecycleCloseAction(ctx.directory) - ctx.runTrace.recordScopeClosed({ - at: Date.now(), - monotonicMs: performance.now(), - source: "session.processor.onInterrupt", - reason: "aborted", - propagationPoint: "session.processor.process.onInterrupt", - ...(lifecycleAction ? lifecycleCloseActionMeta(lifecycleAction) : {}), - }) - ctx.trace.recordAbortState({ - provenanceSource: "session.processor.onInterrupt", - provenanceReason: "aborted", - provenanceMode: "hard", - provenanceRecordedAt: Date.now(), + if ( + attemptID && + retrySignal.retryable && + decision.recommendation === "auto_retry_once" && + automaticStreamRetriesUsed === 0 + ) { + const beforeRetry = yield* retryStillAllowed("before_backoff") + if (beforeRetry.allowed) { + automaticStreamRetriesUsed += 1 + const next = Date.now() + SAFE_RECOVERY_AUTO_RETRY_BACKOFF_MS + yield* status.set(ctx.sessionID, { + type: "retry", + attempt: ctx.attemptCount, + message: retrySignal.message ?? "Retrying interrupted stream", + next, }) - if (!ctx.assistantMessage.error) { - yield* halt(new DOMException("Aborted", "AbortError"), processAttemptID) + yield* Effect.sleep(`${SAFE_RECOVERY_AUTO_RETRY_BACKOFF_MS} millis`).pipe( + Effect.onInterrupt(() => recordProcessInterrupt(attemptID)), + ) + const afterRetry = yield* retryStillAllowed("after_backoff") + if (afterRetry.allowed) { + ctx.runTrace.recordAutoRetryAttempted({ + attemptID, + at: Date.now(), + monotonicMs: performance.now(), + }) + continue } - }), - ), - Effect.catchCauseIf( - (cause) => !Cause.hasInterruptsOnly(cause), - (cause) => Effect.fail(Cause.squash(cause)), - ), - Effect.retry( - SessionRetry.policy({ - parse, - set: (info) => - status.set(ctx.sessionID, { - type: "retry", - attempt: info.attempt, - message: info.message, - next: info.next, - }), - signalTerminal: (classification) => { - ctx.terminalClassification = classification - }, - }), - ), - Effect.catch((error) => halt(error, processAttemptID)), - Effect.ensuring(cleanup()), - ) + yield* halt(result.error, attemptID, { + recordFailure: false, + interruptionMessage: afterRetry.interruptionMessage, + }) + break + } + yield* halt(result.error, attemptID, { + recordFailure: false, + interruptionMessage: beforeRetry.interruptionMessage, + }) + break + } + + yield* halt(result.error, attemptID, { + recordFailure: false, + interruptionMessage: recoveryInterruptionMessage(decision), + }) + break + } if (ctx.needsCompaction) return "compact" if (ctx.blocked || ctx.assistantMessage.error) return "stop" return "continue" - }) + }).pipe(Effect.ensuring(cleanup())) }) const recordSyntheticBlock = Effect.fn("SessionProcessor.recordSyntheticBlock")(function* (input: { diff --git a/packages/opencode/src/session/run-incident/derive.ts b/packages/opencode/src/session/run-incident/derive.ts index 2eccb9db7..e290ac786 100644 --- a/packages/opencode/src/session/run-incident/derive.ts +++ b/packages/opencode/src/session/run-incident/derive.ts @@ -128,6 +128,9 @@ function factsFromEvidence( const materializedToolBoundary = summarizeMaterializedToolBoundaries(input.materializedToolBoundaries, attemptID) const has = (eventType: string) => scopedEvidence.some((event) => event.event_type === eventType) const count = (eventType: string) => scopedEvidence.filter((event) => event.event_type === eventType).length + const sideEffectFactsComplete = attemptID + ? scopedSideEffectFactsComplete(scopedEvidence, materializedToolBoundary) + : input.sideEffectFactsComplete return { provider_progress_seen: has("provider_progress_seen"), visible_output_seen: has("visible_output_seen"), @@ -145,7 +148,7 @@ function factsFromEvidence( materialized_tool_requires_confirmation: materializedToolBoundary ? materializedToolBoundary.effect.unsafe || !materializedToolBoundary.effect.complete : undefined, - side_effect_facts_complete: input.sideEffectFactsComplete, + side_effect_facts_complete: sideEffectFactsComplete, lifecycle_close_seen: has("lifecycle_close_seen"), user_cancel_seen: has("user_cancel_seen"), watchdog_fired: has("watchdog_fired"), @@ -153,6 +156,19 @@ function factsFromEvidence( } } +function scopedSideEffectFactsComplete( + scopedEvidence: IncidentEvidenceEvent[], + materializedToolBoundary: MaterializedToolBoundary | undefined, +) { + if (scopedEvidence.some((event) => event.event_type === "provider_executed_tool_boundary")) return false + if (materializedToolBoundary && !materializedToolBoundary.effect.complete) return false + if (scopedEvidence.some((event) => event.tool_effect_complete === false)) return false + const snapshots = scopedEvidence.flatMap((event) => + event.side_effect_boundary_snapshot ? [event.side_effect_boundary_snapshot] : [], + ) + return snapshots.every((snapshot) => snapshot.proof_result === "complete") +} + function evidenceAtOrBefore(event: IncidentEvidenceEvent, terminal: IncidentEvidenceEvent) { if (event.monotonic_ms !== undefined && terminal.monotonic_ms !== undefined) { if (event.monotonic_ms < terminal.monotonic_ms) return true diff --git a/packages/opencode/src/session/run-incident/policy.ts b/packages/opencode/src/session/run-incident/policy.ts index ced83f402..faf92c342 100644 --- a/packages/opencode/src/session/run-incident/policy.ts +++ b/packages/opencode/src/session/run-incident/policy.ts @@ -18,7 +18,7 @@ export function recoveryFor(input: { reason: "local_lifecycle_close", } } - if (!input.facts.side_effect_facts_complete) { + if (!terminalFacts.side_effect_facts_complete) { return { ...base, recommendation: "ask_user_before_retry", @@ -26,7 +26,7 @@ export function recoveryFor(input: { reason: "side_effect_facts_incomplete", } } - if (input.facts.unsafe_side_effect_started) { + if (terminalFacts.unsafe_side_effect_started) { return { ...base, recommendation: "ask_user_before_retry", @@ -34,11 +34,11 @@ export function recoveryFor(input: { reason: "unsafe_side_effect_started", } } - if (input.facts.tool_execution_started) { + if (terminalFacts.tool_execution_started) { return { ...base, recommendation: "ask_user_before_retry", confidence: "medium", reason: "tool_execution_started" } } - if (input.facts.tool_call_materialized) { - if (!input.facts.side_effect_facts_complete || input.facts.materialized_tool_effect_kind === "unknown") { + if (terminalFacts.tool_call_materialized) { + if (!terminalFacts.side_effect_facts_complete || terminalFacts.materialized_tool_effect_kind === "unknown") { return { ...base, recommendation: "ask_user_before_retry", @@ -46,7 +46,7 @@ export function recoveryFor(input: { reason: "side_effect_facts_incomplete", } } - if (input.facts.materialized_tool_requires_confirmation) { + if (terminalFacts.materialized_tool_requires_confirmation) { return { ...base, recommendation: "ask_user_before_retry", @@ -69,7 +69,7 @@ export function recoveryFor(input: { reason: "partial_tool_input_without_execution", } } - if (input.facts.visible_output_seen) { + if (terminalFacts.visible_output_seen) { return { ...base, recommendation: "offer_continue", @@ -77,7 +77,13 @@ export function recoveryFor(input: { reason: "visible_output_without_tool_execution", } } - if (input.cause.category === "provider_transport_disconnect") { + if (input.facts.user_cancel_seen) { + return { ...base, recommendation: "do_not_retry", confidence: "high", reason: "user_cancel" } + } + if (input.facts.lifecycle_close_seen) { + return { ...base, recommendation: "do_not_retry", confidence: "high", reason: "local_lifecycle_close" } + } + if (input.cause.category === "provider_transport_disconnect" || input.cause.category === "watchdog_timeout") { return { ...base, recommendation: "auto_retry_once", diff --git a/packages/opencode/src/session/run-incident/types.ts b/packages/opencode/src/session/run-incident/types.ts index 3991e58f5..ee0ba122a 100644 --- a/packages/opencode/src/session/run-incident/types.ts +++ b/packages/opencode/src/session/run-incident/types.ts @@ -5,6 +5,7 @@ import type { RunID, SafeErrorFingerprint, SafeToolName, + SideEffectBoundarySnapshot, ToolEffect, ToolEffectKind, } from "../run-observability/types" @@ -41,6 +42,7 @@ export type IncidentEvidenceEvent = { tool_effect_kind?: ToolEffectKind tool_effect_unsafe?: boolean tool_effect_complete?: boolean + side_effect_boundary_snapshot?: SideEffectBoundarySnapshot interruption_phase?: "tool_input_generation" | "tool_call_materialized_without_execution" | "tool_execution" tool_execution_started?: boolean } diff --git a/packages/opencode/src/session/run-observability/index.ts b/packages/opencode/src/session/run-observability/index.ts index c05604cd1..6d6be584a 100644 --- a/packages/opencode/src/session/run-observability/index.ts +++ b/packages/opencode/src/session/run-observability/index.ts @@ -24,6 +24,7 @@ export namespace RunObservability { export type Summary = Types.Summary export type Recorder = Types.Recorder export type RecorderInput = Types.RecorderInput + export type SideEffectBoundarySnapshot = Types.SideEffectBoundarySnapshot export type Classification = Types.Classification export type SummaryKey = Types.SummaryKey } diff --git a/packages/opencode/src/session/run-observability/recorder.ts b/packages/opencode/src/session/run-observability/recorder.ts index 424d719fd..cbc681ed0 100644 --- a/packages/opencode/src/session/run-observability/recorder.ts +++ b/packages/opencode/src/session/run-observability/recorder.ts @@ -7,6 +7,7 @@ import { type RecorderInput, RunID, SCHEMA_VERSION, + type SideEffectBoundarySnapshot, type Summary, type SummaryKey, type LifecycleKind, @@ -53,12 +54,15 @@ export function createRecorder(input: RecorderInput): Recorder { let readOnlyToolStarted = false let unsafeSideEffectStarted = false let sideEffectFactsComplete = true + let sideEffectBoundarySnapshot: SideEffectBoundarySnapshot | undefined const materializedToolBoundaries: RunIncident.MaterializedToolBoundary[] = [] let lastEventMonotonicMs = input.monotonicStartMs let failure: Failure | undefined let lifecycleFailure: ScopeClosedFailure | undefined let pendingToolPartsInterrupted = 0 const evidence: RunIncident.EvidenceEvent[] = [] + const recoveredAttemptIDs = new Set() + const recoveredIncidents: RunIncident.Summary[] = [] const rememberEvent = (monotonicMs: number) => { lastEventMonotonicMs = Math.max(lastEventMonotonicMs, monotonicMs) @@ -93,6 +97,104 @@ export function createRecorder(input: RecorderInput): Recorder { toolExecutionCompleted: has("tool_execution_completed"), } } + const terminalEvidence = () => + evidence.map((event) => { + if (!event.attempt_id || !recoveredAttemptIDs.has(event.attempt_id) || !event.terminal_candidate) return event + const { cause: _cause, ...nonTerminalEvent } = event + return { ...nonTerminalEvent, terminal_candidate: false } + }) + const deriveCurrentIncident = ( + completedAt?: number, + options?: { includeRecoveredTerminal?: boolean }, + ) => { + const incidentLifecycle = lifecycleSummary(lifecycleFailure) + return RunIncident.derive({ + runID: input.runID, + traceID: input.traceID, + sessionID: input.sessionID, + messageID: input.messageID, + parentMessageID: input.parentMessageID, + createdAt: input.createdAt, + completedAt, + evidence: options?.includeRecoveredTerminal ? evidence : terminalEvidence(), + unsafeSideEffectKinds: unsafeKinds, + sideEffectFactsComplete, + materializedToolBoundaries, + lifecycle: incidentLifecycle, + missingProvenance: lifecycleFailure && !incidentLifecycle ? ["lifecycle.close_requested"] : undefined, + }) + } + const unknownRecovery = (): RunIncident.Recovery => ({ + recommendation: "unknown", + confidence: "low", + reason: "unknown", + safety_scope: "visible_output_and_tool_side_effects", + }) + const recordTransportFailureEvidence = (next: { + attemptID?: AttemptID + at: number + monotonicMs: number + error: unknown + evidence?: string[] + }) => { + const error = safeErrorFingerprint(next.error) + const factsAtFailure = transportFactsAt(next.attemptID, next.monotonicMs) + failure ??= { + type: "transport", + at: next.at, + monotonicMs: next.monotonicMs, + error: next.error, + evidence: next.evidence ?? [], + attemptID: next.attemptID, + } + appendEvidence({ + monotonic_ms: next.monotonicMs, + source: "provider_stream", + attempt_id: next.attemptID, + event_type: "provider_transport_failure", + terminal_candidate: true, + confidence: error.cause_code === "UND_ERR_SOCKET" ? "high" : "medium", + error, + cause: RunIncident.transportCause({ + error, + providerProgressSeen: factsAtFailure.providerProgressSeen, + toolInputStarted: factsAtFailure.toolInputStarted, + toolInputCompleted: factsAtFailure.toolInputCompleted, + toolCallMaterialized: factsAtFailure.toolCallMaterialized, + toolExecutionStarted: factsAtFailure.toolExecutionStarted, + toolExecutionCompleted: factsAtFailure.toolExecutionCompleted, + }), + }) + rememberEvent(next.monotonicMs) + } + const recordWatchdogFailureEvidence = (next: { + attemptID?: AttemptID + at: number + monotonicMs: number + error: unknown + phase: "connect" | "silent_stream" | "unknown" + }) => { + const error = safeErrorFingerprint(next.error) + failure ??= { + type: "transport", + at: next.at, + monotonicMs: next.monotonicMs, + error: next.error, + evidence: ["watchdog_fired", "iterator_error"], + attemptID: next.attemptID, + } + appendEvidence({ + monotonic_ms: next.monotonicMs, + source: "watchdog", + attempt_id: next.attemptID, + event_type: "watchdog_fired", + terminal_candidate: true, + confidence: "high", + error, + cause: { category: "watchdog_timeout", subcategory: next.phase, confidence: "high" }, + }) + rememberEvent(next.monotonicMs) + } return { beginAttempt(next) { @@ -365,37 +467,51 @@ export function createRecorder(input: RecorderInput): Recorder { }) rememberEvent(next.monotonicMs) }, - recordTransportFailure(next) { - const error = safeErrorFingerprint(next.error) - const factsAtFailure = transportFactsAt(next.attemptID, next.monotonicMs) - failure ??= { - type: "transport", - at: next.at, - monotonicMs: next.monotonicMs, - error: next.error, - evidence: next.evidence ?? [], - attemptID: next.attemptID, + recordSideEffectBoundarySnapshot(next) { + sideEffectBoundarySnapshot = { ...next.snapshot } + if (next.snapshot.proof_result === "incomplete") sideEffectFactsComplete = false + appendEvidence({ + monotonic_ms: next.monotonicMs, + source: "processor", + attempt_id: next.attemptID, + event_type: "side_effect_boundary_snapshot", + terminal_candidate: false, + confidence: next.snapshot.proof_result === "complete" ? "high" : "medium", + side_effect_boundary_snapshot: { ...next.snapshot }, + }) + rememberEvent(next.monotonicMs) + }, + recordAttemptFailureAndDeriveRecovery(next) { + if (next.watchdog) { + recordWatchdogFailureEvidence({ ...next, phase: next.watchdog.phase }) + } else { + recordTransportFailureEvidence(next) } + return deriveCurrentIncident(next.at, { includeRecoveredTerminal: true })?.recovery ?? unknownRecovery() + }, + recordAutoRetryAttempted(next) { appendEvidence({ monotonic_ms: next.monotonicMs, - source: "provider_stream", + source: "recovery", attempt_id: next.attemptID, - event_type: "provider_transport_failure", - terminal_candidate: true, - confidence: error.cause_code === "UND_ERR_SOCKET" ? "high" : "medium", - error, - cause: RunIncident.transportCause({ - error, - providerProgressSeen: factsAtFailure.providerProgressSeen, - toolInputStarted: factsAtFailure.toolInputStarted, - toolInputCompleted: factsAtFailure.toolInputCompleted, - toolCallMaterialized: factsAtFailure.toolCallMaterialized, - toolExecutionStarted: factsAtFailure.toolExecutionStarted, - toolExecutionCompleted: factsAtFailure.toolExecutionCompleted, - }), + event_type: "auto_retry_attempted", + terminal_candidate: false, + confidence: "high", }) + const incident = deriveCurrentIncident(next.at, { includeRecoveredTerminal: true }) + if (incident && !recoveredIncidents.some((entry) => entry.phase.terminal_attempt_id === next.attemptID)) { + recoveredIncidents.push({ + ...incident, + incident_id: `${incident.incident_id}:recovered:${next.attemptID}`, + }) + } + recoveredAttemptIDs.add(next.attemptID) + if (failure && "attemptID" in failure && failure.attemptID === next.attemptID) failure = undefined rememberEvent(next.monotonicMs) }, + recordTransportFailure(next) { + recordTransportFailureEvidence(next) + }, recordSetupFailure(next) { failure ??= { type: "setup", at: next.at, monotonicMs: next.monotonicMs, error: next.error } const error = safeErrorFingerprint(next.error) @@ -455,22 +571,7 @@ export function createRecorder(input: RecorderInput): Recorder { }, finalize(final) { const lifecycle = lifecycleSummary(lifecycleFailure) - const incidentLifecycle = lifecycleSummary(lifecycleFailure) - const incident = RunIncident.derive({ - runID: input.runID, - traceID: input.traceID, - sessionID: input.sessionID, - messageID: input.messageID, - parentMessageID: input.parentMessageID, - createdAt: input.createdAt, - completedAt: final.completedAt, - evidence, - unsafeSideEffectKinds: unsafeKinds, - sideEffectFactsComplete, - materializedToolBoundaries, - lifecycle: incidentLifecycle, - missingProvenance: lifecycleFailure && !lifecycle ? ["lifecycle.close_requested"] : undefined, - }) + const incident = deriveCurrentIncident(final.completedAt) const classification = incident ? classificationForIncident(incident.terminal_cause) : classify(failure) const missingProvenance = classification === "unknown_scope_close" ? ["lifecycle.close_requested"] : undefined const summaryKey = summaryKeyFor( @@ -479,16 +580,19 @@ export function createRecorder(input: RecorderInput): Recorder { ? summarySuffixForIncident(incident.terminal_cause, { providerProgressSeen }) : summarySuffix({ failure, providerProgressSeen }), ) + const terminalAttempt = incident?.phase.terminal_attempt_id ? getAttempt(incident.phase.terminal_attempt_id) : undefined const retrySafety = retrySafetyFor({ classification, - visibleOutputSeen, - toolExecutionStarted, - unsafeSideEffectStarted, + visibleOutputSeen: terminalAttempt?.visible_output_seen ?? visibleOutputSeen, + toolExecutionStarted: terminalAttempt?.tool_execution_started ?? toolExecutionStarted, + unsafeSideEffectStarted: terminalAttempt?.unsafe_side_effect_started ?? unsafeSideEffectStarted, }) const completedAt = final.completedAt const failureMonotonicMs = failure?.monotonicMs const error = failure && "error" in failure ? safeErrorFingerprint(failure.error) : undefined - const terminalAttemptID = failure && "attemptID" in failure ? failure.attemptID : attempts.at(-1)?.attempt_id + const terminalAttemptID = incident + ? (incident.phase.terminal_attempt_id ?? (failure && "attemptID" in failure ? failure.attemptID : undefined)) + : undefined return { schema_version: SCHEMA_VERSION, run_id: input.runID, @@ -516,8 +620,10 @@ export function createRecorder(input: RecorderInput): Recorder { unsafe_side_effect_started: unsafeSideEffectStarted, unsafe_side_effect_kinds: unsafeKinds, side_effect_facts_complete: sideEffectFactsComplete, + side_effect_boundary_snapshot: sideEffectBoundarySnapshot ? { ...sideEffectBoundarySnapshot } : undefined, pending_tool_parts_interrupted: pendingToolPartsInterrupted || undefined, incident, + recovered_incidents: recoveredIncidents.length ? recoveredIncidents : undefined, lifecycle, missing_provenance: missingProvenance, durations_ms: { @@ -559,6 +665,7 @@ function classify(failure: Failure | undefined): Classification { function classificationForIncident(cause: RunIncident.TerminalCause): Classification { switch (cause.category) { case "provider_transport_disconnect": + case "watchdog_timeout": return "external_stream_disconnect" case "local_lifecycle_close": if (cause.subcategory === "unknown_lifecycle_close") return "unknown_scope_close" @@ -581,6 +688,7 @@ function classificationForIncident(cause: RunIncident.TerminalCause): Classifica } function summarySuffixForIncident(cause: RunIncident.TerminalCause, input: { providerProgressSeen: boolean }) { + if (cause.category === "watchdog_timeout") return "watchdog_timeout" if (cause.category === "provider_transport_disconnect") { if (!input.providerProgressSeen) return "transport_failure" if (cause.subcategory === "during_tool_input_generation") return "provider_progress_tool_input_disconnect" diff --git a/packages/opencode/src/session/run-observability/types.ts b/packages/opencode/src/session/run-observability/types.ts index 71b21fac4..2789e3ce8 100644 --- a/packages/opencode/src/session/run-observability/types.ts +++ b/packages/opencode/src/session/run-observability/types.ts @@ -65,6 +65,22 @@ export type ToolEffect = { complete: boolean } +export type SideEffectBoundarySnapshot = { + exposed_tool_count: number + unknown_tool_count: number + unclassified_effect_count: number + provider_executed_capability_present: boolean + external_boundary_present: boolean + proof_result: "complete" | "incomplete" + proof_reason: + | "all_boundaries_classified" + | "unknown_tool_boundary" + | "unclassified_effect" + | "provider_executed_capability" + | "external_boundary" + | "unknown" +} + export type AttemptSummary = { attempt_id: AttemptID attempt_index: number @@ -108,8 +124,10 @@ export type Summary = { unsafe_side_effect_started: boolean unsafe_side_effect_kinds: ToolEffectKind[] side_effect_facts_complete: boolean + side_effect_boundary_snapshot?: SideEffectBoundarySnapshot pending_tool_parts_interrupted?: number incident?: RunIncident.Summary + recovered_incidents?: RunIncident.Summary[] lifecycle?: { action_id: string kind: LifecycleKind @@ -188,6 +206,21 @@ export type Recorder = { interruptionPhase?: RunIncident.EvidenceEvent["interruption_phase"] toolExecutionStarted?: boolean }): void + recordSideEffectBoundarySnapshot(input: { + attemptID?: AttemptID + at: number + monotonicMs: number + snapshot: SideEffectBoundarySnapshot + }): void + recordAttemptFailureAndDeriveRecovery(input: { + attemptID?: AttemptID + at: number + monotonicMs: number + error: unknown + evidence?: string[] + watchdog?: { phase: "connect" | "silent_stream" | "unknown" } + }): RunIncident.Recovery + recordAutoRetryAttempted(input: { attemptID: AttemptID; at: number; monotonicMs: number }): void recordTransportFailure(input: { attemptID?: AttemptID at: number diff --git a/packages/opencode/test/session/export.test.ts b/packages/opencode/test/session/export.test.ts index 98c13cdf4..aabdeb55e 100644 --- a/packages/opencode/test/session/export.test.ts +++ b/packages/opencode/test/session/export.test.ts @@ -1995,6 +1995,164 @@ describe("redactPart", () => { expect(serialized).not.toContain("sk-secret") }) + test("exports sanitized side-effect boundary snapshots without raw request data", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_side_effect_snapshot_sanitize"), + traceID: MessageID.make("msg_side_effect_snapshot_sanitize"), + sessionID: SessionID.make("ses_side_effect_snapshot_sanitize"), + messageID: MessageID.make("msg_side_effect_snapshot_sanitize"), + providerID: "test", + modelID: "test-model", + createdAt: 1, + monotonicStartMs: 100, + }) + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 2, monotonicMs: 110 }) + recorder.recordSideEffectBoundarySnapshot({ + attemptID: attempt.attemptID, + at: 3, + monotonicMs: 120, + snapshot: { + exposed_tool_count: 1, + unknown_tool_count: 1, + unclassified_effect_count: 1, + provider_executed_capability_present: false, + external_boundary_present: false, + proof_result: "incomplete", + proof_reason: "unknown_tool_boundary", + }, + }) + recorder.recordAttemptFailureAndDeriveRecovery({ + attemptID: attempt.attemptID, + at: 4, + monotonicMs: 130, + error: new Error("raw provider body with /Users/alice/project and sk-secret"), + evidence: ["iterator_error"], + }) + const summary = recorder.finalize({ completedAt: 5, monotonicMs: 140 }) + + const sanitized = Export.sanitizeSnapshot({ + schema_version: 1, + format: "pawwork-session-export", + exported_at: 1, + root_session_id: SessionID.make("ses_side_effect_snapshot_sanitize"), + runtime_context: { + app_version: "test", + runtime_namespace: "pawwork", + platform: process.platform, + os_version: "test", + locale: "en-US", + timezone: "UTC", + instruction_sources: [], + model_refs: {}, + stats: { session_count: 1, message_count: 0, part_count: 0, omitted_attachment_count: 0 }, + }, + diagnostics: { run_observability_schema_version: 1, run_observability: [summary] }, + session: { + info: { + id: SessionID.make("ses_side_effect_snapshot_sanitize"), + version: "0.0.0", + time: { created: 1, updated: 1 }, + title: "x", + directory: "/tmp/project", + } as SessionNs.Info, + had_cloud_share: false, + diffs: [], + messages: [], + children: [], + }, + }) + + expect(sanitized.diagnostics.run_observability?.[0]?.side_effect_boundary_snapshot).toMatchObject({ + exposed_tool_count: 1, + unknown_tool_count: 1, + unclassified_effect_count: 1, + proof_result: "incomplete", + proof_reason: "unknown_tool_boundary", + }) + expect( + sanitized.diagnostics.run_incidents?.[0]?.evidence?.find( + (event) => event.event_type === "side_effect_boundary_snapshot", + )?.side_effect_boundary_snapshot, + ).toMatchObject({ + proof_result: "incomplete", + proof_reason: "unknown_tool_boundary", + }) + const serialized = JSON.stringify(sanitized.diagnostics) + expect(serialized).toContain("unknown_tool_boundary") + expect(serialized).not.toContain("/Users/alice") + expect(serialized).not.toContain("sk-secret") + expect(serialized).not.toContain("raw provider body") + }) + + test("exports recovered stream incidents without making final run observability terminal", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_recovered_incident_export"), + traceID: MessageID.make("msg_recovered_incident_export"), + sessionID: SessionID.make("ses_recovered_incident_export"), + messageID: MessageID.make("msg_recovered_incident_export"), + providerID: "test", + modelID: "test-model", + createdAt: 1, + monotonicStartMs: 100, + }) + const first = recorder.beginAttempt({ attemptIndex: 1, at: 2, monotonicMs: 110 }) + recorder.recordAttemptFailureAndDeriveRecovery({ + attemptID: first.attemptID, + at: 3, + monotonicMs: 120, + error: new Error("LLM stream connection timed out after 120000ms without provider progress"), + evidence: ["watchdog_fired", "iterator_error"], + watchdog: { phase: "connect" }, + }) + recorder.recordAutoRetryAttempted({ attemptID: first.attemptID, at: 4, monotonicMs: 130 }) + const second = recorder.beginAttempt({ attemptIndex: 2, at: 5, monotonicMs: 140 }) + recorder.recordVisibleOutput({ attemptID: second.attemptID, at: 6, monotonicMs: 150 }) + const summary = recorder.finalize({ completedAt: 7, monotonicMs: 160 }) + + const sanitized = Export.sanitizeSnapshot({ + schema_version: 1, + format: "pawwork-session-export", + exported_at: 1, + root_session_id: SessionID.make("ses_recovered_incident_export"), + runtime_context: { + app_version: "test", + runtime_namespace: "pawwork", + platform: process.platform, + os_version: "test", + locale: "en-US", + timezone: "UTC", + instruction_sources: [], + model_refs: {}, + stats: { session_count: 1, message_count: 0, part_count: 0, omitted_attachment_count: 0 }, + }, + diagnostics: { run_observability_schema_version: 1, run_observability: [summary] }, + session: { + info: { + id: SessionID.make("ses_recovered_incident_export"), + version: "0.0.0", + time: { created: 1, updated: 1 }, + title: "x", + directory: "/tmp/project", + } as SessionNs.Info, + had_cloud_share: false, + diffs: [], + messages: [], + children: [], + }, + }) + + expect(sanitized.diagnostics.run_observability?.[0]?.classification).toBe("success") + expect(sanitized.diagnostics.run_observability?.[0]?.incident).toBeUndefined() + expect(sanitized.diagnostics.run_observability?.[0]?.recovered_incidents?.[0]?.terminal_cause).toMatchObject({ + category: "watchdog_timeout", + subcategory: "connect", + }) + expect(sanitized.diagnostics.run_incidents?.[0]?.terminal_cause).toMatchObject({ + category: "watchdog_timeout", + subcategory: "connect", + }) + }) + test("sanitizes generated lifecycle incident provenance and chains", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_generated_chain_sanitize"), diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 2c0f10576..3891d1e66 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -17,6 +17,7 @@ import { LLM } from "../../src/session/llm" import { MessageV2 } from "../../src/session/message-v2" import { SessionProcessor } from "../../src/session/processor" import { SessionDiagnostics } from "../../src/session/diagnostics" +import { createLifecycleCloseAction, withLifecycleCloseAction } from "../../src/session/lifecycle-provenance" import { MessageID, PartID, SessionID } from "../../src/session/schema" import { SessionStatus } from "../../src/session/status" import { SessionSummary } from "../../src/session/summary" @@ -986,6 +987,65 @@ it.live("session.processor effect tests retry recognized structured json errors" ), ) +it.live("retryable API errors stop after one safe recovery retry", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const seen = defer() + const { processors, session, provider } = yield* boot() + const bus = yield* Bus.Service + + yield* llm.error(503, { error: "temporarily unavailable" }) + yield* llm.error(503, { error: "still unavailable" }) + yield* llm.text("third attempt should not run") + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "retry api twice") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const off = yield* bus.subscribeCallback(Session.Event.Error, (evt) => { + if (evt.properties.sessionID !== chat.id) return + if (!evt.properties.error) return + seen.resolve() + }) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + const value = yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies MessageV2.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "retry api twice" }], + tools: {}, + }) + + yield* Effect.promise(() => seen.promise) + const parts = MessageV2.parts(msg.id) + off() + + expect(value).toBe("stop") + expect(yield* llm.calls).toBe(2) + expect(parts.some((part) => part.type === "text" && part.text === "third attempt should not run")).toBe(false) + expect(handle.message.error?.name).toBe("APIError") + expect(handle.message.diagnostics?.run_observability?.attempts).toHaveLength(2) + expect(handle.message.diagnostics?.run_observability?.recovered_incidents).toHaveLength(1) + }), + { git: true, config: (url) => providerCfg(url) }, + ), +) + it.live("session.processor effect tests publish retry status updates", () => provideTmpdirServer( ({ dir, llm }) => @@ -1038,6 +1098,391 @@ it.live("session.processor effect tests publish retry status updates", () => ), ) +it.live("connect timeout before provider progress auto retries once and succeeds", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + + yield* llm.hang + yield* llm.text("after retry") + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "auto retry connect timeout") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + const value = yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies MessageV2.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "auto retry connect timeout" }], + tools: {}, + connectTimeoutMs: 20, + streamTimeoutMs: 1_000, + }) + + const parts = MessageV2.parts(msg.id) + const stored = (yield* session.messages({ sessionID: chat.id })).find( + (message) => message.info.role === "assistant" && message.info.id === msg.id, + ) + + expect(value).toBe("continue") + expect(yield* llm.calls).toBe(2) + expect(parts.some((part) => part.type === "text" && part.text === "after retry")).toBe(true) + expect(handle.message.error).toBeUndefined() + expect(stored?.info.role).toBe("assistant") + if (stored?.info.role === "assistant") { + expect(stored.info.error).toBeUndefined() + const observability = stored.info.diagnostics?.run_observability + expect(observability?.classification).toBe("success") + expect(String(observability?.summary_key)).toBe("success.completed") + expect(observability?.terminal_attempt_id).toBeUndefined() + expect(observability?.incident).toBeUndefined() + expect(observability?.recovered_incidents?.[0]?.terminal_cause).toMatchObject({ + category: "watchdog_timeout", + subcategory: "connect", + }) + expect(observability?.recovered_incidents?.[0]?.recovery).toMatchObject({ + recommendation: "auto_retry_once", + reason: "no_visible_output_or_tool_execution", + }) + expect(observability?.attempts).toHaveLength(2) + expect(observability?.attempts[0]).toMatchObject({ + attempt_index: 1, + provider_progress_seen: false, + visible_output_seen: false, + tool_call_materialized: false, + tool_execution_started: false, + }) + expect(observability?.attempts[1]).toMatchObject({ + attempt_index: 2, + visible_output_seen: true, + }) + } + }), + { git: true, config: (url) => providerCfg(url) }, + ), +) + +it.live("connect timeout auto retry stops if lifecycle closes during backoff", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const retrySeen = defer() + const { processors, session, provider } = yield* boot() + const bus = yield* Bus.Service + + yield* llm.hang + yield* llm.text("should not run") + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "auto retry lifecycle close") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const off = yield* bus.subscribeCallback(SessionStatus.Event.Status, (evt) => { + if (evt.properties.sessionID !== chat.id) return + if (evt.properties.status.type === "retry") retrySeen.resolve() + }) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + const run = yield* handle + .process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies MessageV2.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "auto retry lifecycle close" }], + tools: {}, + connectTimeoutMs: 20, + streamTimeoutMs: 1_000, + }) + .pipe(Effect.forkChild) + + yield* Effect.promise(() => retrySeen.promise) + const action = createLifecycleCloseAction("instance_reload", { + affectedDirectories: [path.resolve(dir)], + origin: { source: "runtime", operation: "instance.reload", reason: "test_retry_backoff" }, + }) + yield* Effect.promise(() => + withLifecycleCloseAction([path.resolve(dir)], action, async () => { + await Bun.sleep(1_200) + }), + ) + const value = yield* Fiber.join(run) + off() + + const stored = (yield* session.messages({ sessionID: chat.id })).find( + (message) => message.info.role === "assistant" && message.info.id === msg.id, + ) + expect(value).toBe("stop") + expect(yield* llm.calls).toBe(1) + expect(stored?.info.role).toBe("assistant") + if (stored?.info.role === "assistant") { + expect(stored.info.error?.data.message).toContain("local lifecycle close") + expect(stored.info.error?.data.message).not.toContain("provider connection") + expect(stored.info.diagnostics?.run_observability?.incident?.facts.lifecycle_close_seen).toBe(true) + expect(stored.info.diagnostics?.run_observability?.incident?.recovery).toMatchObject({ + recommendation: "do_not_retry", + reason: "local_lifecycle_close", + }) + } + }), + { git: true, config: (url) => providerCfg(url) }, + ), +) + +it.live("connect timeout auto retry records abort if interrupted during backoff", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const retrySeen = defer() + const { processors, session, provider } = yield* boot() + const bus = yield* Bus.Service + + yield* llm.hang + yield* llm.text("should not run") + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "auto retry backoff interrupt") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const off = yield* bus.subscribeCallback(SessionStatus.Event.Status, (evt) => { + if (evt.properties.sessionID !== chat.id) return + if (evt.properties.status.type === "retry") retrySeen.resolve() + }) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + const run = yield* handle + .process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies MessageV2.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "auto retry backoff interrupt" }], + tools: {}, + connectTimeoutMs: 20, + streamTimeoutMs: 1_000, + }) + .pipe(Effect.forkChild) + + yield* Effect.promise(() => retrySeen.promise) + yield* Fiber.interrupt(run) + const exit = yield* Fiber.await(run) + off() + + const stored = (yield* session.messages({ sessionID: chat.id })).find( + (message) => message.info.role === "assistant" && message.info.id === msg.id, + ) + expect(Exit.isFailure(exit)).toBe(true) + expect(yield* llm.calls).toBe(1) + expect(handle.message.error?.name).toBe("MessageAbortedError") + expect(stored?.info.role).toBe("assistant") + if (stored?.info.role === "assistant") { + const observability = stored.info.diagnostics?.run_observability + expect(stored.info.error?.name).toBe("MessageAbortedError") + expect(observability?.classification).not.toBe("success") + expect(String(observability?.summary_key)).not.toBe("success.completed") + expect(observability?.recovered_incidents).toBeUndefined() + } + }), + { git: true, config: (url) => providerCfg(url) }, + ), +) + +it.live("disabled unknown tools do not block safe connect-timeout auto retry", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + + yield* llm.hang + yield* llm.text("after retry") + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "disabled unknown tool retry") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + const value = yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + tools: { mcp_write: false }, + } satisfies MessageV2.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "disabled unknown tool retry" }], + tools: { + read: tool({ + description: "read", + inputSchema: z.object({}), + }), + mcp_write: tool({ + description: "unknown disabled tool", + inputSchema: z.object({}), + }), + }, + connectTimeoutMs: 20, + streamTimeoutMs: 1_000, + }) + + const stored = (yield* session.messages({ sessionID: chat.id })).find( + (message) => message.info.role === "assistant" && message.info.id === msg.id, + ) + expect(value).toBe("continue") + expect(yield* llm.calls).toBe(2) + expect(stored?.info.role).toBe("assistant") + if (stored?.info.role === "assistant") { + const snapshot = stored.info.diagnostics?.run_observability?.side_effect_boundary_snapshot + expect(snapshot).toMatchObject({ + exposed_tool_count: 1, + unknown_tool_count: 0, + unclassified_effect_count: 0, + proof_result: "complete", + }) + expect(stored.info.diagnostics?.run_observability?.recovered_incidents?.[0]?.recovery).toMatchObject({ + recommendation: "auto_retry_once", + reason: "no_visible_output_or_tool_execution", + }) + } + }), + { git: true, config: (url) => providerCfg(url) }, + ), +) + +it.live("retryable stream error after visible output does not replay the assistant message", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + + yield* llm.push( + raw({ + head: [ + { + id: "chatcmpl-visible-retry-guard", + object: "chat.completion.chunk", + choices: [{ delta: { role: "assistant" } }], + }, + { + id: "chatcmpl-visible-retry-guard", + object: "chat.completion.chunk", + choices: [{ delta: { content: "visible" } }], + }, + ], + tail: [ + { + error: { + message: "stream terminated", + type: "server_error", + code: "stream_terminated", + }, + }, + ], + }), + ) + yield* llm.text("replayed") + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "visible output retry guard") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + const value = yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies MessageV2.User, + sessionID: chat.id, + model: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "visible output retry guard" }], + tools: {}, + }) + + const textParts = MessageV2.parts(msg.id).filter((part): part is MessageV2.TextPart => part.type === "text") + const stored = (yield* session.messages({ sessionID: chat.id })).find( + (message) => message.info.role === "assistant" && message.info.id === msg.id, + ) + + expect(value).toBe("stop") + expect(yield* llm.calls).toBe(1) + expect(textParts.map((part) => part.text)).toContain("visible") + expect(textParts.map((part) => part.text)).not.toContain("replayed") + expect(stored?.info.role).toBe("assistant") + if (stored?.info.role === "assistant") { + expect(stored.info.error?.data.message).toContain("interrupted after output started") + expect(stored.info.error?.data.message).not.toContain("stream terminated") + expect(stored.info.diagnostics?.run_observability?.incident?.recovery).toMatchObject({ + recommendation: "offer_continue", + reason: "visible_output_without_tool_execution", + }) + } + }), + { git: true, config: (url) => providerCfg(url) }, + ), +) + it.live("session.processor effect tests compact on structured context overflow", () => provideTmpdirServer( ({ dir, llm }) => @@ -1531,7 +1976,7 @@ it.live("session.processor effect tests record aborted errors and idle state", ( ), ) -it.live("connect timeout writes assistant info.error and flips session_status idle", () => +it.live("connect timeout writes assistant info.error and flips session_status idle after retry also fails", () => provideTmpdirServer( ({ dir, llm }) => Effect.gen(function* () { @@ -1540,6 +1985,7 @@ it.live("connect timeout writes assistant info.error and flips session_status id const bus = yield* Bus.Service const sts = yield* SessionStatus.Service + yield* llm.hang yield* llm.hang const chat = yield* session.create({}) @@ -1584,6 +2030,7 @@ it.live("connect timeout writes assistant info.error and flips session_status id off() expect(result).toBe("stop") + expect(yield* llm.calls).toBe(2) expect(handle.message.error).toBeTruthy() expect(stored.info.role).toBe("assistant") if (stored.info.role === "assistant") { diff --git a/packages/opencode/test/session/run-observability.test.ts b/packages/opencode/test/session/run-observability.test.ts index e7a14d284..25bc41834 100644 --- a/packages/opencode/test/session/run-observability.test.ts +++ b/packages/opencode/test/session/run-observability.test.ts @@ -45,6 +45,82 @@ describe("RunObservability", () => { expect(String(summary.summary_key)).toBe("external_stream_disconnect.transport_failure") }) + test("derives watchdog timeout recovery from attempt failure evidence before first provider progress", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_watchdog_connect_timeout"), + traceID: MessageID.make("msg_watchdog_connect_timeout"), + sessionID: SessionID.make("ses_watchdog_connect_timeout"), + messageID: MessageID.make("msg_watchdog_connect_timeout"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + const decision = recorder.recordAttemptFailureAndDeriveRecovery({ + attemptID: attempt.attemptID, + at: 130, + monotonicMs: 230, + error: new Error("LLM stream connection timed out after 120000ms without provider progress"), + evidence: ["watchdog_fired", "iterator_error"], + watchdog: { phase: "connect" }, + }) + + const summary = recorder.finalize({ completedAt: 131, monotonicMs: 231 }) + expect(decision).toMatchObject({ + recommendation: "auto_retry_once", + reason: "no_visible_output_or_tool_execution", + }) + expect(summary.incident?.terminal_cause).toMatchObject({ + category: "watchdog_timeout", + subcategory: "connect", + }) + expect(summary.incident?.facts.watchdog_fired).toBe(true) + expect(summary.incident?.phase).toMatchObject({ + stream_phase: "before_first_provider_progress", + tool_phase: "none", + terminal_attempt_id: attempt.attemptID, + }) + }) + + test("auto retried attempt is recovered instead of becoming the final incident", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_recovered_watchdog_timeout"), + traceID: MessageID.make("msg_recovered_watchdog_timeout"), + sessionID: SessionID.make("ses_recovered_watchdog_timeout"), + messageID: MessageID.make("msg_recovered_watchdog_timeout"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + + const first = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + recorder.recordAttemptFailureAndDeriveRecovery({ + attemptID: first.attemptID, + at: 12, + monotonicMs: 120, + error: new Error("LLM stream connection timed out after 120000ms without provider progress"), + evidence: ["watchdog_fired", "iterator_error"], + watchdog: { phase: "connect" }, + }) + recorder.recordAutoRetryAttempted({ attemptID: first.attemptID, at: 13, monotonicMs: 130 }) + const second = recorder.beginAttempt({ attemptIndex: 2, at: 14, monotonicMs: 140 }) + recorder.recordVisibleOutput({ attemptID: second.attemptID, at: 15, monotonicMs: 150 }) + + const summary = recorder.finalize({ completedAt: 16, monotonicMs: 160 }) + expect(summary.classification).toBe("success") + expect(String(summary.summary_key)).toBe("success.completed") + expect(summary.terminal_attempt_id).toBeUndefined() + expect(summary.incident).toBeUndefined() + expect(summary.recovered_incidents).toHaveLength(1) + expect(summary.recovered_incidents?.[0]?.terminal_cause).toMatchObject({ + category: "watchdog_timeout", + subcategory: "connect", + }) + }) + test("classifies completed runs without failure as success", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_success"), @@ -715,7 +791,7 @@ describe("RunObservability", () => { expect(summary.error?.message).toBe("redacted") }) - test("retry safety is denied when any earlier attempt emitted visible output", () => { + test("retry safety is denied when the failed attempt emitted visible output", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_retry_aggregate"), traceID: MessageID.make("msg_retry_aggregate"), @@ -727,11 +803,10 @@ describe("RunObservability", () => { monotonicStartMs: 100, }) - const first = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) - recorder.recordVisibleOutput({ attemptID: first.attemptID, at: 12, monotonicMs: 120 }) - const second = recorder.beginAttempt({ attemptIndex: 2, at: 20, monotonicMs: 200 }) + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + recorder.recordVisibleOutput({ attemptID: attempt.attemptID, at: 12, monotonicMs: 120 }) recorder.recordTransportFailure({ - attemptID: second.attemptID, + attemptID: attempt.attemptID, at: 21, monotonicMs: 210, error: { name: "TypeError", message: "terminated", cause: { code: "UND_ERR_SOCKET" } }, @@ -739,13 +814,63 @@ describe("RunObservability", () => { }) const summary = recorder.finalize({ completedAt: 22, monotonicMs: 220 }) - expect(summary.attempts).toHaveLength(2) + expect(summary.attempts).toHaveLength(1) expect(summary.visible_output_seen).toBe(true) expect(summary.retry_safety.recommendation).toBe("do_not_auto_retry") expect(summary.retry_safety.reason).toBe("visible_output_seen") }) - test("terminal attempt facts drive transport cause while run facts drive recovery safety", () => { + test("independent assistant messages do not share tool execution facts for recovery safety", () => { + const previous = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_previous_assistant_step"), + traceID: MessageID.make("msg_previous_assistant_step"), + sessionID: SessionID.make("ses_same_session"), + messageID: MessageID.make("msg_previous_assistant_step"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + const previousAttempt = previous.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + previous.recordToolExecutionStarted({ + attemptID: previousAttempt.attemptID, + at: 12, + monotonicMs: 120, + toolName: RunObservability.safeToolName("read"), + effect: RunObservability.toolEffect("read"), + }) + previous.recordToolCompleted({ attemptID: previousAttempt.attemptID, at: 13, monotonicMs: 130 }) + const previousSummary = previous.finalize({ completedAt: 14, monotonicMs: 140 }) + expect(previousSummary.tool_execution_started).toBe(true) + + const current = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_current_assistant_step"), + traceID: MessageID.make("msg_current_assistant_step"), + sessionID: SessionID.make("ses_same_session"), + messageID: MessageID.make("msg_current_assistant_step"), + parentMessageID: MessageID.make("msg_previous_assistant_step"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 20, + monotonicStartMs: 200, + }) + const currentAttempt = current.beginAttempt({ attemptIndex: 1, at: 21, monotonicMs: 210 }) + const decision = current.recordAttemptFailureAndDeriveRecovery({ + attemptID: currentAttempt.attemptID, + at: 22, + monotonicMs: 220, + error: new Error("LLM stream connection timed out after 120000ms without provider progress"), + evidence: ["watchdog_fired", "iterator_error"], + watchdog: { phase: "connect" }, + }) + + expect(decision).toMatchObject({ + recommendation: "auto_retry_once", + reason: "no_visible_output_or_tool_execution", + }) + }) + + test("terminal attempt facts drive transport recovery while message facts stay diagnostic", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_attempt_scoped_terminal"), traceID: MessageID.make("msg_attempt_scoped_terminal"), @@ -778,9 +903,17 @@ describe("RunObservability", () => { tool_phase: "none", terminal_attempt_id: second.attemptID, }) + expect(summary.incident?.facts).toMatchObject({ + visible_output_seen: true, + tool_input_started: true, + }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "offer_continue", - reason: "visible_output_without_tool_execution", + recommendation: "auto_retry_once", + reason: "no_visible_output_or_tool_execution", + }) + expect(summary.retry_safety).toMatchObject({ + recommendation: "candidate_safe_auto_retry", + reason: "no_visible_output_or_tool_execution", }) }) @@ -814,6 +947,67 @@ describe("RunObservability", () => { expect(summary.incident?.evidence?.map((event) => event.event_type)).toContain("provider_executed_tool_boundary") }) + test("unknown request side-effect boundary prevents auto retry before local tool events", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_unknown_request_boundary"), + traceID: MessageID.make("msg_unknown_request_boundary"), + sessionID: SessionID.make("ses_unknown_request_boundary"), + messageID: MessageID.make("msg_unknown_request_boundary"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + recorder.recordSideEffectBoundarySnapshot({ + attemptID: attempt.attemptID, + at: 12, + monotonicMs: 120, + snapshot: { + exposed_tool_count: 1, + unknown_tool_count: 1, + unclassified_effect_count: 1, + provider_executed_capability_present: false, + external_boundary_present: false, + proof_result: "incomplete", + proof_reason: "unknown_tool_boundary", + }, + }) + const decision = recorder.recordAttemptFailureAndDeriveRecovery({ + attemptID: attempt.attemptID, + at: 13, + monotonicMs: 130, + error: { + name: "TypeError", + message: "terminated", + cause: { name: "SocketError", message: "other side closed", code: "UND_ERR_SOCKET" }, + }, + evidence: ["iterator_error"], + }) + + const summary = recorder.finalize({ completedAt: 14, monotonicMs: 140 }) + expect(summary.visible_output_seen).toBe(false) + expect(summary.tool_call_materialized).toBe(false) + expect(summary.tool_execution_started).toBe(false) + expect(summary.side_effect_facts_complete).toBe(false) + expect(summary.side_effect_boundary_snapshot).toMatchObject({ + exposed_tool_count: 1, + unknown_tool_count: 1, + unclassified_effect_count: 1, + proof_result: "incomplete", + proof_reason: "unknown_tool_boundary", + }) + expect(decision).toMatchObject({ + recommendation: "ask_user_before_retry", + reason: "side_effect_facts_incomplete", + }) + expect(summary.incident?.recovery).toMatchObject({ + recommendation: "ask_user_before_retry", + reason: "side_effect_facts_incomplete", + }) + }) + test("transport failure after tool input end is not classified as text generation", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_after_tool_input_end"), @@ -1459,7 +1653,7 @@ describe("RunObservability", () => { }) }) - test("earlier attempt unsafe materialized tool prevents later auto retry", () => { + test("earlier attempt unsafe materialized tool does not prevent later clean attempt auto retry", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_cross_attempt_unsafe_materialized_tool"), traceID: MessageID.make("msg_cross_attempt_unsafe_materialized_tool"), @@ -1492,12 +1686,12 @@ describe("RunObservability", () => { subcategory: "before_first_provider_progress", }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "ask_user_before_retry", - reason: "tool_call_materialized_without_execution", + recommendation: "auto_retry_once", + reason: "no_visible_output_or_tool_execution", }) }) - test("earlier attempt safe materialized tool prevents later auto retry", () => { + test("earlier attempt safe materialized tool does not prevent later clean attempt auto retry", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_cross_attempt_safe_materialized_tool"), traceID: MessageID.make("msg_cross_attempt_safe_materialized_tool"), @@ -1526,12 +1720,12 @@ describe("RunObservability", () => { const summary = recorder.finalize({ completedAt: 22, monotonicMs: 220 }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "offer_continue", - reason: "tool_call_materialized_without_execution", + recommendation: "auto_retry_once", + reason: "no_visible_output_or_tool_execution", }) }) - test("earlier attempt unknown materialized tool prevents later auto retry", () => { + test("earlier attempt unknown materialized tool does not prevent later clean attempt auto retry", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_cross_attempt_unknown_materialized_tool"), traceID: MessageID.make("msg_cross_attempt_unknown_materialized_tool"), @@ -1561,8 +1755,8 @@ describe("RunObservability", () => { const summary = recorder.finalize({ completedAt: 23, monotonicMs: 230 }) expect(summary.incident?.recovery).toMatchObject({ - recommendation: "ask_user_before_retry", - reason: "side_effect_facts_incomplete", + recommendation: "auto_retry_once", + reason: "no_visible_output_or_tool_execution", }) })