test(server): update instance fixture replacement syntax

Merge 3e9b009642 without changing shutdown scheduling or recovery behavior. Migrate regression fixtures to opaque LayerNode replacements and retain fresh Instance selection around each stub runner, including the upstream helper compile fix.

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(

packages/core/src/session/run-coordinator.ts

@@ -9,9 +9,9 @@ export interface Coordinator<Key, E, Reason = never> {
   readonly active: Effect.Effect<ReadonlySet<Key>>
   /** Checks ownership for one key, including cleanup and terminal settlement. */
   readonly isActive: (key: Key) => Effect.Effect<boolean>
-  /** 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<void, E>
-  /** 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<void>
   /**
    * Stops the active execution and clears its doorbell. No-op when idle. Resolves once the
@@ -61,10 +61,19 @@ export const make = <Key, E, Reason = never>(options: {
    * drain and before the execution settles (waiters resolve after it completes).
    */
   readonly settled?: (key: Key, exit: Exit.Exit<void, E>, reason?: Reason) => Effect.Effect<void>
+  /** Preserves a pending successor when shutdown prevents starting it, after the terminal hook completes. */
+  readonly suspended?: (key: Key) => Effect.Effect<void>
 }): Effect.Effect<Coordinator<Key, E, Reason>, never, Scope.Scope> =>
   Effect.gen(function* () {
     const executions = new Map<Key, Execution<E, Reason>>()
     const fork = yield* FiberSet.makeRuntime<never, void, never>()
+    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<E, Reason>, force: boolean): Effect.Effect<void, E> =>
       Effect.suspend(() => options.drain(key, force, execution.scope)).pipe(
@@ -89,16 +98,29 @@ export const make = <Key, E, Reason = never>(options: {
       // The leading yield lets `owner` be assigned before the drain can settle, and keeps
       // failing self-waking executions from growing the stack across successor starts.
       // Drains start one tick after wake; callers observe progress through events or run.
+      // Mask the yield with startup so shutdown cannot skip the write-ahead started hook.
       execution.owner = fork(
         Effect.yieldNow.pipe(
-          Effect.andThen(Effect.uninterruptible(options.started?.(key) ?? Effect.void)),
+          Effect.andThen(options.started?.(key) ?? Effect.void),
+          Effect.uninterruptible,
           Effect.andThen(loop(key, execution, force)),
           Effect.onExit((exit) =>
             Effect.sync(() => {
               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 Effect.suspend(() => options.suspended?.(key) ?? Effect.void).pipe(
+                  Effect.ensuring(Effect.sync(() => settle(key, execution, exit))),
+                )
+              // Keep this decision and settlement synchronous. An Effect boundary could yield
+              // to shutdown after skipping suspension but before starting the successor.
+              settle(key, execution, exit)
+              return Effect.void
+            }),
+          ),
           Effect.exit,
           Effect.asVoid,
         ),
@@ -107,9 +129,9 @@ export const make = <Key, E, Reason = never>(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<E, Reason>, exit: Exit.Exit<void, E>) => {
-      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 +140,7 @@ export const make = <Key, E, Reason = never>(options: {
 
     const run = (key: Key): Effect.Effect<void, E> =>
       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 +153,7 @@ export const make = <Key, E, Reason = never>(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 +165,7 @@ export const make = <Key, E, Reason = never>(options: {
 
     const interrupt = (key: Key, reason?: Reason): Effect.Effect<boolean> =>
       Effect.sync(() => {
+        if (closing) return false
         const execution = executions.get(key)
         if (execution === undefined || execution.stopping) return false
         if (execution.owner === undefined) {

packages/core/test/session-execution.test.ts

@@ -7,6 +7,7 @@ import { Bus } from "@opencode-ai/core/bus"
 import { Instance } from "@opencode-ai/core/instance/service"
 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"
@@ -19,12 +20,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(
@@ -160,6 +164,129 @@ describe("SessionExecution lifecycle", () => {
     }),
   )
 
+  for (const boundary of ["cleanup", "successor"]) {
+    it.live(`recovers a cleanup-era move when shutdown interrupts ${boundary}`, () =>
+      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<void>()
+        const cleanup = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        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.replace(Layer.succeed(Database.Service, database)),
+              Bus.node.replace(Layer.succeed(Bus.Service, bus)),
+              SessionStore.node.replace(Layer.succeed(SessionStore.Service, store)),
+              SessionInbox.node.replace(Layer.succeed(SessionInbox.Service, admission)),
+              Job.node.replace(Layer.succeed(Job.Service, jobs)),
+              // The shared Bus already has the production Session projectors.
+              SessionProjector.node.replace(Layer.empty),
+              Project.node.replace(globalProjectNode),
+              SessionExecution.node.replace(Layer.succeed(SessionExecution.Service, execution)),
+              LocationServiceMap.node.replace(
+                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), {
+                        replacements: [Project.node.replace(globalProjectNode)],
+                      }) as unknown as Layer.Layer<LocationServices>,
+                  ),
+                ),
+              ),
+            ]).pipe(Layer.fresh),
+            scope,
+          ),
+          Session.Service,
+        )
+
+        yield* execution.wake(sessionID)
+        yield* Deferred.await(draining)
+        // A joined caller observes the retiring execution's exit after its successor is forked.
+        const joined = yield* execution.resume(sessionID).pipe(
+          Effect.onExit(() => (boundary === "successor" ? Scope.close(scope, Exit.void) : Effect.void)),
+          Effect.exit,
+          Effect.forkChild({ startImmediately: true }),
+        )
+        yield* sessions.interrupt(sessionID)
+        yield* Deferred.await(cleanup)
+        yield* sessions.move({ sessionID, directory: destination })
+        expect(yield* sessions.inbox(sessionID)).toMatchObject([{ type: "move" }])
+
+        const closing =
+          boundary === "cleanup"
+            ? yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild({ startImmediately: true }))
+            : undefined
+        if (closing) yield* Effect.yieldNow
+        yield* Deferred.succeed(release, undefined)
+        if (closing) yield* Fiber.join(closing)
+        yield* Fiber.join(joined)
+
+        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(
+          boundary === "cleanup" ? ["started", "user"] : ["started", "user", "started", "shutdown"],
+        )
+        // 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<void>()
+        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
@@ -1374,7 +1501,9 @@ function buildExecution(
         Layer.provide(Layer.succeed(Job.Service, jobs)),
         // Do not reuse the outer harness's selector with its already-captured Location map.
         Layer.provide(
-          LayerNode.compile(Instance.byLocationNode, [[LocationServiceMap.node, locations]]).pipe(Layer.fresh),
+          LayerNode.compile(Instance.byLocationNode, {
+            replacements: [LocationServiceMap.node.replace(locations)],
+          }).pipe(Layer.fresh),
         ),
       ),
       scope,

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, Scheduler, 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,164 @@ 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<void>()
+            const cleanup = yield* Deferred.make<void>()
+            const release = yield* Deferred.make<void>()
+            const scope = yield* Scope.make(strategy)
+            yield* Effect.addFinalizer(() =>
+              Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))),
+            )
+            const lifecycle: Array<string | undefined> = []
+            const coordinator = yield* SessionRunCoordinator.make<string, never, string>({
+              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())
+          }),
+      )
+    }
+  }
+
+  for (const outcome of ["success", "effect defect", "synchronous throw"]) {
+    it.effect(`settles a shutdown-deferred successor after suspension hook ${outcome}`, () =>
+      Effect.gen(function* () {
+        const settling = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        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: () => {
+            lifecycle.push("suspended")
+            if (outcome === "synchronous throw") throw new Error("suspension failed")
+            return outcome === "effect defect" ? Effect.die(new Error("suspension failed")) : Effect.void
+          },
+        }).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")
+      }),
+    )
+  }
+
+  for (const cutoff of Array.from({ length: 40 }, (_, index) => index + 1)) {
+    it.effect(`shutdown at settlement operation ${cutoff} preserves its pending successor`, () =>
+      Effect.gen(function* () {
+        const running = yield* Deferred.make<void>()
+        const cleanup = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        const closeNow = yield* Deferred.make<void>()
+        const scope = yield* Scope.make()
+        yield* Effect.addFinalizer(() =>
+          Deferred.succeed(release, undefined).pipe(Effect.andThen(Scope.close(scope, Exit.void))),
+        )
+        const base = new Scheduler.MixedScheduler()
+        const dispatcher = base.makeDispatcher()
+        const lifecycle: string[] = []
+        let owner: number | undefined
+        let seen = 0
+        const scheduler: Scheduler.Scheduler = {
+          executionMode: base.executionMode,
+          makeDispatcher: () => base.makeDispatcher(),
+          shouldYield: (fiber) => {
+            if (fiber.id === owner && ++seen === cutoff) {
+              // Close on the next scheduler tick, never reentrantly inside the current operation.
+              dispatcher.scheduleTask(() => Deferred.doneUnsafe(closeNow, Exit.void), 0)
+              return true
+            }
+            return base.shouldYield(fiber)
+          },
+        }
+        const coordinator = yield* SessionRunCoordinator.make({
+          started: () => Effect.sync(() => void lifecycle.push("started")),
+          drain: () =>
+            Deferred.succeed(running, undefined).pipe(
+              Effect.andThen(Effect.never),
+              Effect.onInterrupt(() =>
+                Deferred.succeed(cleanup, undefined).pipe(Effect.andThen(Deferred.await(release))),
+              ),
+            ),
+          settled: () =>
+            Effect.withFiber((fiber) => {
+              owner ??= fiber.id
+              return Effect.void
+            }),
+          suspended: () => Effect.sync(() => void lifecycle.push("suspended")),
+        }).pipe(Scope.provide(scope), Effect.provideService(Scheduler.Scheduler, scheduler))
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(running)
+        yield* coordinator.interrupt("session")
+        yield* Deferred.await(cleanup)
+        yield* coordinator.wake("session")
+        const closing = yield* Deferred.await(closeNow).pipe(
+          Effect.andThen(Scope.close(scope, Exit.void)),
+          Effect.forkChild({ startImmediately: true }),
+        )
+        yield* Deferred.succeed(release, undefined)
+        yield* Effect.yieldNow
+        // If settlement finished before the cutoff, close the already-scheduled successor.
+        yield* Deferred.succeed(closeNow, undefined)
+        yield* Fiber.join(closing)
+
+        expect(lifecycle.length).toBe(2)
+        expect(yield* coordinator.active).toEqual(new Set())
+      }),
+    )
+  }
+
   it.effect("coalesces wakes received during active execution", () =>
     Effect.gen(function* () {
       const firstStarted = yield* Deferred.make<void>()

packages/server/test/session-instances.test.ts

@@ -52,18 +52,19 @@ it.live(
       const cell = PluginRuntime.makeCell()
       // Host and private instances must reuse the same global layer identities.
       const replacements: LayerNode.Replacements = [
-        [Global.node, tempGlobalLayer],
-        [Database.node, Database.node],
-        [Bus.node, Bus.node],
-        [App.node, App.node],
-        [ModelsDev.node, ModelsDev.configured({ fetch: false })],
-        [Watcher.node, Watcher.configured({ enabled: false })],
-        [PluginRuntime.node, PluginRuntime.layerWithCell(cell)],
-        [PluginRuntime.providerNode, PluginRuntime.providerNodeWithCell(cell)],
-        [llmClient, Layer.succeed(LLMClient.Service, llm)],
-        [SessionRunnerModel.node, Layer.succeed(SessionRunnerModel.Service, { resolve: () => Effect.succeed(model) })],
-        [
-          Instance.byLocationNode,
+        Global.node.replace(tempGlobalLayer),
+        Database.node.replace(Database.node),
+        Bus.node.replace(Bus.node),
+        App.node.replace(App.node),
+        ModelsDev.node.replace(ModelsDev.configured({ fetch: false })),
+        Watcher.node.replace(Watcher.configured({ enabled: false })),
+        PluginRuntime.node.replace(PluginRuntime.layerWithCell(cell)),
+        PluginRuntime.providerNode.replace(PluginRuntime.providerNodeWithCell(cell)),
+        llmClient.replace(Layer.succeed(LLMClient.Service, llm)),
+        SessionRunnerModel.node.replace(
+          Layer.succeed(SessionRunnerModel.Service, { resolve: () => Effect.succeed(model) }),
+        ),
+        Instance.byLocationNode.replace(
           Layer.effect(
             Instance.Service,
             Effect.gen(function* () {
@@ -151,7 +152,7 @@ it.live(
               })
             }),
           ),
-        ],
+        ),
       ]
       const context = yield* Layer.build(
         createEmbeddedRoutes({}, replacements).pipe(Layer.provide(HttpServer.layerServices)),