refactor(core): advance sessions before running steps (#45358)

Centralize control dispatch and first-Step preparation in advanceToStep. Keep input delivery outside logical-Step retries and preserve queue ordering, Location handoff, context refresh, and durable settlement.

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

@@ -78,83 +78,143 @@ const layer = Layer.effect(
       readonly continuation?: Continuation
       readonly promotable?: SessionInbox.Promotable
     }) {
+      const sessionID = input.sessionID
       let force = input.force
-      let continuation = input.continuation
+      let continuing = input.continuation !== undefined
+      let step = input.continuation?.step ?? 1
+      let entering = true
       const promotable = input.promotable ?? "input"
-      if (!force && !continuation && !(yield* eligible(input.sessionID, promotable))) return DrainResult.Complete()
-      yield* plugins.flush
-      yield* settleStaleToolCalls(input.sessionID)
-      while (true) {
-        // Scope gates input promotion, not a between-step control that is next in line.
-        if (yield* runPendingCompaction(input.sessionID, "input")) {
-          force = false
-          continue
-        }
-        if (yield* runPendingMove(input.sessionID, "input")) return DrainResult.Moved({})
-        if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable)))
+      if (!force && !continuing) {
+        const pending = yield* SessionInbox.nextPromotable(db, sessionID, "input")
+        if (
+          !pending ||
+          (pending.delivery === "queue" &&
+            promotable === "steer" &&
+            pending.type !== "compaction" &&
+            pending.type !== "move")
+        )
           return DrainResult.Complete()
-        const result = yield* runSteps(input.sessionID, continuation, promotable)
-        if (result._tag === "Moved") return result
-        force = false
-        continuation = undefined
       }
-    })
+      yield* plugins.flush
+      yield* settleStaleToolCalls(sessionID)
 
-    const eligible = Effect.fnUntraced(function* (sessionID: SessionSchema.ID, promotable: SessionInbox.Promotable) {
-      if (yield* SessionInbox.has(db, sessionID, promotable)) return true
-      if (promotable === "input") return false
-      const next = yield* SessionInbox.nextPromotable(db, sessionID, "input")
-      return next?.type === "compaction" || next?.type === "move"
-    })
+      const advanceToStep = Effect.fn("SessionRunner.advanceToStep")(() =>
+        Effect.uninterruptibleMask((restore) =>
+          Effect.gen(function* () {
+            while (true) {
+              // Location entry and idle boundaries allow queued controls, not necessarily queued prompts.
+              const pending = yield* SessionInbox.serialized(
+                sessionID,
+                Effect.gen(function* () {
+                  const next = yield* SessionInbox.nextPromotable(
+                    db,
+                    sessionID,
+                    entering || !continuing ? "input" : "steer",
+                  )
+                  if (next?.type === "compaction")
+                    yield* bus.publishAll([
+                      [SessionEvent.InboxDelivered, { sessionID, inboxID: next.id }],
+                      [SessionEvent.Compaction.Started, { sessionID, reason: "manual", recent: "", inputID: next.id }],
+                    ])
+                  if (next?.type === "move")
+                    yield* restore(
+                      Effect.gen(function* () {
+                        yield* modelTransport.close(sessionID)
+                        yield* bus.publishAll([
+                          [SessionEvent.InboxDelivered, { sessionID, inboxID: next.id }],
+                          [SessionEvent.Moved, { sessionID, ...next.payload }],
+                        ])
+                      }),
+                    )
+                  return next
+                }),
+              )
+              if (!continuing && pending?.delivery !== "steer") {
+                entering = true
+                step = 1
+              }
+              if (pending?.type === "move")
+                return DrainResult.Moved({ continuation: !entering && continuing ? { step } : undefined })
+              if (pending?.type === "compaction") {
+                const session = yield* store.get(sessionID)
+                if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
+                const compacted = yield* restore(
+                  Effect.gen(function* () {
+                    return yield* compaction.compactManual({
+                      session,
+                      messages: yield* store.context(sessionID),
+                      inputID: pending.id,
+                      started: true,
+                    })
+                  }),
+                ).pipe(Effect.exit)
+                if (Exit.isFailure(compacted)) {
+                  yield* bus.publish(SessionEvent.Compaction.Failed, {
+                    sessionID,
+                    reason: "manual",
+                    error: Cause.hasInterruptsOnly(compacted.cause)
+                      ? { type: "aborted", message: "Compaction cancelled" }
+                      : { type: "compaction.failed", message: Cause.pretty(compacted.cause) },
+                    inputID: pending.id,
+                  })
+                  return yield* Effect.failCause(compacted.cause)
+                }
+                force = false
+                continue
+              }
+              if (!force && !continuing && (!pending || (pending.delivery === "queue" && promotable === "steer")))
+                return DrainResult.Complete()
+              return yield* restore(
+                Effect.gen(function* () {
+                  const selected = yield* prepareContext(sessionID)
+                  const promoted = yield* SessionInbox.promote(
+                    db,
+                    bus,
+                    sessionID,
+                    entering && !continuing ? promotable : "steer",
+                  )
+                  if (promoted > 0 && !selected.session.parentID && SessionTitle.isUntitled(selected.session))
+                    yield* FiberMap.run(titles, sessionID, title.generate(sessionID).pipe(Effect.ignore), {
+                      onlyIfMissing: true,
+                    })
+                  if (promoted > 0) step = 1
+                  return { _tag: "Ready" as const, context: yield* context.load(selected) }
+                }),
+              )
+            }
+          }),
+        ),
+      )
 
