diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts
index 9656228cf8d..cd0ea5913e1 100644
--- a/packages/core/src/session/runner/llm.ts
+++ b/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 })
}),
)
diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts
index f8257111002..657d5edbb70 100644
--- a/packages/core/test/session-runner.test.ts
+++ b/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("\nEntry summary\n")
+ 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("\nOverflow summary\n")
+ 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