From cf347cd5e41d6fdab2c34df59ae021cb39c2e8d4 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 26 Aug 2026 15:25:13 -0400 Subject: [PATCH] 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 | 263 ++++++++++------------ packages/core/test/session-runner.test.ts | 226 +++++++++++++++++++ 2 files changed, 349 insertions(+), 140 deletions(-) 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