fix(core): separate interrupt successor waiters

AGENTS.md

@@ -143,7 +143,7 @@ const table = sqliteTable("session", {
 
 - Keep durable prompt admission separate from model execution. `SessionV2.prompt(...)` admits one durable `session_input` row before scheduling advisory `SessionExecution.wake(sessionID)` unless `resume: false` requests admit-only behavior. The serialized runner promotes admitted inputs into visible user messages at safe boundaries.
 - Reusing a Session ID adopts the existing Session. Reusing a prompt message ID reconciles an exact retry only when Session, prompt, and delivery mode match; conflicting reuse fails. Historical projected prompts lazily synthesize promoted inbox records during exact retry.
-- Keep `SessionExecution` process-global and Session-ID based. It discovers placement through the read-side `SessionStore` and `LocationServiceMap.get(session.location)`; no layer should take a Session ID.
+- Keep `SessionExecution` process-global and Session-ID based. Its local implementation owns the process-local Session coordinator and discovers placement through `SessionStore` plus `LocationServiceMap.get(session.location)` only when a drain starts; no layer should take a Session ID. V2 interruption targets the active process-local ownership chain for that Session; idle or missing interruption is a no-op.
 - Keep `SessionRunner`, model resolution, tool registry, permissions, and filesystem Location-scoped. Omitted `Location.workspaceID` means implicit-local placement; explicit workspace identity remains reserved for future placement semantics.
 - Preserve one explicit `llm.stream(request)` call per provider turn and reload projected history before durable continuation. Do not bridge through legacy `SessionPrompt.loop(...)` or delegate orchestration to an in-memory tool loop.
 - Keep local Session drains process-local until clustering is implemented. `SessionRunCoordinator` joins explicit same-Session resumes, coalesces prompt wakeups, and allows different Sessions to run concurrently. Advisory wakes drain eligible durable inbox rows only; post-crash activity recovery requires a separate explicit design before it may retry provider work.

packages/core/src/location-layer.ts

@@ -39,7 +39,6 @@ import { LLMClient } from "@opencode-ai/llm"
 import { RequestExecutor } from "@opencode-ai/llm/route"
 import * as SessionRunnerLLM from "./session/runner/llm"
 import { SessionRunnerModel } from "./session/runner/model"
-import { SessionRunCoordinator } from "./session/run-coordinator"
 import { SystemContextBuiltIns } from "./system-context-builtins"
 import { FetchHttpClient } from "effect/unstable/http"
 
@@ -81,7 +80,6 @@ export class LocationServiceMap extends LayerMap.Service<LocationServiceMap>()("
     )
     const model = SessionRunnerModel.locationLayer.pipe(Layer.provide(services))
     const runner = SessionRunnerLLM.defaultLayer.pipe(Layer.provide(services), Layer.provide(model))
-    const coordinator = SessionRunCoordinator.layer.pipe(Layer.provide(runner))
     return Layer.mergeAll(
       services,
       commits,
@@ -91,7 +89,6 @@ export class LocationServiceMap extends LayerMap.Service<LocationServiceMap>()("
       questions,
       model,
       runner,
-      coordinator,
       builtInTools,
     ).pipe(Layer.fresh)
   },

packages/core/src/public/opencode.ts

@@ -51,6 +51,7 @@ export const layer = Layer.effect(
           }),
         get: sessions.get,
         list: sessions.list,
+        interrupt: sessions.interrupt,
         prompt: (input) =>
           sessions.prompt({
             id: input.id,

packages/core/src/public/session.ts

@@ -84,6 +84,8 @@ export interface Interface {
   readonly get: (sessionID: ID) => Effect.Effect<Info, NotFoundError>
   readonly list: (input?: ListInput) => Effect.Effect<Info[]>
   readonly prompt: (input: PromptInput) => Effect.Effect<Admission, NotFoundError | PromptConflictError>
+  /** Interrupt the active V2 execution chain for one Session on this process. Interrupting an idle or missing Session is a no-op. */
+  readonly interrupt: (sessionID: ID) => Effect.Effect<void>
   readonly messages: (input: MessagesInput) => Effect.Effect<Message[], NotFoundError | MessageDecodeError>
   readonly message: (input: MessageInput) => Effect.Effect<Message | undefined>
   readonly context: (sessionID: ID) => Effect.Effect<Message[], NotFoundError | MessageDecodeError>

packages/core/src/session.ts

@@ -155,6 +155,7 @@ export interface Interface {
   readonly compact: (input: CompactInput) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
   readonly wait: (id: SessionSchema.ID) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
+  readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/v2/Session") {}
@@ -399,6 +400,9 @@ export const layer = Layer.effect(
         yield* result.get(sessionID)
         yield* execution.resume(sessionID)
       }),
+      interrupt: Effect.fn("V2Session.interrupt")(function* (sessionID) {
+        yield* execution.interrupt(sessionID)
+      }),
     })
 
     return result

packages/core/src/session/execution.ts

@@ -9,10 +9,15 @@ export interface Interface {
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
   /** Schedule a drain after durable work is recorded. Repeated wakeups may coalesce. */
   readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
+  /** Interrupt active work owned by this process. Idle interruption is a no-op. */
+  readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
 }
 
 /** Routes execution from a Session ID to the runner owned by that Session's Location. */
 export class Service extends Context.Service<Service, Interface>()("@opencode/v2/SessionExecution") {}
 
 /** Low-level compatibility layer for callers that only need durable Session recording. */
-export const noopLayer = Layer.succeed(Service, Service.of({ resume: () => Effect.void, wake: () => Effect.void }))
+export const noopLayer = Layer.succeed(
+  Service,
+  Service.of({ resume: () => Effect.void, wake: () => Effect.void, interrupt: () => Effect.void }),
+)

packages/core/src/session/execution/local.ts

@@ -1,6 +1,7 @@
 import { Effect, Layer } from "effect"
 import { LocationServiceMap } from "../../location-layer"
 import { SessionRunCoordinator } from "../run-coordinator"
+import { SessionRunner } from "../runner"
 import { SessionSchema } from "../schema"
 import { SessionStore } from "../store"
 import { SessionExecution } from "../execution"
@@ -11,25 +12,25 @@ export const layer = Layer.effect(
   Effect.gen(function* () {
     const store = yield* SessionStore.Service
     const locations = yield* LocationServiceMap
-    const scope = yield* Effect.scope
-    const withCoordinator = Effect.fnUntraced(function* <A, E>(
-      sessionID: SessionSchema.ID,
-      use: (coordinator: SessionRunCoordinator.Interface) => Effect.Effect<A, E>,
-    ) {
-      const session = yield* store.get(sessionID)
-      if (!session) return yield* Effect.die(`Session not found: ${sessionID}`)
-      return yield* SessionRunCoordinator.Service.use(use).pipe(Effect.provide(locations.get(session.location)))
+    const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, void, SessionRunner.RunError>({
+      drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, mode) {
+        const session = yield* store.get(sessionID)
+        if (!session) return yield* Effect.die(`Session not found: ${sessionID}`)
+        return yield* SessionRunner.Service.use((runner) => runner.run({ sessionID, force: mode === "run" })).pipe(
+          Effect.provide(locations.get(session.location)),
+        )
+      }),
+      onFailure: (sessionID, cause) =>
+        Effect.logError("Failed to drain Session").pipe(
+          Effect.annotateLogs("sessionID", sessionID),
+          Effect.annotateLogs("cause", cause),
+        ),
     })
 
     return SessionExecution.Service.of({
-      resume: Effect.fn("SessionExecution.resume")(function* (sessionID) {
-        return yield* withCoordinator(sessionID, (coordinator) => coordinator.run(sessionID))
-      }),
-      wake: Effect.fn("SessionExecution.wake")(function* (sessionID) {
-        yield* withCoordinator(sessionID, (coordinator) =>
-          coordinator.wake(sessionID).pipe(Effect.andThen(coordinator.awaitIdle(sessionID))),
-        ).pipe(Effect.forkIn(scope), Effect.asVoid)
-      }),
+      interrupt: coordinator.interrupt,
+      resume: coordinator.run,
+      wake: coordinator.wake,
     })
   }),
 )

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

@@ -1,6 +1,6 @@
 export * as SessionRunCoordinator from "./run-coordinator"
 
-import { Cause, Context, Deferred, Effect, Exit, FiberSet, Layer, Scope } from "effect"
+import { Cause, Context, Deferred, Effect, Exit, Fiber, FiberSet, Layer, Scope } from "effect"
 import { SessionRunner } from "./runner"
 import { SessionSchema } from "./schema"
 
@@ -18,6 +18,9 @@ export type Mode = "run" | "wake"
  *
  * `wake` reports that durable work may now be available. It starts a chain while idle or
  * requests one coalesced follow-up while draining. Repeated wakes collapse together.
+ *
+ * `interrupt` stops the current ownership chain. Wakes and explicit runs arriving after the
+ * interruption request become a fresh successor; previously queued reruns are suppressed.
  */
 export interface Coordinator<Key, A, E> {
   /** Starts or joins one explicit drain generation. */
@@ -26,13 +29,19 @@ export interface Coordinator<Key, A, E> {
   readonly wake: (key: Key) => Effect.Effect<void>
   /** Waits until the current ownership chain settles. */
   readonly awaitIdle: (key: Key) => Effect.Effect<void, E>
+  /** Interrupts the active ownership chain. Later requests may start a fresh successor. */
+  readonly interrupt: (key: Key) => Effect.Effect<void>
 }
 
 type Entry<A, E> = {
   readonly done: Deferred.Deferred<A, E>
+  readonly settled: Deferred.Deferred<Exit.Exit<A, E>>
   mode: Mode
   rerun?: Mode
   explicit?: Deferred.Deferred<A, E>
+  successorExplicit?: Deferred.Deferred<A, E>
+  owner?: Fiber.Fiber<void, never>
+  stopping: boolean
 }
 
 const strongest = (left: Mode | undefined, right: Mode): Mode => (left === "run" || right === "run" ? "run" : "wake")
@@ -44,7 +53,7 @@ export const make = <Key, A, E>(options: {
 }): Effect.Effect<Coordinator<Key, A, E>, never, Scope.Scope> =>
   Effect.gen(function* () {
     const active = new Map<Key, Entry<A, E>>()
-    const scope = yield* Effect.scope
+    const report = yield* FiberSet.makeRuntime<never, void, never>()
     const fork = yield* FiberSet.makeRuntime<never, void, never>()
     const shutdown = Deferred.makeUnsafe<void>()
     let closed = false
@@ -58,49 +67,77 @@ export const make = <Key, A, E>(options: {
 
     const makeEntry = (mode: Mode, explicit?: Deferred.Deferred<A, E>): Entry<A, E> => ({
       done: Deferred.makeUnsafe<A, E>(),
+      settled: Deferred.makeUnsafe<Exit.Exit<A, E>>(),
       mode,
       explicit,
+      stopping: false,
     })
 
-    const start = (key: Key, entry: Entry<A, E>, mode: Mode) => {
-      fork(own(key, entry, mode))
+    const start = (key: Key, entry: Entry<A, E>, mode: Mode, successor = false) => {
+      const ready = Deferred.makeUnsafe<void>()
+      const drain = Effect.suspend(() => options.drain(key, mode))
+      // Initial work retains immediate-start behavior but cannot run before ownership is published.
+      // Observer-started successors yield once so synchronous drains cannot recurse on the JS stack.
+      const owner = fork(
+        (successor ? Effect.yieldNow.pipe(Effect.andThen(drain)) : Deferred.await(ready).pipe(Effect.andThen(drain))).pipe(
+          Effect.onExit((exit) => Effect.sync(() => settle(key, entry, mode, exit))),
+          Effect.exit,
+          Effect.asVoid,
+        ),
+      )
+      entry.owner = owner
+      if (!successor) Deferred.doneUnsafe(ready, Effect.void)
     }
 
-    const own = (key: Key, entry: Entry<A, E>, mode: Mode): Effect.Effect<void> =>
-      Effect.suspend(() => options.drain(key, mode)).pipe(
-        Effect.exit,
-        Effect.flatMap((exit) => {
-          if (closed) return Deferred.done(entry.done, exit).pipe(Effect.asVoid)
-          if (mode === "run" && entry.explicit !== undefined) {
-            Deferred.doneUnsafe(entry.explicit, exit)
-            entry.explicit = undefined
-          }
-          if (exit._tag === "Success") {
-            if (active.get(key) !== entry) return Deferred.done(entry.done, exit).pipe(Effect.asVoid)
-            if (entry.rerun !== undefined) {
-              const mode = entry.rerun
-              entry.rerun = undefined
-              entry.mode = mode
-              return own(key, entry, mode)
-            }
-            active.delete(key)
-            return Deferred.done(entry.done, exit).pipe(Effect.asVoid)
-          }
+    const settle = (key: Key, entry: Entry<A, E>, mode: Mode, exit: Exit.Exit<A, E>) => {
+      if (closed) {
+        Deferred.doneUnsafe(entry.done, exit)
+        Deferred.doneUnsafe(entry.settled, Effect.succeed(exit))
+        return
+      }
+      if (mode === "run" && entry.explicit !== undefined) {
+        Deferred.doneUnsafe(entry.explicit, exit)
+        entry.explicit = undefined
+      }
+      if (entry.stopping && mode === "wake" && entry.explicit !== undefined) {
+        Deferred.doneUnsafe(entry.explicit, exit)
+        entry.explicit = undefined
+      }
+      if (active.get(key) !== entry) {
+        Deferred.doneUnsafe(entry.done, exit)
+        Deferred.doneUnsafe(entry.settled, Effect.succeed(exit))
+        return
+      }
+      if (exit._tag === "Success" && !entry.stopping) {
+        if (entry.rerun !== undefined) {
+          const mode = entry.rerun
+          entry.rerun = undefined
+          entry.mode = mode
+          start(key, entry, mode, true)
+          return
+        }
+        active.delete(key)
+        Deferred.doneUnsafe(entry.done, exit)
+        Deferred.doneUnsafe(entry.settled, Effect.succeed(exit))
+        return
+      }
 
-          const successor =
-            active.get(key) === entry && entry.rerun !== undefined ? makeEntry(entry.rerun, entry.explicit) : undefined
-          if (successor === undefined) active.delete(key)
-          else {
-            active.set(key, successor)
-          }
-          if (successor !== undefined) start(key, successor, successor.mode)
-          const report =
-            mode === "wake" && options.onFailure !== undefined
-              ? options.onFailure(key, exit.cause).pipe(Effect.forkIn(scope), Effect.asVoid)
-              : Effect.void
-          return Deferred.done(entry.done, exit).pipe(Effect.andThen(report), Effect.asVoid)
-        }),
-      )
+      const successorExplicit = entry.successorExplicit ?? (mode === "wake" ? entry.explicit : undefined)
+      const successor = entry.rerun !== undefined ? makeEntry(entry.rerun, successorExplicit) : undefined
+      if (successor === undefined) active.delete(key)
+      else active.set(key, successor)
+      if (successor !== undefined) start(key, successor, successor.mode, true)
+      Deferred.doneUnsafe(entry.done, exit)
+      Deferred.doneUnsafe(entry.settled, Effect.succeed(exit))
+      if (
+        exit._tag === "Failure" &&
+        !(entry.stopping && Cause.hasInterruptsOnly(exit.cause)) &&
+        mode === "wake" &&
+        options.onFailure !== undefined
+      ) {
+        report(Effect.suspend(() => options.onFailure!(key, exit.cause)))
+      }
+    }
 
     const wake = (key: Key) =>
       Effect.sync(() => {
@@ -123,7 +160,7 @@ export const make = <Key, A, E>(options: {
           const entry = active.get(key)
           if (entry === undefined) break
           const exit = yield* Effect.raceFirst(
-            Deferred.await(entry.done).pipe(Effect.exit),
+            Deferred.await(entry.settled),
             Deferred.await(shutdown).pipe(Effect.as(Exit.void)),
           )
           if (closed) break
@@ -132,13 +169,29 @@ export const make = <Key, A, E>(options: {
         if (firstFailure !== undefined) return yield* Effect.failCause(firstFailure)
       })
 
-    return { run, wake, awaitIdle }
+    const interrupt = (key: Key): Effect.Effect<void> =>
+      Effect.suspend(() => {
+        const entry = active.get(key)
+        if (entry?.owner === undefined) return Effect.void
+        if (!entry.stopping) {
+          entry.stopping = true
+          entry.rerun = undefined
+        }
+        return Fiber.interrupt(entry.owner)
+      })
+
+    return { run, wake, awaitIdle, interrupt }
 
     function run(key: Key): Effect.Effect<A, E> {
       return Effect.uninterruptibleMask((restore) => {
         if (closed) return Effect.interrupt
         const entry = active.get(key)
         if (entry !== undefined) {
+          if (entry.stopping) {
+            entry.rerun = strongest(entry.rerun, "run")
+            entry.successorExplicit ??= Deferred.makeUnsafe<A, E>()
+            return restore(awaitRun(entry.successorExplicit))
+          }
           if (entry.mode === "wake") {
             entry.rerun = "run"
             entry.explicit ??= Deferred.makeUnsafe<A, E>()
@@ -165,19 +218,17 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/v2
 
 export const layer = Layer.effect(
   Service,
-  Effect.gen(function* () {
-    const runner = yield* SessionRunner.Service
-    return Service.of(
-      yield* make<SessionSchema.ID, void, SessionRunner.RunError>({
+  SessionRunner.Service.pipe(
+    Effect.flatMap((runner) =>
+      make<SessionSchema.ID, void, SessionRunner.RunError>({
         drain: (sessionID, mode) => runner.run({ sessionID, force: mode === "run" }),
         onFailure: (sessionID, cause) =>
-          Cause.hasInterruptsOnly(cause)
-            ? Effect.void
-            : Effect.logError("Failed to drain Session").pipe(
-                Effect.annotateLogs("sessionID", sessionID),
-                Effect.annotateLogs("cause", cause),
-              ),
+          Effect.logError("Failed to drain Session").pipe(
+            Effect.annotateLogs("sessionID", sessionID),
+            Effect.annotateLogs("cause", cause),
+          ),
       }),
-    )
-  }),
+    ),
+    Effect.map(Service.of),
+  ),
 )

packages/core/src/session/runner/llm.ts

@@ -16,6 +16,7 @@ import { SessionInput } from "../input"
 import { QuestionV2 } from "../../question"
 import { SystemContextRegistry } from "../../system-context-registry"
 import { SessionContextEpoch } from "../context-epoch"
+import { Location } from "../../location"
 
 /**
  * Runs one durable coding-agent Session until it settles.
@@ -87,6 +88,7 @@ export const layer = Layer.effect(
     const tools = yield* ToolRegistry.Service
     const models = yield* SessionRunnerModel.Service
     const store = yield* SessionStore.Service
+    const location = yield* Location.Service
     const systemContext = yield* SystemContextRegistry.Service
     const db = (yield* Database.Service).db
     const getSession = Effect.fn("SessionRunner.getSession")(function* (sessionID: SessionSchema.ID) {
@@ -132,6 +134,8 @@ export const layer = Layer.effect(
       promotion: "steer" | "queue" | undefined,
     ) {
       const session = yield* getSession(sessionID)
+      if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID)
+        return yield* Effect.interrupt
       const initialized = yield* SessionContextEpoch.initialize(db, systemContext, session.id, session.location)
       const model = yield* models.resolve(session)
       const toolFibers = yield* FiberSet.make<void, never>()

packages/core/test/public-opencode.test.ts

@@ -17,6 +17,7 @@ describe("public native OpenCode API", () => {
         "create",
         "events",
         "get",
+        "interrupt",
         "list",
         "message",
         "messages",

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

@@ -23,6 +23,7 @@ const events = EventV2.layer.pipe(Layer.provide(database))
 const projector = SessionProjector.layer.pipe(Layer.provide(events), Layer.provide(database))
 const store = SessionStore.layer.pipe(Layer.provide(database))
 const executionCalls: SessionV2.ID[] = []
+const interruptCalls: SessionV2.ID[] = []
 const wakeCalls: SessionV2.ID[] = []
 const execution = Layer.succeed(
   SessionExecution.Service,
@@ -31,6 +32,10 @@ const execution = Layer.succeed(
       Effect.sync(() => {
         executionCalls.push(sessionID)
       }),
+    interrupt: (sessionID) =>
+      Effect.sync(() => {
+        interruptCalls.push(sessionID)
+      }),
     wake: (sessionID) =>
       Effect.sync(() => {
         wakeCalls.push(sessionID)
@@ -108,6 +113,27 @@ describe("SessionV2.prompt", () => {
     }),
   )
 
+  it.effect("delegates interruption through SessionExecution", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      interruptCalls.length = 0
+
+      yield* session.interrupt(sessionID)
+      expect(interruptCalls).toEqual([sessionID])
+    }),
+  )
+
+  it.effect("delegates interruption without requiring a recorded Session", () =>
+    Effect.gen(function* () {
+      const session = yield* SessionV2.Service
+      interruptCalls.length = 0
+
+      yield* session.interrupt(SessionV2.ID.make("ses_missing"))
+      expect(interruptCalls).toEqual([SessionV2.ID.make("ses_missing")])
+    }),
+  )
+
   it.effect("durably admits one user message before transcript promotion", () =>
     Effect.gen(function* () {
       yield* setup

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

@@ -41,6 +41,371 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("does nothing when interrupted while idle", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const coordinator = yield* SessionRunCoordinator.make({ drain: () => Effect.void })
+
+        yield* coordinator.interrupt("session")
+      }),
+    ),
+  )
+
+  it.effect("interrupts only the requested key", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const secondStarted = yield* Deferred.make<void>()
+        const secondGate = yield* Deferred.make<void>()
+        const secondInterrupted = yield* Deferred.make<void>()
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (key: string) =>
+            key === "first"
+              ? Deferred.succeed(firstStarted, undefined).pipe(Effect.andThen(Effect.never))
+              : Deferred.succeed(secondStarted, undefined).pipe(
+                  Effect.andThen(Deferred.await(secondGate)),
+                  Effect.onInterrupt(() => Deferred.succeed(secondInterrupted, undefined)),
+                ),
+        })
+
+        yield* coordinator.wake("first")
+        yield* coordinator.wake("second")
+        yield* Effect.all([Deferred.await(firstStarted), Deferred.await(secondStarted)])
+
+        yield* coordinator.interrupt("first")
+        expect(yield* Deferred.isDone(secondInterrupted)).toBeFalse()
+        yield* Deferred.succeed(secondGate, undefined)
+        yield* coordinator.awaitIdle("second")
+      }),
+    ),
+  )
+
+  it.effect("interrupts the active drain and suppresses its queued wake", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const interrupted = yield* Deferred.make<void>()
+        let runs = 0
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () =>
+            Effect.sync(() => ++runs).pipe(
+              Effect.flatMap((run) =>
+                run === 1
+                  ? Deferred.succeed(firstStarted, undefined).pipe(
+                      Effect.andThen(Effect.never),
+                      Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined)),
+                    )
+                  : Effect.void,
+              ),
+            ),
+        })
+
+        const run = yield* coordinator.run("session").pipe(Effect.forkChild)
+        yield* Deferred.await(firstStarted)
+        yield* coordinator.wake("session")
+
+        yield* coordinator.interrupt("session")
+        yield* Deferred.await(interrupted)
+        yield* coordinator.awaitIdle("session")
+        const exit = yield* Fiber.await(run)
+        expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue()
+        expect(runs).toBe(1)
+        yield* coordinator.interrupt("session")
+      }),
+    ),
+  )
+
+  it.effect("allows a wake received after interruption to start a successor", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const firstInterrupted = yield* Deferred.make<void>()
+        const secondStarted = yield* Deferred.make<void>()
+        let runs = 0
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () =>
+            Effect.sync(() => ++runs).pipe(
+              Effect.flatMap((run) =>
+                run === 1
+                  ? Deferred.succeed(firstStarted, undefined).pipe(
+                      Effect.andThen(Effect.never),
+                      Effect.onInterrupt(() => Deferred.succeed(firstInterrupted, undefined)),
+                    )
+                  : Deferred.succeed(secondStarted, undefined),
+              ),
+            ),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(firstStarted)
+        const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        yield* Effect.yieldNow
+        yield* coordinator.wake("session")
+        yield* Deferred.await(firstInterrupted)
+        yield* Fiber.join(interrupt)
+        yield* Deferred.await(secondStarted)
+        yield* coordinator.awaitIdle("session")
+
+        expect(runs).toBe(2)
+      }),
+    ),
+  )
+
+  it.effect("interrupts an explicit run queued before the interruption request", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        let runs = 0
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () =>
+            Effect.sync(() => ++runs).pipe(
+              Effect.flatMap((run) =>
+                run === 1
+                  ? Deferred.succeed(firstStarted, undefined).pipe(Effect.andThen(Effect.never))
+                  : Effect.void,
+              ),
+            ),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(firstStarted)
+        const run = yield* coordinator.run("session").pipe(Effect.forkChild)
+        yield* Effect.yieldNow
+
+        yield* coordinator.interrupt("session")
+        const exit = yield* Fiber.await(run)
+        expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue()
+        expect(runs).toBe(1)
+      }),
+    ),
+  )
+
+  it.effect("settles a pre-interrupt explicit run only after active wake cleanup", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const cleanupStarted = yield* Deferred.make<void>()
+        const cleanupGate = yield* Deferred.make<void>()
+        const runSettled = yield* Deferred.make<void>()
+        const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
+          drain: () =>
+            Deferred.succeed(started, undefined).pipe(
+              Effect.andThen(Effect.never),
+              Effect.onInterrupt(() =>
+                Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+              ),
+            ),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(started)
+        const run = yield* coordinator
+          .run("session")
+          .pipe(Effect.exit, Effect.ensuring(Deferred.succeed(runSettled, undefined)), Effect.forkChild)
+        const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        yield* Deferred.await(cleanupStarted)
+
+        expect(yield* Deferred.isDone(runSettled)).toBeFalse()
+        yield* Deferred.succeed(cleanupGate, undefined)
+        const runExit = yield* Fiber.join(run)
+        expect(Exit.isFailure(runExit) && Cause.hasInterruptsOnly(runExit.cause)).toBeTrue()
+        yield* Fiber.join(interrupt)
+      }),
+    ),
+  )
+
+  it.effect("routes an explicit run arriving during interrupt cleanup to the successor", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const cleanupStarted = yield* Deferred.make<void>()
+        const cleanupGate = yield* Deferred.make<void>()
+        const secondStarted = yield* Deferred.make<void>()
+        let runs = 0
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () =>
+            Effect.sync(() => ++runs).pipe(
+              Effect.flatMap((run) =>
+                run === 1
+                  ? Deferred.succeed(firstStarted, undefined).pipe(
+                      Effect.andThen(Effect.never),
+                      Effect.onInterrupt(() =>
+                        Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+                      ),
+                    )
+                  : Deferred.succeed(secondStarted, undefined),
+              ),
+            ),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(firstStarted)
+        const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        yield* Deferred.await(cleanupStarted)
+        const run = yield* coordinator.run("session").pipe(Effect.forkChild)
+        yield* Deferred.succeed(cleanupGate, undefined)
+        yield* Fiber.join(interrupt)
+        yield* Fiber.join(run)
+        yield* Deferred.await(secondStarted)
+        expect(runs).toBe(2)
+      }),
+    ),
+  )
+
+  it.effect("separates pre-interrupt and post-interrupt explicit waiters", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const cleanupStarted = yield* Deferred.make<void>()
+        const cleanupGate = yield* Deferred.make<void>()
+        const secondStarted = yield* Deferred.make<void>()
+        let runs = 0
+        const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
+          drain: () =>
+            Effect.sync(() => ++runs).pipe(
+              Effect.flatMap((run) =>
+                run === 1
+                  ? Deferred.succeed(firstStarted, undefined).pipe(
+                      Effect.andThen(Effect.never),
+                      Effect.onInterrupt(() =>
+                        Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+                      ),
+                    )
+                  : Deferred.succeed(secondStarted, undefined),
+              ),
+            ),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(firstStarted)
+        const before = yield* coordinator.run("session").pipe(Effect.exit, Effect.forkChild)
+        const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        yield* Deferred.await(cleanupStarted)
+        const after = yield* coordinator.run("session").pipe(Effect.exit, Effect.forkChild)
+        yield* Deferred.succeed(cleanupGate, undefined)
+
+        const beforeExit = yield* Fiber.join(before)
+        expect(Exit.isFailure(beforeExit) && Cause.hasInterruptsOnly(beforeExit.cause)).toBeTrue()
+        yield* Fiber.join(interrupt)
+        yield* Fiber.join(after)
+        yield* Deferred.await(secondStarted)
+        expect(runs).toBe(2)
+      }),
+    ),
+  )
+
+  it.effect("waits for interrupt cleanup before settling callers", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const cleanupStarted = yield* Deferred.make<void>()
+        const cleanupGate = yield* Deferred.make<void>()
+        const runSettled = yield* Deferred.make<void>()
+        const idleSettled = yield* Deferred.make<void>()
+        const interruptSettled = yield* Deferred.make<void>()
+        const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
+          drain: () =>
+            Deferred.succeed(started, undefined).pipe(
+              Effect.andThen(Effect.never),
+              Effect.onInterrupt(() =>
+                Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+              ),
+            ),
+        })
+
+        const run = yield* coordinator
+          .run("session")
+          .pipe(Effect.ensuring(Deferred.succeed(runSettled, undefined)), Effect.forkChild)
+        yield* Deferred.await(started)
+        const idle = yield* coordinator
+          .awaitIdle("session")
+          .pipe(Effect.exit, Effect.ensuring(Deferred.succeed(idleSettled, undefined)), Effect.forkChild)
+        const interrupt = yield* coordinator
+          .interrupt("session")
+          .pipe(Effect.ensuring(Deferred.succeed(interruptSettled, undefined)), Effect.forkChild)
+        yield* Deferred.await(cleanupStarted)
+
+        expect(yield* Deferred.isDone(runSettled)).toBeFalse()
+        expect(yield* Deferred.isDone(idleSettled)).toBeFalse()
+        expect(yield* Deferred.isDone(interruptSettled)).toBeFalse()
+        yield* Deferred.succeed(cleanupGate, undefined)
+        const runExit = yield* Fiber.await(run)
+        const idleExit = yield* Fiber.join(idle)
+        expect(Exit.isFailure(runExit) && Cause.hasInterruptsOnly(runExit.cause)).toBeTrue()
+        expect(Exit.isFailure(idleExit) && Cause.hasInterruptsOnly(idleExit.cause)).toBeTrue()
+        yield* Fiber.join(interrupt)
+      }),
+    ),
+  )
+
+  it.effect("joins concurrent interruption requests for one active drain", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const cleanupStarted = yield* Deferred.make<void>()
+        const cleanupGate = yield* Deferred.make<void>()
+        const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
+          drain: () =>
+            Deferred.succeed(started, undefined).pipe(
+              Effect.andThen(Effect.never),
+              Effect.onInterrupt(() =>
+                Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+              ),
+            ),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(started)
+        const first = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        yield* Deferred.await(cleanupStarted)
+        const second = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        yield* Deferred.succeed(cleanupGate, undefined)
+
+        yield* Fiber.join(first)
+        yield* Fiber.join(second)
+      }),
+    ),
+  )
+
+  it.effect("does not discard a post-interrupt successor when interrupted again", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const cleanupStarted = yield* Deferred.make<void>()
+        const cleanupGate = yield* Deferred.make<void>()
+        const secondStarted = yield* Deferred.make<void>()
+        let runs = 0
+        const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
+          drain: () =>
+            Effect.sync(() => ++runs).pipe(
+              Effect.flatMap((run) =>
+                run === 1
+                  ? Deferred.succeed(firstStarted, undefined).pipe(
+                      Effect.andThen(Effect.never),
+                      Effect.onInterrupt(() =>
+                        Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+                      ),
+                    )
+                  : Deferred.succeed(secondStarted, undefined),
+              ),
+            ),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(firstStarted)
+        const firstInterrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        yield* Deferred.await(cleanupStarted)
+        const run = yield* coordinator.run("session").pipe(Effect.forkChild)
+        const secondInterrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        yield* Deferred.succeed(cleanupGate, undefined)
+
+        yield* Effect.all([Fiber.join(firstInterrupt), Fiber.join(secondInterrupt), Fiber.join(run)])
+        yield* Deferred.await(secondStarted)
+        expect(runs).toBe(2)
+      }),
+    ),
+  )
+
   it.effect("coalesces wakes received during an active run", () =>
     Effect.scoped(
       Effect.gen(function* () {
@@ -381,4 +746,103 @@ describe("SessionRunCoordinator", () => {
       }),
     ),
   )
+
+  it.effect("reports an advisory drain failure exactly once", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const failure = new Error("wake failed")
+        const reported: Cause.Cause<Error>[] = []
+        const reportedOnce = yield* Deferred.make<void>()
+        const coordinator = yield* SessionRunCoordinator.make<string, void, Error>({
+          drain: () => Effect.fail(failure),
+          onFailure: (_key, cause) =>
+            Effect.sync(() => reported.push(cause)).pipe(Effect.andThen(Deferred.succeed(reportedOnce, undefined))),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(reportedOnce)
+        yield* Effect.yieldNow
+
+        expect(reported).toHaveLength(1)
+        expect(Cause.squash(reported[0]!)).toBe(failure)
+      }),
+    ),
+  )
+
+  it.effect("contains defects thrown while constructing an advisory failure report", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const coordinator = yield* SessionRunCoordinator.make<string, void, Error>({
+          drain: () => Effect.fail(new Error("wake failed")),
+          onFailure: () => {
+            throw new Error("report defect")
+          },
+        })
+
+        yield* coordinator.wake("session")
+        yield* coordinator.awaitIdle("session").pipe(Effect.exit)
+        yield* coordinator.wake("session")
+        yield* coordinator.awaitIdle("session").pipe(Effect.exit)
+      }),
+    ),
+  )
+
+  it.effect("reports an independently interrupted advisory drain", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const reported = yield* Deferred.make<Cause.Cause<never>>()
+        const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
+          drain: () => Effect.interrupt,
+          onFailure: (_key, cause) => Deferred.succeed(reported, cause).pipe(Effect.asVoid),
+        })
+
+        yield* coordinator.wake("session")
+
+        expect(Cause.hasInterruptsOnly(yield* Deferred.await(reported))).toBeTrue()
+      }),
+    ),
+  )
+
+  it.effect("does not report deliberate interruption as an advisory failure", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const reported: Cause.Cause<never>[] = []
+        const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
+          drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)),
+          onFailure: (_key, cause) => Effect.sync(() => reported.push(cause)),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(started)
+        yield* coordinator.interrupt("session")
+        yield* Effect.yieldNow
+
+        expect(reported).toEqual([])
+      }),
+    ),
+  )
+
+  it.effect("trampolines many synchronous self-waking drains", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const limit = 20_000
+        let runs = 0
+        let wake: (key: string) => Effect.Effect<void> = () => Effect.void
+        const coordinator = yield* SessionRunCoordinator.make<string, void, never>({
+          drain: (key) =>
+            Effect.sync(() => ++runs).pipe(
+              Effect.tap((run) => (run < limit ? wake(key) : Effect.void)),
+              Effect.asVoid,
+            ),
+        })
+        wake = coordinator.wake
+
+        yield* coordinator.wake("session")
+        yield* coordinator.awaitIdle("session")
+
+        expect(runs).toBe(limit)
+      }),
+    ),
+  )
 })

packages/core/test/session-runner-recorded.test.ts

@@ -20,6 +20,7 @@ import { ToolRegistry } from "@opencode-ai/core/tool/registry"
 import { SessionTable } from "@opencode-ai/core/session/sql"
 import { SessionStore } from "@opencode-ai/core/session/store"
 import { SystemContextRegistry } from "@opencode-ai/core/system-context-registry"
+import { Location } from "@opencode-ai/core/location"
 import { describe, expect } from "bun:test"
 import { eq } from "drizzle-orm"
 import { Effect, Layer } from "effect"
@@ -57,6 +58,7 @@ const model = OpenAIChat.route
   .model({ id: "gpt-4o-mini" })
 const models = SessionRunnerModel.layerWith(() => Effect.succeed(model))
 const systemContext = SystemContextRegistry.layer
+const location = Location.layer({ directory: AbsolutePath.make("/project") }).pipe(Layer.provide(Project.defaultLayer))
 const runner = SessionRunnerLLM.defaultLayer.pipe(
   Layer.provide(database),
   Layer.provide(store),
@@ -65,12 +67,15 @@ const runner = SessionRunnerLLM.defaultLayer.pipe(
   Layer.provide(registry),
   Layer.provide(models),
   Layer.provide(systemContext),
+  Layer.provide(location),
 )
 const coordinator = SessionRunCoordinator.layer.pipe(Layer.provide(runner))
 const execution = Layer.effect(
   SessionExecution.Service,
   SessionRunCoordinator.Service.pipe(
-    Effect.map((coordinator) => SessionExecution.Service.of({ resume: coordinator.run, wake: coordinator.wake })),
+    Effect.map((coordinator) =>
+      SessionExecution.Service.of({ resume: coordinator.run, wake: coordinator.wake, interrupt: coordinator.interrupt }),
+    ),
   ),
 ).pipe(Layer.provide(coordinator))
 const sessions = SessionV2.layer.pipe(
@@ -92,6 +97,7 @@ const it = testEffect(
     registry,
     models,
     systemContext,
+    location,
     runner,
     coordinator,
     execution,

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

@@ -42,6 +42,7 @@ import { SessionStore } from "@opencode-ai/core/session/store"
 import { SystemContext } from "@opencode-ai/core/system-context"
 import { SystemContextRegistry } from "@opencode-ai/core/system-context-registry"
 import { ModelV2 } from "@opencode-ai/core/model"
+import { Location } from "@opencode-ai/core/location"
 import { ProviderV2 } from "@opencode-ai/core/provider"
 import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect"
 import { asc, eq } from "drizzle-orm"
@@ -180,6 +181,7 @@ const systemContext = Layer.effectDiscard(
     ),
   ),
 ).pipe(Layer.provideMerge(SystemContextRegistry.layer))
+const location = Location.layer({ directory: AbsolutePath.make("/project") }).pipe(Layer.provide(Project.defaultLayer))
 const runner = SessionRunnerLLM.layer.pipe(
   Layer.provide(database),
   Layer.provide(store),
@@ -188,12 +190,15 @@ const runner = SessionRunnerLLM.layer.pipe(
   Layer.provide(registry),
   Layer.provide(models),
   Layer.provide(systemContext),
+  Layer.provide(location),
 )
 const coordinator = SessionRunCoordinator.layer.pipe(Layer.provide(runner))
 const execution = Layer.effect(
   SessionExecution.Service,
   SessionRunCoordinator.Service.pipe(
-    Effect.map((coordinator) => SessionExecution.Service.of({ resume: coordinator.run, wake: coordinator.wake })),
+    Effect.map((coordinator) =>
+      SessionExecution.Service.of({ resume: coordinator.run, wake: coordinator.wake, interrupt: coordinator.interrupt }),
+    ),
   ),
 ).pipe(Layer.provide(coordinator))
 const sessions = SessionV2.layer.pipe(
@@ -217,6 +222,7 @@ const it = testEffect(
     echo,
     models,
     systemContext,
+    location,
     runner,
     coordinator,
     execution,
@@ -596,7 +602,7 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
-  it.effect("requires a complete new baseline after a Session moves", () =>
+  it.effect("interrupts a source Location runner after a Session moves", () =>
     Effect.gen(function* () {
       yield* setup
       const session = yield* SessionV2.Service
@@ -620,12 +626,10 @@ describe("SessionRunnerLLM", () => {
           .get(),
       ).toBeUndefined()
 
-      systemUnavailable = true
       yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Second" }), resume: false })
       const exit = yield* session.resume(sessionID).pipe(Effect.exit)
 
-      expect(Exit.isFailure(exit)).toBe(true)
-      if (Exit.isFailure(exit)) expect(Cause.squash(exit.cause)).toBeInstanceOf(SystemContext.InitializationBlocked)
+      expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true)
       expect(requests).toHaveLength(1)
       expect(yield* SessionInput.hasPending(db, sessionID, "steer")).toBe(true)
     }),
@@ -1558,6 +1562,48 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("preserves durable queued input for a later wake after interruption", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      const coordinator = yield* SessionRunCoordinator.Service
+      yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Interrupt current work" }), resume: false })
+
+      requests.length = 0
+      responses = [
+        [],
+        [
+          LLMEvent.stepStart({ index: 0 }),
+          LLMEvent.stepFinish({ index: 0, reason: "stop" }),
+          LLMEvent.finish({ reason: "stop" }),
+        ],
+      ]
+      streamGate = yield* Deferred.make<void>()
+      streamStarted = yield* Deferred.make<void>()
+
+      const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
+      yield* Deferred.await(streamStarted)
+      yield* session.prompt({
+        sessionID,
+        prompt: new Prompt({ text: "Run after interrupt" }),
+        delivery: "queue",
+      })
+      yield* session.interrupt(sessionID)
+      expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" })
+      expect(requests).toHaveLength(1)
+      yield* coordinator.wake(sessionID)
+      while (requests.length < 2) yield* Effect.yieldNow
+      yield* Deferred.succeed(streamGate, undefined)
+      yield* coordinator.awaitIdle(sessionID)
+      streamGate = undefined
+      streamStarted = undefined
+
+      expect(requests).toHaveLength(2)
+      expect(userTexts(requests[0]!)).toEqual(["Interrupt current work"])
+      expect(userTexts(requests[1]!)).toEqual(["Interrupt current work", "Run after interrupt"])
+    }),
+  )
+
   it.effect("runs queued active inputs as separate FIFO activities", () =>
     Effect.gen(function* () {
       yield* setup
@@ -2269,13 +2315,13 @@ describe("SessionRunnerLLM", () => {
         Stream.never,
       )
 
-      const runner = yield* SessionRunner.Service
-      const run = yield* runner.run({ sessionID, force: true }).pipe(Effect.forkChild)
+      const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
       while (executions.length === 0) yield* Effect.yieldNow
-      yield* Fiber.interrupt(run)
+      yield* session.interrupt(sessionID)
       toolExecutionGate = undefined
 
       expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" })
+      yield* session.interrupt(sessionID)
       expect(yield* session.context(sessionID)).toMatchObject([
         { type: "user", text: "Interrupt blocked tool" },
         {
@@ -2304,6 +2350,29 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("interrupts a blocked provider turn without local tool activity", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Interrupt provider" }), resume: false })
+      requests.length = 0
+      response = []
+      streamGate = yield* Deferred.make<void>()
+      streamStarted = yield* Deferred.make<void>()
+
+      const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
+      yield* Deferred.await(streamStarted)
+      yield* session.interrupt(sessionID)
+      const exit = yield* Fiber.await(run)
+      streamGate = undefined
+      streamStarted = undefined
+
+      expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue()
+      expect(requests).toHaveLength(1)
+      yield* session.interrupt(sessionID)
+    }),
+  )
+
   it.effect("durably fails blocked local tools when interrupted while awaiting settlement", () =>
     Effect.gen(function* () {
       yield* setup

specs/v2/session.md

@@ -18,6 +18,13 @@ sessions.prompt({ id?, sessionID, prompt, delivery?, resume? })
   -> exact retry schedules another wake unless resume is false
   -> resume omitted or true schedules execution after admission
   -> resume false admits only
+
+sessions.interrupt(sessionID)
+  -> interrupts the active ownership chain on this process
+  -> waits for active drain cleanup and settlement
+  -> suppresses reruns already queued before interruption
+  -> preserves durable inbox rows for a later fresh wake or resume
+  -> idle or missing Session is a no-op
 ```
 
 `session_input` is the durable admission inbox. Admitted inputs remain outside model-visible Session history until the serialized runner publishes `PromptLifecycle.Promoted`. The projector atomically writes the visible user message and marks its inbox row promoted in the same event transaction. The legacy V1-to-V2 shadow bridge continues publishing ordinary `Prompted` events for already-visible V1 prompts.
@@ -111,7 +118,7 @@ Execution has two entry points:
 
 Post-crash activity recovery is intentionally deferred. A wake does not infer that ambiguous provider work is safe to retry after an input has already been promoted. Explicit `run` may deliberately continue from durable projected history. A future recovery slice should model durable activity identity, provider-dispatch ambiguity, required continuation, queue-opener reservation, retry policy, and visible recovery status together.
 
-A location-scoped `SessionRunCoordinator` serializes each Session drain chain while allowing different Sessions to drain concurrently. Automatic startup discovery, durable multi-node ownership, stale-owner fencing, interruption controls, and retry policy remain future work.
+A process-global `SessionRunCoordinator` serializes each local Session drain chain while allowing different Sessions to drain concurrently. It enters the Session's current Location only when a drain starts, so interruption targets process execution ownership rather than Location cache identity. Interruption establishes a local ownership-chain boundary by stopping the current chain while preserving pending/unpromoted durable inbox rows for a later fresh wake and projected history for explicit resume. A Location runner also fences every new provider turn against its captured Location so a moved Session cannot begin another turn through source-Location tools or context. An already-dispatched provider turn may still settle source-Location calls until a future move-control slice interrupts active ownership. Automatic startup discovery, durable multi-node ownership, stale-owner fencing, and retry policy remain future work.
 
 Inbox promotion coalesces pending steers in durable admission order and opens one queued activity at a time in FIFO order. Add explicit inbox backlog and steering-batch limits before exposing broad multi-caller admission or untrusted queue growth.
 

specs/v2/todo.md

@@ -33,7 +33,7 @@ through legacy `SessionPrompt.loop(...)`:
 Prompt admission now uses a durable `session_input` inbox rather than immediate
 transcript projection. `steer` inputs coalesce into the active activity at the
 next safe provider-turn boundary. `queue` inputs form a FIFO of future activities
-that open one at a time. A location-scoped `SessionRunCoordinator` coalesces process-local wakeups
+that open one at a time. A process-global `SessionRunCoordinator` coalesces process-local wakeups
 around settlement races. Explicit `run` resumes perform at least one provider
 attempt; advisory `wake` notifications call the provider only for eligible inbox
 work. Steers coalesce into the active activity at
@@ -55,7 +55,7 @@ Next reviewed slices:
 - integrate the new BackgroundJob service with V2 tool execution: support background
   bash jobs and background agent dispatch with durable status observation,
   completion delivery, and explicit cancellation / continuation semantics
-- add compaction, interruption, retries, and stale-owner fencing
+- add compaction, durable/clustered interruption, retries, and stale-owner fencing
   only as their slices become concrete
 
 ### Deferred durable activity recovery