refactor(core): move wake scope into coordinator

packages/core/src/session.ts

@@ -789,7 +789,10 @@ const layer = Layer.effect(
             return false
           }),
         )
-        if (recovered) return
+        if (recovered) {
+          yield* execution.wakeActive(input.sessionID)
+          return
+        }
         yield* execution.wake(input.sessionID)
       }),
       compact: Effect.fn("Session.compact")(function* (input) {
@@ -875,8 +878,7 @@ const layer = Layer.effect(
       interrupt: Effect.fn("Session.interrupt")((sessionID, options) =>
         Effect.uninterruptible(
           Effect.gen(function* () {
-            yield* execution.interrupt(sessionID)
-            if (options?.continue && (yield* SessionInbox.has(db, sessionID, "any"))) yield* execution.wake(sessionID)
+            yield* execution.interrupt(sessionID, options)
           }),
         ),
       ),

packages/core/src/session/execution.ts

@@ -1,7 +1,8 @@
 export * as SessionExecution from "./execution.js"
 
-import { Cause, Context, Effect, Exit, Layer, Stream } from "effect"
+import { Cause, Context, Effect, Exit, Layer } from "effect"
 import { Bus } from "../bus.js"
+import { Database } from "../database/database.js"
 import { LocationServiceMap } from "../location-service-map.js"
 import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
 import { SessionEvent } from "./event.js"
@@ -11,6 +12,7 @@ import { SessionSchema } from "./schema.js"
 import { SessionStore } from "./store.js"
 import { toSessionError } from "./to-session-error.js"
 import { UserInterruptedError } from "./error.js"
+import { SessionInbox } from "./inbox.js"
 
 export interface Interface {
   /** Snapshots active execution owned by this process. */
@@ -19,8 +21,10 @@ export interface Interface {
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
   /** Registers newly recorded work. Repeated wakeups may coalesce. */
   readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void>
+  /** Wakes only an active execution, preserving its current input eligibility. */
+  readonly wakeActive: (sessionID: SessionSchema.ID) => Effect.Effect<void>
   /** Interrupt active work owned by this process. Idle interruption is a no-op. */
-  readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
+  readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<void>
   /** Resolves once this process owns no active execution for the Session. Returns immediately when idle and never starts work. */
   readonly awaitIdle: (sessionID: SessionSchema.ID) => Effect.Effect<void>
 }
@@ -45,6 +49,7 @@ export const layer = Layer.effect(
     const store = yield* SessionStore.Service
     const locations = yield* LocationServiceMap.Service
     const bus = yield* Bus.Service
+    const db = (yield* Database.Service).db
     const reportLifecycle = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) =>
       effect.pipe(
         Effect.tapCause((cause) =>
@@ -71,12 +76,13 @@ export const layer = Layer.effect(
       sessionID: SessionSchema.ID,
       force: boolean,
       continuation?: SessionRunner.Continuation,
+      promotable: SessionInbox.Promotable = "input",
     ): Effect.Effect<void, SessionRunner.RunError> {
       return Effect.gen(function* () {
         const session = yield* store.get(sessionID)
         if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
         const result = yield* SessionRunner.Service.use((runner) =>
-          runner.drain({ sessionID, force, continuation }),
+          runner.drain({ sessionID, force, continuation, promotable }),
         ).pipe(
           Effect.provide(locations.get(session.location)),
           Effect.tapCause((cause) =>
@@ -86,7 +92,7 @@ export const layer = Layer.effect(
           ),
         )
         if (result.type === "complete") return
-        return yield* drain(sessionID, false, result.continuation)
+        return yield* drain(sessionID, false, result.continuation, promotable)
       })
     }
     const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError, InterruptReason>({
@@ -95,7 +101,7 @@ export const layer = Layer.effect(
           sessionID,
           bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)),
         ),
-      drain: (sessionID, force) => drain(sessionID, force),
+      drain: (sessionID, force, promotable) => drain(sessionID, force, undefined, promotable),
       // One terminal observation per busy period, covering every coalesced drain.
       settled: (sessionID, exit, reason) =>
         reportLifecycle(
@@ -127,16 +133,20 @@ export const layer = Layer.effect(
           }),
         ),
     })
-    yield* bus.subscribe(SessionEvent.Moved).pipe(
-      Stream.runForEach((event) => coordinator.wake(event.data.sessionID)),
-      Effect.forkScoped,
-    )
 
     return Service.of({
       active: coordinator.active,
-      interrupt: (sessionID) => coordinator.interrupt(sessionID, "user"),
+      interrupt: (sessionID, options) =>
+        coordinator.interrupt(
+          sessionID,
+          "user",
+          options?.continue
+            ? { continue: { request: "steer", when: SessionInbox.has(db, sessionID, "steer") } }
+            : undefined,
+        ),
       resume: coordinator.run,
       wake: coordinator.wake,
+      wakeActive: coordinator.wakeActive,
       awaitIdle: coordinator.awaitIdle,
     })
   }),
@@ -145,7 +155,7 @@ export const layer = Layer.effect(
 export const node = makeGlobalNode({
   service: Service,
   layer,
-  deps: [SessionStore.node, LocationServiceMap.node, Bus.node],
+  deps: [SessionStore.node, LocationServiceMap.node, Bus.node, Database.node],
 })
 
 /** Low-level compatibility layer for callers that only need durable Session recording. */
@@ -155,6 +165,7 @@ export const noopLayer = Layer.succeed(
     active: Effect.succeed(new Set()),
     resume: () => Effect.void,
     wake: () => Effect.void,
+    wakeActive: () => Effect.void,
     interrupt: () => Effect.void,
     awaitIdle: () => Effect.void,
   }),

packages/core/src/session/inbox.ts

@@ -349,6 +349,14 @@ export const nextSteer = Effect.fn("SessionInbox.nextSteer")(function* (
   return row ? fromRow(row) : undefined
 })
 
+export const nextPromotable = Effect.fn("SessionInbox.nextPromotable")(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  promotable: Promotable,
+) {
+  return (yield* nextSteer(db, sessionID)) ?? (promotable === "input" ? yield* nextQueued(db, sessionID) : undefined)
+})
+
 /**
  * Which pending rows count: "any" counts every row, while "input" means any
  * item in either delivery mode.

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

@@ -1,6 +1,7 @@
 export * as SessionRunCoordinator from "./run-coordinator.js"
 
 import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
+import type { Promotable } from "./inbox.js"
 
 /** Serializes execution for each key while allowing different keys to run concurrently. */
 export interface Coordinator<Key, E, Reason = never> {
@@ -9,26 +10,41 @@ export interface Coordinator<Key, E, Reason = never> {
   /** Starts an execution while idle, or joins the active execution and returns its exit. */
   readonly run: (key: Key) => Effect.Effect<void, E>
   /** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
-  readonly wake: (key: Key) => Effect.Effect<void>
+  readonly wake: (key: Key, request?: Request) => Effect.Effect<void>
+  /** Rings the current execution's doorbell with its existing request. Idle keys remain idle. */
+  readonly wakeActive: (key: Key) => Effect.Effect<void>
   /** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
-  readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<void>
+  readonly interrupt: (
+    key: Key,
+    reason?: Reason,
+    options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect<boolean> } },
+  ) => Effect.Effect<void>
   /** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
   readonly awaitIdle: (key: Key) => Effect.Effect<void>
 }
 
+export type Request = Promotable
+
 /**
  * One execution is a busy period for one key: one fiber that drains from the first wake
  * until the key would stay idle. `pendingWake` is the doorbell: work recorded during the
- * execution rings it, and the execution loop drains again instead of ending. The doorbell
- * closes the gap between a drain's last eligibility check and the idle transition, since
- * those cannot be one atomic step. `done` resolves joiners with this execution's exit.
+ * execution rings it with its eligibility request, and the execution loop drains again
+ * instead of ending. The doorbell closes the gap between a drain's last eligibility check
+ * and the idle transition, since those cannot be one atomic step. `done` resolves joiners
+ * with this execution's exit.
  */
 type Execution<E, Reason> = {
   readonly done: Deferred.Deferred<void, E>
   owner?: Fiber.Fiber<void>
-  pendingWake: boolean
+  request: Request
+  pendingWake?: Request
   stopping: boolean
   interruptionReason?: Reason
+  continuation?: {
+    readonly request: Request
+    readonly when: Effect.Effect<boolean>
+    signaled: boolean
+  }
 }
 
 /**
@@ -43,7 +59,7 @@ type Execution<E, Reason> = {
  * ```
  */
 export const make = <Key, E, Reason = never>(options: {
-  readonly drain: (key: Key, force: boolean) => Effect.Effect<void, E>
+  readonly drain: (key: Key, force: boolean, request: Request) => Effect.Effect<void, E>
   /** Runs once when a process-local busy period begins, before its first drain. */
   readonly started?: (key: Key) => Effect.Effect<void>
   /**
@@ -55,23 +71,26 @@ export const make = <Key, E, Reason = never>(options: {
   Effect.gen(function* () {
     const executions = new Map<Key, Execution<E, Reason>>()
     const fork = yield* FiberSet.makeRuntime<never, void, never>()
+    const merge = (left: Request, right: Request): Request =>
+      left === "input" || right === "input" ? "input" : "steer"
 
     const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> =>
-      Effect.suspend(() => options.drain(key, force)).pipe(
+      Effect.suspend(() => options.drain(key, force, execution.request)).pipe(
         Effect.flatMap(() =>
           Effect.suspend(() => {
-            if (execution.stopping || !execution.pendingWake) return Effect.void
-            execution.pendingWake = false
+            if (execution.stopping || execution.pendingWake === undefined) return Effect.void
+            execution.request = execution.pendingWake
+            execution.pendingWake = undefined
             // Trampoline so drains that complete synchronously cannot grow the stack.
             return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
           }),
         ),
       )
 
-    const start = (key: Key, force: boolean) => {
+    const start = (key: Key, force: boolean, request: Request) => {
       const execution: Execution<E, Reason> = {
         done: Deferred.makeUnsafe<void, E>(),
-        pendingWake: false,
+        request,
         stopping: false,
       }
       executions.set(key, execution)
@@ -87,7 +106,7 @@ export const make = <Key, E, Reason = never>(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) => finish(key, execution, exit)),
           Effect.exit,
           Effect.asVoid,
         ),
@@ -97,12 +116,22 @@ 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.
-    const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
-      if (execution.pendingWake) start(key, false)
+    const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>, continuation: boolean) => {
+      if (continuation && execution.continuation) start(key, false, execution.continuation.request)
+      else if (execution.pendingWake) start(key, false, execution.pendingWake)
       else executions.delete(key)
       Deferred.doneUnsafe(execution.done, exit)
     }
 
+    const finish = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
+      if (!execution.continuation) return Effect.sync(() => settle(key, execution, exit, false))
+      return execution.continuation.when.pipe(
+        Effect.flatMap((ready) =>
+          Effect.sync(() => settle(key, execution, exit, ready || execution.continuation?.signaled === true)),
+        ),
+      )
+    }
+
     const run = (key: Key): Effect.Effect<void, E> =>
       Effect.suspend(() => {
         const execution = executions.get(key)
@@ -111,26 +140,57 @@ export const make = <Key, E, Reason = never>(options: {
           if (execution.stopping) return Deferred.await(execution.done).pipe(Effect.andThen(run(key)))
           return Deferred.await(execution.done)
         }
-        return Deferred.await(start(key, true).done)
+        return Deferred.await(start(key, true, "input").done)
       })
 
-    const wake = (key: Key) =>
+    const wake = (key: Key, request: Request = "input") =>
       Effect.sync(() => {
         const execution = executions.get(key)
         if (execution !== undefined) {
-          execution.pendingWake = true
+          if (execution.stopping) {
+            if (execution.continuation) execution.continuation.signaled = true
+            else execution.continuation = { request, when: Effect.succeed(true), signaled: true }
+            return
+          }
+          execution.pendingWake = execution.pendingWake ? merge(execution.pendingWake, request) : request
           return
         }
-        start(key, false)
+        start(key, false, request)
       })
 
-    const interrupt = (key: Key, reason?: Reason): Effect.Effect<void> =>
+    const wakeActive = (key: Key) =>
       Effect.suspend(() => {
         const execution = executions.get(key)
-        if (execution?.owner === undefined || execution.stopping) return Effect.void
+        return execution ? wake(key, execution.request) : Effect.void
+      })
+
+    const interrupt = (
+      key: Key,
+      reason?: Reason,
+      options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect<boolean> } },
+    ): Effect.Effect<void> =>
+      Effect.suspend(() => {
+        const execution = executions.get(key)
+        if (execution === undefined) return Effect.void
+        if (execution.stopping) {
+          if (options?.continue)
+            execution.continuation = {
+              ...options.continue,
+              signaled: execution.continuation?.signaled ?? false,
+            }
+          return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid)
+        }
+        if (execution.owner === undefined) {
+          if (!options?.continue) return Effect.void
+          execution.stopping = true
+          execution.pendingWake = undefined
+          execution.continuation = { ...options.continue, signaled: false }
+          return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid)
+        }
         execution.stopping = true
-        execution.pendingWake = false
+        execution.pendingWake = undefined
         execution.interruptionReason = reason
+        if (options?.continue) execution.continuation = { ...options.continue, signaled: false }
         return Fiber.interrupt(execution.owner)
       })
 
@@ -143,5 +203,5 @@ export const make = <Key, E, Reason = never>(options: {
         return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key)))
       })
 
-    return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle }
+    return { active: Effect.sync(() => new Set(executions.keys())), run, wake, wakeActive, interrupt, awaitIdle }
   })

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

@@ -3,6 +3,7 @@ export * as SessionRunner from "./index.js"
 import type { AIError } from "@opencode-ai/ai"
 import { Context, Effect } from "effect"
 import { SessionSchema } from "../schema.js"
+import type { Promotable } from "../inbox.js"
 import type { AgentNotFoundError, MessageDecodeError, StepFailedError, UserInterruptedError } from "../error.js"
 import { SessionRunnerModel } from "./model.js"
 import type { Instructions } from "../../instructions/index.js"
@@ -29,6 +30,8 @@ export interface Interface {
     readonly sessionID: SessionSchema.ID
     readonly force: boolean
     readonly continuation?: Continuation
+    /** "steer" settles the active intent without promoting queued next-turn work. */
+    readonly promotable?: Promotable
   }) => Effect.Effect<DrainResult, RunError>
 }
 

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

@@ -128,22 +128,25 @@ const layer = Layer.effect(
       readonly sessionID: SessionSchema.ID
       readonly force: boolean
       readonly continuation?: Continuation
+      readonly promotable?: SessionInbox.Promotable
     }) {
       let force = input.force
       let continuation = input.continuation
-      if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "any")))
+      const promotable = input.promotable ?? "input"
+      if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable)))
         return { type: "complete" as const }
       yield* settleStaleToolCalls(input.sessionID)
       while (true) {
-        if (yield* runPendingCompaction(input.sessionID)) {
+        if (yield* runPendingCompaction(input.sessionID, promotable)) {
           force = false
           continue
         }
-        if (yield* runPendingMove(input.sessionID, "input")) return { type: "moved" as const }
-        if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "input")))
+        if (yield* runPendingMove(input.sessionID, promotable)) return { type: "moved" as const }
+        if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable)))
           return { type: "complete" as const }