-    /** Queued inputs wait until the current model work reaches idle; later Steps absorb only steers. */
-    const runSteps = Effect.fn("SessionRunner.runSteps")(function* (
-      sessionID: SessionSchema.ID,
-      continuation: Continuation | undefined,
-      drainPromotable: SessionInbox.Promotable,
-    ) {
-      let promotable: SessionInbox.Promotable = continuation ? "steer" : drainPromotable
-      let step = continuation?.step ?? 1
-      let next = continuation
-      let first = true
       while (true) {
-        if (yield* runPendingCompaction(sessionID, "steer")) continue
-        if (yield* runPendingMove(sessionID, "steer")) return DrainResult.Moved({ continuation: next })
-        if (!first && !next && !(yield* SessionInbox.has(db, sessionID, "steer"))) return DrainResult.Complete()
-        const result = yield* runStep(sessionID, promotable, step)
-        first = false
-        promotable = "steer"
-        step = result.step + 1
-        next = result.needsContinuation ? { step } : undefined
+        const next = yield* advanceToStep()
+        if (next._tag !== "Ready") return next
+        continuing = yield* runStep(next.context, step)
+        step++
+        force = false
+        entering = false
       }
     })
 
+    const prepareContext = Effect.fn("SessionRunner.prepareContext")(function* (sessionID: SessionSchema.ID) {
+      const selected = yield* context.select(sessionID)
+      // A blocked initial instruction baseline must leave admitted input pending.
+      yield* InstructionState.prepare(db, bus, selected.instructions, sessionID)
+      return selected
+    })
+
     /** Owns logical Step policy; each attempt owns its streaming, tools, and durable settlement. */
