refactor(core): simplify session runner loop and pending input scopes (#38602)

packages/core/src/session/pending.ts

@@ -1,6 +1,6 @@
 export * as SessionPending from "./pending"
 
-import { and, asc, eq } from "drizzle-orm"
+import { and, asc, eq, or } from "drizzle-orm"
 import { DateTime, Effect, Schema } from "effect"
 import {
   Compaction,
@@ -26,6 +26,13 @@ type DatabaseService = Database.Interface["db"]
 
 export { Compaction, Delivery, Info, Message, Synthetic, SyntheticData, User, UserData }
 
+/**
+ * Which pending input `promote` may consume: "steer" promotes steers only (a step
+ * boundary mid-work), while "input" also allows one queued input when no steers are
+ * waiting (the idle boundary, where the Session picks up fresh work).
+ */
+export type Promotable = "input" | "steer"
+
 const decodeUser = Schema.decodeUnknownSync(UserData)
 const encodeUser = Schema.encodeSync(UserData)
 const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData)
@@ -355,16 +362,32 @@ export const list = Effect.fn("SessionPending.list")(function* (db: DatabaseServ
   return rows.map(fromRow)
 })
 
+/**
+ * Which pending rows count: "any" counts every row including compaction, while
+ * delivery scopes are blocked behind a pending compaction barrier. "input" means
+ * any model-facing input, steered or queued.
+ */
+export type Scope = "any" | "input" | Delivery
+
 export const has = Effect.fn("SessionPending.has")(function* (
   db: DatabaseService,
   sessionID: SessionSchema.ID,
-  delivery: Delivery,
+  scope: Scope,
 ) {
-  if (yield* compaction(db, sessionID)) return false
+  if (scope !== "any" && (yield* compaction(db, sessionID))) return false
   const row = yield* db
     .select({ id: SessionPendingTable.id })
     .from(SessionPendingTable)
-    .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, delivery)))
+    .where(
+      and(
+        eq(SessionPendingTable.session_id, sessionID),
+        scope === "any"
+          ? undefined
+          : scope === "input"
+            ? or(eq(SessionPendingTable.delivery, "steer"), eq(SessionPendingTable.delivery, "queue"))
+            : eq(SessionPendingTable.delivery, scope),
+      ),
+    )
     .limit(1)
     .get()
     .pipe(Effect.orDie)
@@ -394,65 +417,73 @@ const publish = Effect.fn("SessionPending.publish")(function* (
   sessionID: SessionSchema.ID,
   rows: ReadonlyArray<typeof SessionPendingTable.$inferSelect>,
 ) {
-  return yield* inboxLocks.withLock(sessionID)(
-    Effect.gen(function* () {
-      if (yield* compaction(db, sessionID)) return 0
-      yield* Effect.forEach(
-        rows,
-        (row) => {
-          const entry = fromRow(row)
-          if (entry.type === "compaction") return Effect.die(new LifecycleConflict({ id: entry.id }))
-          return events
-            .publish(SessionEvent.InputPromoted, {
-              sessionID,
-              inputID: entry.id,
-            })
-            .pipe(
-              Effect.catchDefect((defect) =>
-                defect instanceof LifecycleConflict
-                  ? promotedFromHistory(db, sessionID, entry.id).pipe(
-                      Effect.flatMap((stored) => (stored !== undefined ? Effect.void : Effect.die(defect))),
-                    )
-                  : Effect.die(defect),
-              ),
-            )
-        },
-        { discard: true },
-      )
-      return rows.length
-    }),
+  if (yield* compaction(db, sessionID)) return 0
+  yield* Effect.forEach(
+    rows,
+    (row) => {
+      const entry = fromRow(row)
+      if (entry.type === "compaction") return Effect.die(new LifecycleConflict({ id: entry.id }))
+      return events
+        .publish(SessionEvent.InputPromoted, {
+          sessionID,
+          inputID: entry.id,
+        })
+        .pipe(
+          Effect.catchDefect((defect) =>
+            defect instanceof LifecycleConflict
+              ? promotedFromHistory(db, sessionID, entry.id).pipe(
+                  Effect.flatMap((stored) => (stored !== undefined ? Effect.void : Effect.die(defect))),
+                )
+              : Effect.die(defect),
+          ),
+        )
+    },
+    { discard: true },
   )
+  return rows.length
 })
 
-export const promoteSteers = Effect.fn("SessionPending.promoteSteers")(function* (
+/**
+ * Promotes pending input into visible messages and returns the promoted count.
+ * Steers always go first; only the "input" scope may fall through to one queued
+ * input, and it then collects steers that arrived during promotion.
+ */
+export const promote = Effect.fn("SessionPending.promote")(function* (
   db: DatabaseService,
   events: EventV2.Interface,
   sessionID: SessionSchema.ID,
+  scope: Promotable,
 ) {
-  if (yield* compaction(db, sessionID)) return 0
-  const rows = yield* db
-    .select()
-    .from(SessionPendingTable)
-    .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, "steer")))
-    .orderBy(asc(SessionPendingTable.admitted_seq))
-    .all()
-    .pipe(Effect.orDie)
-  return yield* publish(db, events, sessionID, rows)
-})
+  return yield* inboxLocks.withLock(sessionID)(
+    Effect.gen(function* () {
+      if (yield* compaction(db, sessionID)) return 0
+      const steers = yield* db
+        .select()
+        .from(SessionPendingTable)
+        .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, "steer")))
+        .orderBy(asc(SessionPendingTable.admitted_seq))
+        .all()
+        .pipe(Effect.orDie)
+      if (steers.length > 0 || scope === "steer") return yield* publish(db, events, sessionID, steers)
 