-        const result = yield* runSteps(input.sessionID, continuation)
+        const result = yield* runSteps(input.sessionID, continuation, promotable)
         if (result.type === "moved") return result
+        if (promotable === "steer") return { type: "complete" as const }
         force = false
         continuation = undefined
       }
@@ -156,13 +159,14 @@ const layer = Layer.effect(
     const runSteps = Effect.fn("SessionRunner.runSteps")(function* (
       sessionID: SessionSchema.ID,
       continuation?: Continuation,
+      initialPromotable: SessionInbox.Promotable = continuation ? "steer" : "input",
     ) {
       // Fresh work may promote queued input; later steps absorb steers only.
-      let promotable: SessionInbox.Promotable = continuation ? "steer" : "input"
+      let promotable = initialPromotable
       let step = continuation?.step ?? 1
       let next = continuation
       while (true) {
-        if (yield* runPendingCompaction(sessionID)) continue
+        if (yield* runPendingCompaction(sessionID, "steer")) continue
         if (yield* runPendingMove(sessionID, "steer")) return { type: "moved" as const, continuation: next }
         const result = yield* runStep(sessionID, promotable, step)
         next = result.needsContinuation ? { step: result.step + 1 } : undefined
@@ -515,14 +519,14 @@ const layer = Layer.effect(
     /** Executes a previously admitted manual compaction request, if one is pending. */
     const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* (
       sessionID: SessionSchema.ID,
+      promotable: SessionInbox.Promotable,
     ) {
       return yield* Effect.uninterruptibleMask((restore) =>
         Effect.gen(function* () {
           const pending = yield* SessionInbox.serialized(
             sessionID,
             Effect.gen(function* () {
-              const selected =
-                (yield* SessionInbox.nextSteer(db, sessionID)) ?? (yield* SessionInbox.nextQueued(db, sessionID))
+              const selected = yield* SessionInbox.nextPromotable(db, sessionID, promotable)
               if (selected?.type !== "compaction") return
               yield* bus.publishAll([
                 [SessionEvent.InboxDelivered, { sessionID, inboxID: selected.id }],
@@ -564,9 +568,7 @@ const layer = Layer.effect(
       return yield* SessionInbox.serialized(
         sessionID,
         Effect.gen(function* () {
-          const pending =
-            (yield* SessionInbox.nextSteer(db, sessionID)) ??
-            (promotable === "input" ? yield* SessionInbox.nextQueued(db, sessionID) : undefined)
+          const pending = yield* SessionInbox.nextPromotable(db, sessionID, promotable)
           if (pending?.type !== "move") return false
           yield* modelTransport.close(sessionID)
           yield* bus.publishAll([

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

@@ -31,6 +31,7 @@ import { testEffect } from "./lib/effect"
 
 const executionCalls: Session.ID[] = []
 const interruptCalls: Session.ID[] = []
+const interruptContinuations: Array<boolean | undefined> = []
 const wakeCalls: Session.ID[] = []
 const activeSessions = new Set<Session.ID>()
 const execution = Layer.succeed(
@@ -41,14 +42,16 @@ const execution = Layer.succeed(
       Effect.sync(() => {
         executionCalls.push(sessionID)
       }),
-    interrupt: (sessionID) =>
+    interrupt: (sessionID, options) =>
       Effect.sync(() => {
         interruptCalls.push(sessionID)
+        interruptContinuations.push(options?.continue)
       }),
     wake: (sessionID) =>
       Effect.sync(() => {
         wakeCalls.push(sessionID)
       }),
+    wakeActive: () => Effect.void,
     awaitIdle: () => Effect.void,
   }),
 )
@@ -177,31 +180,18 @@ describe("Session.prompt", () => {
     }),
   )
 
-  it.effect("continues after interruption when pending work remains", () =>
-    Effect.gen(function* () {
-      yield* setup
-      const session = yield* Session.Service
-      yield* session.synthetic({ sessionID, text: "Continue after interrupt", resume: false })
-      interruptCalls.length = 0
-      wakeCalls.length = 0
-
-      yield* session.interrupt(sessionID, { continue: true })
-
-      expect(interruptCalls).toEqual([sessionID])
-      expect(wakeCalls).toEqual([sessionID])
-    }),
-  )
-
-  it.effect("does not continue after interruption without pending work", () =>
+  it.effect("forwards interrupt continuation policy", () =>
     Effect.gen(function* () {
       yield* setup
       const session = yield* Session.Service
       interruptCalls.length = 0
+      interruptContinuations.length = 0
       wakeCalls.length = 0
 
       yield* session.interrupt(sessionID, { continue: true })
 
       expect(interruptCalls).toEqual([sessionID])
+      expect(interruptContinuations).toEqual([true])
       expect(wakeCalls).toEqual([])
     }),
   )

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

@@ -269,6 +269,35 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("replaces a settlement-window wake with a steer continuation", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const settling = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        const requests: SessionRunCoordinator.Request[] = []
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (_key, _force, request) => Effect.sync(() => requests.push(request)),
+          settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))),
+        })
+
+        yield* coordinator.wake("session", "input")
+        yield* Deferred.await(settling)
+        yield* coordinator.wake("session", "input")
+        const interrupted = yield* coordinator
+          .interrupt("session", undefined, {
+            continue: { request: "steer", when: Effect.succeed(true) },
+          })
+          .pipe(Effect.forkChild)
+        yield* Effect.yieldNow
+        yield* Deferred.succeed(release, undefined)
+        yield* Fiber.join(interrupted)
+        yield* coordinator.awaitIdle("session")
+
+        expect(requests).toEqual(["input", "steer"])
+      }),
+    ),
+  )
+
   it.effect("interrupts active execution and clears its pending wake", () =>
     Effect.scoped(
       Effect.gen(function* () {
@@ -342,6 +371,193 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("coalesces drain requests with input taking precedence", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        const requests: SessionRunCoordinator.Request[] = []
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (_key, _force, request) =>
+            Effect.gen(function* () {
+              requests.push(request)
+              if (requests.length !== 1) return
+              yield* Deferred.succeed(firstStarted, undefined)
+              yield* Deferred.await(release)
+            }),
+        })
+
+        yield* coordinator.wake("session", "steer")
+        yield* Deferred.await(firstStarted)
+        yield* coordinator.wake("session", "steer")
+        yield* coordinator.wake("session", "input")
+        yield* Deferred.succeed(release, undefined)
+        yield* coordinator.awaitIdle("session")
+
+        expect(requests).toEqual(["steer", "input"])
+      }),
+    ),
+  )
+
+  it.effect("does not carry a completed input request into a steer drain", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        const requests: SessionRunCoordinator.Request[] = []
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (_key, _force, request) =>
+            Effect.gen(function* () {
+              requests.push(request)
+              if (requests.length !== 1) return
+              yield* Deferred.succeed(firstStarted, undefined)
+              yield* Deferred.await(release)
+            }),
+        })
+
+        yield* coordinator.wake("session", "input")
+        yield* Deferred.await(firstStarted)
+        yield* coordinator.wake("session", "steer")
+        yield* Deferred.succeed(release, undefined)
+        yield* coordinator.awaitIdle("session")
+
+        expect(requests).toEqual(["input", "steer"])
+      }),
+    ),
+  )
+
+  it.effect("an active wake inherits scope without starting idle work", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        const requests: SessionRunCoordinator.Request[] = []
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (_key, _force, request) =>
+            Effect.gen(function* () {
+              requests.push(request)
+              if (requests.length !== 1) return
+              yield* Deferred.succeed(firstStarted, undefined)
+              yield* Deferred.await(release)
+            }),
+        })
+
+        yield* coordinator.wakeActive("session")
+        yield* coordinator.wake("session", "steer")
+        yield* Deferred.await(firstStarted)
+        yield* coordinator.wakeActive("session")
+        yield* Deferred.succeed(release, undefined)
+        yield* coordinator.awaitIdle("session")
+
+        expect(requests).toEqual(["steer", "steer"])
+      }),
+    ),
+  )
+
+  it.effect("coalesces overlapping interrupt continuations into one steer 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 requests: SessionRunCoordinator.Request[] = []
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (_key, _force, request) =>
+            Effect.gen(function* () {
+              requests.push(request)
+              if (requests.length !== 1) return
+              yield* Deferred.succeed(firstStarted, undefined)
+              yield* Effect.never.pipe(
+                Effect.onInterrupt(() =>
+                  Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+                ),
+              )
+            }),
+        })
+        const continuation = { continue: { request: "steer" as const, when: Effect.succeed(false) } }
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(firstStarted)
+        const first = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild)
+        yield* Deferred.await(cleanupStarted)
+        const second = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild)
+        yield* Effect.yieldNow
+        yield* coordinator.wake("session", "input")
+        yield* Deferred.succeed(cleanupGate, undefined)
+        yield* Effect.all([Fiber.join(first), Fiber.join(second)])
+        yield* coordinator.awaitIdle("session")
+
+        expect(requests).toEqual(["input", "steer"])
+      }),
+    ),
+  )
+
+  it.effect("a continuing interrupt replaces a cleanup-era input wake", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const cleanupStarted = yield* Deferred.make<void>()
+        const cleanupGate = yield* Deferred.make<void>()
+        const requests: SessionRunCoordinator.Request[] = []
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (_key, _force, request) =>
+            Effect.gen(function* () {
+              requests.push(request)
+              if (requests.length !== 1) return
+              yield* Deferred.succeed(firstStarted, undefined)
+              yield* Effect.never.pipe(
+                Effect.onInterrupt(() =>
+                  Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+                ),
+              )
+            }),
+        })
+
+        yield* coordinator.wake("session", "input")
+        yield* Deferred.await(firstStarted)
+        const plain = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        yield* Deferred.await(cleanupStarted)
+        yield* coordinator.wake("session", "input")
+        const continuing = yield* coordinator
+          .interrupt("session", undefined, {
+            continue: { request: "steer", when: Effect.succeed(false) },
+          })
+          .pipe(Effect.forkChild)
+        yield* Effect.yieldNow
+        yield* Deferred.succeed(cleanupGate, undefined)
+        yield* Effect.all([Fiber.join(plain), Fiber.join(continuing)])
+        yield* coordinator.awaitIdle("session")
+
+        expect(requests).toEqual(["input", "steer"])
+      }),
+    ),
+  )
+
+  it.effect("does not start a conditional continuation without eligible work", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const requests: SessionRunCoordinator.Request[] = []
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (_key, _force, request) =>
+            Effect.sync(() => requests.push(request)).pipe(
+              Effect.andThen(Deferred.succeed(started, undefined)),
+              Effect.andThen(Effect.never),
+            ),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(started)
+        yield* coordinator.interrupt("session", undefined, {
+          continue: { request: "steer", when: Effect.succeed(false) },
+        })
+        yield* coordinator.awaitIdle("session")
+
+        expect(requests).toEqual(["input"])
+      }),
+    ),
+  )
+
   it.effect("starts a resume registered during interruption cleanup", () =>
     Effect.scoped(
       Effect.gen(function* () {

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

@@ -126,7 +126,8 @@ const execution = (llmClient: Layer.Layer<typeof LLMClient.Service>) =>
         active: coordinator.active,
         resume: coordinator.run,
         wake: coordinator.wake,
-        interrupt: coordinator.interrupt,
+        wakeActive: coordinator.wakeActive,
+        interrupt: (sessionID) => coordinator.interrupt(sessionID),
         awaitIdle: coordinator.awaitIdle,
       })
     }),

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