-    const runStep = Effect.fn("SessionRunner.runStep")(function* (
-      sessionID: SessionSchema.ID,
-      promotable: SessionInbox.Promotable,
-      step: number,
-    ) {
+    const runStep = Effect.fn("SessionRunner.runStep")(function* (first: SessionContext.Loaded, step: number) {
+      const sessionID = first.session.id
       let assistantMessageID = SessionMessage.ID.create()
       const retry = yield* Schedule.toStepWithSleep(SessionRunnerRetry.schedule(bus, sessionID))
-      let currentPromotable: SessionInbox.Promotable | undefined = promotable
-      let currentStep = step
+      let initial: SessionContext.Loaded | undefined = first
       let recoverOverflow = true
       let recoverContinuation = true
       while (true) {
-        const selected = yield* context.select(sessionID)
-        // A blocked initial instruction baseline must leave admitted input pending.
-        yield* InstructionState.prepare(db, bus, selected.instructions, selected.session.id)
-        const promoted = currentPromotable
-          ? yield* SessionInbox.promote(db, bus, selected.session.id, currentPromotable)
-          : 0
-        if (promoted > 0 && !selected.session.parentID && SessionTitle.isUntitled(selected.session))
-          yield* FiberMap.run(titles, sessionID, title.generate(sessionID).pipe(Effect.ignore), {
-            onlyIfMissing: true,
-          })
-        currentStep = promoted > 0 ? 1 : currentStep
-        currentPromotable = undefined
-        const loaded = yield* context.load(selected)
+        // Reuse boundary preparation once; retries refresh context without delivering more input.
+        const loaded = initial ?? (yield* prepareContext(sessionID).pipe(Effect.flatMap(context.load)))
+        initial = undefined
         const compactionInput = { session: loaded.session, messages: loaded.messages, resolved: loaded.model }
         if (compaction.required(compactionInput)) {
           const compacted = yield* compaction.compact(compactionInput)
@@ -162,7 +222,7 @@ const layer = Layer.effect(
           assistantMessageID = SessionMessage.ID.create()
           continue
         }
-        const stepLimitReached = loaded.agent.info.steps !== undefined && currentStep >= loaded.agent.info.steps
+        const stepLimitReached = loaded.agent.info.steps !== undefined && step >= loaded.agent.info.steps
         const transcript = SessionModelRequest.baseTranscript({
           agent: loaded.agent.info,
           model: loaded.model,
@@ -197,7 +257,7 @@ const layer = Layer.effect(
               : Effect.succeed(false),
           ),
         })
-        if (outcome._tag === "Completed") return { needsContinuation: outcome.needsContinuation, step: currentStep }
+        if (outcome._tag === "Completed") return outcome.needsContinuation
         if (outcome._tag === "Retry" || outcome._tag === "Continue") {
           yield* retry({ cause: outcome.cause, error: outcome.error, assistantMessageID }).pipe(
             Pull.catchDone(() =>
@@ -223,77 +283,6 @@ const layer = Layer.effect(
       }
     })
 
-    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.nextPromotable(db, sessionID, promotable)
-              if (selected?.type !== "compaction") return
-              yield* bus.publishAll([
-                [SessionEvent.InboxDelivered, { sessionID, inboxID: selected.id }],
-                [SessionEvent.Compaction.Started, { sessionID, reason: "manual", recent: "", inputID: selected.id }],
-              ])
-              return selected
-            }),
-          )
-          if (pending?.type !== "compaction") return false
-          const session = yield* getSession(sessionID)
-          const compacted = yield* restore(
-            Effect.gen(function* () {
-              return yield* compaction.compactManual({
-                session,
-                messages: yield* store.context(sessionID),
-                inputID: pending.id,
-                started: true,
-              })
-            }),
-          ).pipe(Effect.exit)
-          if (Exit.isSuccess(compacted)) return true
-          yield* bus.publish(SessionEvent.Compaction.Failed, {
-            sessionID,
-            reason: "manual",
-            error: Cause.hasInterruptsOnly(compacted.cause)
-              ? { type: "aborted", message: "Compaction cancelled" }
-              : { type: "compaction.failed", message: Cause.pretty(compacted.cause) },
-            inputID: pending.id,
-          })
-          return yield* Effect.failCause(compacted.cause)
-        }),
-      )
-    })
-
-    const runPendingMove = Effect.fn("SessionRunner.runPendingMove")(function* (
-      sessionID: SessionSchema.ID,
-      promotable: SessionInbox.Promotable,
-    ) {
-      return yield* SessionInbox.serialized(
-        sessionID,
-        Effect.gen(function* () {
-          const pending = yield* SessionInbox.nextPromotable(db, sessionID, promotable)
-          if (pending?.type !== "move") return false
-          yield* modelTransport.close(sessionID)
-          yield* bus.publishAll([
-            [SessionEvent.InboxDelivered, { sessionID, inboxID: pending.id }],
-            [
-              SessionEvent.Moved,
-              {
-                sessionID,
-                location: pending.payload.location,
-                projectID: pending.payload.projectID,
-                subpath: pending.payload.subpath,
-              },
-            ],
-          ])
-          return true
-        }),
-      )
-    })
-
     const settleStaleToolCalls = Effect.fn("SessionRunner.settleStaleToolCalls")(function* (
       sessionID: SessionSchema.ID,
     ) {
@@ -319,12 +308,6 @@ const layer = Layer.effect(
       }
     })
 
-    const getSession = Effect.fn("SessionRunner.getSession")(function* (sessionID: SessionSchema.ID) {
-      const session = yield* store.get(sessionID)
-      if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
-      return session
-    })
-
     return Service.of({ drain })
   }),
 )

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