-export const promoteNextQueued = Effect.fn("SessionPending.promoteNextQueued")(function* (
-  db: DatabaseService,
-  events: EventV2.Interface,
-  sessionID: SessionSchema.ID,
-) {
-  if (yield* compaction(db, sessionID)) return false
-  const row = yield* db
-    .select()
-    .from(SessionPendingTable)
-    .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, "queue")))
-    .orderBy(asc(SessionPendingTable.admitted_seq))
-    .limit(1)
-    .get()
-    .pipe(Effect.orDie)
-  return row === undefined ? false : yield* publish(db, events, sessionID, [row]).pipe(Effect.as(true))
+      const queued = yield* db
+        .select()
+        .from(SessionPendingTable)
+        .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, "queue")))
+        .orderBy(asc(SessionPendingTable.admitted_seq))
+        .limit(1)
+        .get()
+        .pipe(Effect.orDie)
+      if (!queued) return 0
+      const promoted = yield* publish(db, events, sessionID, [queued])
+      const arrivedSteers = yield* db
+        .select()
+        .from(SessionPendingTable)
+        .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.delivery, "steer")))
+        .orderBy(asc(SessionPendingTable.admitted_seq))
+        .all()
+        .pipe(Effect.orDie)
+      return promoted + (yield* publish(db, events, sessionID, arrivedSteers))
+    }),
+  )
 })

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