@@ -413,7 +413,8 @@ const execution = Layer.effect(
       active: coordinator.active,
       resume: coordinator.run,
       wake: coordinator.wake,
-      interrupt: coordinator.interrupt,
+      wakeActive: coordinator.wakeActive,
+      interrupt: (sessionID) => coordinator.interrupt(sessionID),
       awaitIdle: coordinator.awaitIdle,
     })
   }),
@@ -3088,6 +3089,24 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("stops a steer-scoped drain before queued input", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      const { db } = yield* Database.Service
+      yield* session.prompt({ sessionID, text: "Queue for later", delivery: "queue", resume: false })
+      yield* session.prompt({ sessionID, text: "Steer now", resume: false })
+      yield* TestLLM.push(TestLLM.stop())
+
+      const runner = yield* SessionRunner.Service
+      yield* runner.drain({ sessionID, force: false, promotable: "steer" })
+
+      expect(requests).toHaveLength(1)
+      expect(userTexts(requests[0])).toEqual(["Steer now"])
+      expect(yield* SessionInbox.has(db, sessionID, "steer")).toBe(false)
+      expect(yield* SessionInbox.has(db, sessionID, "queue")).toBe(true)
+    }),
+  )
+
   it.effect("promotes queued input after steering continuation ends", () =>
     Effect.gen(function* () {
       const session = yield* setup

packages/core/test/tool-shell.test.ts

@@ -115,6 +115,7 @@ const executionNode = makeGlobalNode({
         active: Effect.succeed(new Set()),
         resume: complete,
         wake: () => Effect.void,
+        wakeActive: () => Effect.void,
         interrupt: () => Effect.void,
         awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid),
       })

packages/core/test/tool-subagent.test.ts

@@ -86,6 +86,7 @@ const executionNode = makeGlobalNode({
         active: Effect.succeed(new Set()),
         resume: complete,
         wake: () => Effect.void,
+        wakeActive: () => Effect.void,
         interrupt: () => Effect.void,
         awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid),
       })

packages/protocol/src/groups/session.ts

@@ -660,7 +660,7 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
             identifier: "v2.session.interrupt",
             summary: "Interrupt session execution",
             description:
-              "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes if durable inbox work remains after interruption.",
+              "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes pending steering input while queued work remains parked.",
           }),
         ),
     )