diff --git a/packages/opencode/src/effect/runner.ts b/packages/opencode/src/effect/runner.ts index f6d60b4a5..1ec524c52 100644 --- a/packages/opencode/src/effect/runner.ts +++ b/packages/opencode/src/effect/runner.ts @@ -36,6 +36,8 @@ interface PendingHandle { export interface InterruptMeta { source?: string reason?: string + lifecycleActionID?: string + lifecycleKind?: string // Reserved for future paths that originate from a tool or model ctx.abort signal instead of // an explicit session.cancel call. viaCtxAbort?: boolean @@ -55,7 +57,7 @@ export const make = ( onIdle?: Effect.Effect onBusy?: Effect.Effect onInterrupt?: (meta?: InterruptMeta) => Effect.Effect - interruptFallback?: InterruptMeta + interruptFallback?: InterruptMeta | (() => InterruptMeta) busy?: () => never }, ): Runner => { @@ -79,6 +81,8 @@ export const make = ( source: "runner.interrupt_without_meta", reason: "fiber_interrupt_without_meta", } + const getInterruptFallback = () => + typeof interruptFallback === "function" ? interruptFallback() : interruptFallback const complete = (done: Deferred.Deferred, exit: Exit.Exit) => Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause) @@ -117,7 +121,7 @@ export const make = ( const resolveInterrupt = (interruptMeta: Ref.Ref): Effect.Effect => Effect.gen(function* () { - const meta = withRecordedInterruptMeta(yield* Ref.get(interruptMeta), interruptFallback) + const meta = withRecordedInterruptMeta(yield* Ref.get(interruptMeta), getInterruptFallback()) if (onInterrupt) return yield* onInterrupt(meta) return yield* Effect.die(new Cancelled()) }) diff --git a/packages/opencode/src/project/instance-store.ts b/packages/opencode/src/project/instance-store.ts index 43ad29b23..f44c4f394 100644 --- a/packages/opencode/src/project/instance-store.ts +++ b/packages/opencode/src/project/instance-store.ts @@ -8,6 +8,11 @@ import { InstanceBootstrap } from "./bootstrap-service" import { type InstanceContext } from "./instance-context" import { Project } from "./project" import { State } from "./state" +import { + createLifecycleCloseAction, + type LifecycleCloseAction, + withLifecycleCloseAction, +} from "@/session/lifecycle-provenance" export interface LoadInput { directory: string @@ -111,19 +116,22 @@ export const layer = Layer.effect( }), ) - const disposeContext = (ctx: InstanceContext) => + const disposeContext = (ctx: InstanceContext, action?: LifecycleCloseAction) => Effect.gen(function* () { yield* Effect.promise(async () => { - await State.dispose(ctx.directory) - await runDisposers(ctx.directory) + const closeAction = action ?? createLifecycleCloseAction("instance_dispose") + await withLifecycleCloseAction([ctx.directory], closeAction, async () => { + await State.dispose(ctx.directory) + await runDisposers(ctx.directory) + }) }) yield* emitDisposed(ctx) }) - const disposeEntry = (directory: string, entry: Entry, ctx: InstanceContext) => + const disposeEntry = (directory: string, entry: Entry, ctx: InstanceContext, action?: LifecycleCloseAction) => Effect.gen(function* () { if (entries.get(directory) !== entry) return false - yield* disposeContext(ctx) + yield* disposeContext(ctx, action) if (entries.get(directory) !== entry) return false entries.delete(directory) return true @@ -140,7 +148,7 @@ export const layer = Layer.effect( yield* Effect.gen(function* () { if (previous) { const exit = yield* Deferred.await(previous.deferred).pipe(Effect.exit) - if (Exit.isSuccess(exit)) yield* disposeContext(exit.value) + if (Exit.isSuccess(exit)) yield* disposeContext(exit.value, createLifecycleCloseAction("instance_reload")) else yield* removeEntry(directory, previous) } yield* completeLoad(directory, input, entry) @@ -184,7 +192,7 @@ export const layer = Layer.effect( const exit = yield* Deferred.await(entry.deferred).pipe(Effect.exit) if (Exit.isFailure(exit)) return yield* removeEntry(directory, entry).pipe(Effect.asVoid) if (exit.value !== ctx) return - yield* disposeEntry(directory, entry, ctx).pipe(Effect.asVoid) + yield* disposeEntry(directory, entry, ctx, createLifecycleCloseAction("instance_dispose")).pipe(Effect.asVoid) }) const disposeDirectory = (inputDirectory: string) => @@ -195,10 +203,13 @@ export const layer = Layer.effect( const exit = yield* Deferred.await(entry.deferred).pipe(Effect.exit) if (Exit.isFailure(exit)) return yield* removeEntry(directory, entry).pipe(Effect.asVoid) - yield* disposeEntry(directory, entry, exit.value).pipe(Effect.asVoid) + yield* disposeEntry(directory, entry, exit.value, createLifecycleCloseAction("instance_dispose_directory")).pipe( + Effect.asVoid, + ) }) const disposeAllOnce = Effect.gen(function* () { + const action = createLifecycleCloseAction("instance_dispose_all") yield* Effect.forEach( [...entries.entries()], ([directory, entry]) => @@ -208,7 +219,7 @@ export const layer = Layer.effect( yield* removeEntry(directory, entry) return } - yield* disposeEntry(directory, entry, exit.value) + yield* disposeEntry(directory, entry, exit.value, action) }), { discard: true }, ) diff --git a/packages/opencode/src/session/lifecycle-provenance.ts b/packages/opencode/src/session/lifecycle-provenance.ts new file mode 100644 index 000000000..84c9ad062 --- /dev/null +++ b/packages/opencode/src/session/lifecycle-provenance.ts @@ -0,0 +1,45 @@ +import type { LifecycleKind } from "./run-observability/types" + +export type LifecycleCloseAction = { + actionID: string + kind: LifecycleKind +} + +let nextActionID = 0 +const activeByDirectory = new Map() + +export function createLifecycleCloseAction(kind: LifecycleKind): LifecycleCloseAction { + nextActionID += 1 + return { + actionID: `lifecycle:${kind}:${Date.now().toString(36)}:${nextActionID.toString(36)}`, + kind, + } +} + +export async function withLifecycleCloseAction( + directories: string[], + action: LifecycleCloseAction, + fn: () => Promise, +): Promise { + for (const directory of directories) { + const stack = activeByDirectory.get(directory) ?? [] + stack.push(action) + activeByDirectory.set(directory, stack) + } + try { + return await fn() + } finally { + for (const directory of directories) { + const stack = activeByDirectory.get(directory) + if (!stack) continue + const index = stack.lastIndexOf(action) + if (index >= 0) stack.splice(index, 1) + if (stack.length) activeByDirectory.set(directory, stack) + else activeByDirectory.delete(directory) + } + } +} + +export function currentLifecycleCloseAction(directory: string): LifecycleCloseAction | undefined { + return activeByDirectory.get(directory)?.at(-1) +} diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 59132168a..8c6b171bf 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -22,9 +22,11 @@ import { ExternalResult } from "@/tool/external-result" import { errorMessage } from "@/util/error" import { Log } from "@opencode-ai/core/util/log" import { isRecord } from "@/util/record" +import { InstanceState } from "@/effect/instance-state" import { TurnChange } from "./turn-change" import { LLMTrace } from "./llm-trace" import { RunObservability } from "./run-observability" +import { currentLifecycleCloseAction } from "./lifecycle-provenance" const log = Log.create({ service: "session.processor" }) const TOOL_CLEANUP_TIMEOUT_MS = 1_000 @@ -120,6 +122,7 @@ type PendingLoopAction = { } interface ProcessorContext extends Input { + directory: string toolcalls: Record pendingLoopActions: Record pendingToolUpdates: Record MessageV2.ToolPart>> @@ -175,10 +178,12 @@ export const layer: Layer.Layer< // may execute tools internally before emitting start-step events, // so capturing inside the event handler can be too late. const initialSnapshot = yield* snapshot.track() + const instanceContext = yield* InstanceState.context const ctx: ProcessorContext = { assistantMessage: input.assistantMessage, sessionID: input.sessionID, model: input.model, + directory: instanceContext.directory, toolcalls: {}, pendingLoopActions: {}, pendingToolUpdates: {}, @@ -1073,12 +1078,15 @@ export const layer: Layer.Layer< 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", + lifecycleActionID: lifecycleAction?.actionID, + lifecycleKind: lifecycleAction?.kind, }) ctx.trace.recordAbortState({ provenanceSource: "session.processor.onInterrupt", diff --git a/packages/opencode/src/session/run-observability/recorder.ts b/packages/opencode/src/session/run-observability/recorder.ts index 6fbce870d..3a1f7bf44 100644 --- a/packages/opencode/src/session/run-observability/recorder.ts +++ b/packages/opencode/src/session/run-observability/recorder.ts @@ -9,6 +9,7 @@ import { SCHEMA_VERSION, type Summary, type SummaryKey, + type LifecycleKind, type ToolEffectKind, } from "./types" import { safeErrorFingerprint } from "./sanitize" @@ -25,6 +26,7 @@ type Failure = source?: string reason?: string lifecycleActionID?: string + lifecycleKind?: LifecycleKind } | { type: "tool"; at: number; monotonicMs: number; error?: unknown; attemptID?: AttemptID } @@ -155,6 +157,7 @@ export function createRecorder(input: RecorderInput): Recorder { source: next.source, reason: next.reason, lifecycleActionID: next.lifecycleActionID, + lifecycleKind: next.lifecycleKind, } rememberEvent(next.monotonicMs) }, @@ -196,6 +199,7 @@ export function createRecorder(input: RecorderInput): Recorder { unsafe_side_effect_started: unsafeSideEffectStarted, unsafe_side_effect_kinds: unsafeKinds, side_effect_facts_complete: sideEffectFactsComplete, + lifecycle: lifecycleSummary(failure), missing_provenance: missingProvenance, durations_ms: { total: duration(input.monotonicStartMs, final.monotonicMs), @@ -219,8 +223,16 @@ 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") + if (failure.type === "scope_closed") { + if (failure.lifecycleKind === "instance_reload") return "local_instance_reload" + if ( + failure.lifecycleKind === "instance_dispose" || + failure.lifecycleKind === "instance_dispose_directory" || + failure.lifecycleKind === "instance_dispose_all" + ) + return "local_instance_dispose" return failure.lifecycleActionID ? "known_lifecycle_close" : "unknown_scope_close" + } if (failure.type === "transport") return "external_stream_disconnect" return "unknown_failure" } @@ -232,7 +244,10 @@ function summarySuffix(input: { failure: Failure | undefined; providerProgressSe 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 === "scope_closed") { + if (input.failure.lifecycleActionID) return "lifecycle_close" + 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" @@ -243,6 +258,16 @@ export function summaryKeyFor(classification: Classification, suffix: string): S return `${classification}.${suffix}` as SummaryKey } +function lifecycleSummary(failure: Failure | undefined): Summary["lifecycle"] { + if (failure?.type !== "scope_closed" || !failure.lifecycleActionID || !failure.lifecycleKind) return undefined + return { + action_id: failure.lifecycleActionID, + kind: failure.lifecycleKind, + source: failure.source, + reason: failure.reason, + } +} + export function isProviderProgressEvent(event: { type: string }) { switch (event.type) { case "text-start": diff --git a/packages/opencode/src/session/run-observability/types.ts b/packages/opencode/src/session/run-observability/types.ts index 02a343c67..dfb6a203e 100644 --- a/packages/opencode/src/session/run-observability/types.ts +++ b/packages/opencode/src/session/run-observability/types.ts @@ -12,6 +12,8 @@ export type AttemptID = z.infer export const Classification = z.enum([ "success", "external_stream_disconnect", + "local_instance_reload", + "local_instance_dispose", "known_lifecycle_close", "unknown_scope_close", "request_setup_failure", @@ -48,6 +50,8 @@ export type SafeErrorFingerprint = { cause_code?: string } +export type LifecycleKind = "instance_reload" | "instance_dispose" | "instance_dispose_directory" | "instance_dispose_all" + export type ToolEffectKind = "read_only" | "local_file_write" | "local_process" | "unknown" export type ToolEffect = { kind: ToolEffectKind @@ -91,6 +95,12 @@ export type Summary = { unsafe_side_effect_started: boolean unsafe_side_effect_kinds: ToolEffectKind[] side_effect_facts_complete: boolean + lifecycle?: { + action_id: string + kind: LifecycleKind + source?: string + reason?: string + } missing_provenance?: string[] durations_ms: { total?: number @@ -147,6 +157,7 @@ export type Recorder = { reason?: string propagationPoint?: string lifecycleActionID?: string + lifecycleKind?: LifecycleKind }): void finalize(input: { completedAt?: number; monotonicMs: number }): Summary } diff --git a/packages/opencode/src/session/run-state.ts b/packages/opencode/src/session/run-state.ts index 22a542695..3915ae4f1 100644 --- a/packages/opencode/src/session/run-state.ts +++ b/packages/opencode/src/session/run-state.ts @@ -5,6 +5,7 @@ import * as Session from "./session" import { MessageV2 } from "./message-v2" import { SessionID } from "./schema" import { SessionStatus } from "./status" +import { currentLifecycleCloseAction } from "./lifecycle-provenance" export interface Interface { readonly assertNotBusy: (sessionID: SessionID) => Effect.Effect @@ -30,17 +31,30 @@ export const layer = Layer.effect( const status = yield* SessionStatus.Service const state = yield* InstanceState.make( - Effect.fn("SessionRunState.state")(function* () { + Effect.fn("SessionRunState.state")(function* (ctx) { const scope = yield* Scope.Scope const runners = new Map>() + let scopeCloseAction = currentLifecycleCloseAction(ctx.directory) + const lifecycleAction = () => currentLifecycleCloseAction(ctx.directory) + const interruptFallback = () => { + const action = lifecycleAction() ?? scopeCloseAction + return { + source: "session.run_state.scope", + reason: "scope_closed_without_cancel_meta", + ...(action ? { lifecycleActionID: action.actionID, lifecycleKind: action.kind } : {}), + } satisfies InterruptMeta + } yield* Effect.addFinalizer( Effect.fnUntraced(function* () { + const action = lifecycleAction() + scopeCloseAction = action ?? scopeCloseAction yield* Effect.forEach( runners.values(), (runner) => runner.cancelWith({ source: "session.run_state.finalizer", reason: "scope_finalizer", + ...(action ? { lifecycleActionID: action.actionID, lifecycleKind: action.kind } : {}), }), { concurrency: "unbounded", @@ -50,7 +64,7 @@ export const layer = Layer.effect( runners.clear() }), ) - return { runners, scope } + return { runners, scope, interruptFallback } }), ) @@ -68,10 +82,7 @@ export const layer = Layer.effect( }), onBusy: status.set(sessionID, { type: "busy" }), onInterrupt, - interruptFallback: { - source: "session.run_state.scope", - reason: "scope_closed_without_cancel_meta", - }, + interruptFallback: data.interruptFallback, busy: () => { throw new Session.BusyError(sessionID) }, diff --git a/packages/opencode/test/session/run-observability.test.ts b/packages/opencode/test/session/run-observability.test.ts index 1e653f47f..f7a47da6f 100644 --- a/packages/opencode/test/session/run-observability.test.ts +++ b/packages/opencode/test/session/run-observability.test.ts @@ -237,6 +237,39 @@ describe("RunObservability", () => { expect(summary.retry_safety.recommendation).toBe("do_not_auto_retry") }) + test("classifies known instance reload lifecycle closes with parent provenance", () => { + const recorder = RunObservability.createRecorder({ + runID: RunObservability.RunID.make("run_instance_reload"), + traceID: MessageID.make("msg_instance_reload"), + sessionID: SessionID.make("ses_instance_reload"), + messageID: MessageID.make("msg_instance_reload"), + providerID: "openai", + modelID: "gpt-5.5", + createdAt: 10, + monotonicStartMs: 100, + }) + recorder.recordScopeClosed({ + at: 20, + monotonicMs: 200, + source: "session.run_state.finalizer", + reason: "scope_finalizer", + propagationPoint: "session.prompt.loop.onInterrupt", + lifecycleActionID: "lifecycle:instance_reload:abc123", + lifecycleKind: "instance_reload", + }) + + const summary = recorder.finalize({ completedAt: 21, monotonicMs: 210 }) + expect(summary.classification).toBe("local_instance_reload") + expect(String(summary.summary_key)).toBe("local_instance_reload.lifecycle_close") + expect(summary.lifecycle).toEqual({ + action_id: "lifecycle:instance_reload:abc123", + kind: "instance_reload", + source: "session.run_state.finalizer", + reason: "scope_finalizer", + }) + expect(summary.missing_provenance).toBeUndefined() + }) + test("records tool execution effect facts conservatively", () => { const recorder = RunObservability.createRecorder({ runID: RunObservability.RunID.make("run_tool_effects"), diff --git a/packages/opencode/test/session/run-state.test.ts b/packages/opencode/test/session/run-state.test.ts index 87230e1ce..e730b5a83 100644 --- a/packages/opencode/test/session/run-state.test.ts +++ b/packages/opencode/test/session/run-state.test.ts @@ -1,7 +1,12 @@ -import { describe, expect } from "bun:test" +import { describe, expect, test } from "bun:test" import { Effect, Exit, Fiber, Layer } from "effect" import * as CrossSpawnSpawner from "@opencode-ai/core/cross-spawn-spawner" import { Instance } from "../../src/project/instance" +import { + createLifecycleCloseAction, + currentLifecycleCloseAction, + withLifecycleCloseAction, +} from "../../src/session/lifecycle-provenance" import { SessionRunState } from "../../src/session/run-state" import { SessionID } from "../../src/session/schema" import { provideTmpdirInstance } from "../fixture/fixture" @@ -10,8 +15,43 @@ import { testEffect } from "../lib/effect" const it = testEffect(Layer.mergeAll(CrossSpawnSpawner.defaultLayer, SessionRunState.defaultLayer)) describe("SessionRunState", () => { - it.live("annotates runner interrupts caused by the run-state scope closing", () => { - let captured: { source?: string; reason?: string; recordedAt?: number } | undefined + test("keeps overlapping lifecycle actions isolated per directory", async () => { + const directory = "/tmp/pawwork-lifecycle-overlap" + const first = createLifecycleCloseAction("instance_reload") + const second = createLifecycleCloseAction("instance_dispose_all") + let resolveFirst!: () => void + let resolveSecond!: () => void + + const firstDone = withLifecycleCloseAction([directory], first, async () => { + await new Promise((resolve) => { + resolveFirst = resolve + }) + }) + const secondDone = withLifecycleCloseAction([directory], second, async () => { + await new Promise((resolve) => { + resolveSecond = resolve + }) + }) + + expect(currentLifecycleCloseAction(directory)).toBe(second) + resolveFirst() + await firstDone + expect(currentLifecycleCloseAction(directory)).toBe(second) + resolveSecond() + await secondDone + expect(currentLifecycleCloseAction(directory)).toBeUndefined() + }) + + it.live("annotates runner interrupts caused by instance disposal with lifecycle provenance", () => { + let captured: + | { + source?: string + reason?: string + recordedAt?: number + lifecycleActionID?: string + lifecycleKind?: string + } + | undefined return provideTmpdirInstance( () => @@ -38,10 +78,54 @@ describe("SessionRunState", () => { expect(captured).toMatchObject({ source: "session.run_state.scope", reason: "scope_closed_without_cancel_meta", + lifecycleKind: "instance_dispose", }) + expect(captured?.lifecycleActionID).toStartWith("lifecycle:instance_dispose:") expect(typeof captured?.recordedAt).toBe("number") }), { git: true }, ) }) + + it.live("fans out one disposeAll lifecycle action to multiple in-flight runs", () => { + const captured = new Map() + + return provideTmpdirInstance( + () => + Effect.gen(function* () { + const run = yield* SessionRunState.Service + const firstID = SessionID.make("ses_dispose_all_first") + const secondID = SessionID.make("ses_dispose_all_second") + const start = (sessionID: SessionID) => + run + .ensureRunning( + sessionID, + (meta) => + Effect.sync(() => { + captured.set(sessionID, meta) + return {} as never + }), + Effect.never, + ) + .pipe(Effect.forkChild) + + const firstFiber = yield* start(firstID) + const secondFiber = yield* start(secondID) + + yield* Effect.sleep("10 millis") + yield* Effect.promise(() => Instance.disposeAll()) + + expect(Exit.isSuccess(yield* Fiber.await(firstFiber))).toBe(true) + expect(Exit.isSuccess(yield* Fiber.await(secondFiber))).toBe(true) + + const first = captured.get(firstID) + const second = captured.get(secondID) + expect(first).toMatchObject({ lifecycleKind: "instance_dispose_all" }) + expect(second).toMatchObject({ lifecycleKind: "instance_dispose_all" }) + expect(first?.lifecycleActionID).toBe(second?.lifecycleActionID) + expect(first?.lifecycleActionID).toStartWith("lifecycle:instance_dispose_all:") + }), + { git: true }, + ) + }) })