diff --git a/AGENTS.md b/AGENTS.md index bbf855df640..56c2c23ba72 100644 --- a/AGENTS.md +++ b/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. diff --git a/packages/core/src/session/inbox.ts b/packages/core/src/session/inbox.ts index be8581ad876..727389c9644 100644 --- a/packages/core/src/session/inbox.ts +++ b/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 + }), + ) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 1000f0c7723..c83b462aeed 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/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 } }), ), diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index f77849a2ecc..7f4baea3c24 100644 --- a/packages/core/test/session-runner.test.ts +++ b/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() + const release = yield* Deferred.make() + 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( diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 87f214f2bb4..2429392b479 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/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 | undefined + function Probe() { + client = useClient() + rows = createSessionRows(() => sessionID) + return + } + const app = await testRender(() => ( + + + + + + + + + + )) + 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" diff --git a/packages/www/src/docs/content/compaction.mdx b/packages/www/src/docs/content/compaction.mdx index 7d7ff3c30c3..31a7bb20fcc 100644 --- a/packages/www/src/docs/content/compaction.mdx +++ b/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