diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 3842614395e..5b3c10df30c 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -109,6 +109,10 @@ export const layer = Layer.effect( bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)), ), drain: (sessionID, force, promotable) => drain(sessionID, force, undefined, promotable), + // Claim after the old terminal: user cancellation must release its claim and retry budget + // before a deferred successor becomes fresh recovery intent, without reporting a start. + // This retains ID-only recovery: restart does not preserve the drain's promotable scope. + suspended: (sessionID) => reportLifecycle(sessionID, store.claim(sessionID)), // One terminal observation per busy period, covering every coalesced drain. settled: (sessionID, exit, reason) => reportLifecycle( diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index 7130f5ed5e2..7f6b5638266 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -9,9 +9,9 @@ export interface Coordinator { readonly active: Effect.Effect> /** Checks ownership for one key, including cleanup and terminal settlement. */ readonly isActive: (key: Key) => Effect.Effect - /** Starts an execution while idle, or joins the active execution and returns its exit. */ + /** Starts while idle or joins the active execution. Interrupts new runs once shutdown begins. */ readonly run: (key: Key) => Effect.Effect - /** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */ + /** Rings the doorbell: starts while idle or drains again before settling. No-op once shutdown begins. */ readonly wake: (key: Key, scope?: Promotable) => Effect.Effect /** * Stops the active execution and clears its doorbell. No-op when idle. Resolves once the @@ -61,10 +61,19 @@ export const make = (options: { * drain and before the execution settles (waiters resolve after it completes). */ readonly settled?: (key: Key, exit: Exit.Exit, reason?: Reason) => Effect.Effect + /** Preserves a pending successor when shutdown prevents starting it, after the terminal hook completes. */ + readonly suspended?: (key: Key) => Effect.Effect }): Effect.Effect, never, Scope.Scope> => Effect.gen(function* () { const executions = new Map>() const fork = yield* FiberSet.makeRuntime() + let closing = false + // Finalizers run in reverse order: stop scheduling before FiberSet closes and interrupts its owners. + yield* Effect.addFinalizer(() => + Effect.sync(() => { + closing = true + }), + ) const loop = (key: Key, execution: Execution, force: boolean): Effect.Effect => Effect.suspend(() => options.drain(key, force, execution.scope)).pipe( @@ -98,7 +107,16 @@ export const make = (options: { execution.owner = undefined }).pipe(Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void)), ), - Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))), + Effect.onExit((exit) => + Effect.suspend(() => { + if (closing && execution.pendingWake) + return (options.suspended?.(key) ?? Effect.void).pipe( + Effect.ensuring(Effect.sync(() => settle(key, execution, exit))), + ) + settle(key, execution, exit) + return Effect.void + }), + ), Effect.exit, Effect.asVoid, ), @@ -107,9 +125,9 @@ export const make = (options: { } // A doorbell that survives the execution loop (rung after the loop decided to end, or - // during failure or interruption cleanup) starts a fresh execution for the remaining work. + // during failure or interruption cleanup) starts fresh work, unless shutdown suspends it. const settle = (key: Key, execution: Execution, exit: Exit.Exit) => { - if (execution.pendingWake) start(key, false, execution.pendingWake) + if (execution.pendingWake && !closing) start(key, false, execution.pendingWake) else executions.delete(key) Deferred.doneUnsafe(execution.done, exit) } @@ -118,6 +136,7 @@ export const make = (options: { const run = (key: Key): Effect.Effect => Effect.suspend(() => { + if (closing) return Effect.interrupt const execution = executions.get(key) if (execution !== undefined) { // A stopping execution refuses joiners: wait out its cleanup, then run fresh. @@ -130,6 +149,7 @@ export const make = (options: { const wake = (key: Key, scope: Promotable = "input") => Effect.sync(() => { + if (closing) return const execution = executions.get(key) if (execution !== undefined) { // Coalesced wakes keep the widest scope: "input" subsumes "steer". @@ -141,6 +161,7 @@ export const make = (options: { const interrupt = (key: Key, reason?: Reason): Effect.Effect => Effect.sync(() => { + if (closing) return false const execution = executions.get(key) if (execution === undefined || execution.stopping) return false if (execution.owner === undefined) { diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index fab76d0e316..23fa86d195c 100644 --- a/packages/core/test/session-execution.test.ts +++ b/packages/core/test/session-execution.test.ts @@ -6,6 +6,7 @@ import { LayerNode } from "@opencode-ai/util/effect/layer-node" import { Bus } from "@opencode-ai/core/bus" import { Job } from "@opencode-ai/core/job" import { KV } from "@opencode-ai/core/kv" +import { Location } from "@opencode-ai/core/location" import { LocationServiceMap } from "@opencode-ai/core/location-service-map" import type { LocationServices } from "@opencode-ai/core/location-services" import { Project } from "@opencode-ai/core/project" @@ -18,12 +19,15 @@ import { UserInterruptedError } from "@opencode-ai/core/session/error" import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionInbox } from "@opencode-ai/core/session/inbox" import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionRunner } from "@opencode-ai/core/session/runner/index" import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" import { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect" import { eq } from "drizzle-orm" import { testEffect } from "./lib/effect" +import { globalProjectNode } from "./lib/project" +import { tmpdirScoped } from "./fixture/tmpdir" const it = testEffect( AppNodeBuilder.build( @@ -159,6 +163,112 @@ describe("SessionExecution lifecycle", () => { }), ) + it.live("recovers a move admitted during user-interruption cleanup when shutdown prevents its successor", () => + Effect.gen(function* () { + const database = yield* Database.Service + const bus = yield* Bus.Service + const store = yield* SessionStore.Service + const admission = yield* SessionInbox.Service + const jobs = yield* Job.Service + const destination = AbsolutePath.make((yield* tmpdirScoped()).path) + const sessionID = Session.ID.make("ses_shutdown_successor") + yield* seedSessions(database, [sessionID], { resume_attempts: 2 }) + const lifecycle: string[] = [] + yield* bus.project(SessionEvent.Execution.Started, () => Effect.sync(() => void lifecycle.push("started"))) + yield* bus.project(SessionEvent.Execution.Interrupted, (event) => + Effect.sync(() => void lifecycle.push(event.data.reason)), + ) + + const draining = yield* Deferred.make() + const cleanup = yield* Deferred.make() + const release = yield* Deferred.make() + const scope = yield* Scope.make() + yield* Effect.addFinalizer(() => + Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))), + ) + const drains: string[] = [] + const context = yield* buildExecution(scope, () => + Effect.sync(() => void drains.push("original")).pipe( + Effect.andThen(Deferred.succeed(draining, undefined)), + Effect.andThen(Effect.never), + Effect.onInterrupt(() => Deferred.succeed(cleanup, undefined).pipe(Effect.andThen(Deferred.await(release)))), + ), + ) + const execution = Context.get(context, SessionExecution.Service) + const sessions = Context.get( + yield* Layer.buildWithScope( + AppNodeBuilder.build(Session.node, [ + [Database.node, Layer.succeed(Database.Service, database)], + [Bus.node, Layer.succeed(Bus.Service, bus)], + [SessionStore.node, Layer.succeed(SessionStore.Service, store)], + [SessionInbox.node, Layer.succeed(SessionInbox.Service, admission)], + [Job.node, Layer.succeed(Job.Service, jobs)], + // The shared Bus already has the production Session projectors. + [SessionProjector.node, Layer.empty], + [Project.node, globalProjectNode], + [SessionExecution.node, Layer.succeed(SessionExecution.Service, execution)], + [ + LocationServiceMap.node, + Layer.effect( + LocationServiceMap.Service, + LayerMap.make( + (ref: Location.Ref) => + // Move validation only needs Location from the destination graph. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + LayerNode.compile(Location.boundNode(ref), [ + [Project.node, globalProjectNode], + ]) as unknown as Layer.Layer, + ), + ), + ], + ]).pipe(Layer.fresh), + scope, + ), + Session.Service, + ) + + yield* execution.wake(sessionID) + yield* Deferred.await(draining) + yield* sessions.interrupt(sessionID) + yield* Deferred.await(cleanup) + yield* sessions.move({ sessionID, directory: destination }) + expect(yield* sessions.inbox(sessionID)).toMatchObject([{ type: "move" }]) + + const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild({ startImmediately: true })) + yield* Effect.yieldNow + yield* Deferred.succeed(release, undefined) + yield* Fiber.join(closing) + + expect({ active: yield* execution.isActive(sessionID), claimed: (yield* claims(database))[sessionID] }).toEqual({ + active: false, + claimed: true, + }) + yield* execution.awaitIdle(sessionID) + expect(drains).toEqual(["original"]) + expect(lifecycle).toEqual(["started", "user"]) + // The interrupted intent releases its old recovery budget; the pending move is new work. + expect(yield* attempts(database, sessionID)).toBe(0) + + const restartedScope = yield* Scope.make() + yield* Effect.addFinalizer(() => Scope.close(restartedScope, Exit.void)) + const resumed = yield* Deferred.make() + const pending: SessionInbox.Item["type"][] = [] + const restarted = yield* buildExecution(restartedScope, () => + Effect.gen(function* () { + drains.push("restarted") + pending.push(...(yield* SessionInbox.list(database.db, sessionID)).map((item) => item.type)) + yield* Deferred.succeed(resumed, undefined) + }), + ) + yield* Context.get(restarted, SessionRestart.Service).resumeSuspendedSessions + yield* Deferred.await(resumed) + yield* Context.get(restarted, SessionExecution.Service).awaitIdle(sessionID) + expect(drains).toEqual(["original", "restarted"]) + expect(pending).toEqual(["move"]) + expect((yield* claims(database))[sessionID]).toBe(false) + }), + ) + it.effect("does not resume a user-cancelled background child whose notification was not admitted", () => Effect.gen(function* () { const database = yield* Database.Service diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index e04ea8e16bd..bad5b80ce2d 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -1,5 +1,5 @@ import { describe, expect } from "bun:test" -import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect" +import { Cause, Deferred, Effect, Exit, Fiber, Layer, Scope } from "effect" import { SessionInbox } from "@opencode-ai/core/session/inbox" import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator" import { testEffect } from "./lib/effect" @@ -158,6 +158,92 @@ describe("SessionRunCoordinator", () => { }), ) + for (const strategy of ["sequential", "parallel"] as const) { + for (const pending of [false, true]) { + it.effect( + `${strategy} shutdown ${pending ? "suspends a cleanup-era wake" : "leaves cancelled work stopped"}`, + () => + Effect.gen(function* () { + const started = yield* Deferred.make() + const cleanup = yield* Deferred.make() + const release = yield* Deferred.make() + const scope = yield* Scope.make(strategy) + yield* Effect.addFinalizer(() => + Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))), + ) + const lifecycle: Array = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => + Deferred.succeed(cleanup, undefined).pipe(Effect.andThen(Deferred.await(release))), + ), + ), + started: () => Effect.sync(() => void lifecycle.push("started")), + settled: (_key, _exit, reason) => Effect.sync(() => void lifecycle.push(reason)), + suspended: () => Effect.sync(() => void lifecycle.push("suspended")), + }).pipe(Scope.provide(scope)) + + yield* coordinator.wake("session") + yield* Deferred.await(started) + yield* coordinator.interrupt("session", "user") + yield* Deferred.await(cleanup) + if (pending) yield* coordinator.wake("session") + const idle = yield* coordinator.awaitIdle("session").pipe(Effect.forkChild) + const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild({ startImmediately: true })) + yield* Effect.yieldNow + yield* Deferred.succeed(release, undefined) + yield* Fiber.join(closing) + + expect(yield* coordinator.active).toEqual(new Set()) + yield* Fiber.join(idle) + expect(lifecycle).toEqual(pending ? ["started", "user", "suspended"] : ["started", "user"]) + + // A retained reference to the closed coordinator must not create ghost ownership. + yield* coordinator.wake("session") + expect(yield* coordinator.interrupt("session", "user")).toBe(false) + const exit = yield* coordinator.run("session").pipe(Effect.exit) + expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true) + expect(yield* coordinator.active).toEqual(new Set()) + }), + ) + } + } + + it.effect("suspends a settlement-window wake after its terminal hook completes", () => + Effect.gen(function* () { + const settling = yield* Deferred.make() + const release = yield* Deferred.make() + const scope = yield* Scope.make() + yield* Effect.addFinalizer(() => + Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))), + ) + const lifecycle: string[] = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => Effect.sync(() => void lifecycle.push("drained")), + settled: () => + Deferred.succeed(settling, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.andThen(Effect.sync(() => void lifecycle.push("settled"))), + ), + suspended: () => Effect.sync(() => void lifecycle.push("suspended")), + }).pipe(Scope.provide(scope)) + + yield* coordinator.wake("session") + yield* Deferred.await(settling) + yield* coordinator.wake("session") + const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild({ startImmediately: true })) + yield* Effect.yieldNow + yield* Deferred.succeed(release, undefined) + yield* Fiber.join(closing) + + expect(lifecycle).toEqual(["drained", "settled", "suspended"]) + expect(yield* coordinator.active).toEqual(new Set()) + yield* coordinator.awaitIdle("session") + }), + ) + it.effect("coalesces wakes received during active execution", () => Effect.gen(function* () { const firstStarted = yield* Deferred.make()