fix(core): keep interrupted sessions stopped

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