feat(core): acknowledge session interruption immediately (#43552)

packages/core/src/session/execution.ts

@@ -21,7 +21,11 @@ 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>
-  /** Interrupt active work owned by this process. Idle interruption is a no-op. */
+  /**
+   * Interrupt active work owned by this process. Idle interruption is a no-op. Resolves once
+   * the interruption is accepted; cleanup settles asynchronously in the execution fiber.
+   * Compose with `awaitIdle` when settlement matters.
+   */
   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>
@@ -141,6 +145,11 @@ export const layer = Layer.effect(
           // Resume steering input and between-turn control work from the interrupted
           // intent. Queued next-turn prompts stay parked: a steer-scoped drain never
           // promotes them, and a control item behind a queued prompt waits its turn.
+          // Interruption acknowledges before cleanup settles, so this wake usually lands
+          // on the stopping execution's doorbell and starts the successor at settle.
+          // Reading the inbox concurrently with the dying drain is safe: delivery consumes
+          // rows inside uninterruptible publications, so a steer row is either still
+          // promotable here or was fully delivered and needs no resumption.
           const next = yield* SessionInbox.nextPromotable(db, sessionID, "input")
           if (next === undefined) return
           if (next.delivery === "steer" || next.type === "compaction" || next.type === "move")

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

@@ -11,7 +11,11 @@ export interface Coordinator<Key, E, Reason = never> {
   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, scope?: Promotable) => Effect.Effect<void>
-  /** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
+  /**
+   * Stops the active execution and clears its doorbell. No-op when idle. Resolves once the
+   * interruption is accepted, not when cleanup settles: the execution fiber finishes its
+   * finalizers and settled hook on its own time. Compose with `awaitIdle` for settlement.
+   */
   readonly interrupt: (key: Key, reason?: Reason) => 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>
@@ -26,7 +30,13 @@ export interface Coordinator<Key, E, Reason = never> {
  * with this execution's exit.
  */
 type Execution<E, Reason> = {
-  readonly done: Deferred.Deferred<void, E>
+  /**
+   * Resolves with the execution's exit as a success value. Success-valued on purpose:
+   * completing a Deferred with an interrupted exit interrupts suspended waiters as it
+   * resumes them, and can starve later waiters of their resume entirely
+   * (Effect-TS/effect#7364). Joiners flatten the exit; idleness waiters just await.
+   */
+  readonly done: Deferred.Deferred<Exit.Exit<void, E>>
   owner?: Fiber.Fiber<void>
   scope: Promotable
   pendingWake?: Promotable
@@ -74,7 +84,7 @@ export const make = <Key, E, Reason = never>(options: {
 
     const start = (key: Key, force: boolean, scope: Promotable) => {
       const execution: Execution<E, Reason> = {
-        done: Deferred.makeUnsafe<void, E>(),
+        done: Deferred.makeUnsafe<Exit.Exit<void, E>>(),
         scope,
         stopping: false,
       }
@@ -104,7 +114,7 @@ export const make = <Key, E, Reason = never>(options: {
     const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
       if (execution.pendingWake) start(key, false, execution.pendingWake)
       else executions.delete(key)
-      Deferred.doneUnsafe(execution.done, exit)
+      Deferred.doneUnsafe(execution.done, Exit.succeed(exit))
     }
 
     const run = (key: Key): Effect.Effect<void, E> =>
@@ -113,9 +123,9 @@ export const make = <Key, E, Reason = never>(options: {
         if (execution !== undefined) {
           // A stopping execution refuses joiners: wait out its cleanup, then run fresh.
           if (execution.stopping) return Deferred.await(execution.done).pipe(Effect.andThen(run(key)))
-          return Deferred.await(execution.done)
+          return Deferred.await(execution.done).pipe(Effect.flatten)
         }
-        return Deferred.await(start(key, true, "input").done)
+        return Deferred.await(start(key, true, "input").done).pipe(Effect.flatten)
       })
 
     const wake = (key: Key, scope: Promotable = "input") =>
@@ -132,13 +142,23 @@ export const make = <Key, E, Reason = never>(options: {
     const interrupt = (key: Key, reason?: Reason): Effect.Effect<void> =>
       Effect.suspend(() => {
         const execution = executions.get(key)
-        if (execution?.owner === undefined || execution.stopping) return Effect.void
+        if (execution === undefined || execution.stopping) return Effect.void
+        if (execution.owner === undefined) {
+          // Settlement window: the owner exited but the settled hook has not finished. The
+          // terminal outcome is already decided, so no reason attaches — but the interrupt
+          // still claims the recorded wakes so settle does not start a dead-intent successor.
+          execution.pendingWake = undefined
+          return Effect.void
+        }
         execution.stopping = true
         // Wakes recorded so far belong to the interrupted intent; the interrupt claims them.
         // Wakes arriving during cleanup are new admissions and restart normally at settle.
         execution.pendingWake = undefined
         execution.interruptionReason = reason
-        return Fiber.interrupt(execution.owner)
+        // Fire and forget: nobody benefits from waiting out cleanup here, and callers like
+        // the interrupt endpoint must acknowledge immediately even when finalizers are slow.
+        fork(Fiber.interrupt(execution.owner))
+        return Effect.void
       })
 
     // One execution's `done` already spans coalesced continuations; re-check after it
@@ -147,7 +167,7 @@ export const make = <Key, E, Reason = never>(options: {
       Effect.suspend(() => {
         const execution = executions.get(key)
         if (execution === undefined) return Effect.void
-        return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key)))
+        return Deferred.await(execution.done).pipe(Effect.andThen(awaitIdle(key)))
       })
 
     return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle }

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

@@ -622,6 +622,7 @@ describe("SessionRunCoordinator", () => {
         yield* coordinator.wake("session")
         yield* Deferred.await(started)
         yield* coordinator.interrupt("session")
+        yield* coordinator.awaitIdle("session")
 
         expect(settled).toHaveLength(1)
         expect(settled[0] !== undefined && Exit.isFailure(settled[0]) && Cause.hasInterrupts(settled[0].cause)).toBe(
@@ -632,6 +633,69 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("acknowledges interruption before cleanup settles", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const cleanupStarted = yield* Deferred.make<void>()
+        const cleanupGate = yield* Deferred.make<void>()
+        const settled: 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(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+              ),
+            ),
+          settled: (_key, _exit, reason) => Effect.sync(() => void settled.push(reason)),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(started)
+        // Interrupt resolves while the cleanup gate is still closed: acceptance, not settlement.
+        yield* coordinator.interrupt("session", "user")
+        expect(settled).toHaveLength(0)
+        expect(Array.from(yield* coordinator.active)).toEqual(["session"])
+        // Repeating the interrupt during cleanup stays an immediate no-op.
+        yield* coordinator.interrupt("session", "user")
+
+        yield* Deferred.await(cleanupStarted)
+        yield* Deferred.succeed(cleanupGate, undefined)
+        yield* coordinator.awaitIdle("session")
+
+        expect(settled).toEqual(["user"])
+        expect(yield* coordinator.active).toEqual(new Set())
+      }),
+    ),
+  )
+
+  it.effect("an interrupt during terminal settlement claims the recorded wake", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const settling = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        let drains = 0
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () => Effect.sync(() => void drains++),
+          settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))),
+        })
+
+        yield* coordinator.wake("session")
+        yield* Deferred.await(settling)
+        // The owner has exited; this wake lands on the settling execution's doorbell.
+        yield* coordinator.wake("session")
+        // The interrupt claims it: settle must not start a successor for the dead intent.
+        yield* coordinator.interrupt("session")
+        yield* Deferred.succeed(release, undefined)
+        yield* coordinator.awaitIdle("session")
+
+        expect(drains).toBe(1)
+        expect(yield* coordinator.active).toEqual(new Set())
+      }),
+    ),
+  )
+
   it.effect("trampolines synchronous self-waking execution", () =>
     Effect.scoped(
       Effect.gen(function* () {

packages/tui/src/mini/runtime.ts

@@ -115,7 +115,6 @@ type ResolvedSession = {
 type RuntimeState = {
   sdk: RunInput["sdk"]
   shown: boolean
-  aborting: boolean
   model: RunInput["model"]
   defaultModel: RunInput["model"]
   providers: RunProvider[]
@@ -213,7 +212,6 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep
   const state: RuntimeState = {
     sdk: ctx.sdk,
     shown: !session.first,
-    aborting: false,
     model: ctx.model ?? session.model,
     defaultModel: undefined,
     providers: [],
@@ -368,20 +366,17 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep
       }
     },
     onInterrupt: () => {
-      if (!state.sessionID || state.aborting) {
+      if (!state.sessionID) {
         return false
       }
 
-      state.aborting = true
+      // No in-flight guard: interruption acknowledges immediately server-side and repeating
+      // it is an idempotent no-op, so repeated presses are never swallowed.
       void (
         state.stream
           ? state.stream.then((item) => item.handle.interruptActiveTurn())
           : state.sdk.session.interrupt({ sessionID: state.sessionID, continue: true })
-      )
-        .catch(() => {})
-        .finally(() => {
-          state.aborting = false
-        })
+      ).catch(() => {})
       return true
     },
     onBackground: () => {

packages/tui/src/mini/stream-v2.transport.ts

@@ -143,6 +143,8 @@ type State = {
   closed: boolean
   initial: boolean
   rootActive: boolean
+  /** Bumped on every root execution lifecycle event; guards paints against stale acks. */
+  executionEpoch: number
   buffered?: ReplayBuffer
   errors: Set<string>
   pending: Map<string, FooterQueuedPrompt>
@@ -474,6 +476,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
     closed: false,
     initial: true,
     rootActive: false,
+    executionEpoch: 0,
     errors: new Set(),
     pending: new Map(),
     admitted: new Set(),
@@ -512,6 +515,13 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
   })
   controller.signal.addEventListener("abort", () => subagents.close(), { once: true })
 
+  // The one "go idle" transition, shared by settlement, terminal events, and the
+  // interrupt ack so the flag and the paint cannot drift apart.
+  const paintIdle = (status: string) => {
+    state.rootActive = false
+    write([], { phase: "idle", status })
+  }
+
   const write = (commits: StreamCommit[], patch?: { phase?: "idle" | "running"; status?: string; usage?: string }) => {
     if (state.closed || controller.signal.aborted || input.footer.isClosed) return
     if (!state.initial && state.buffered === undefined)
@@ -790,8 +800,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
   const settleSession = async (client: OpenCodeClient) => {
     await client.session.wait({ sessionID: input.sessionID }, { signal: controller.signal })
     for (const message of await projectedMessages(client, controller.signal)) renderMessage(message, true, true)
-    state.rootActive = false
-    write([], { phase: "idle", status: blockerStatus(state.view) })
+    paintIdle(blockerStatus(state.view))
     await input.footer.idle()
   }
 
@@ -1297,6 +1306,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
       return
     }
     if (event.type === "session.execution.started") {
+      state.executionEpoch++
       state.rootActive = true
       write([], { phase: "running" })
       return
@@ -1306,8 +1316,8 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
       event.type === "session.execution.failed" ||
       event.type === "session.execution.interrupted"
     ) {
-      state.rootActive = false
-      write([], { phase: "idle", status: "" })
+      state.executionEpoch++
+      paintIdle("")
       const current = state.wait
       if (!current) return
       if (current.interrupted && event.type === "session.execution.interrupted" && event.data.reason === "user") {
@@ -1780,7 +1790,17 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
         return
       }
       if (state.wait) state.wait.interrupted = true
-      await sdk.session.interrupt({ sessionID: input.sessionID, continue: true }).catch(() => {})
+      // Paint idle at the ack, not at settlement: the server accepts interruption immediately
+      // while cleanup finishes asynchronously, and the terminal execution event re-confirms.
+      // A failed request paints nothing, so the two-press gesture stays available for retry,
+      // and a lifecycle event racing the ack wins via the epoch guard.
+      const epoch = state.executionEpoch
+      await sdk.session.interrupt({ sessionID: input.sessionID, continue: true }).then(
+        () => {
+          if (state.executionEpoch === epoch) paintIdle(blockerStatus(state.view))
+        },
+        () => {},
+      )
     },
     selectSubagent(sessionID) {
       subagents.select(sdk, sessionID)