diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 506caffb658..03a66db99ab 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -819,7 +819,7 @@ const layer = Layer.effect( const admitted = yield* SessionInbox.admitCompaction(db, bus, { id: inputID, sessionID: input.sessionID, - delivery: input.delivery ?? "queue", + delivery: input.delivery ?? "steer", }).pipe( Effect.catchDefect((defect) => defect instanceof SessionInbox.LifecycleConflict diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 1872c7dbf5a..af6ed3e4555 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -179,8 +179,9 @@ const layer = Layer.effect( yield* settleStaleToolCalls(input.sessionID) while (true) { // Between-turn control items run under any drain scope: scope gates which user - // input may promote, not whether admitted housekeeping runs. Enqueue order still - // holds — a control item behind a queued prompt is not the next eligible item. + // input may promote, not whether admitted housekeeping runs. Steered control + // items go ahead of any queued input; only a queue-delivered control item + // parked behind a queued prompt is not the next eligible item. if (yield* runPendingCompaction(input.sessionID, "input")) { force = false continue @@ -216,15 +217,23 @@ const layer = Layer.effect( let promotable: SessionInbox.Promotable = continuation ? "steer" : drainPromotable let step = continuation?.step ?? 1 let next = continuation + // The drain admitted this work, so the first step always runs — even after a + // control item consumed at this boundary (unlike drain's one-shot force). + let first = true + // Every boundary has the same shape: control items first, then one exit decision, + // then the model. The turn continues only while the first step, a continuation, or + // steer input is owed. Deciding after control items means consuming the last + // steered compaction ends the turn instead of issuing an input-free model call. while (true) { if (yield* runPendingCompaction(sessionID, "steer")) continue if (yield* runPendingMove(sessionID, "steer")) return { type: "moved" as const, continuation: next } - const result = yield* runStep(sessionID, promotable, step) - next = result.needsContinuation ? { step: result.step + 1 } : undefined - if (!result.needsContinuation && !(yield* SessionInbox.has(db, sessionID, "steer"))) + if (!first && !next && !(yield* SessionInbox.has(db, sessionID, "steer"))) return { type: "complete" as const } + const result = yield* runStep(sessionID, promotable, step) + first = false promotable = "steer" step = result.step + 1 + next = result.needsContinuation ? { step } : undefined } }) diff --git a/packages/core/test/session-compact.test.ts b/packages/core/test/session-compact.test.ts index ae4ac575ea1..0eeea9fec4c 100644 --- a/packages/core/test/session-compact.test.ts +++ b/packages/core/test/session-compact.test.ts @@ -106,13 +106,13 @@ describe("Session.compact", () => { expect(second.id).toBe(first.id) expect(requests).toHaveLength(0) expect(yield* session.inbox(created.id)).toEqual([ - expect.objectContaining({ id: first.id, type: "compaction", delivery: "queue" }), + expect.objectContaining({ id: first.id, type: "compaction", delivery: "steer" }), ]) expect((yield* session.context(created.id)).find((message) => message.id === first.id)).toBeUndefined() - const steered = yield* session.create({ location }) - const steer = yield* session.compact({ sessionID: steered.id, delivery: "steer" }) - expect(steer).toMatchObject({ type: "compaction", delivery: "steer" }) + const queued = yield* session.create({ location }) + const queue = yield* session.compact({ sessionID: queued.id, delivery: "queue" }) + expect(queue).toMatchObject({ type: "compaction", delivery: "queue" }) }), ) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index be66da5de8d..a836edaa448 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -2115,7 +2115,7 @@ describe("SessionRunnerLLM", () => { const active = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* stream.started - const first = yield* session.compact({ sessionID }) + const first = yield* session.compact({ sessionID, delivery: "queue" }) expect(yield* SessionInbox.find((yield* Database.Service).db, first.id)).toMatchObject({ id: first.id, }) @@ -2232,6 +2232,68 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("runs manual compaction at the next step boundary before queued prompts", () => + Effect.gen(function* () { + const session = yield* setup + currentModel = recoveryModel + const stream = yield* TestLLM.gate + yield* TestLLM.push( + TestLLM.text("Active complete", "text-active-steer-compact"), + [LLMEvent.textDelta({ id: "summary", text: "durable summary" })], + TestLLM.text("Queue complete", "text-queue-after-compact"), + ) + yield* admit(session, "Active work") + const active = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* stream.started + + const compaction = yield* session.compact({ sessionID }) + yield* session.prompt({ sessionID, text: "Queued prompt", delivery: "queue", resume: false }) + yield* stream.release + yield* Fiber.join(active) + + // Steer-delivered compaction runs at the boundary after the active step, ahead of + // the queued prompt, and consuming it does not trigger an input-free model call. + expect(requests).toHaveLength(3) + expect(userTexts(requests[1])[0]).toContain("Create a new anchored summary") + expect(userTexts(requests[2])).toContain("Queued prompt") + expect(yield* SessionInbox.find((yield* Database.Service).db, compaction.id)).toBeUndefined() + expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({ + type: "compaction", + status: "completed", + summary: "durable summary", + }) + }), + ) + + it.effect("runs manual compaction before the continuation of an active tool turn", () => + Effect.gen(function* () { + const session = yield* setup + currentModel = recoveryModel + const stream = yield* TestLLM.gate + yield* TestLLM.push( + TestLLM.tool("call-active", "echo", { text: "active" }), + [LLMEvent.textDelta({ id: "summary", text: "durable summary" })], + TestLLM.text("Continued", "text-continued-after-compact"), + ) + yield* admit(session, "Active work") + const active = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* stream.started + + const compaction = yield* session.compact({ sessionID }) + yield* stream.release + yield* Fiber.join(active) + + // The compaction summary is requested before the tool turn's continuation step. + expect(requests).toHaveLength(3) + expect(userTexts(requests[1])[0]).toContain("Create a new anchored summary") + expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({ + type: "compaction", + status: "completed", + summary: "durable summary", + }) + }), + ) + it.effect("preserves provider errors from manual compaction", () => Effect.gen(function* () { const session = yield* setup diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 17aad57cf1d..2062c600384 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -440,7 +440,8 @@ export const makeSessionGroup = (sessionLo OpenApi.annotations({ identifier: "v2.session.compact", summary: "Compact session", - description: "Queue a durable session compaction request.", + description: + "Durably admit a session compaction request. Steers by default: it runs at the next step boundary instead of waiting behind queued prompts.", }), ), )