diff --git a/packages/opencode/src/session/export.ts b/packages/opencode/src/session/export.ts index dd8744858..0466fb1e3 100644 --- a/packages/opencode/src/session/export.ts +++ b/packages/opencode/src/session/export.ts @@ -26,6 +26,7 @@ import { Glob } from "@/util/glob" import { safeToolFailureMetadata } from "./tool-failure" import { LLMTrace } from "./llm-trace" import { safeErrorFingerprint, safeProviderCorrelation } from "./llm-trace/stream-diagnostics" +import { RunObservability } from "./run-observability" export function getRuntimeNamespace(): "pawwork" | "opencode" { return Runtime.isPawWork() ? "pawwork" : "opencode" @@ -155,6 +156,8 @@ export namespace Export { } llm_trace_schema_version?: typeof LLMTrace.SCHEMA_VERSION llm_traces?: LLMTrace.Summary[] + run_observability_schema_version?: typeof RunObservability.SCHEMA_VERSION + run_observability?: RunObservability.Summary[] aborts?: Array<{ session_id: SessionID message_id: MessageID @@ -205,6 +208,8 @@ export namespace Export { } llm_trace_schema_version?: typeof LLMTrace.SCHEMA_VERSION llm_traces?: LLMTrace.Summary[] + run_observability_schema_version?: typeof RunObservability.SCHEMA_VERSION + run_observability?: RunObservability.Summary[] aborts?: Array<{ session_id: SessionID message_id: MessageID @@ -289,11 +294,15 @@ export namespace Export { } walk(node) const llm_traces = collectLLMTraces(node) + const run_observability = collectRunObservability(node) const aborts = collectAbortDiagnostics(node) const title_generations = collectTitleGenerations(node) return { ...(last ? { loop: { last } } : {}), ...(llm_traces.length ? { llm_trace_schema_version: LLMTrace.SCHEMA_VERSION, llm_traces } : {}), + ...(run_observability.length + ? { run_observability_schema_version: RunObservability.SCHEMA_VERSION, run_observability } + : {}), ...(aborts.length ? { aborts } : {}), ...(title_generations.length ? { title_generations } : {}), } @@ -316,6 +325,23 @@ export namespace Export { }) } + function collectRunObservability(node: Tree) { + const traces: RunObservability.Summary[] = [] + const walk = (t: Tree) => { + for (const message of t.messages ?? []) { + if (message.info.role !== "assistant") continue + const summary = message.info.diagnostics?.run_observability + if (summary) traces.push(summary) + } + for (const child of t.children ?? []) walk(child) + } + walk(node) + return traces.sort((a, b) => { + if (a.session_id !== b.session_id) return a.session_id.localeCompare(b.session_id) + return a.message_id.localeCompare(b.message_id) + }) + } + function collectAbortDiagnostics(node: Tree) { const aborts: NonNullable = [] const walk = (t: Tree) => { @@ -945,6 +971,9 @@ export namespace Export { llm_trace: msg.info.diagnostics.llm_trace ? sanitizeLLMTrace(msg.info.diagnostics.llm_trace) : undefined, + run_observability: msg.info.diagnostics.run_observability + ? sanitizeRunObservability(msg.info.diagnostics.run_observability) + : undefined, }, }, parts: msg.parts.map(partFn), @@ -998,6 +1027,14 @@ export namespace Export { : redact("title-generation-error-message", String(index), trace.error_message), })), llm_traces: diagnostics.llm_traces?.map(sanitizeLLMTrace), + run_observability: diagnostics.run_observability?.map(sanitizeRunObservability), + } + } + + function sanitizeRunObservability(summary: RunObservability.Summary): RunObservability.Summary { + return { + ...summary, + error: summary.error, } } diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 7001a8386..0651d33c7 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -19,6 +19,7 @@ import { Effect } from "effect" import { EffectLogger } from "@/effect" import { isMedia } from "@/util/media" import { LLMTrace } from "./llm-trace" +import { RunObservability } from "./run-observability" export { isMedia } from "@/util/media" function truncateToolOutput(text: string, maxChars?: number) { @@ -275,9 +276,7 @@ export const SubtaskPart = PartBase.extend({ parent_message_id: z.string().optional(), subagent_session_id: z.string().optional(), // lifecycle (NEW) - status: z - .enum(["running", "completed", "completed_empty", "failed", "canceled_by_user"]) - .default("completed"), + status: z.enum(["running", "completed", "completed_empty", "failed", "canceled_by_user"]).default("completed"), started_at: z.number().optional(), updated_at: z.number().optional(), ended_at: z.number().optional(), @@ -492,10 +491,7 @@ export const Assistant = Base.extend({ agent: z.string(), // Pre-design messages serialised path as a single absolute string. Readers lift it to // {cwd, root} where cwd === root; writers still emit only the modern object shape. - path: z.union([ - z.object({ cwd: z.string(), root: z.string() }), - z.string().transform((s) => ({ cwd: s, root: s })), - ]), + path: z.union([z.object({ cwd: z.string(), root: z.string() }), z.string().transform((s) => ({ cwd: s, root: s }))]), summary: z.boolean().optional(), cost: z.number(), tokens: z.object({ @@ -514,6 +510,7 @@ export const Assistant = Base.extend({ diagnostics: z .object({ llm_trace: LLMTrace.Summary.optional(), + run_observability: z.any().optional(), abort: z .object({ source: z.string().optional(), @@ -887,10 +884,7 @@ export const toModelMessagesEffect = Effect.fnUntraced(function* ( state: "output-error", toolCallId: part.callID, input: part.state.input, - errorText: formatToolFailureForModel( - part.state.error, - part.state.metadata?.diagnostics?.failure, - ), + errorText: formatToolFailureForModel(part.state.error, part.state.metadata?.diagnostics?.failure), ...(part.metadata?.providerExecuted ? { providerExecuted: true } : {}), ...(differentModel ? {} : { callProviderMetadata: providerMeta(part.metadata) }), }) diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index bbfd15a76..59132168a 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -24,6 +24,7 @@ import { Log } from "@opencode-ai/core/util/log" import { isRecord } from "@/util/record" import { TurnChange } from "./turn-change" import { LLMTrace } from "./llm-trace" +import { RunObservability } from "./run-observability" const log = Log.create({ service: "session.processor" }) const TOOL_CLEANUP_TIMEOUT_MS = 1_000 @@ -47,6 +48,9 @@ export interface Handle { attachments?: MessageV2.FilePart[] }, ) => Effect.Effect + readonly recordToolExecutionStarted?: (input: { tool: string; toolCallID: string }) => Effect.Effect + readonly recordToolExecutionCompleted?: (input: { toolCallID: string }) => Effect.Effect + readonly recordToolExecutionFailed?: (input: { toolCallID: string; error?: unknown }) => Effect.Effect readonly process: (streamInput: LLM.StreamInput) => Effect.Effect readonly errorRecords: (parentID: MessageV2.Assistant["parentID"]) => SessionDiagnostics.ToolErrorRecord[] readonly syntheticBlockSigKeys: (parentID: MessageV2.Assistant["parentID"]) => string[] @@ -126,6 +130,9 @@ interface ProcessorContext extends Input { currentText: MessageV2.TextPart | undefined reasoningMap: Record trace: LLMTrace.Recorder + runTrace: RunObservability.Recorder + attemptCount: number + currentAttemptID: RunObservability.AttemptID | undefined streamError: boolean /** Set by policy() signalTerminal when free_quota_exhausted is detected. Read * and reset to undefined at the start of halt() to avoid cross-call staling. */ @@ -192,6 +199,19 @@ export const layer: Layer.Layer< variant: input.assistantMessage.variant, createdAt: input.assistantMessage.time.created, }), + runTrace: RunObservability.createRecorder({ + runID: RunObservability.makeRunID(input.assistantMessage.id), + traceID: input.assistantMessage.id, + sessionID: input.sessionID, + messageID: input.assistantMessage.id, + parentMessageID: input.assistantMessage.parentID, + providerID: input.model.providerID, + modelID: input.model.id, + createdAt: input.assistantMessage.time.created, + monotonicStartMs: performance.now(), + }), + attemptCount: 0, + currentAttemptID: undefined, streamError: false, terminalClassification: undefined, } @@ -225,6 +245,47 @@ export const layer: Layer.Layer< return { call, part } }) + const recordToolExecutionStarted = Effect.fn("SessionProcessor.recordToolExecutionStarted")(function* (input: { + tool: string + toolCallID: string + }) { + void input.toolCallID + if (!ctx.currentAttemptID) return + ctx.runTrace.recordToolExecutionStarted({ + attemptID: ctx.currentAttemptID, + at: Date.now(), + monotonicMs: performance.now(), + toolName: RunObservability.safeToolName(input.tool), + effect: RunObservability.toolEffect(input.tool), + }) + }) + + const recordToolExecutionCompleted = Effect.fn("SessionProcessor.recordToolExecutionCompleted")( + function* (input: { toolCallID: string }) { + void input.toolCallID + if (!ctx.currentAttemptID) return + ctx.runTrace.recordToolCompleted({ + attemptID: ctx.currentAttemptID, + at: Date.now(), + monotonicMs: performance.now(), + }) + }, + ) + + const recordToolExecutionFailed = Effect.fn("SessionProcessor.recordToolExecutionFailed")(function* (input: { + toolCallID: string + error?: unknown + }) { + void input.toolCallID + if (!ctx.currentAttemptID) return + ctx.runTrace.recordToolFailed({ + attemptID: ctx.currentAttemptID, + at: Date.now(), + monotonicMs: performance.now(), + error: input.error, + }) + }) + const updateToolCall = Effect.fn("SessionProcessor.updateToolCall")(function* ( toolCallID: string, update: (part: MessageV2.ToolPart) => MessageV2.ToolPart, @@ -244,7 +305,9 @@ export const layer: Layer.Layer< return part }) - const applyPendingToolUpdates = Effect.fn("SessionProcessor.applyPendingToolUpdates")(function* (toolCallID: string) { + const applyPendingToolUpdates = Effect.fn("SessionProcessor.applyPendingToolUpdates")(function* ( + toolCallID: string, + ) { const pending = ctx.pendingToolUpdates[toolCallID] if (!pending?.length) return const match = yield* readToolCall(toolCallID) @@ -311,8 +374,7 @@ export const layer: Layer.Layer< if (part.type !== "tool") return [] const loop = toolDiagnostics(part)?.loop if (!loop) return [] - const isLoopRelevant = - !!loop.errorFingerprint || loop.loopAction === "block" || loop.loopAction === "stop" + const isLoopRelevant = !!loop.errorFingerprint || loop.loopAction === "block" || loop.loopAction === "stop" if (!isLoopRelevant) return [] const targetHash = loop.targetHashIsFallback ? undefined : loop.targetHash return [ @@ -567,6 +629,19 @@ export const layer: Layer.Layer< const handleEvent = Effect.fnUntraced(function* (value: StreamEvent) { ctx.trace.observeEvent(value) + if (ctx.currentAttemptID) { + const now = Date.now() + const monotonicMs = performance.now() + if (RunObservability.isProviderProgressEvent(value)) { + ctx.runTrace.recordProviderProgress({ attemptID: ctx.currentAttemptID, at: now, monotonicMs }) + } + if (value.type === "text-start" || value.type === "text-delta" || value.type === "reasoning-start") { + ctx.runTrace.recordVisibleOutput({ attemptID: ctx.currentAttemptID, at: now, monotonicMs }) + } + if (value.type === "tool-input-start" || value.type === "tool-call") { + ctx.runTrace.recordToolCall({ attemptID: ctx.currentAttemptID, at: now, monotonicMs }) + } + } switch (value.type) { case "start": yield* status.set(ctx.sessionID, { type: "busy" }) @@ -871,9 +946,7 @@ export const layer: Layer.Layer< // (cancelled before answered), don't claim whether the user saw it // — they may have. See issue #419. const errorText = - part.tool === "question" - ? "Question cancelled before the user answered it." - : "Tool execution aborted" + part.tool === "question" ? "Question cancelled before the user answered it." : "Tool execution aborted" yield* session.updatePart({ ...part, state: { @@ -884,6 +957,13 @@ export const layer: Layer.Layer< time: { start: "time" in part.state ? part.state.time.start : end, end }, }, }) + if (ctx.currentAttemptID) { + ctx.runTrace.recordToolInterrupted({ + attemptID: ctx.currentAttemptID, + at: end, + monotonicMs: performance.now(), + }) + } } ctx.toolcalls = {} ctx.assistantMessage.time.completed = Date.now() @@ -901,6 +981,10 @@ export const layer: Layer.Layer< streamError: ctx.streamError, aborted, }), + run_observability: ctx.runTrace.finalize({ + completedAt: ctx.assistantMessage.time.completed, + monotonicMs: performance.now(), + }), } yield* session.updateMessage(ctx.assistantMessage) yield* turnChange.finalize({ sessionID: ctx.sessionID, messageID: ctx.assistantMessage.id }) @@ -909,6 +993,15 @@ export const layer: Layer.Layer< const halt = Effect.fn("SessionProcessor.halt")(function* (e: unknown) { slog.error("process", { error: errorMessage(e), stack: e instanceof Error ? e.stack : undefined }) ctx.streamError = true + if (ctx.currentAttemptID) { + ctx.runTrace.recordTransportFailure({ + attemptID: ctx.currentAttemptID, + at: Date.now(), + monotonicMs: performance.now(), + error: e, + evidence: ["iterator_error"], + }) + } // Read-then-reset: free_quota path leaves a value here; all other paths // leave it undefined. Resetting at entry guards against "previous halt @@ -949,11 +1042,24 @@ export const layer: Layer.Layer< yield* Effect.gen(function* () { ctx.currentText = undefined ctx.reasoningMap = {} - const stream = llm.stream({ - ...ProviderTransform.streamTimeouts(streamInput.model), - ...streamInput, - trace: ctx.trace, + ctx.attemptCount++ + const attempt = ctx.runTrace.beginAttempt({ + attemptIndex: ctx.attemptCount, + at: Date.now(), + monotonicMs: performance.now(), }) + ctx.currentAttemptID = 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)), @@ -967,6 +1073,13 @@ export const layer: Layer.Layer< Effect.onInterrupt(() => Effect.gen(function* () { aborted = true + ctx.runTrace.recordScopeClosed({ + at: Date.now(), + monotonicMs: performance.now(), + source: "session.processor.onInterrupt", + reason: "aborted", + propagationPoint: "session.processor.process.onInterrupt", + }) ctx.trace.recordAbortState({ provenanceSource: "session.processor.onInterrupt", provenanceReason: "aborted", @@ -1192,6 +1305,9 @@ export const layer: Layer.Layer< }, updateToolCall, completeToolCall, + recordToolExecutionStarted, + recordToolExecutionCompleted, + recordToolExecutionFailed, process, errorRecords, syntheticBlockSigKeys, diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 2b9534f9f..415312264 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -99,7 +99,11 @@ export function reconcileTitleGenerationStateAfterCompletion(input: { abortRecordedAt?: number completedAt?: number }): TitleGenerationState | undefined { - if (input.state === "in_flight" && typeof input.abortRecordedAt === "number" && typeof input.completedAt === "number") { + if ( + input.state === "in_flight" && + typeof input.abortRecordedAt === "number" && + typeof input.completedAt === "number" + ) { return input.completedAt <= input.abortRecordedAt ? "completed_before_abort" : "completed_after_abort" } return input.state @@ -320,10 +324,7 @@ export const layer = Layer.effect( } satisfies AgentPromptOps }) - const cancel = Effect.fn("SessionPrompt.cancel")(function* ( - sessionID: SessionID, - options?: { source?: string }, - ) { + const cancel = Effect.fn("SessionPrompt.cancel")(function* (sessionID: SessionID, options?: { source?: string }) { const source = options?.source ?? "session.prompt.cancel" yield* elog.info("cancel", { sessionID, source }) yield* state.cancel(sessionID, { @@ -403,7 +404,9 @@ export const layer = Layer.effect( let assistant: MessageV2.WithParts | undefined for (let attempt = 0; attempt < 50; attempt++) { const messages = yield* sessions.messages({ sessionID: input.session.id }) - assistant = messages.find((message) => message.info.role === "assistant" && message.info.parentID === firstInfo.id) + assistant = messages.find( + (message) => message.info.role === "assistant" && message.info.parentID === firstInfo.id, + ) if (assistant) break yield* Effect.sleep("10 millis") } @@ -501,7 +504,9 @@ export const layer = Layer.effect( .map((line) => line.trim()) .find((line) => line.length > 0) if (!cleaned) { - yield* recordTitleTrace({ completedAt, success: true, applied: false }).pipe(Effect.catchCause(() => Effect.void)) + yield* recordTitleTrace({ completedAt, success: true, applied: false }).pipe( + Effect.catchCause(() => Effect.void), + ) return } const t = cleaned.length > 100 ? cleaned.substring(0, 97) + "..." : cleaned @@ -791,9 +796,9 @@ NOTE: At any point in time through this workflow you should feel free to ask the const abortHandler = () => { const result = ExternalResult.abortPendingSync({ sessionID, messageID, callID }) if (!result.ok) return - run.promise( - Deferred.fail(result.deferred, new ExternalResult.Error({ reason: "aborted" })), - ).catch(() => {}) + run + .promise(Deferred.fail(result.deferred, new ExternalResult.Error({ reason: "aborted" }))) + .catch(() => {}) } const signal = options.abortSignal if (signal) { @@ -841,7 +846,24 @@ NOTE: At any point in time through this workflow you should feel free to ask the { tool: item.id, sessionID: ctx.sessionID, callID: ctx.callID }, { args }, ) - const result = yield* item.execute(args, ctx) + if (input.processor.recordToolExecutionStarted) { + yield* input.processor.recordToolExecutionStarted({ + tool: item.id, + toolCallID: options.toolCallId, + }) + } + let result: Tool.ExecuteResult + try { + result = yield* item.execute(args, ctx) + } catch (error) { + if (input.processor.recordToolExecutionFailed) { + yield* input.processor.recordToolExecutionFailed({ toolCallID: options.toolCallId, error }) + } + throw error + } + if (input.processor.recordToolExecutionCompleted) { + yield* input.processor.recordToolExecutionCompleted({ toolCallID: options.toolCallId }) + } const output = { ...result, attachments: result.attachments?.map((attachment) => ({ @@ -898,7 +920,21 @@ NOTE: At any point in time through this workflow you should feel free to ask the ) const result: Awaited>> = yield* Effect.gen(function* () { yield* ctx.ask({ permission: key, metadata: {}, patterns: ["*"], always: ["*"] }) - return yield* Effect.promise(() => execute(args, opts)) + if (input.processor.recordToolExecutionStarted) { + yield* input.processor.recordToolExecutionStarted({ tool: key, toolCallID: opts.toolCallId }) + } + try { + const result = yield* Effect.promise(() => execute(args, opts)) + if (input.processor.recordToolExecutionCompleted) { + yield* input.processor.recordToolExecutionCompleted({ toolCallID: opts.toolCallId }) + } + return result + } catch (error) { + if (input.processor.recordToolExecutionFailed) { + yield* input.processor.recordToolExecutionFailed({ toolCallID: opts.toolCallId, error }) + } + throw error + } }).pipe( Effect.withSpan("Tool.execute", { attributes: { @@ -2079,11 +2115,11 @@ NOTE: At any point in time through this workflow you should feel free to ask the "", "At reply/task closeout, if the user explicitly stated a stable long-lived preference, workflow, project convention, or durable fact worth future recall, update this MEMORY.md file with existing file editing tools.", "Only write stable explicit facts. Do not write temporary tasks, emotions, one-off decisions, guesses, or unconfirmed facts. If a new fact conflicts with an old memory, update the old memory instead of appending a contradiction.", - "Record only what the user has stated explicitly. Do not extrapolate preferences from how the user phrases questions or what tasks they request. For example, do not turn \"the user asked me to change one PR line\" into \"the user prefers small PRs.\"", + 'Record only what the user has stated explicitly. Do not extrapolate preferences from how the user phrases questions or what tasks they request. For example, do not turn "the user asked me to change one PR line" into "the user prefers small PRs."', "Never write passwords, API keys, tokens, private keys, ID/passport/license numbers, credit card/bank/CVV data, private health records, home addresses, or private phone numbers.", "Keep Profile short because it is loaded at session start. Put longer history in Archive.", - "If this is the first time you auto-write to memory and MEMORY.md was effectively empty or still only had the default template, mention it briefly and naturally in the normal reply, e.g. \"I'll remember this preference for future chats.\" After this first time, subsequent auto-writes are silent.", - "Do not show toast, dialog, or inline UI feedback. If the user explicitly asks you to remember something, acknowledge naturally in the normal reply, e.g. \"Got it, I'll remember that.\"", + 'If this is the first time you auto-write to memory and MEMORY.md was effectively empty or still only had the default template, mention it briefly and naturally in the normal reply, e.g. "I\'ll remember this preference for future chats." After this first time, subsequent auto-writes are silent.', + 'Do not show toast, dialog, or inline UI feedback. If the user explicitly asks you to remember something, acknowledge naturally in the normal reply, e.g. "Got it, I\'ll remember that."', "", ].join("\n") }).pipe( @@ -2095,7 +2131,12 @@ NOTE: At any point in time through this workflow you should feel free to ask the ), ) : undefined - const system = [...env, ...(skills ? [skills] : []), ...instructions, ...(memoryProfile ? [memoryProfile] : [])] + const system = [ + ...env, + ...(skills ? [skills] : []), + ...instructions, + ...(memoryProfile ? [memoryProfile] : []), + ] const format = lastUser.format ?? { type: "text" as const } if (format.type === "json_schema") system.push(STRUCTURED_OUTPUT_SYSTEM_PROMPT) const result = yield* handle.process({ @@ -2154,7 +2195,13 @@ NOTE: At any point in time through this workflow you should feel free to ask the const loop: (input: z.infer) => Effect.Effect = Effect.fn( "SessionPrompt.loop", )(function* (input: z.infer) { - const onInterrupt = (meta?: { source?: string; reason?: string; viaCtxAbort?: boolean; propagationPoint?: string; recordedAt?: number }) => + const onInterrupt = (meta?: { + source?: string + reason?: string + viaCtxAbort?: boolean + propagationPoint?: string + recordedAt?: number + }) => Effect.gen(function* () { interruptedSessions.add(input.sessionID) const assistant = yield* currentTurnTarget(input.sessionID) diff --git a/packages/opencode/src/session/run-observability/index.ts b/packages/opencode/src/session/run-observability/index.ts new file mode 100644 index 000000000..c05604cd1 --- /dev/null +++ b/packages/opencode/src/session/run-observability/index.ts @@ -0,0 +1,29 @@ +import { + createRecorder as createRunRecorder, + isProviderProgressEvent as isProviderProgressStreamEvent, + makeRunID as makeRunIdentifier, + summaryKeyFor as makeSummaryKey, +} from "./recorder" +import { safeToolName as makeSafeToolName, toolEffect as classifyToolEffect } from "./sanitize" +import { SCHEMA_VERSION as VERSION, RunID as RunIDSchema, AttemptID as AttemptIDSchema } from "./types" +import * as Types from "./types" + +export namespace RunObservability { + export const SCHEMA_VERSION = VERSION + export const RunID = { ...RunIDSchema, make: (value: string) => RunIDSchema.parse(value) } + export const AttemptID = { ...AttemptIDSchema, make: (value: string) => AttemptIDSchema.parse(value) } + export const createRecorder = createRunRecorder + export const makeRunID = makeRunIdentifier + export const summaryKeyFor = makeSummaryKey + export const isProviderProgressEvent = isProviderProgressStreamEvent + export const safeToolName = makeSafeToolName + export const toolEffect = classifyToolEffect + + export type RunID = Types.RunID + export type AttemptID = Types.AttemptID + export type Summary = Types.Summary + export type Recorder = Types.Recorder + export type RecorderInput = Types.RecorderInput + 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 new file mode 100644 index 000000000..6fbce870d --- /dev/null +++ b/packages/opencode/src/session/run-observability/recorder.ts @@ -0,0 +1,306 @@ +import { MessageID } from "../schema" +import { + AttemptID, + type AttemptSummary, + type Classification, + type Recorder, + type RecorderInput, + RunID, + SCHEMA_VERSION, + type Summary, + type SummaryKey, + type ToolEffectKind, +} from "./types" +import { safeErrorFingerprint } from "./sanitize" + +type AttemptMutable = AttemptSummary & { lastMonotonicMs: number } + +type Failure = + | { type: "transport"; at: number; monotonicMs: number; error: unknown; evidence: string[]; attemptID?: AttemptID } + | { type: "setup"; at: number; monotonicMs: number; error: unknown } + | { + type: "scope_closed" + at: number + monotonicMs: number + source?: string + reason?: string + lifecycleActionID?: string + } + | { type: "tool"; at: number; monotonicMs: number; error?: unknown; attemptID?: AttemptID } + +export function createRecorder(input: RecorderInput): Recorder { + const attempts: AttemptMutable[] = [] + const unsafeKinds: ToolEffectKind[] = [] + let providerProgressSeen = false + let visibleOutputSeen = false + let toolCallSeen = false + let toolExecutionStarted = false + let readOnlyToolStarted = false + let unsafeSideEffectStarted = false + let sideEffectFactsComplete = true + let lastEventMonotonicMs = input.monotonicStartMs + let failure: Failure | undefined + + const rememberEvent = (monotonicMs: number) => { + lastEventMonotonicMs = Math.max(lastEventMonotonicMs, monotonicMs) + } + const getAttempt = (attemptID: AttemptID | undefined) => attempts.find((attempt) => attempt.attempt_id === attemptID) + const updateAttempt = (attemptID: AttemptID | undefined, fn: (attempt: AttemptMutable) => void) => { + const attempt = getAttempt(attemptID) + if (attempt) fn(attempt) + } + + return { + beginAttempt(next) { + const attemptID = AttemptID.parse(`${input.runID}:attempt:${next.attemptIndex}`) + attempts.push({ + attempt_id: attemptID, + attempt_index: next.attemptIndex, + started_at: next.at, + provider_progress_seen: false, + visible_output_seen: false, + tool_call_seen: false, + tool_execution_started: false, + unsafe_side_effect_started: false, + lastMonotonicMs: next.monotonicMs, + }) + rememberEvent(next.monotonicMs) + return { attemptID } + }, + recordProviderProgress(next) { + providerProgressSeen = true + updateAttempt(next.attemptID, (attempt) => { + attempt.provider_progress_seen = true + attempt.lastMonotonicMs = Math.max(attempt.lastMonotonicMs, next.monotonicMs) + }) + rememberEvent(next.monotonicMs) + }, + recordVisibleOutput(next) { + visibleOutputSeen = true + updateAttempt(next.attemptID, (attempt) => { + attempt.visible_output_seen = true + attempt.lastMonotonicMs = Math.max(attempt.lastMonotonicMs, next.monotonicMs) + }) + rememberEvent(next.monotonicMs) + }, + recordToolCall(next) { + toolCallSeen = true + updateAttempt(next.attemptID, (attempt) => { + attempt.tool_call_seen = true + attempt.lastMonotonicMs = Math.max(attempt.lastMonotonicMs, next.monotonicMs) + }) + rememberEvent(next.monotonicMs) + }, + recordToolExecutionStarted(next) { + void next.toolName + toolExecutionStarted = true + if (next.effect.kind === "read_only") readOnlyToolStarted = true + if (!next.effect.complete) sideEffectFactsComplete = false + if (next.effect.unsafe) { + unsafeSideEffectStarted = true + if (!unsafeKinds.includes(next.effect.kind)) unsafeKinds.push(next.effect.kind) + } + updateAttempt(next.attemptID, (attempt) => { + attempt.tool_execution_started = true + attempt.unsafe_side_effect_started ||= next.effect.unsafe + attempt.lastMonotonicMs = Math.max(attempt.lastMonotonicMs, next.monotonicMs) + }) + rememberEvent(next.monotonicMs) + }, + recordToolCompleted(next) { + updateAttempt(next.attemptID, (attempt) => { + attempt.last_tool_completed_at = next.at + attempt.lastMonotonicMs = Math.max(attempt.lastMonotonicMs, next.monotonicMs) + }) + rememberEvent(next.monotonicMs) + }, + recordToolFailed(next) { + if (failure?.type === "setup" || failure?.type === "scope_closed") return + failure = { + type: "tool", + at: next.at, + monotonicMs: next.monotonicMs, + error: next.error, + attemptID: next.attemptID, + } + rememberEvent(next.monotonicMs) + }, + recordToolInterrupted(next) { + if (failure?.type === "setup" || failure?.type === "scope_closed") return + failure = { type: "tool", at: next.at, monotonicMs: next.monotonicMs, attemptID: next.attemptID } + rememberEvent(next.monotonicMs) + }, + recordTransportFailure(next) { + if (failure?.type === "scope_closed" || failure?.type === "setup" || failure?.type === "tool") return + failure = { + type: "transport", + at: next.at, + monotonicMs: next.monotonicMs, + error: next.error, + evidence: next.evidence ?? [], + attemptID: next.attemptID, + } + rememberEvent(next.monotonicMs) + }, + recordSetupFailure(next) { + if (failure?.type === "scope_closed") return + failure = { type: "setup", at: next.at, monotonicMs: next.monotonicMs, error: next.error } + rememberEvent(next.monotonicMs) + }, + recordScopeClosed(next) { + failure = { + type: "scope_closed", + at: next.at, + monotonicMs: next.monotonicMs, + source: next.source, + reason: next.reason, + lifecycleActionID: next.lifecycleActionID, + } + rememberEvent(next.monotonicMs) + }, + finalize(final) { + const classification = classify(failure) + const missingProvenance = classification === "unknown_scope_close" ? ["lifecycle.close_requested"] : undefined + const summaryKey = summaryKeyFor(classification, summarySuffix({ failure, providerProgressSeen })) + const retrySafety = retrySafetyFor({ + classification, + visibleOutputSeen, + toolExecutionStarted, + 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 + return { + schema_version: SCHEMA_VERSION, + run_id: input.runID, + trace_id: input.traceID, + session_id: input.sessionID, + message_id: input.messageID, + parent_message_id: input.parentMessageID, + provider: input.providerID, + model: input.modelID, + created_at: input.createdAt, + completed_at: completedAt, + classification, + summary_key: summaryKey, + retry_safety: retrySafety, + attempts: attempts.map(({ lastMonotonicMs, ...attempt }) => attempt), + terminal_attempt_id: terminalAttemptID, + provider_progress_seen: providerProgressSeen, + visible_output_seen: visibleOutputSeen, + tool_call_seen: toolCallSeen, + tool_execution_started: toolExecutionStarted, + read_only_tool_started: readOnlyToolStarted, + unsafe_side_effect_started: unsafeSideEffectStarted, + unsafe_side_effect_kinds: unsafeKinds, + side_effect_facts_complete: sideEffectFactsComplete, + missing_provenance: missingProvenance, + durations_ms: { + total: duration(input.monotonicStartMs, final.monotonicMs), + last_event_to_failure: + failureMonotonicMs === undefined + ? undefined + : duration(lastEventBeforeFailure(failureMonotonicMs), failureMonotonicMs), + }, + error, + } satisfies Summary + }, + } + + function lastEventBeforeFailure(failureMonotonicMs: number) { + const candidates = attempts.map((attempt) => attempt.lastMonotonicMs).filter((value) => value <= failureMonotonicMs) + return candidates.length ? Math.max(...candidates) : lastEventMonotonicMs + } +} + +function classify(failure: Failure | undefined): Classification { + if (!failure) return "success" + if (failure.type === "setup") return "request_setup_failure" + if (failure.type === "tool") return "tool_failure" + if (failure.type === "scope_closed") + return failure.lifecycleActionID ? "known_lifecycle_close" : "unknown_scope_close" + if (failure.type === "transport") return "external_stream_disconnect" + return "unknown_failure" +} + +function summarySuffix(input: { failure: Failure | undefined; providerProgressSeen: boolean }) { + if (input.failure?.type === "transport") { + const error = safeErrorFingerprint(input.failure.error) + if (input.providerProgressSeen && error.cause_code === "UND_ERR_SOCKET") return "provider_progress_socket_closed" + if (input.providerProgressSeen) return "provider_progress_transport_failure" + return "transport_failure" + } + if (input.failure?.type === "scope_closed") return "missing_lifecycle_provenance" + if (input.failure?.type === "setup") return "request_setup_failed" + if (input.failure?.type === "tool") return "tool_execution_failed" + if (!input.failure) return "completed" + return "unknown" +} + +export function summaryKeyFor(classification: Classification, suffix: string): SummaryKey { + return `${classification}.${suffix}` as SummaryKey +} + +export function isProviderProgressEvent(event: { type: string }) { + switch (event.type) { + case "text-start": + case "text-delta": + case "reasoning-start": + case "reasoning-delta": + case "tool-input-start": + case "tool-input-delta": + case "tool-call": + case "tool-result": + case "tool-error": + return true + default: + return false + } +} + +function retrySafetyFor(input: { + classification: Classification + visibleOutputSeen: boolean + toolExecutionStarted: boolean + unsafeSideEffectStarted: boolean +}): Summary["retry_safety"] { + const base = { safety_scope: "user_visible_and_tool_side_effects" as const } + if (input.classification === "success") { + return { ...base, recommendation: "unknown", confidence: "high", reason: "completed_without_failure" } + } + if (input.visibleOutputSeen) { + return { ...base, recommendation: "do_not_auto_retry", confidence: "high", reason: "visible_output_seen" } + } + if (input.unsafeSideEffectStarted) { + return { ...base, recommendation: "do_not_auto_retry", confidence: "high", reason: "unsafe_side_effect_started" } + } + if (input.toolExecutionStarted) { + return { ...base, recommendation: "ask_user", confidence: "medium", reason: "tool_execution_started" } + } + if (input.classification === "external_stream_disconnect") { + return { + ...base, + recommendation: "candidate_safe_auto_retry", + confidence: "medium", + reason: "no_visible_output_or_tool_execution", + } + } + if (input.classification === "known_lifecycle_close" || input.classification === "unknown_scope_close") { + return { + ...base, + recommendation: "do_not_auto_retry", + confidence: "medium", + reason: "local_abort_or_lifecycle_close", + } + } + return { ...base, recommendation: "unknown", confidence: "low", reason: "unknown" } +} + +function duration(start: number | undefined, end: number) { + if (start === undefined) return undefined + return Math.max(0, end - start) +} + +export const makeRunID = (messageID: MessageID): RunID => RunID.parse(`run:${messageID}`) diff --git a/packages/opencode/src/session/run-observability/sanitize.ts b/packages/opencode/src/session/run-observability/sanitize.ts new file mode 100644 index 000000000..d75bf035e --- /dev/null +++ b/packages/opencode/src/session/run-observability/sanitize.ts @@ -0,0 +1,71 @@ +import { isRecord } from "@/util/record" +import type { SafeErrorFingerprint, SafeToolName, ToolEffect } from "./types" + +export const TOOL_READ = "read" +export const TOOL_GLOB = "glob" +export const TOOL_GREP = "grep" +export const TOOL_WEBFETCH = "webfetch" +export const TOOL_APPLY_PATCH = "apply_patch" +export const TOOL_BASH = "bash" + +const READ_ONLY_TOOLS = new Set([TOOL_READ, TOOL_GLOB, TOOL_GREP, TOOL_WEBFETCH]) + +export function safeToolName(value: unknown): SafeToolName { + if (typeof value !== "string") return "unknown" as SafeToolName + const trimmed = value.trim().slice(0, 80) + if (!trimmed) return "unknown" as SafeToolName + if (/https?:\/\//i.test(trimmed) || /[/\\?]/.test(trimmed) || /token|secret|bearer|sk-|cookie/i.test(trimmed)) { + return "redacted" as SafeToolName + } + const safe = trimmed.replace(/[^a-zA-Z0-9_.:-]/g, "_") + return (safe || "unknown") as SafeToolName +} + +export function toolEffect(toolName: string): ToolEffect { + if (READ_ONLY_TOOLS.has(toolName)) { + return { kind: "read_only", unsafe: false, complete: true } + } + if (toolName === TOOL_APPLY_PATCH) return { kind: "local_file_write", unsafe: true, complete: true } + if (toolName === TOOL_BASH) return { kind: "local_process", unsafe: true, complete: true } + return { kind: "unknown", unsafe: true, complete: false } +} + +export function safeErrorFingerprint(error: unknown): SafeErrorFingerprint { + const record = isRecord(error) ? error : undefined + const cause = record && isRecord(record.cause) ? record.cause : undefined + return compact({ + name: safeLowCardinality(record?.name ?? record?.constructor?.name), + message: safeErrorMessage(record?.message ?? (typeof error === "string" ? error : undefined)), + code: safeLowCardinality(record?.code), + cause_name: safeLowCardinality(cause?.name ?? cause?.constructor?.name), + cause_message: safeErrorMessage(cause?.message), + cause_code: safeLowCardinality(cause?.code), + }) +} + +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 safeErrorMessage(value: unknown) { + if (typeof value !== "string") return undefined + const normalized = value.trim().toLowerCase() + if (!normalized) return undefined + if (normalized === "terminated") return "terminated" + if (normalized === "aborted") return "aborted" + if (normalized === "other side closed") return "other side closed" + if (normalized.includes("tool execution aborted")) return "tool execution aborted" + if (normalized.includes("socket") && normalized.includes("closed")) return "socket closed" + if (normalized.includes("timeout") || normalized.includes("timed out")) return "timeout" + if (normalized.includes("rate limit")) return "rate_limit" + if (normalized.includes("unauthorized") || normalized.includes("forbidden")) return "auth_failure" + return "redacted" +} + +function compact>(value: T): T { + return Object.fromEntries(Object.entries(value).filter(([, entry]) => entry !== undefined)) as T +} diff --git a/packages/opencode/src/session/run-observability/types.ts b/packages/opencode/src/session/run-observability/types.ts new file mode 100644 index 000000000..02a343c67 --- /dev/null +++ b/packages/opencode/src/session/run-observability/types.ts @@ -0,0 +1,155 @@ +import { MessageID, SessionID } from "../schema" +import z from "zod" + +export const SCHEMA_VERSION = 1 + +export const RunID = z.string().brand<"RunID">() +export type RunID = z.infer + +export const AttemptID = z.string().brand<"RunAttemptID">() +export type AttemptID = z.infer + +export const Classification = z.enum([ + "success", + "external_stream_disconnect", + "known_lifecycle_close", + "unknown_scope_close", + "request_setup_failure", + "tool_failure", + "unknown_failure", +]) +export type Classification = z.infer + +export const SummaryKey = z.string().brand<"RunObservabilitySummaryKey">() +export type SummaryKey = z.infer + +export type Confidence = "low" | "medium" | "high" + +export type RetrySafety = { + recommendation: "candidate_safe_auto_retry" | "do_not_auto_retry" | "ask_user" | "unknown" + confidence: Confidence + reason: + | "completed_without_failure" + | "no_visible_output_or_tool_execution" + | "visible_output_seen" + | "tool_execution_started" + | "unsafe_side_effect_started" + | "local_abort_or_lifecycle_close" + | "unknown" + safety_scope: "user_visible_and_tool_side_effects" +} + +export type SafeErrorFingerprint = { + name?: string + message?: string + code?: string + cause_name?: string + cause_message?: string + cause_code?: string +} + +export type ToolEffectKind = "read_only" | "local_file_write" | "local_process" | "unknown" +export type ToolEffect = { + kind: ToolEffectKind + unsafe: boolean + complete: boolean +} + +export type AttemptSummary = { + attempt_id: AttemptID + attempt_index: number + started_at: number + last_tool_completed_at?: number + provider_progress_seen: boolean + visible_output_seen: boolean + tool_call_seen: boolean + tool_execution_started: boolean + unsafe_side_effect_started: boolean +} + +export type Summary = { + schema_version: typeof SCHEMA_VERSION + run_id: RunID + trace_id: MessageID + session_id: SessionID + message_id: MessageID + parent_message_id?: MessageID + provider: string + model: string + created_at: number + completed_at?: number + classification: Classification + summary_key: SummaryKey + retry_safety: RetrySafety + attempts: AttemptSummary[] + terminal_attempt_id?: AttemptID + provider_progress_seen: boolean + visible_output_seen: boolean + tool_call_seen: boolean + tool_execution_started: boolean + read_only_tool_started: boolean + unsafe_side_effect_started: boolean + unsafe_side_effect_kinds: ToolEffectKind[] + side_effect_facts_complete: boolean + missing_provenance?: string[] + durations_ms: { + total?: number + last_event_to_failure?: number + } + error?: SafeErrorFingerprint +} + +export type RecorderInput = { + runID: RunID + traceID: MessageID + sessionID: SessionID + messageID: MessageID + parentMessageID?: MessageID + providerID: string + modelID: string + createdAt: number + monotonicStartMs: number +} + +export type BeginAttemptInput = { + attemptIndex: number + at: number + monotonicMs: number +} + +export type Recorder = { + beginAttempt(input: BeginAttemptInput): { attemptID: AttemptID } + recordProviderProgress(input: { attemptID: AttemptID; at: number; monotonicMs: number }): void + recordVisibleOutput(input: { attemptID: AttemptID; at: number; monotonicMs: number }): void + recordToolCall(input: { attemptID: AttemptID; at: number; monotonicMs: number }): void + recordToolExecutionStarted(input: { + attemptID: AttemptID + at: number + monotonicMs: number + toolName: SafeToolName + effect: ToolEffect + }): void + recordToolCompleted(input: { attemptID: AttemptID; at: number; monotonicMs: number }): void + recordToolFailed(input: { attemptID: AttemptID; at: number; monotonicMs: number; error?: unknown }): void + recordToolInterrupted(input: { attemptID: AttemptID; at: number; monotonicMs: number }): void + recordTransportFailure(input: { + attemptID?: AttemptID + at: number + monotonicMs: number + error: unknown + evidence?: string[] + }): void + recordSetupFailure(input: { at: number; monotonicMs: number; error: unknown }): void + recordScopeClosed(input: { + at: number + monotonicMs: number + source?: string + reason?: string + propagationPoint?: string + lifecycleActionID?: string + }): void + finalize(input: { completedAt?: number; monotonicMs: number }): Summary +} + +export const SafeToolName = z.string().brand<"SafeToolName">() +export type SafeToolName = z.infer diff --git a/packages/opencode/test/session/export.test.ts b/packages/opencode/test/session/export.test.ts index 953e72958..d6ef9ac8e 100644 --- a/packages/opencode/test/session/export.test.ts +++ b/packages/opencode/test/session/export.test.ts @@ -13,6 +13,7 @@ import { tmpdir } from "../fixture/fixture" import { Config } from "../../src/config" import { TOOL_FAILURE_HINTS } from "../../src/session/tool-failure" import { LLMTrace } from "../../src/session/llm-trace" +import { RunObservability } from "../../src/session/run-observability" const projectRoot = path.join(__dirname, "../..") void Log.init({ print: false }) @@ -689,6 +690,80 @@ describe("Export.session", () => { }) }) + test("collects run observability diagnostics as a top-level projection", async () => { + await Instance.provide({ + directory: projectRoot, + fn: async () => { + const root = await SessionNs.create({ title: "run observability" }) + const userID = MessageID.make("msg_run_obs_user") + const assistantID = MessageID.make("msg_run_obs_assistant") + const summary: RunObservability.Summary = { + schema_version: 1, + run_id: RunObservability.RunID.make("run_export"), + trace_id: assistantID, + session_id: root.id, + message_id: assistantID, + parent_message_id: userID, + provider: "test", + model: "test-model", + created_at: 10, + completed_at: 20, + classification: "external_stream_disconnect", + summary_key: RunObservability.summaryKeyFor("external_stream_disconnect", "provider_progress_socket_closed"), + retry_safety: { + recommendation: "candidate_safe_auto_retry", + confidence: "medium", + reason: "no_visible_output_or_tool_execution", + safety_scope: "user_visible_and_tool_side_effects", + }, + attempts: [], + provider_progress_seen: true, + visible_output_seen: false, + tool_call_seen: false, + tool_execution_started: false, + read_only_tool_started: false, + unsafe_side_effect_started: false, + unsafe_side_effect_kinds: [], + side_effect_facts_complete: true, + durations_ms: { total: 10 }, + error: { name: "TypeError", message: "terminated", cause_code: "UND_ERR_SOCKET" }, + } + try { + await SessionNs.updateMessage({ + id: userID, + sessionID: root.id, + role: "user", + time: { created: Date.now() }, + agent: "build", + model: { providerID: "test", modelID: "test-model" }, + } as MessageV2.User) + await SessionNs.updateMessage({ + id: assistantID, + role: "assistant", + sessionID: root.id, + mode: "build", + agent: "build", + path: { cwd: projectRoot, root: projectRoot }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + modelID: "test-model", + providerID: "test", + parentID: userID, + time: { created: 10, completed: 20 }, + finish: "error", + diagnostics: { run_observability: summary }, + } as MessageV2.Assistant) + + const result = await AppRuntime.runPromise(Export.session(root.id)) + expect(result.diagnostics.run_observability_schema_version).toBe(1) + expect(result.diagnostics.run_observability).toEqual([summary]) + } finally { + await SessionNs.remove(root.id) + } + }, + }) + }) + test("collects abort and title generation diagnostics from assistant messages", async () => { await Instance.provide({ directory: projectRoot, @@ -1582,6 +1657,99 @@ describe("redactPart", () => { ).toEqual({ "x-request-id": "req_123" }) }) + test("sanitizeSnapshot preserves safe run observability error fingerprints", () => { + const summary: RunObservability.Summary = { + schema_version: 1, + run_id: RunObservability.RunID.make("run_sanitize"), + trace_id: MessageID.make("msg_sanitize"), + session_id: SessionID.make("ses_sanitize"), + message_id: MessageID.make("msg_sanitize"), + provider: "test", + model: "test-model", + created_at: 1, + completed_at: 2, + classification: "external_stream_disconnect", + summary_key: RunObservability.summaryKeyFor("external_stream_disconnect", "provider_progress_socket_closed"), + retry_safety: { + recommendation: "candidate_safe_auto_retry", + confidence: "medium", + reason: "no_visible_output_or_tool_execution", + safety_scope: "user_visible_and_tool_side_effects", + }, + attempts: [], + provider_progress_seen: true, + visible_output_seen: false, + tool_call_seen: false, + tool_execution_started: false, + read_only_tool_started: false, + unsafe_side_effect_started: false, + unsafe_side_effect_kinds: [], + side_effect_facts_complete: true, + durations_ms: { total: 1 }, + error: { name: "TypeError", message: "terminated", cause_code: "UND_ERR_SOCKET" }, + } + const fakeSnapshot: Export.Snapshot = { + schema_version: 1, + format: "pawwork-session-export", + exported_at: 1, + root_session_id: SessionID.make("ses_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: 1, part_count: 0, omitted_attachment_count: 0 }, + }, + diagnostics: { run_observability_schema_version: 1, run_observability: [summary] }, + session: { + info: { + id: SessionID.make("ses_sanitize"), + version: "0.0.0", + time: { created: 1, updated: 1 }, + title: "x", + directory: "/tmp/project", + } as SessionNs.Info, + had_cloud_share: false, + diffs: [], + messages: [ + { + info: { + id: MessageID.make("msg_sanitize"), + role: "assistant", + sessionID: SessionID.make("ses_sanitize"), + parentID: MessageID.make("msg_parent_sanitize"), + mode: "build", + agent: "build", + path: { cwd: "/tmp/project", root: "/tmp/project" }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + modelID: "test-model", + providerID: "test", + time: { created: 1 }, + diagnostics: { run_observability: summary }, + } as MessageV2.Assistant, + parts: [], + }, + ], + children: [], + }, + } + + const sanitized = Export.sanitizeSnapshot(fakeSnapshot) + expect(sanitized.diagnostics.run_observability?.[0]?.error?.cause_code).toBe("UND_ERR_SOCKET") + expect( + sanitized.session.messages[0].info.role === "assistant" + ? sanitized.session.messages[0].info.diagnostics?.run_observability?.error?.cause_code + : undefined, + ).toBe("UND_ERR_SOCKET") + expect(JSON.stringify(sanitized)).not.toContain("/Users/") + expect(JSON.stringify(sanitized)).not.toContain("sk-") + }) + test("redacts data: url inside completed tool attachments", () => { const ctx = { count: { omitted: 0 } } const part: MessageV2.ToolPart = { diff --git a/packages/opencode/test/session/run-observability.test.ts b/packages/opencode/test/session/run-observability.test.ts new file mode 100644 index 000000000..1e653f47f --- /dev/null +++ b/packages/opencode/test/session/run-observability.test.ts @@ -0,0 +1,359 @@ +import { describe, expect, test } from "bun:test" +import { MessageID, SessionID } from "../../src/session/schema" +import { RunObservability } from "../../src/session/run-observability" + +describe("RunObservability", () => { + test("does not treat stream lifecycle events as provider progress", () => { + expect(RunObservability.isProviderProgressEvent({ type: "start" })).toBe(false) + expect(RunObservability.isProviderProgressEvent({ type: "finish-step" })).toBe(false) + expect(RunObservability.isProviderProgressEvent({ type: "text-delta" })).toBe(true) + expect(RunObservability.isProviderProgressEvent({ type: "tool-call" })).toBe(true) + }) + + test("does not use lifecycle-only stream events for provider-progress transport summaries", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_start_disconnect"), + traceID: MessageID.make("msg_start_disconnect"), + sessionID: SessionID.make("ses_start_disconnect"), + messageID: MessageID.make("msg_start_disconnect"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + if (RunObservability.isProviderProgressEvent({ type: "start" })) { + recorder.recordProviderProgress({ attemptID: attempt.attemptID, at: 12, monotonicMs: 120 }) + } + recorder.recordTransportFailure({ + attemptID: attempt.attemptID, + at: 25, + monotonicMs: 250, + error: { + name: "TypeError", + message: "terminated", + cause: { name: "SocketError", message: "other side closed", code: "UND_ERR_SOCKET" }, + }, + evidence: ["iterator_error"], + }) + + const summary = recorder.finalize({ completedAt: 26, monotonicMs: 260 }) + expect(summary.provider_progress_seen).toBe(false) + expect(summary.attempts[0]?.provider_progress_seen).toBe(false) + expect(String(summary.summary_key)).toBe("external_stream_disconnect.transport_failure") + }) + + test("classifies completed runs without failure as success", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_success"), + traceID: MessageID.make("msg_success"), + sessionID: SessionID.make("ses_success"), + messageID: MessageID.make("msg_success"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + + recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + + const summary = recorder.finalize({ completedAt: 20, monotonicMs: 200 }) + expect(summary.classification).toBe("success") + expect(String(summary.summary_key)).toBe("success.completed") + expect(summary.retry_safety).toEqual({ + recommendation: "unknown", + confidence: "high", + reason: "completed_without_failure", + safety_scope: "user_visible_and_tool_side_effects", + }) + expect(summary.error).toBeUndefined() + }) + + test("does not write attempt completed_at for text-only success", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_text_only_completion"), + traceID: MessageID.make("msg_text_only_completion"), + sessionID: SessionID.make("ses_text_only_completion"), + messageID: MessageID.make("msg_text_only_completion"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + recorder.recordVisibleOutput({ attemptID: attempt.attemptID, at: 12, monotonicMs: 120 }) + + const summary = recorder.finalize({ completedAt: 20, monotonicMs: 200 }) + expect(summary.attempts[0]).not.toHaveProperty("completed_at") + expect(summary.attempts[0]).not.toHaveProperty("last_tool_completed_at") + }) + + test("records tool completion with tool-specific attempt field", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_tool_completion"), + traceID: MessageID.make("msg_tool_completion"), + sessionID: SessionID.make("ses_tool_completion"), + messageID: MessageID.make("msg_tool_completion"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + recorder.recordToolExecutionStarted({ + attemptID: attempt.attemptID, + at: 12, + monotonicMs: 120, + toolName: RunObservability.safeToolName("read"), + effect: RunObservability.toolEffect("read"), + }) + recorder.recordToolCompleted({ attemptID: attempt.attemptID, at: 13, monotonicMs: 130 }) + + const summary = recorder.finalize({ completedAt: 20, monotonicMs: 200 }) + expect(summary.attempts[0]).not.toHaveProperty("completed_at") + expect(summary.attempts[0]).toMatchObject({ last_tool_completed_at: 13 }) + }) + + test("keeps tool completion timestamps scoped to their attempts", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_multi_attempt_completion"), + traceID: MessageID.make("msg_multi_attempt_completion"), + sessionID: SessionID.make("ses_multi_attempt_completion"), + messageID: MessageID.make("msg_multi_attempt_completion"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + + const first = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + recorder.recordToolCompleted({ attemptID: first.attemptID, at: 13, monotonicMs: 130 }) + const second = recorder.beginAttempt({ attemptIndex: 2, at: 20, monotonicMs: 200 }) + recorder.recordVisibleOutput({ attemptID: second.attemptID, at: 21, monotonicMs: 210 }) + + const summary = recorder.finalize({ completedAt: 30, monotonicMs: 300 }) + expect(summary.attempts).toHaveLength(2) + expect(summary.attempts[0]).toMatchObject({ last_tool_completed_at: 13 }) + expect(summary.attempts[1]).not.toHaveProperty("completed_at") + expect(summary.attempts[1]).not.toHaveProperty("last_tool_completed_at") + }) + + test("classifies external stream disconnect from run-level aggregate facts", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_external"), + traceID: MessageID.make("msg_external"), + sessionID: SessionID.make("ses_external"), + messageID: MessageID.make("msg_external"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + recorder.recordProviderProgress({ attemptID: attempt.attemptID, at: 12, monotonicMs: 120 }) + recorder.recordTransportFailure({ + attemptID: attempt.attemptID, + at: 25, + monotonicMs: 250, + error: { + name: "TypeError", + message: "terminated", + cause: { name: "SocketError", message: "other side closed", code: "UND_ERR_SOCKET" }, + }, + evidence: ["provider_progress_seen", "iterator_error"], + }) + + const summary = recorder.finalize({ completedAt: 26, monotonicMs: 260 }) + expect(summary.classification).toBe("external_stream_disconnect") + expect(String(summary.summary_key)).toBe("external_stream_disconnect.provider_progress_socket_closed") + expect(summary.retry_safety).toEqual({ + recommendation: "candidate_safe_auto_retry", + confidence: "medium", + reason: "no_visible_output_or_tool_execution", + safety_scope: "user_visible_and_tool_side_effects", + }) + expect(summary.visible_output_seen).toBe(false) + expect(summary.tool_execution_started).toBe(false) + expect(summary.durations_ms.last_event_to_failure).toBe(130) + }) + + test("retry safety is denied when any earlier attempt emitted visible output", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_retry_aggregate"), + traceID: MessageID.make("msg_retry_aggregate"), + sessionID: SessionID.make("ses_retry_aggregate"), + messageID: MessageID.make("msg_retry_aggregate"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + 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 }) + recorder.recordTransportFailure({ + attemptID: second.attemptID, + at: 21, + monotonicMs: 210, + error: { name: "TypeError", message: "terminated", cause: { code: "UND_ERR_SOCKET" } }, + evidence: ["iterator_error"], + }) + + const summary = recorder.finalize({ completedAt: 22, monotonicMs: 220 }) + expect(summary.attempts).toHaveLength(2) + 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("classifies local scope close with missing lifecycle provenance separately from user cancel", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_scope_close"), + traceID: MessageID.make("msg_scope_close"), + sessionID: SessionID.make("ses_scope_close"), + messageID: MessageID.make("msg_scope_close"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + recorder.recordScopeClosed({ + at: 20, + monotonicMs: 200, + source: "session.run_state.scope", + reason: "scope_closed_without_cancel_meta", + propagationPoint: "session.prompt.loop.onInterrupt", + }) + + const summary = recorder.finalize({ completedAt: 21, monotonicMs: 210 }) + expect(summary.classification).toBe("unknown_scope_close") + expect(String(summary.summary_key)).toBe("unknown_scope_close.missing_lifecycle_provenance") + expect(summary.missing_provenance).toEqual(["lifecycle.close_requested"]) + expect(summary.retry_safety.recommendation).toBe("do_not_auto_retry") + }) + + test("records tool execution effect facts conservatively", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_tool_effects"), + traceID: MessageID.make("msg_tool_effects"), + sessionID: SessionID.make("ses_tool_effects"), + messageID: MessageID.make("msg_tool_effects"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + + recorder.recordToolExecutionStarted({ + attemptID: attempt.attemptID, + at: 12, + monotonicMs: 120, + toolName: RunObservability.safeToolName("bash /Users/alice/.ssh/id_rsa?token=secret"), + effect: RunObservability.toolEffect("bash"), + }) + recorder.recordToolInterrupted({ attemptID: attempt.attemptID, at: 13, monotonicMs: 130 }) + + const summary = recorder.finalize({ completedAt: 14, monotonicMs: 140 }) + expect(summary.tool_execution_started).toBe(true) + expect(summary.unsafe_side_effect_started).toBe(true) + expect(summary.unsafe_side_effect_kinds).toEqual(["local_process"]) + expect(summary.side_effect_facts_complete).toBe(true) + expect(JSON.stringify(summary)).not.toContain("/Users/alice") + expect(JSON.stringify(summary)).not.toContain("secret") + }) + + test("redacts arbitrary error messages to low-cardinality values", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_safe_error"), + traceID: MessageID.make("msg_safe_error"), + sessionID: SessionID.make("ses_safe_error"), + messageID: MessageID.make("msg_safe_error"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + recorder.recordTransportFailure({ + attemptID: attempt.attemptID, + at: 12, + monotonicMs: 120, + error: { + name: "TypeError", + message: "private prompt: email alice@example.com password=hunter2 file C:\\Users\\Alice\\secret.txt", + cause: { message: "api_key=secret and /var/private/project/file.ts" }, + }, + }) + + const summary = recorder.finalize({ completedAt: 13, monotonicMs: 130 }) + expect(summary.error?.message).toBe("redacted") + expect(summary.error?.cause_message).toBe("redacted") + const serialized = JSON.stringify(summary) + expect(serialized).not.toContain("alice@example.com") + expect(serialized).not.toContain("hunter2") + expect(serialized).not.toContain("api_key") + expect(serialized).not.toContain("secret.txt") + }) + + test("does not let generic transport handling overwrite tool or setup failure provenance", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_precedence"), + traceID: MessageID.make("msg_precedence"), + sessionID: SessionID.make("ses_precedence"), + messageID: MessageID.make("msg_precedence"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 11, monotonicMs: 110 }) + recorder.recordToolFailed({ + attemptID: attempt.attemptID, + at: 12, + monotonicMs: 120, + error: new Error("tool failed"), + }) + recorder.recordTransportFailure({ + attemptID: attempt.attemptID, + at: 13, + monotonicMs: 130, + error: new Error("terminated"), + }) + + const summary = recorder.finalize({ completedAt: 14, monotonicMs: 140 }) + expect(summary.classification).toBe("tool_failure") + expect(summary.summary_key).toBe(RunObservability.summaryKeyFor("tool_failure", "tool_execution_failed")) + }) + + test("monotonic durations never go negative when wall clock moves backward", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_clock"), + traceID: MessageID.make("msg_clock"), + sessionID: SessionID.make("ses_clock"), + messageID: MessageID.make("msg_clock"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 1_000, + monotonicStartMs: 500, + }) + const attempt = recorder.beginAttempt({ attemptIndex: 1, at: 900, monotonicMs: 510 }) + recorder.recordProviderProgress({ attemptID: attempt.attemptID, at: 800, monotonicMs: 520 }) + recorder.recordTransportFailure({ + attemptID: attempt.attemptID, + at: 700, + monotonicMs: 515, + error: { name: "TypeError", message: "terminated" }, + evidence: ["iterator_error"], + }) + + const summary = recorder.finalize({ completedAt: 600, monotonicMs: 505 }) + expect(summary.durations_ms.total).toBe(5) + expect(summary.durations_ms.last_event_to_failure).toBe(0) + }) +}) diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 3609cd2bb..9cc99b7ec 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -585,6 +585,7 @@ export type AssistantMessage = { completed_at?: number stream?: unknown } + run_observability?: unknown abort?: { source?: string reason?: string