@@ -1444,6 +1444,49 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("delivers controls without preflighting unavailable initial instructions", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      const database = yield* Database.Service
+      const bus = yield* Bus.Service
+      const runner = yield* SessionRunner.Service
+      systemUnavailable = true
+      let reads = 0
+      systemLoadHook = Effect.sync(() => {
+        reads++
+      })
+      const compaction = yield* SessionInbox.admitCompaction(database.db, bus, {
+        id: SessionMessage.ID.create(),
+        sessionID,
+        delivery: "queue",
+      })
+      yield* SessionInbox.admit(database.db, bus, {
+        id: SessionMessage.ID.create(),
+        sessionID,
+        item: {
+          type: "move",
+          payload: {
+            location: Location.Ref.make({ directory: AbsolutePath.make("/moved") }),
+            projectID: Project.ID.global,
+          },
+          delivery: "queue",
+        },
+      })
+
+      expect(yield* runner.drain({ sessionID, force: false })).toEqual(SessionRunner.DrainResult.Moved({}))
+
+      expect(reads).toBe(0)
+      expect(requests).toHaveLength(0)
+      expect(yield* session.inbox(sessionID)).toEqual([])
+      expect((yield* session.get(sessionID)).location.directory).toBe(AbsolutePath.make("/moved"))
+      expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({
+        type: "compaction",
+        status: "failed",
+        error: { type: "compaction.unavailable", message: "Nothing to compact yet" },
+      })
+    }),
+  )
+
   it.effect("delivers a queued move atomically at the idle boundary", () =>
     Effect.gen(function* () {
       const session = yield* setup
@@ -1554,6 +1597,56 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("runs a queued control on Location entry before a carried continuation", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      const database = yield* Database.Service
+      const bus = yield* Bus.Service
+      const runner = yield* SessionRunner.Service
+      yield* admit(session, "Echo before moving")
+      yield* TestLLM.push(
+        TestLLM.tool("call-entry", "echo", { text: "moving" }),
+        TestLLM.text("Entry summary", "entry-summary"),
+        TestLLM.text("Continued", "entry-continuation"),
+      )
+      const stream = yield* TestLLM.gate
+      const run = yield* runner.drain({ sessionID, force: false }).pipe(Effect.forkChild)
+      yield* stream.started
+      const compaction = yield* SessionInbox.admitCompaction(database.db, bus, {
+        id: SessionMessage.ID.create(),
+        sessionID,
+        delivery: "queue",
+      })
+      yield* SessionInbox.admit(database.db, bus, {
+        id: SessionMessage.ID.create(),
+        sessionID,
+        item: {
+          type: "move",
+          payload: {
+            location: Location.Ref.make({ directory: AbsolutePath.make("/project") }),
+            projectID: Project.ID.global,
+          },
+          delivery: "steer",
+        },
+      })
+      yield* stream.release
+      const moved = yield* Fiber.join(run)
+
+      expect(moved).toEqual(SessionRunner.DrainResult.Moved({ continuation: { step: 2 } }))
+      expect(requests).toHaveLength(1)
+      expect(yield* SessionInbox.find(database.db, compaction.id)).toMatchObject({ id: compaction.id })
+      if (moved._tag !== "Moved") throw new Error("Expected a Location handoff")
+
+      // Location entry considers queued controls even when model work carries across the move.
+      yield* runner.drain({ sessionID, force: false, continuation: moved.continuation })
+
+      expect(requests).toHaveLength(3)
+      expect(userTexts(requests[1])[0]).toContain("Create a new anchored summary")
+      expect(userTexts(requests[2])[0]).toContain("<summary>\nEntry summary\n</summary>")
+      expect(yield* session.inbox(sessionID)).toEqual([])
+    }),
+  )
+
   it.effect("seeds a fork with the parent's newest instruction values", () =>
     Effect.gen(function* () {
       const session = yield* setup
@@ -2562,6 +2655,74 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("refreshes preparation after overflow compaction without promoting new input", () =>
+    Effect.gen(function* () {
+      const session = yield* setupOverflowRecovery
+      const bus = yield* Bus.Service
+      let reads = 0
+      let resolutions = 0
+      systemLoadHook = Effect.sync(() => {
+        reads++
+      })
+      modelResolveHook = Effect.sync(() => {
+        resolutions++
+      })
+      yield* admit(session, "Continue")
+      yield* TestLLM.push(
+        [LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" })],
+        TestLLM.text("Overflow summary", "overflow-summary"),
+        TestLLM.text("Recovered", "overflow-recovered"),
+        TestLLM.stop(),
+        TestLLM.stop(),
+      )
+      const first = yield* TestLLM.gate
+      const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
+      yield* first.started
+      expect(reads).toBe(1)
+      expect(resolutions).toBe(1)
+      expect(requests[0]?.model).toBe(recoveryModel)
+
+      const summary = yield* TestLLM.gate
+      yield* first.release
+      yield* summary.started
+      systemBaseline = "Changed during compaction"
+      yield* bus.publish(SessionEvent.ModelSelected, {
+        sessionID,
+        model: { id: ID.make("replacement"), providerID: Provider.ID.make("fake") },
+      })
+      const queued = yield* session.prompt({
+        sessionID,
+        text: "Queued during compaction",
+        delivery: "queue",
+        resume: false,
+      })
+      const steered = yield* admit(session, "Steered during compaction")
+      const retry = yield* TestLLM.gate
+      yield* summary.release
+      yield* retry.started
+
+      expect(reads).toBe(2)
+      expect(resolutions).toBe(2)
+      expect(requests).toHaveLength(3)
+      expect(requests[2]?.model).toBe(replacementModel)
+      expect(requests[2]?.system.map((part) => part.text)).toEqual([defaultSystem, "Initial context"])
+      expect(systemTexts(requests[2])).toContain("Changed during compaction")
+      expect(userTexts(requests[2])[0]).toContain("<summary>\nOverflow summary\n</summary>")
+      expect(userTexts(requests[2]).join("\n")).not.toContain("Queued during compaction")
+      expect(userTexts(requests[2]).join("\n")).not.toContain("Steered during compaction")
+      expect((yield* session.inbox(sessionID)).map((item) => item.id)).toEqual([queued.id, steered.id])
+
+      yield* retry.release
+      yield* Fiber.join(run)
+
+      expect(requests).toHaveLength(5)
+      expect(userTexts(requests[3])).toContain("Steered during compaction")
+      expect(userTexts(requests[3])).not.toContain("Queued during compaction")
+      expect(userTexts(requests[4])).toContain("Queued during compaction")
+      expect(yield* session.inbox(sessionID)).toEqual([])
+    }),
+  )
+
   it.effect("does not recover provider context overflow when automatic compaction is disabled", () =>
     Effect.gen(function* () {
       const session = yield* setupOverflowRecovery
@@ -3320,6 +3481,71 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("keeps queued input parked when a steer is cancelled during preparation", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      const runner = yield* SessionRunner.Service
+      yield* admit(session, "A")
+      yield* TestLLM.push(TestLLM.stop(), TestLLM.stop(), TestLLM.stop())
+      const stream = yield* TestLLM.gate
+      const run = yield* runner.drain({ sessionID, force: false }).pipe(Effect.forkChild)
+      yield* stream.started
+
+      yield* session.prompt({ sessionID, text: "B", delivery: "queue", resume: false })
+      const steer = yield* admit(session, "S")
+      systemLoadHook = Effect.gen(function* () {
+        systemLoadHook = Effect.void
+        yield* session.cancelInbox({ sessionID, inboxID: steer.id }).pipe(Effect.orDie)
+      })
+      yield* stream.release
+      yield* Fiber.join(run)
+
+      expect(requests.map(userTexts)).toEqual([["A"], ["A"], ["A", "B"]])
+      expect((yield* session.messages({ sessionID })).some((message) => message.id === steer.id)).toBe(false)
+      expect(yield* session.inbox(sessionID)).toEqual([])
+    }),
+  )
+
+  it.effect("dispatches a queued move when a steer is cancelled during preparation", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      const runner = yield* SessionRunner.Service
+      const database = yield* Database.Service
+      const bus = yield* Bus.Service
+      const location = Location.Ref.make({ directory: AbsolutePath.make("/moved") })
+      yield* admit(session, "A")
+      yield* TestLLM.push(TestLLM.stop(), TestLLM.stop())
+      const stream = yield* TestLLM.gate
+      const run = yield* runner.drain({ sessionID, force: false }).pipe(Effect.forkChild)
+      yield* stream.started
+
+      yield* SessionInbox.admit(database.db, bus, {
+        id: SessionMessage.ID.create(),
+        sessionID,
+        item: {
+          type: "move",
+          payload: { location, projectID: Project.ID.global },
+          delivery: "queue",
+        },
+      })
+      const steer = yield* admit(session, "S")
+      systemLoadHook = Effect.gen(function* () {
+        systemLoadHook = Effect.void
+        yield* session.cancelInbox({ sessionID, inboxID: steer.id }).pipe(Effect.orDie)
+      })
+      yield* stream.release
+
+      expect({ result: yield* Fiber.join(run), location: (yield* session.get(sessionID)).location }).toEqual({
+        result: SessionRunner.DrainResult.Moved({}),
+        location,
+      })
+      expect(requests.map(userTexts)).toEqual([["A"], ["A"]])
+      expect(closedTransports).toEqual([sessionID])
+      expect(yield* recordedEventTypes(sessionID)).toContain(Bus.versionedType(SessionEvent.Moved.type, 1))
+      expect(yield* session.inbox(sessionID)).toEqual([])
+    }),
+  )
+
   it.effect("preserves durable queued input for a later wake after interruption", () =>
     Effect.gen(function* () {
       const session = yield* setup