chore(core): refresh queued control base

Merge v2 at 3e9b009642 while preserving queued-control dispatch behavior and regression tests. Update the execution test fixture to the opaque LayerNode replacement API.

packages/core/src/session/execution.ts

@@ -159,8 +159,7 @@ export const layer = Layer.effect(
           // promotable here or was fully delivered and needs no resumption.
           const next = yield* SessionInbox.nextPromotable(db, sessionID, "input")
           if (next === undefined) return interrupted
-          if (next.delivery === "steer" || next.type === "compaction" || next.type === "move")
-            yield* coordinator.wake(sessionID, "steer")
+          if (next.delivery === "steer" || SessionInbox.isControl(next)) yield* coordinator.wake(sessionID, "steer")
           return interrupted
         }),
       resume: coordinator.run,

packages/core/src/session/inbox.ts

@@ -47,6 +47,9 @@ export {
  */
 export type Promotable = "input" | "steer"
 
+export const isControl = (item: Pick<Item, "type">): item is Pick<Compaction | Move, "type"> =>
+  item.type === "compaction" || item.type === "move"
+
 const decodeUser = Schema.decodeUnknownSync(UserPayload)
 const encodeUser = Schema.encodeSync(UserPayload)
 const decodeSynthetic = Schema.decodeUnknownSync(SyntheticPayload)
@@ -505,7 +508,7 @@ export const promote = Effect.fn("SessionInbox.promote")(function* (
     Effect.gen(function* () {
       const steers = yield* pendingSteers(db, sessionID)
       if (steers.length > 0 || scope === "steer") {
-        const control = steers.findIndex((row) => row.type === "compaction" || row.type === "move")
+        const control = steers.findIndex(isControl)
         return yield* publish(db, bus, sessionID, control === -1 ? steers : steers.slice(0, control))
       }
 
@@ -517,10 +520,10 @@ export const promote = Effect.fn("SessionInbox.promote")(function* (
         .limit(1)
         .get()
         .pipe(Effect.orDie)
-      if (!queued) return 0
+      if (!queued || isControl(queued)) return 0
       const promoted = yield* publish(db, bus, sessionID, [queued])
       const arrivedSteers = yield* pendingSteers(db, sessionID)
-      const control = arrivedSteers.findIndex((row) => row.type === "compaction" || row.type === "move")
+      const control = arrivedSteers.findIndex(isControl)
       return (
         promoted +
         (yield* publish(db, bus, sessionID, control === -1 ? arrivedSteers : arrivedSteers.slice(0, control)))

packages/core/src/session/projector.ts

@@ -594,7 +594,7 @@ const layer = Layer.effectDiscard(
           id: event.data.inboxID,
           sessionID: event.data.sessionID,
         })
-        if (input.type === "compaction" || input.type === "move") return
+        if (SessionInbox.isControl(input)) return
         yield* insertMessage(
           db,
           event,

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

@@ -54,8 +54,8 @@ const layer = Layer.effect(
       if (!force && !continuing) {
         const pending = yield* SessionInbox.nextPromotable(db, sessionID, "input")
         if (!pending) return DrainResult.Complete()
-        const control = pending.type === "compaction" || pending.type === "move"
-        if (promotable === "steer" && pending.delivery === "queue" && !control) return DrainResult.Complete()
+        if (promotable === "steer" && pending.delivery === "queue" && !SessionInbox.isControl(pending))
+          return DrainResult.Complete()
       }
       yield* plugins.flush
       yield* settleStaleToolCalls(sessionID)
@@ -128,15 +128,16 @@ const layer = Layer.effect(
               }
               if (!force && !continuing && (!pending || (pending.delivery === "queue" && promotable === "steer")))
                 return DrainResult.Complete()
-              return yield* restore(
+              const ready = yield* restore(
                 Effect.gen(function* () {
                   const selected = yield* prepareContext(sessionID)
-                  const promoted = yield* SessionInbox.promote(
-                    db,
-                    bus,
-                    sessionID,
-                    entering && !continuing ? promotable : "steer",
-                  )
+                  const scope = entering && !continuing ? promotable : "steer"
+                  const promoted = yield* SessionInbox.promote(db, bus, sessionID, scope)
+                  if (promoted === 0) {
+                    // Cancellation during preparation can expose a control instead of input.
+                    const next = yield* SessionInbox.nextPromotable(db, sessionID, scope)
+                    if (next && SessionInbox.isControl(next)) return undefined
+                  }
                   if (promoted > 0 && !selected.session.parentID && SessionTitle.isUntitled(selected.session))
                     yield* FiberMap.run(titles, sessionID, title.generate(sessionID), {
                       onlyIfMissing: true,
@@ -145,6 +146,7 @@ const layer = Layer.effect(
                   return { _tag: "Ready" as const, context: yield* context.load(selected) }
                 }),
               )
+              if (ready) return ready
             }
           }),
         ),

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

@@ -1422,6 +1422,67 @@ describe("SessionRunnerLLM", () => {
     })
   })
 
+  scenario("dispatches a queued move exposed by cancellation during preparation", function* (s) {
+    const runner = yield* SessionRunner.Service
+    const prompt = yield* s.session.prompt({ sessionID, text: "Cancel me", delivery: "queue", resume: false })
+    const location = Location.Ref.make({ directory: AbsolutePath.make("/moved") })
+    yield* s.sessionInbox.admit({
+      id: SessionMessage.ID.create(),
+      sessionID,
+      item: {
+        type: "move",
+        payload: { location, projectID: Project.ID.global },
+        delivery: "queue",
+      },
+    })
+    s.systemLoadHook = Effect.gen(function* () {
+      s.systemLoadHook = Effect.void
+      yield* s.session.cancelInbox({ sessionID, inboxID: prompt.id }).pipe(Effect.orDie)
+    })
+
+    expect(yield* runner.drain({ sessionID, force: false })).toEqual(SessionRunner.DrainResult.Moved({}))
+    expect((yield* s.session.get(sessionID)).location).toEqual(location)
+    expect(s.closedTransports).toEqual([sessionID])
+    expect(s.requests).toHaveLength(0)
+    expect(yield* s.messages).toMatchObject([{ type: "location-switched", location }])
+    expect(yield* s.inbox).toEqual([])
+    expect((yield* recordedEventTypes(sessionID)).slice(-2)).toEqual([
+      Bus.versionedType(SessionEvent.InboxDelivered.type, 1),
+      Bus.versionedType(SessionEvent.Moved.type, 1),
+    ])
+  })
+
+  scenario("dispatches a queued compaction exposed by cancellation during preparation", function* (s) {
+    const runner = yield* SessionRunner.Service
+    const prompt = yield* s.session.prompt({ sessionID, text: "Cancel me", delivery: "queue", resume: false })
+    const compaction = yield* s.sessionInbox.admitCompaction({
+      id: SessionMessage.ID.create(),
+      sessionID,
+      delivery: "queue",
+    })
+    s.systemLoadHook = Effect.gen(function* () {
+      s.systemLoadHook = Effect.void
+      yield* s.session.cancelInbox({ sessionID, inboxID: prompt.id }).pipe(Effect.orDie)
+    })
+
+    expect(yield* runner.drain({ sessionID, force: false })).toEqual(SessionRunner.DrainResult.Complete())
+    expect(s.requests).toHaveLength(0)
+    expect(yield* s.inbox).toEqual([])
+    expect(yield* s.messages).toMatchObject([
+      {
+        id: compaction.id,
+        type: "compaction",
+        status: "failed",
+        error: { type: "compaction.unavailable", message: "Nothing to compact yet" },
+      },
+    ])
+    expect((yield* recordedEventTypes(sessionID)).slice(-3)).toEqual([
+      Bus.versionedType(SessionEvent.InboxDelivered.type, 1),
+      Bus.versionedType(SessionEvent.Compaction.Started.type, 1),
+      Bus.versionedType(SessionEvent.Compaction.Failed.type, 1),
+    ])
+  })
+
   scenario("delivers a queued move atomically at the idle boundary", function* (s) {
     const inboxID = SessionMessage.ID.create()
     yield* s.sessionInbox.admit({