fix(core): steer manual compaction by default (#43724)

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

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
       }
     })
 

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" })
     }),
   )
 

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

packages/protocol/src/groups/session.ts

@@ -440,7 +440,8 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(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.",
           }),
         ),
     )