fix(core): keep interrupted sessions stopped

packages/core/src/session.ts

@@ -379,7 +379,7 @@ const layer = Layer.effect(
             )
             if (!SessionInput.equivalent(admitted, expected))
               return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
-            if (input.resume !== false) yield* execution.wake(admitted.sessionID)
+            if (input.resume !== false) yield* execution.wake(admitted.sessionID, admitted.admittedSeq)
             return admitted
           }),
         ),
@@ -428,7 +428,12 @@ const layer = Layer.effect(
         yield* execution.resume(sessionID)
       }),
       interrupt: Effect.fn("V2Session.interrupt")((sessionID) =>
-        Effect.uninterruptible(execution.interrupt(sessionID)),
+        Effect.uninterruptible(
+          Effect.gen(function* () {
+            if (!(yield* store.get(sessionID))) return yield* execution.interrupt(sessionID)
+            yield* execution.interrupt(sessionID, yield* EventV2.latestSequence(db, sessionID))
+          }),
+        ),
       ),
       revert: {
         stage: Effect.fn("V2Session.revert.stage")(function* (input) {

packages/core/src/session/execution.ts

@@ -12,9 +12,9 @@ export interface Interface {
   /** Starts execution while idle or joins the active execution. */
   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>
+  readonly wake: (sessionID: SessionSchema.ID, seq?: number) => 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, seq?: number) => Effect.Effect<void>
 }
 
 /** Routes execution from a Session ID to the runner owned by that Session's Location. */

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

@@ -9,28 +9,39 @@ export interface Coordinator<Key, E> {
   /** Starts execution while idle or joins the active execution. */
   readonly run: (key: Key) => Effect.Effect<void, E>
   /** Registers one coalesced follow-up after newly recorded work. */
-  readonly wake: (key: Key) => Effect.Effect<void>
+  readonly wake: (key: Key, seq?: number) => Effect.Effect<void>
   /** Stops active execution and waits for its cleanup. */
-  readonly interrupt: (key: Key) => Effect.Effect<void>
+  readonly interrupt: (key: Key, seq?: number) => Effect.Effect<void>
 }
 
+type Wake = { readonly seq?: number }
+
 type Entry<E> = {
   readonly done: Deferred.Deferred<void, E>
+  currentWakeSeq?: number
   owner?: Fiber.Fiber<void, never>
-  pendingWake: boolean
+  pendingWake?: Wake
+  interruptSeq?: number
   stopping: boolean
 }
 
+const coalesceWake = (left: Wake | undefined, seq: number | undefined): Wake => {
+  if (left === undefined) return { seq }
+  if (left.seq === undefined || seq === undefined) return {}
+  return { seq: Math.max(left.seq, seq) }
+}
+
 export const make = <Key, E>(options: {
   readonly drain: (key: Key, force: boolean) => Effect.Effect<void, E>
 }): Effect.Effect<Coordinator<Key, E>, never, Scope.Scope> =>
   Effect.gen(function* () {
     const active = new Map<Key, Entry<E>>()
+    const interruptSeq = new Map<Key, number>()
     const fork = yield* FiberSet.makeRuntime<never, void, never>()
 
-    const makeEntry = (): Entry<E> => ({
+    const makeEntry = (currentWakeSeq?: number): Entry<E> => ({
       done: Deferred.makeUnsafe<void, E>(),
-      pendingWake: false,
+      currentWakeSeq,
       stopping: false,
     })
 
@@ -49,13 +60,15 @@ export const make = <Key, E>(options: {
     }
 
     const settle = (key: Key, entry: Entry<E>, exit: Exit.Exit<void, E>) => {
-      if (Exit.isSuccess(exit) && !entry.stopping && entry.pendingWake) {
+      if (Exit.isSuccess(exit) && !entry.stopping && entry.pendingWake !== undefined) {
-        entry.pendingWake = false
+        const pending = entry.pendingWake
+        entry.pendingWake = undefined
+        entry.currentWakeSeq = pending.seq
         start(key, entry, false, true)
         return
       }
 
-      const successor = entry.pendingWake ? makeEntry() : undefined
+      const successor = entry.pendingWake === undefined ? undefined : makeEntry(entry.pendingWake.seq)
       if (successor === undefined) active.delete(key)
       else {
         active.set(key, successor)
@@ -78,25 +91,33 @@ export const make = <Key, E>(options: {
         return restore(Deferred.await(next.done))
       })
 
-    const wake = (key: Key) =>
+    const wake = (key: Key, seq?: number) =>
       Effect.sync(() => {
+        const latest = interruptSeq.get(key)
+        if (latest !== undefined && (seq === undefined || seq <= latest)) return
         const entry = active.get(key)
         if (entry !== undefined) {
-          entry.pendingWake = true
+          if (entry.stopping && entry.interruptSeq !== undefined && (seq === undefined || seq <= entry.interruptSeq))
+            return
+          entry.pendingWake = coalesceWake(entry.pendingWake, seq)
           return
         }
 
-        const next = makeEntry()
+        const next = makeEntry(seq)
         active.set(key, next)
         start(key, next, false)
       })
 
-    const interrupt = (key: Key): Effect.Effect<void> =>
+    const interrupt = (key: Key, seq?: number): Effect.Effect<void> =>
       Effect.suspend(() => {
+        if (seq !== undefined) interruptSeq.set(key, Math.max(interruptSeq.get(key) ?? seq, seq))
         const entry = active.get(key)
         if (entry?.owner === undefined) return Effect.void
+        if (seq !== undefined && entry.currentWakeSeq !== undefined && entry.currentWakeSeq > seq) return Effect.void
         entry.stopping = true
-        entry.pendingWake = false
+        entry.interruptSeq = seq
+        if (seq === undefined || entry.pendingWake?.seq === undefined || entry.pendingWake.seq <= seq)
+          entry.pendingWake = undefined
         return Fiber.interrupt(entry.owner)
       })
 

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

@@ -22,7 +22,9 @@ import { testEffect } from "./lib/effect"
 
 const executionCalls: SessionV2.ID[] = []
 const interruptCalls: SessionV2.ID[] = []
+const interruptSeqs: Array<number | undefined> = []
 const wakeCalls: SessionV2.ID[] = []
+const wakeSeqs: Array<number | undefined> = []
 const activeSessions = new Set<SessionV2.ID>()
 const execution = Layer.succeed(
   SessionExecution.Service,
@@ -32,13 +34,15 @@ const execution = Layer.succeed(
       Effect.sync(() => {
         executionCalls.push(sessionID)
       }),
-    interrupt: (sessionID) =>
+    interrupt: (sessionID, seq) =>
       Effect.sync(() => {
         interruptCalls.push(sessionID)
+        interruptSeqs.push(seq)
       }),
-    wake: (sessionID) =>
+    wake: (sessionID, seq) =>
       Effect.sync(() => {
         wakeCalls.push(sessionID)
+        wakeSeqs.push(seq)
       }),
   }),
 )
@@ -123,9 +127,11 @@ describe("SessionV2.prompt", () => {
       yield* setup
       const session = yield* SessionV2.Service
       interruptCalls.length = 0
+      interruptSeqs.length = 0
 
       yield* session.interrupt(sessionID)
       expect(interruptCalls).toEqual([sessionID])
+      expect(interruptSeqs).toEqual([-1])
       expect(yield* session.messages({ sessionID })).toEqual([])
     }),
   )
@@ -134,9 +140,11 @@ describe("SessionV2.prompt", () => {
     Effect.gen(function* () {
       const session = yield* SessionV2.Service
       interruptCalls.length = 0
+      interruptSeqs.length = 0
 
       yield* session.interrupt(SessionV2.ID.make("ses_missing"))
       expect(interruptCalls).toEqual([SessionV2.ID.make("ses_missing")])
+      expect(interruptSeqs).toEqual([undefined])
     }),
   )
 
@@ -542,11 +550,13 @@ describe("SessionV2.prompt", () => {
       const session = yield* SessionV2.Service
       executionCalls.length = 0
       wakeCalls.length = 0
+      wakeSeqs.length = 0
 
-      yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Run by default" }) })
+      const message = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Run by default" }) })
 
       expect(executionCalls).toEqual([])
       expect(wakeCalls).toEqual([sessionID])
+      expect(wakeSeqs).toEqual([message.admittedSeq])
     }),
   )
 

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

@@ -244,6 +244,39 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("suppresses a stale wake registered during interruption cleanup", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const cleanupStarted = yield* Deferred.make<void>()
+        const cleanupGate = yield* Deferred.make<void>()
+        let runs = 0
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: () =>
+            Effect.sync(() => ++runs).pipe(
+              Effect.andThen(Deferred.succeed(firstStarted, undefined)),
+              Effect.andThen(Effect.never),
+              Effect.onInterrupt(() =>
+                Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
+              ),
+            ),
+        })
+
+        yield* coordinator.wake("session", 1)
+        yield* Deferred.await(firstStarted)
+        const interrupt = yield* coordinator.interrupt("session", 2).pipe(Effect.forkChild)
+        yield* Deferred.await(cleanupStarted)
+        yield* coordinator.wake("session", 1)
+        yield* Deferred.succeed(cleanupGate, undefined)
+        yield* Fiber.join(interrupt)
+        yield* Effect.yieldNow
+
+        expect(runs).toBe(1)
+        expect(Array.from(yield* coordinator.active)).toEqual([])
+      }),
+    ),
+  )
+
   it.effect("runs a wake registered during interruption cleanup", () =>
     Effect.scoped(
       Effect.gen(function* () {
@@ -268,11 +301,11 @@ describe("SessionRunCoordinator", () => {
             ),
         })
 
-        yield* coordinator.wake("session")
+        yield* coordinator.wake("session", 1)
         yield* Deferred.await(firstStarted)
-        const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
+        const interrupt = yield* coordinator.interrupt("session", 2).pipe(Effect.forkChild)
         yield* Deferred.await(cleanupStarted)
-        yield* coordinator.wake("session")
+        yield* coordinator.wake("session", 3)
         yield* Deferred.succeed(cleanupGate, undefined)
         yield* Fiber.join(interrupt)
         yield* Deferred.await(secondStarted)

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

@@ -1898,6 +1898,24 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("does not wake an admitted prompt older than an interrupt", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      const id = SessionMessage.ID.create()
+      const prompt = Prompt.make({ text: "Remain stopped" })
+      requests.length = 0
+      streamGate = yield* Deferred.make<void>()
+
+      yield* session.prompt({ id, sessionID, prompt, resume: false })
+      yield* session.interrupt(sessionID)
+      yield* session.prompt({ id, sessionID, prompt })
+
+      expect(Array.from(yield* session.active)).not.toContain(sessionID)
+      expect(requests).toHaveLength(0)
+    }),
+  )
+
   it.effect("preserves durable queued input for a later wake after interruption", () =>
     Effect.gen(function* () {
       yield* setup