fix(core): prioritize compaction over pending steers (#47340)

AGENTS.md

@@ -183,7 +183,7 @@ const table = sqliteTable("session", {
 - Keep `SessionRunner`, model resolution, tool registry, permissions, and filesystem Location-scoped. Omitted `Location.workspaceID` means implicit-local placement; explicit workspace identity remains reserved for future placement semantics.
 - Preserve one explicit `llm.stream(request)` call per Physical Attempt and reload projected history before durable continuation. A logical Step may use generic pre-output retries, one full-context retry after continuation rejection, incomplete-stream continuation, or one overflow-compaction rebuild. Generic retries retain the logical step number and do not consume another agent-step allowance. Do not delegate orchestration to an in-memory tool loop.
 - Keep local Session drains process-local until clustering is implemented. `SessionRunCoordinator` joins explicit same-Session resumes, coalesces prompt wakeups, and allows different Sessions to run concurrently. A write-ahead execution claim marks a process-local busy period for restart recovery: terminal completion, failure, or user interruption releases it, while shutdown interruption and process death preserve it. Startup recovery resumes claimed top-level Sessions with durable per-execution attempt accounting. The claim is a recovery marker, not clustered ownership, fencing, or an exactly-once guarantee.
-- Keep delivery vocabulary explicit. Prompts steer by default. Steers deliver in enqueue order at safe step boundaries, stopping before compaction or move control items. At an idle boundary, steers take priority; otherwise exactly one queued item delivers before the runner reevaluates continuation. Inbox items may be cancelled or changed between queue and steer before delivery. Promoting new user input resets the selected agent's step allowance; a batch of steers resets it once.
+- Keep delivery vocabulary explicit. Prompts steer by default. At safe step boundaries, steered compaction takes priority up to the first steered move control; other steers retain enqueue order. At an idle boundary, steers take priority; otherwise exactly one queued item delivers before the runner reevaluates continuation. Inbox items may be cancelled or changed between queue and steer before delivery. Promoting new user input resets the selected agent's step allowance; a batch of steers resets it once.
 - One step is one logical LLM call; its durable record covers only the model-visible span. Do not write "provider turn", and do not use bare "turn" for a single call: "turn" is reserved for the future assistant-turn unit containing all steps from prompt promotion until the session would go idle.
 - Keep event replay ownership separate from clustered Session execution ownership.
 - Keep the Instructions algebra and built-ins in `src/instructions`; keep instruction producers with their observed domains, and keep Session History selection plus `InstructionState` and `InstructionEntry` persistence Session-owned. `InstructionDiscovery` observes ambient global and upward-project instructions. The runner composes built-ins, discovery, guidance, and entries explicitly in `loadInstructions`; there is no instruction registry.

packages/core/src/session/inbox.ts

@@ -409,19 +409,17 @@ export const nextPromotable = Effect.fn("SessionInbox.nextPromotable")(function*
   sessionID: SessionSchema.ID,
   promotable: Promotable,
 ) {
-  const next = (delivery: Delivery) =>
-    db
-      .select()
-      .from(SessionInboxTable)
-      .where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, delivery)))
-      .orderBy(asc(SessionInboxTable.enqueued_seq))
-      .limit(1)
-      .get()
-      .pipe(Effect.orDie)
-  const steer = yield* next("steer")
+  const steer = (yield* pendingSteers(db, sessionID))[0]
   if (steer) return fromRow(steer)
   if (promotable !== "input") return undefined
-  const queued = yield* next("queue")
+  const queued = yield* db
+    .select()
+    .from(SessionInboxTable)
+    .where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "queue")))
+    .orderBy(asc(SessionInboxTable.enqueued_seq))
+    .limit(1)
+    .get()
+    .pipe(Effect.orDie)
   return queued ? fromRow(queued) : undefined
 })
 