@@ -20,7 +20,7 @@ export type RunError =
 
 /** Runs one local continuation from already-recorded Session history. */
 export interface Interface {
-  /** Drains eligible durable work. Explicit runs perform one physical attempt even when no work is eligible. */
+  /** Drains eligible durable work. Explicit runs make one model call even when no work is eligible. */
   readonly drain: (input: {
     readonly sessionID: SessionSchema.ID
     readonly force: boolean

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

@@ -1,8 +1,7 @@
 export * as SessionRunnerLLM from "./llm"
 
 import { LLMClient, LLMError, LLMEvent, isContextOverflowFailure, type ProviderErrorEvent } from "@opencode-ai/ai"
-import { SessionError } from "@opencode-ai/schema/session-error"
-import { Cause, Effect, Exit, Fiber, FiberSet, Layer, Option, Semaphore, Stream } from "effect"
+import { Cause, Data, Effect, Exit, Fiber, FiberSet, Layer, Option, Pull, Schedule, Semaphore, Stream } from "effect"
 import { Database } from "../../database/database"
 import { EventV2 } from "../../event"
 import { PermissionV2 } from "../../permission"
@@ -28,6 +27,14 @@ import { toSessionError } from "../to-session-error"
 import { SessionRunnerRetry } from "./retry"
 import { SessionUsage } from "../usage"
 
+/** How one model call ended: settled, awaiting a scheduled retry, or restarted by compaction. */
+type CallOutcome = Data.TaggedEnum<{
+  Completed: { readonly needsContinuation: boolean; readonly step: number }
+  Retry: { readonly step: number; readonly assistantMessageID: SessionMessage.ID }
+  Restart: { readonly step: number; readonly recoveredOverflow: boolean }
+}>
+const CallOutcome = Data.taggedEnum<CallOutcome>()
+
 const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
@@ -43,14 +50,21 @@ const layer = Layer.effect(
     // Title generation is a side effect of the first step; it must not delay step continuation.
     // Tracked per process so repeated wakes before the second user message arrives don't
     // re-fire a redundant LLM call; `SessionTitle` itself is idempotent based on durable history.
-    const titleAttempted = new Set<SessionSchema.ID>()
+    const titleStarted = new Set<SessionSchema.ID>()
     const forkTitle = yield* FiberSet.makeRuntime<never, void, never>()
     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
     })
-    const failInterruptedTools = Effect.fn("SessionRunner.failInterruptedTools")(function* (
+    /** Fires title generation once per process after the first step makes a user message visible. */
+    const startTitleOnce = Effect.fnUntraced(function* (sessionID: SessionSchema.ID) {
+      if (titleStarted.has(sessionID)) return
+      titleStarted.add(sessionID)
+      forkTitle(title.generateForFirstPrompt(yield* getSession(sessionID)).pipe(Effect.ignore))
+    })
+    /** Closes stale tool calls left active by an earlier interrupted drain. */
+    const settleStaleToolCalls = Effect.fn("SessionRunner.settleStaleToolCalls")(function* (
       sessionID: SessionSchema.ID,
     ) {
       for (const message of yield* store.context(sessionID)) {
@@ -76,11 +90,15 @@ const layer = Layer.effect(
           (reason.defect instanceof PermissionV2.DeclinedError || reason.defect instanceof QuestionTool.CancelledError),
       )
 
-    const attemptStep = Effect.fn("SessionRunner.attemptStep")(function* (
+    /**
+     * Prepares and runs at most one model call, executes its local tools, and durably
+     * settles the step. Compaction may instead request that the logical step restart.
+     */
+    const callModel = Effect.fn("SessionRunner.callModel")(function* (
       sessionID: SessionSchema.ID,
-      promotion: SessionPending.Delivery | undefined,
+      promotable: SessionPending.Promotable | undefined,
       step: number,
-      recoverOverflow?: typeof compaction.compact,
+      recoverOverflow: boolean,
       assistantMessageID?: SessionMessage.ID,
     ) {
       const selected = yield* context.select(sessionID)
@@ -88,24 +106,20 @@ const layer = Layer.effect(
       // a blocked first step leaves pending inputs untouched.
       yield* InstructionState.prepare(db, events, selected.instructions, selected.session.id)
       let currentStep = step
-      if (promotion) {
-        let promoted = 0
-        if (promotion === "steer") promoted = yield* SessionPending.promoteSteers(db, events, selected.session.id)
-        if (promotion === "queue") {
-          promoted += Number(yield* SessionPending.promoteNextQueued(db, events, selected.session.id))
-          promoted += yield* SessionPending.promoteSteers(db, events, selected.session.id)
-        }
+      if (promotable) {
+        const promoted = yield* SessionPending.promote(db, events, selected.session.id, promotable)
         if (promoted > 0) currentStep = 1
       }
       const loaded = yield* context.load(selected)
-      const session = loaded.session
-      const agent = loaded.agent
+      const { session, agent } = loaded
       const resolved = loaded.model
       const model = resolved.model
+      // Make room: history must fit the context window before the call. A pending manual
+      // compaction owns this instead; the runner executes it between steps.
       const compactionInput = { session, messages: loaded.messages, model, cost: resolved.cost }
       if (compaction.required(compactionInput) && !(yield* SessionPending.compaction(db, session.id))) {
         const compacted = yield* compaction.compact(compactionInput)
-        if (compacted.status === "completed") return { _tag: "RestartAfterCompaction", step: currentStep } as const
+        if (compacted.status === "completed") return CallOutcome.Restart({ step: currentStep, recoveredOverflow: false })
         return yield* new StepFailedError({ error: compacted.error })
       }
       const prepared = yield* modelRequests.prepare({
@@ -131,6 +145,40 @@ const layer = Layer.effect(
       // mid-event.
       const serialized = <A, E, R>(effect: Effect.Effect<A, E, R>) => publication.withPermit(effect)
       const publish = (event: LLMEvent) => serialized(publisher.publish(event))
+
+      const stepUsage = (settlement: NonNullable<ReturnType<typeof publisher.stepSettlement>>) => ({
+        cost: SessionUsage.calculateCost(resolved.cost, settlement.tokens),
+        tokens: settlement.tokens,
+      })
+
+      const captureStepEnd = Effect.fnUntraced(function* () {
+        const snapshot = yield* snapshots.capture()
+        const files =
+          startSnapshot && snapshot
+            ? yield* snapshots
+                .files({ from: startSnapshot, to: snapshot })
+                .pipe(Effect.catch(() => Effect.succeed(undefined)))
+            : undefined
+        return { snapshot, files }
+      })
+
+      const publishStepEnd = (settlement: NonNullable<ReturnType<typeof publisher.stepSettlement>>) =>
+        Effect.gen(function* () {
+          const end = yield* captureStepEnd()
+          yield* serialized(
+            events.publish(SessionEvent.Step.Ended, {
+              sessionID: session.id,
+              assistantMessageID: yield* publisher.startAssistant(),
+              finish: settlement.finish,
+              ...stepUsage(settlement),
+              ...end,
+            }),
+          )
+        })
+
+      // The stream is defined here but runs inside the settlement mask below: publish each
+      // event durably, fork one fiber per local tool call, and hold back a virgin
+      // context-overflow provider error so settlement may recover it via compaction.
       let overflowFailure: ProviderErrorEvent | undefined
       const providerStream = llm.stream(prepared.request).pipe(
         Stream.runForEach((event) =>
@@ -172,39 +220,10 @@ const layer = Layer.effect(
         Effect.ensuring(serialized(publisher.flush())),
       )
 
-      const stepUsage = (settlement: NonNullable<ReturnType<typeof publisher.stepSettlement>>) => ({
-        cost: SessionUsage.calculateCost(resolved.cost, settlement.tokens),
-        tokens: settlement.tokens,
-      })
-
-      const captureStepEnd = Effect.fnUntraced(function* () {
-        const snapshot = yield* snapshots.capture()
-        const files =
-          startSnapshot && snapshot
-            ? yield* snapshots
-                .files({ from: startSnapshot, to: snapshot })
-                .pipe(Effect.catch(() => Effect.succeed(undefined)))
-            : undefined
-        return { snapshot, files }
-      })
-
-      const publishStepEnd = (settlement: NonNullable<ReturnType<typeof publisher.stepSettlement>>) =>
-        Effect.gen(function* () {
-          const end = yield* captureStepEnd()
-          yield* serialized(
-            events.publish(SessionEvent.Step.Ended, {
-              sessionID: session.id,
-              assistantMessageID: yield* publisher.startAssistant(),
-              finish: settlement.finish,
-              ...stepUsage(settlement),
-              ...end,
-            }),
-          )
-        })
-
+      // Settle: only the stream itself is interruptible (restore); every line after it is
+      // protected so a started call always reaches one durable outcome.
       return yield* Effect.uninterruptibleMask((restore) =>
         Effect.gen(function* () {
-          // Gather the evidence: how did the provider stream end?
           const stream = yield* restore(providerStream).pipe(Effect.exit)
           const streamFailure = Option.getOrUndefined(Exit.findErrorOption(stream))
           // Note: Exit.hasInterrupts is a type guard whose false branch unsoundly narrows
@@ -217,10 +236,10 @@ const layer = Layer.effect(
             recoverOverflow &&
             !publisher.hasRetryEvidence() &&
             isContextOverflowFailure(overflowFailure ?? streamFailure) &&
-            (yield* restore(recoverOverflow({ session, messages: loaded.messages, model, cost: resolved.cost })))
+            (yield* restore(compaction.compact({ session, messages: loaded.messages, model, cost: resolved.cost })))
               .status === "completed"
           )
-            return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const
+            return CallOutcome.Restart({ step: currentStep, recoveredOverflow: true })
 
           // An unrecovered held-back overflow becomes the step's durable provider error. A
           // thrown LLM failure records the assistant failure unless a provider error was
@@ -324,63 +343,59 @@ const layer = Layer.effect(
             return yield* Effect.failCause(settledFailure)
           if (toolsInterrupted && settled._tag === "Failure") return yield* Effect.failCause(settled.cause)
           if (stepFailure) return yield* new StepFailedError({ error: stepFailure })
-          return {
-            _tag: "Completed",
-            needsContinuation,
-            step: currentStep,
-          } as const
+          return CallOutcome.Completed({ needsContinuation, step: currentStep })
         }),
       )
     }, Effect.scoped)
 
+    /** Completes one logical model step, transparently retrying or rebuilding after compaction. */
     const runStep = Effect.fnUntraced(function* (
       sessionID: SessionSchema.ID,
-      promotion: SessionPending.Delivery | undefined,
+      promotable: SessionPending.Promotable,
       step: number,
     ) {
-      // Compaction restarts rebuild the request from compacted history without re-promoting.
-      // Overflow recovery is one-shot: a post-compaction attempt must not recover another
-      // overflow, so the recovery hook is dropped after it fires.
-      let recoverOverflow: typeof compaction.compact | undefined = compaction.compact
-      let currentPromotion = promotion
-      let currentStep = step
-      let assistantMessageID: SessionMessage.ID | undefined
-      while (true) {
-        const attempt = yield* Effect.suspend(() =>
-          attemptStep(sessionID, currentPromotion, currentStep, recoverOverflow, assistantMessageID),
-        ).pipe(
-          Effect.tapError((error) =>
-            error instanceof SessionRunnerRetry.RetryableFailure
-              ? Effect.sync(() => {
-                  currentStep = error.step
-                  assistantMessageID = error.assistantMessageID
-                  currentPromotion = undefined
-                })
-              : Effect.void,
-          ),
-          Effect.retryOrElse(SessionRunnerRetry.schedule(events, sessionID), (error) => {
-            if (!(error instanceof SessionRunnerRetry.RetryableFailure)) return Effect.fail(error)
-            return events
+      const retry = yield* Schedule.toStepWithSleep(SessionRunnerRetry.schedule(events, sessionID))
+      /**
+       * Consumes one retry allowance: sleeps the scheduled backoff and reports what the next
+       * attempt should reuse, or publishes Step.Failed and fails once attempts are exhausted.
+       * The step loop performs the retry itself on the next iteration.
+       */
+      const waitForRetry = (failure: SessionRunnerRetry.RetryableFailure) =>
+        retry(failure).pipe(
+          Effect.as(CallOutcome.Retry({ step: failure.step, assistantMessageID: failure.assistantMessageID })),
+          Pull.catchDone(() =>
+            events
               .publish(SessionEvent.Step.Failed, {
                 sessionID,
-                assistantMessageID: error.assistantMessageID,
-                error: error.error,
+                assistantMessageID: failure.assistantMessageID,
+                error: failure.error,
               })
-              .pipe(Effect.andThen(Effect.fail(error.cause)))
-          }),
+              .pipe(Effect.andThen(Effect.fail(failure.cause))),
+          ),
         )
-        if (attempt._tag === "Completed")
-          return {
-            needsContinuation: attempt.needsContinuation,
-            step: attempt.step,
-          }
-        if (attempt._tag === "RestartAfterOverflowCompaction") recoverOverflow = undefined
-        yield* Effect.yieldNow
-        currentPromotion = undefined
-        currentStep = attempt.step
+      let currentPromotable: SessionPending.Promotable | undefined = promotable
+      let currentStep = step
+      let assistantMessageID: SessionMessage.ID | undefined
+      // Overflow recovery is one-shot: a call after recovery must not recover another overflow.
+      let recoverOverflow = true
+      while (true) {
+        const outcome = yield* callModel(
+          sessionID,
+          currentPromotable,
+          currentStep,
+          recoverOverflow,
+          assistantMessageID,
+        ).pipe(Effect.catchTag("SessionRunner.RetryableFailure", waitForRetry))
+        if (outcome._tag === "Completed") return { needsContinuation: outcome.needsContinuation, step: outcome.step }
+        if (outcome._tag === "Retry") assistantMessageID = outcome.assistantMessageID
+        if (outcome._tag === "Restart" && outcome.recoveredOverflow) recoverOverflow = false
+        // Neither a retry nor a compaction restart re-promotes input.
+        currentPromotable = undefined
+        currentStep = outcome.step
       }
     })
 
+    /** Executes a previously admitted manual compaction request, if one is pending. */
     const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* (
       sessionID: SessionSchema.ID,
     ) {
@@ -399,67 +414,54 @@ const layer = Layer.effect(
             }),
           ).pipe(Effect.exit)
           if (Exit.isSuccess(compacted)) return
-          if (Exit.isFailure(compacted)) {
-            const unsettled = yield* SessionPending.compaction(db, sessionID)
-            if (unsettled)
-              yield* events.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: unsettled.id,
-              })
-            return yield* Effect.failCause(compacted.cause)
-          }
+          const unsettled = yield* SessionPending.compaction(db, sessionID)
+          if (unsettled)
+            yield* events.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: unsettled.id,
+            })
+          return yield* Effect.failCause(compacted.cause)
         }),
       )
     })
 
-    // Execution lifecycle is published per busy period by SessionExecution, not per drain here.
+    /**
+     * Runs logical steps until no tool result or newly admitted steer requires another
+     * model call. Queued inputs remain pending until the current model work reaches idle.
+     */
+    const runSteps = Effect.fn("SessionRunner.runSteps")(function* (sessionID: SessionSchema.ID) {
+      // Fresh work may promote queued input; later steps absorb steers only.
+      let promotable: SessionPending.Promotable = "input"
+      let step = 1
+      while (true) {
+        const result = yield* runStep(sessionID, promotable, step)
+        yield* startTitleOnce(sessionID)
+        yield* runPendingCompaction(sessionID)
+        if (!result.needsContinuation && !(yield* SessionPending.has(db, sessionID, "steer"))) return
+        promotable = "steer"
+        step = result.step + 1
+      }
+    })
+
+    /**
+     * Drains eligible manual compaction and user input until the Session becomes idle.
+     * Execution lifecycle is published per busy period by SessionExecution, not here.
+     */
     const drain = Effect.fn("SessionRunner.drain")(function* (input: {
       readonly sessionID: SessionSchema.ID
       readonly force: boolean
     }) {
+      if (!input.force && !(yield* SessionPending.has(db, input.sessionID, "any"))) return
+      yield* settleStaleToolCalls(input.sessionID)
       yield* runPendingCompaction(input.sessionID)
-      const hasSteer = yield* SessionPending.has(db, input.sessionID, "steer")
-      const hasQueue = hasSteer ? false : yield* SessionPending.has(db, input.sessionID, "queue")
-      if (!input.force && !hasSteer && !hasQueue) return
-      yield* failInterruptedTools(input.sessionID)
-      let promotion: SessionPending.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined
-      let shouldRun = input.force || hasSteer || hasQueue
-      while (shouldRun) {
-        let needsContinuation = true
-        let step = 1
-        // Repeat steps while continuation is needed. A step needs continuation only
-        // when it recorded local tool calls whose results the model has not yet seen;
-        // a provider error suppresses it. Pending steers also continue the loop so
-        // interjections are answered before the session goes idle.
-        while (needsContinuation) {
-          const result = yield* runStep(input.sessionID, promotion, step)
-          // Steer/queue promotion inside runStep has already made the pending input a visible
-          // user message by this point, so the first-user-message check below is reliable.
-          if (!titleAttempted.has(input.sessionID)) {
-            titleAttempted.add(input.sessionID)
-            forkTitle(title.generateForFirstPrompt(yield* getSession(input.sessionID)).pipe(Effect.ignore))
-          }
-          needsContinuation = result.needsContinuation
-          step = result.step + 1
-          if (needsContinuation) {
-            yield* runPendingCompaction(input.sessionID)
-            promotion = "steer"
-            continue
-          }
-          yield* runPendingCompaction(input.sessionID)
-          promotion = "steer"
-          needsContinuation = yield* SessionPending.has(db, input.sessionID, "steer")
-        }
-        yield* runPendingCompaction(input.sessionID)
-        const hasSteer = yield* SessionPending.has(db, input.sessionID, "steer")
-        const hasQueue = hasSteer ? false : yield* SessionPending.has(db, input.sessionID, "queue")
-        shouldRun = hasSteer || hasQueue
-        promotion = hasSteer ? "steer" : hasQueue ? "queue" : undefined
-      }
+      if (!input.force && !(yield* SessionPending.has(db, input.sessionID, "input"))) return
+      do {
+        yield* runSteps(input.sessionID)
+      } while (yield* SessionPending.has(db, input.sessionID, "input"))
     })
 
     return Service.of({ drain })

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

@@ -7,7 +7,6 @@ import { EventV2 } from "../../event"
 import { SessionEvent } from "../event"
 import { SessionMessage } from "../message"
 import { SessionSchema } from "../schema"
-import type { SessionRunner } from "./index"
 
 export class RetryableFailure extends Data.TaggedError("SessionRunner.RetryableFailure")<{
   readonly cause: LLMError
@@ -45,22 +44,18 @@ const retryAfter = (failure: RetryableFailure) => {
 
 export const schedule = (events: EventV2.Interface, sessionID: SessionSchema.ID) =>
   Schedule.max([Schedule.exponential("2 seconds"), Schedule.recurs(4)]).pipe(
-    Schedule.setInputType<RetryableFailure | SessionRunner.RunError>(),
-    Schedule.passthrough,
-    Schedule.while(({ input }) => input instanceof RetryableFailure),
+    Schedule.setInputType<RetryableFailure>(),
     Schedule.modifyDelay(({ input: failure, duration: delay }) => {
-      const minimum = failure instanceof RetryableFailure ? retryAfter(failure) : undefined
+      const minimum = retryAfter(failure)
       return Effect.succeed(minimum === undefined ? delay : Duration.max(delay, Duration.millis(minimum)))
     }),
     Schedule.tap((metadata) =>
-      metadata.input instanceof RetryableFailure
-        ? events.publish(SessionEvent.RetryScheduled, {
-            sessionID,
-            assistantMessageID: metadata.input.assistantMessageID,
-            attempt: metadata.attempt + 1,
-            at: metadata.now + Duration.toMillis(metadata.duration),
-            error: metadata.input.error,
-          })
-        : Effect.void,
+      events.publish(SessionEvent.RetryScheduled, {
+        sessionID,
+        assistantMessageID: metadata.input.assistantMessageID,
+        attempt: metadata.attempt + 1,
+        at: metadata.now + Duration.toMillis(metadata.duration),
+        error: metadata.input.error,
+      }),
     ),
   )

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

@@ -195,9 +195,9 @@ describe("SessionV2.create", () => {
         text: "First",
         resume: false,
       })
-      yield* SessionPending.promoteSteers(db, events, parent.id)
+      yield* SessionPending.promote(db, events, parent.id, "steer")
       yield* session.synthetic({ sessionID: parent.id, text: "parent note", resume: false })
-      yield* SessionPending.promoteSteers(db, events, parent.id)
+      yield* SessionPending.promote(db, events, parent.id, "steer")
 
       const forked = yield* session.fork({ sessionID: parent.id })
       const parentContext = yield* session.context(parent.id)
@@ -232,13 +232,13 @@ describe("SessionV2.create", () => {
         text: "Parent changed",
         resume: false,
       })
-      yield* SessionPending.promoteSteers(db, events, parent.id)
+      yield* SessionPending.promote(db, events, parent.id, "steer")
       yield* session.prompt({
         sessionID: forked.id,
         text: "Child continues",
         resume: false,
       })
-      yield* SessionPending.promoteSteers(db, events, forked.id)
+      yield* SessionPending.promote(db, events, forked.id, "steer")
 
       expect((yield* session.context(parent.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
       expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
@@ -263,13 +263,13 @@ describe("SessionV2.create", () => {
         text: "First",
         resume: false,
       })
-      yield* SessionPending.promoteSteers(db, events, parent.id)
+      yield* SessionPending.promote(db, events, parent.id, "steer")
       const second = yield* session.prompt({
         sessionID: parent.id,
         text: "Second",
         resume: false,
       })
-      yield* SessionPending.promoteSteers(db, events, parent.id)
+      yield* SessionPending.promote(db, events, parent.id, "steer")
       const assistantMessageID = SessionMessage.ID.create()
       const model = ModelV2.Ref.make({ id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") })
       yield* events.publish(SessionEvent.Step.Started, {
@@ -414,7 +414,7 @@ describe("SessionV2.create", () => {
         text: "Hello",
         resume: false,
       })
-      yield* SessionPending.promoteSteers(db, events, created.id)
+      yield* SessionPending.promote(db, events, created.id, "steer")
 
       expect(
         Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(2), Stream.runCollect)),
@@ -440,7 +440,7 @@ describe("SessionV2.create", () => {
         text: "Replay lifecycle",
         resume: false,
       })
-      yield* SessionPending.promoteSteers(sourceDb, sourceEvents, created.id)
+      yield* SessionPending.promote(sourceDb, sourceEvents, created.id, "steer")
       const serialized = (yield* sourceDb
         .select()
         .from(EventTable)

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

@@ -216,7 +216,7 @@ describe("SessionV2.prompt", () => {
         text: "boundary",
         resume: false,
       })
-      yield* SessionPending.promoteSteers(db, events, sessionID)
+      yield* SessionPending.promote(db, events, sessionID, "steer")
       const stale = SessionMessage.ID.make("msg_stale_assistant")
       yield* db.insert(SessionMessageTable).values(assistantRow(stale, 100)).run().pipe(Effect.orDie)
       yield* events.publish(SessionEvent.RevertEvent.Staged, {
@@ -248,7 +248,7 @@ describe("SessionV2.prompt", () => {
         text: "boundary",
         resume: false,
       })
-      yield* SessionPending.promoteSteers(db, events, sessionID)
+      yield* SessionPending.promote(db, events, sessionID, "steer")
       yield* events.publish(SessionEvent.RevertEvent.Staged, {
         sessionID,
         revert: { messageID: boundary.id, files: [] },
@@ -448,7 +448,7 @@ describe("SessionV2.prompt", () => {
 
       yield* session.prompt({ sessionID, text: "First", resume: false })
       yield* session.prompt({ sessionID, text: "Second", resume: false })
-      yield* SessionPending.promoteSteers(db, events, sessionID)
+      yield* SessionPending.promote(db, events, sessionID, "steer")
       const streamed = Array.from(yield* Fiber.join(fiber))
 
       expect(streamed.map((event): [number | undefined, string] => [event.durable?.seq, event.type])).toEqual([
@@ -625,7 +625,7 @@ describe("SessionV2.prompt", () => {
       })
 
       yield* Effect.all(
-        [SessionPending.promoteSteers(db, events, sessionID), SessionPending.promoteSteers(db, events, sessionID)],
+        [SessionPending.promote(db, events, sessionID, "steer"), SessionPending.promote(db, events, sessionID, "steer")],
         { concurrency: "unbounded" },
       )
 
@@ -855,7 +855,7 @@ describe("SessionV2.prompt", () => {
         },
       })
 
-      yield* SessionPending.promoteSteers(db, events, sessionID)
+      yield* SessionPending.promote(db, events, sessionID, "steer")
 
       expect(yield* session.messages({ sessionID })).toMatchObject([
         {
@@ -880,7 +880,7 @@ describe("SessionV2.prompt", () => {
       const entries = yield* Effect.all([session.synthetic(input), session.synthetic(input)], {
         concurrency: "unbounded",
       })
-      yield* SessionPending.promoteSteers(database.db, events, sessionID)
+      yield* SessionPending.promote(database.db, events, sessionID, "steer")
       const promotedRetry = yield* session.synthetic(input)
       const failure = yield* session.synthetic({ ...input, text: "Different completion" }).pipe(Effect.flip)
 
@@ -892,7 +892,7 @@ describe("SessionV2.prompt", () => {
     }),
   )
 
-  it.effect("keeps synthetic queue input pending until the queue boundary", () =>
+  it.effect("keeps queued input pending until the idle boundary", () =>
     Effect.gen(function* () {
       yield* setup
       const session = yield* SessionV2.Service
@@ -907,9 +907,15 @@ describe("SessionV2.prompt", () => {
       })
 
       expect(input.delivery).toBe("queue")
-      expect(yield* SessionPending.promoteSteers(db, events, sessionID)).toBe(0)
+      expect(yield* SessionPending.has(db, sessionID, "input")).toBe(true)
+      expect(
+        yield* SessionPending.promote(db, events, sessionID, "steer"),
+      ).toBe(0)
       expect(yield* session.messages({ sessionID })).toEqual([])
-      expect(yield* SessionPending.promoteNextQueued(db, events, sessionID)).toBe(true)
+      expect(
+        yield* SessionPending.promote(db, events, sessionID, "input"),
+      ).toBe(1)
+      expect(yield* SessionPending.has(db, sessionID, "input")).toBe(false)
       expect(yield* session.messages({ sessionID })).toMatchObject([
         { id: input.id, type: "synthetic", text: "Queued completion" },
       ])
@@ -935,7 +941,7 @@ describe("SessionV2.prompt", () => {
         resume: false,
       })
 
-      yield* SessionPending.promoteSteers(db, events, sessionID)
+      yield* SessionPending.promote(db, events, sessionID, "steer")
 
       expect(
         (yield* session.messages({ sessionID, order: "asc" })).map((message) =>
@@ -978,10 +984,14 @@ describe("SessionV2.pending", () => {
         { id: second.id, type: "user", delivery: "steer" },
       ])
 
-      yield* SessionPending.promoteSteers(db, events, sessionID)
+      expect(
+        yield* SessionPending.promote(db, events, sessionID, "input"),
+      ).toBe(2)
       expect(yield* session.pending(sessionID)).toMatchObject([{ id: queued.id, type: "synthetic" }])
 
-      yield* SessionPending.promoteNextQueued(db, events, sessionID)
+      expect(
+        yield* SessionPending.promote(db, events, sessionID, "input"),
+      ).toBe(1)
       expect(yield* session.pending(sessionID)).toEqual([])
     }),
   )
@@ -993,9 +1003,12 @@ describe("SessionV2.pending", () => {
       const { db } = yield* Database.Service
 
       const barrier = yield* session.compact({ sessionID })
+      expect(yield* SessionPending.has(db, sessionID, "any")).toBe(true)
+      expect(yield* SessionPending.has(db, sessionID, "input")).toBe(false)
       expect(yield* session.pending(sessionID)).toMatchObject([{ id: barrier.id, type: "compaction" }])
 
       yield* SessionPending.settleCompaction(db, { sessionID })
+      expect(yield* SessionPending.has(db, sessionID, "any")).toBe(false)
       expect(yield* session.pending(sessionID)).toEqual([])
     }),
   )

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

@@ -3099,7 +3099,7 @@ describe("SessionRunnerLLM", () => {
       streamFailure = undefined
       streamGate = undefined
       streamStarted = undefined
-      yield* Effect.yieldNow
+      yield* session.wait(sessionID)
 
       expect(requests).toHaveLength(2)
       expect(userTexts(requests[1]!)).toEqual(["Start working", "Recover with this"])
@@ -3111,7 +3111,7 @@ describe("SessionRunnerLLM", () => {
       const session = yield* setup
       const events = yield* EventV2.Service
       yield* admit(session, "Recover interrupted tool")
-      yield* SessionPending.promoteSteers((yield* Database.Service).db, events, sessionID)
+      yield* SessionPending.promote((yield* Database.Service).db, events, sessionID, "steer")
       const assistantMessageID = SessionMessage.ID.create()
       yield* events.publish(SessionEvent.Step.Started, {
         sessionID,
@@ -3168,7 +3168,7 @@ describe("SessionRunnerLLM", () => {
       const session = yield* setup
       const events = yield* EventV2.Service
       yield* admit(session, "Recover interrupted hosted tool")
-      yield* SessionPending.promoteSteers((yield* Database.Service).db, events, sessionID)
+      yield* SessionPending.promote((yield* Database.Service).db, events, sessionID, "steer")
       const assistantMessageID = SessionMessage.ID.create()
       yield* events.publish(SessionEvent.Step.Started, {
         sessionID,
@@ -3219,7 +3219,7 @@ describe("SessionRunnerLLM", () => {
       const session = yield* setup
       const events = yield* EventV2.Service
       yield* admit(session, "Recover interrupted tool input")
-      yield* SessionPending.promoteSteers((yield* Database.Service).db, events, sessionID)
+      yield* SessionPending.promote((yield* Database.Service).db, events, sessionID, "steer")
       const assistantMessageID = SessionMessage.ID.create()
       yield* events.publish(SessionEvent.Step.Started, {
         sessionID,
@@ -4220,7 +4220,7 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
-  it.effect("retries a physical attempt without consuming the logical agent step", () =>
+  it.effect("retries a model call without consuming the logical agent step", () =>
     Effect.gen(function* () {
       const session = yield* setup
       const agents = yield* AgentV2.Service

packages/core/test/tool-subagent.test.ts

@@ -406,7 +406,7 @@ describe("SubagentTool", () => {
             },
           })
           const database = yield* Database.Service
-          yield* SessionPending.promoteSteers(database.db, events, parent.id)
+          yield* SessionPending.promote(database.db, events, parent.id, "steer")
           const synthetic = (yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic")
           expect(synthetic).toHaveLength(1)
           expect(synthetic[0]?.text).toContain(`<subagent id="${childID}" state="completed"`)