@@ -490,9 +488,10 @@ const publish = Effect.fn("SessionInbox.publish")(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.
+ * Promotes pending input into visible messages and returns the promoted count,
+ * or undefined when the runner must first handle a pending control.
+ * Steered compaction takes priority over pending prompts, without crossing a move.
+ * Only the "input" scope may fall through to one queued input.
  */
 export const promote = Effect.fn("SessionInbox.promote")(function* (
   db: DatabaseService,
@@ -506,6 +505,7 @@ export const promote = Effect.fn("SessionInbox.promote")(function* (
       const steers = yield* pendingSteers(db, sessionID)
       if (steers.length > 0 || scope === "steer") {
         const control = steers.findIndex((row) => row.type === "compaction" || row.type === "move")
+        if (control === 0) return undefined
         return yield* publish(db, bus, sessionID, control === -1 ? steers : steers.slice(0, control))
       }
 
@@ -518,6 +518,7 @@ export const promote = Effect.fn("SessionInbox.promote")(function* (
         .get()
         .pipe(Effect.orDie)
       if (!queued) return 0
+      if (queued.type === "compaction" || queued.type === "move") return undefined
       const promoted = yield* publish(db, bus, sessionID, [queued])
       const arrivedSteers = yield* pendingSteers(db, sessionID)
       const control = arrivedSteers.findIndex((row) => row.type === "compaction" || row.type === "move")
@@ -536,4 +537,14 @@ const pendingSteers = (db: DatabaseService, sessionID: SessionSchema.ID) =>
     .where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "steer")))
     .orderBy(asc(SessionInboxTable.enqueued_seq))
     .all()
-    .pipe(Effect.orDie)
+    .pipe(
+      Effect.orDie,
+      Effect.map((rows) => {
+        // A move changes the context's Location: never pull compaction across it.
+        // Within that boundary, compact before promoting even earlier steers so
+        // their text stays verbatim after the checkpoint, not inside its summary.
+        const control = rows.findIndex((row) => row.type === "compaction" || row.type === "move")
+        if (control > 0 && rows[control].type === "compaction") rows.unshift(...rows.splice(control, 1))
+        return rows
+      }),
+    )

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

@@ -147,7 +147,7 @@ const layer = Layer.effect(
               }
               if (!force && !continuing && (!pending || (pending.delivery === "queue" && promotable === "steer")))
                 return DrainResult.Complete()
-              return yield* restore(
+              const ready = yield* restore(
                 Effect.gen(function* () {
                   const selected = yield* prepareContext(sessionID)
                   const promoted = yield* SessionInbox.promote(
@@ -156,6 +156,8 @@ const layer = Layer.effect(
                     sessionID,
                     entering && !continuing ? promotable : "steer",
                   )
+                  // A control admitted during context preparation owns this boundary.
+                  if (promoted === undefined) return undefined
                   if (promoted > 0 && !selected.session.parentID && SessionTitle.isUntitled(selected.session))
                     yield* FiberMap.run(titles, sessionID, title.generate(sessionID), {
                       onlyIfMissing: true,
@@ -164,6 +166,7 @@ const layer = Layer.effect(
                   return { _tag: "Ready" as const, context: yield* context.load(selected) }
                 }),
               )
+              if (ready) return ready
             }
           }),
         ),

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

@@ -1925,6 +1925,185 @@ describe("SessionRunnerLLM", () => {
     ).toEqual(["Replacement context"])
   })
 
+  for (const order of ["before", "between", "after"] as const) {
+    scenario(`prioritizes manual compaction admitted ${order} two steers at the safe boundary`, function* (s) {
+      s.currentModel = recoveryModel
+      yield* s.llm.push(
+        TestLLM.text("Active complete", "active"),
+        TestLLM.text("## Objective\n- Active work checkpoint", "summary"),
+        TestLLM.text("Steers complete", "steers"),
+      )
+      yield* s.admit("Active work")
+      const active = yield* s.resumePaused
+      const compactID = SessionMessage.ID.create()
+      if (order === "before") yield* s.session.compact({ sessionID, id: compactID })
+      const first = yield* s.admit("STEER_A")
+      if (order === "between") yield* s.session.compact({ sessionID, id: compactID })
+      const second = yield* s.admit("STEER_B")
+      if (order === "after") yield* s.session.compact({ sessionID, id: compactID })
+      expect((yield* s.session.compact({ sessionID })).id).toBe(compactID)
+
+      expect(s.requests).toHaveLength(1)
+      expect(yield* s.inbox).toHaveLength(3)
+      expect((yield* s.messages).some((message) => message.type === "compaction")).toBe(false)
+      yield* active.finish
+
+      expect(userTexts(s.requests[1]).at(-1)).toContain("Summarize only the history shown")
+      expect(s.requests).toHaveLength(3)
+      expect(userTexts(s.requests[1])).not.toContain("STEER_A")
+      expect(userTexts(s.requests[1])).not.toContain("STEER_B")
+      expect(userTexts(s.requests[2]).slice(-2)).toEqual(["STEER_A", "STEER_B"])
+      expect(yield* s.inbox).toEqual([])
+      expect((yield* s.messages).filter((message) => message.id === compactID)).toMatchObject([
+        { type: "compaction", status: "completed" },
+      ])
+      expect((yield* s.context).filter((message) => message.type === "user").map((message) => message.id)).toEqual([
+        first.id,
+        second.id,
+      ])
+      // An advisory drain must not redeliver either steer or rerun compaction.
+      const runner = yield* SessionRunner.Service
+      yield* runner.drain({ sessionID, force: false })
+      expect(s.requests).toHaveLength(3)
+    })
+  }
+
+  scenario("waits for active tools before prioritizing compaction over pending steers", function* (s) {
+    yield* s.llm.push(
+      TestLLM.tool("call-active", "echo", { text: "active" }),
+      TestLLM.text("## Objective\n- Tool work checkpoint", "summary"),
+      TestLLM.text("Steers complete", "steers"),
+    )
+    yield* s.admit("Active work")
+    const tools = yield* s.blockTools()
+    const run = yield* s.resume.pipe(Effect.forkChild)
+    yield* tools.started
+    yield* s.admit("STEER_A")
+    yield* s.admit("STEER_B")
+    const compact = yield* s.session.compact({ sessionID })
+    expect(s.requests).toHaveLength(1)
+    expect((yield* s.messages).some((message) => message.id === compact.id)).toBe(false)
+    yield* tools.release
+    yield* Fiber.join(run)
+
+    expect(s.requests).toHaveLength(3)
+    expect(userTexts(s.requests[1]).at(-1)).toContain("Summarize only the history shown")
+    expect(s.requests[1].messages.some((message) => message.role === "tool")).toBe(true)
+    expect(userTexts(s.requests[2]).slice(-2)).toEqual(["STEER_A", "STEER_B"])
+    expect(yield* s.inbox).toEqual([])
+  })
+
+  scenario("rechecks compaction admitted during boundary context preparation", function* (s) {
+    yield* s.runPrompt("Earlier work")
+    yield* s.admit("STEER_A")
+    yield* s.admit("STEER_B")
+    const preparing = yield* Deferred.make<void>()
+    const release = yield* Deferred.make<void>()
+    s.systemLoadHook = Deferred.succeed(preparing, undefined).pipe(Effect.andThen(Deferred.await(release)))
+    yield* s.llm.push(
+      TestLLM.text("## Objective\n- Earlier work checkpoint", "summary"),
+      TestLLM.text("Steers complete", "steers"),
+    )
+    const run = yield* s.resume.pipe(Effect.forkChild)
+    yield* Deferred.await(preparing)
+    yield* s.session.compact({ sessionID })
+    yield* Deferred.succeed(release, undefined)
+    yield* Fiber.join(run)
+    expect(s.requests).toHaveLength(3)
+    expect(userTexts(s.requests[1]).at(-1)).toContain("Summarize only the history shown")
+    expect(userTexts(s.requests[2]).slice(-2)).toEqual(["STEER_A", "STEER_B"])
+    expect(yield* s.inbox).toEqual([])
+  })
+
+  for (const outcome of ["cancelled", "failed"] as const) {
+    scenario(`preserves both earlier steers when prioritized compaction is ${outcome}`, function* (s) {
+      yield* s.llm.push(TestLLM.text("Active complete", "active"))
+      yield* s.admit("Active work")
+      const active = yield* s.resumePaused
+      const first = yield* s.admit("STEER_A")
+      const second = yield* s.admit("STEER_B")
+      const compact = yield* s.session.compact({ sessionID })
+      if (outcome === "cancelled") yield* s.session.cancelInbox({ sessionID, inboxID: compact.id })
+      if (outcome === "failed") yield* s.llm.push([LLMEvent.providerError({ message: "summary unavailable" })])
+      yield* s.llm.push(TestLLM.text("Steers complete", "steers"))
+      yield* active.finish
+
+      expect(s.requests).toHaveLength(outcome === "cancelled" ? 2 : 3)
+      if (outcome === "failed") {
+        expect(userTexts(s.requests[1]).at(-1)).toContain("Summarize only the history shown")
+        expect((yield* s.messages).find((message) => message.id === compact.id)).toMatchObject({
+          status: "failed",
+          error: { type: "provider.error", message: "summary unavailable" },
+        })
+      }
+      if (outcome === "cancelled") expect((yield* s.messages).some((message) => message.id === compact.id)).toBe(false)
+      expect(userTexts(s.requests[s.requests.length - 1]).slice(-2)).toEqual(["STEER_A", "STEER_B"])
+      expect(
+        (yield* s.context)
+          .filter((message) => message.id === first.id || message.id === second.id)
+          .map((message) => message.id),
+      ).toEqual([first.id, second.id])
+      expect(yield* s.inbox).toEqual([])
+    })
+  }
+
+  scenario("keeps steers durable across interrupted priority compaction and replay", function* (s) {
+    yield* s.runPrompt("Earlier work")
+    const first = yield* s.admit("STEER_A")
+    const second = yield* s.admit("STEER_B")
+    yield* s.llm.push(TestLLM.text("## Objective\n- Interrupted checkpoint", "summary"))
+    const summary = yield* s.llm.gate
+    const compact = yield* s.session.compact({ sessionID })
+    yield* summary.started
+    expect(userTexts(s.requests[1]).at(-1)).toContain("Summarize only the history shown")
+    expect((yield* s.inbox).map((item) => item.id)).toEqual([first.id, second.id])
+    yield* s.session.interrupt(sessionID)
+    yield* s.session.wait(sessionID)
+    yield* summary.release
+    expect((yield* s.messages).find((message) => message.id === compact.id)).toMatchObject({ status: "failed" })
+    yield* replaySessionProjection(sessionID)
+    expect((yield* s.inbox).map((item) => item.id)).toEqual([first.id, second.id])
+
+    yield* s.llm.push(TestLLM.text("Recovered steers", "steers"))
+    yield* s.resume
+    expect(s.requests).toHaveLength(3)
+    expect(userTexts(s.requests[2]).slice(-2)).toEqual(["STEER_A", "STEER_B"])
+    expect(yield* s.inbox).toEqual([])
+  })
+
+  scenario("does not pull compaction across an earlier move", function* (s) {
+    yield* s.admit("STEER_A")
+    yield* s.sessionInbox.admit({
+      id: SessionMessage.ID.create(),
+      sessionID,
+      item: {
+        type: "move",
+        payload: {
+          location: Location.Ref.make({ directory: AbsolutePath.make("/project") }),
+          projectID: Project.ID.global,
+        },
+        delivery: "steer",
+      },
+    })
+    yield* s.admit("STEER_B")
+    yield* s.sessionInbox.admitCompaction({ id: SessionMessage.ID.create(), sessionID, delivery: "steer" })
+    yield* s.llm.push(
+      TestLLM.text("First steer complete", "first"),
+      TestLLM.text("## Objective\n- Source work checkpoint", "summary"),
+      TestLLM.text("Second steer complete", "second"),
+    )
+    yield* s.resume
+    expect(s.requests).toHaveLength(3)
+    expect(userTexts(s.requests[0])).toEqual(["STEER_A"])
+    expect(userTexts(s.requests[1]).at(-1)).toContain("Summarize only the history shown")
+    expect(userTexts(s.requests[2]).at(-1)).toBe("STEER_B")
+    expect(
+      (yield* recordedEventTypes(sessionID)).filter(
+        (type) => type === "session.moved.1" || type === "session.compaction.started.1",
+      ),
+    ).toEqual(["session.moved.1", "session.compaction.started.1"])
+  })
+
   scenario("runs steers before queued compaction and later queued input", function* (s) {
     s.currentModel = recoveryModel
     yield* s.llm.push(

packages/tui/test/cli/tui/data.test.tsx

@@ -1579,6 +1579,89 @@ test("tracks session status from active sessions and execution events", async ()
   }
 })
 
+test.each(["before", "between", "after"])("shows compaction admitted %s steers in execution order", async (order) => {
+  const events = createEventStream()
+  const sessionID = "session-compaction-priority"
+  const calls = createFetch((url) => {
+    if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} })
+    return undefined
+  }, events)
+  let rows: SessionRow[] = []
+  let client: ReturnType<typeof useClient> | undefined
+  function Probe() {
+    client = useClient()
+    rows = createSessionRows(() => sessionID)
+    return <box />
+  }
+  const app = await testRender(() => (
+    <TestTuiContexts>
+      <ClientProvider api={createApi(calls.fetch)}>
+        <ProjectProvider>
+          <DataProvider>
+            <Probe />
+          </DataProvider>
+        </ProjectProvider>
+      </ClientProvider>
+    </TestTuiContexts>
+  ))
+  const admissions =
+    order === "before" ? ["compact", "a", "b"] : order === "between" ? ["a", "compact", "b"] : ["a", "b", "compact"]
+  try {
+    await wait(() => client?.connection.status() === "connected")
+    admissions.forEach((id, index) =>
+      emitEvent(events, {
+        id: `evt_admit_${id}`,
+        created: index + 1,
+        type: "session.inbox.enqueued",
+        durable: durable(sessionID, index + 1),
+        data: {
+          sessionID,
+          inboxID: id,
+          item:
+            id === "compact"
+              ? { type: "compaction", payload: {}, delivery: "steer" }
+              : { type: "user", payload: { text: `STEER_${id.toUpperCase()}` }, delivery: "steer" },
+        },
+      }),
+    )
+    await wait(() => rows.length === 3)
+    expect(rows).toEqual([
+      { type: "compaction-queued", inboxID: "compact" },
+      { type: "message", messageID: "a" },
+      { type: "message", messageID: "b" },
+    ])
+    emitEvent(events, {
+      id: "evt_compaction_started",
+      created: 4,
+      type: "session.compaction.started",
+      durable: durable(sessionID, 4),
+      data: { sessionID, reason: "manual", recent: "", inputID: "compact" },
+    })
+    await wait(() => rows[0]?.type === "message")
+    expect(rows).toEqual(["compact", "a", "b"].map((messageID) => ({ type: "message", messageID })))
+    emitEvent(events, {
+      id: "evt_compaction_ended",
+      created: 5,
+      type: "session.compaction.ended",
+      durable: durable(sessionID, 5),
+      data: { sessionID, reason: "manual", text: "## Objective\n- Checkpoint", recent: "" },
+    })
+    for (const [index, id] of ["a", "b"].entries()) {
+      emitEvent(events, {
+        id: `evt_deliver_${id}`,
+        created: index + 6,
+        type: "session.inbox.delivered",
+        durable: durable(sessionID, index + 6),
+        data: { sessionID, inboxID: id },
+      })
+    }
+    await app.renderOnce()
+    expect(rows).toEqual(["compact", "a", "b"].map((messageID) => ({ type: "message", messageID })))
+  } finally {
+    app.renderer.destroy()
+  }
+})
+
 test("restores queued compaction from durable pending input", async () => {
   const events = createEventStream()
   const sessionID = "session-compaction-queued"

packages/www/src/docs/content/compaction.mdx

@@ -37,11 +37,11 @@ Manual compaction is available through session interfaces. See the generated [AP
 operation.
 
 A manual request is durably admitted and wakes the session runner. It can
-compact short histories that would not trigger automatic compaction. If the
-session is busy, compaction runs at the next safe drain boundary before later
-steered or queued prompts are promoted. Repeated requests while one is pending
+compact short histories that would not trigger automatic compaction. By default,
+compaction runs at the next safe step boundary before pending steered or queued
+prompts, even if they were submitted first. Repeated requests while one is pending
 coalesce into that pending request. Whether compaction completes or fails, the
-barrier is then settled so later prompts can proceed.
+barrier is then settled so pending prompts can proceed.
 
 The server operation returns the admitted compaction input; it does not wait
 for summary generation. Clients can then wait for the session or follow the