diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 766334fcc..c70d1415b 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -4,6 +4,7 @@ ### Fixed +- The system baseline written after a compaction now leads the model context when the kept history includes an assistant message, as does a first baseline written after earlier answers, instead of following the new input; Anthropic and OpenAI Codex requests sent it as a mid-conversation system message without an initial prompt or tool set. Other system messages after an assistant message stay in place ([#10904](https://github.com/earendil-works/pi/issues/10904)). - `toBase64`, used by `read` and the Photon image processor, no longer takes about 140 MB of heap to encode a 3 MiB image, which exceeded the 128 MB memory limit of a Cloudflare Durable Object; it now builds a binary string 8 KiB at a time and encodes it with `btoa` ([#10923](https://github.com/earendil-works/pi/issues/10923)). ## [1.2.0] - 2026-10-12 diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 023f26243..efb1842c0 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -281,13 +281,25 @@ Context derivation: fork cut or at an interrupted tail. Drop tool results with no preceding call. 9. Exclude model-less entries and assistant messages with `aborted`, `error`, or `deferred` stop reasons from future provider requests. -10. Move a system message that only user messages precede to the front. A - run's input is committed before generation renders the system prompt, so a - transcript, or the range after a compaction or reset, otherwise starts with - the input. Providers treat only a leading system message as the initial - prompt and tool set; without one, a later tool change rewrites the - request's tool list and invalidates the whole prompt cache. Stored entries - and `contributions` keep the committed order. +10. Move the first system message to the front when only user messages + precede it, or when it is the baseline's: the first `pi.system` entry after + `H`, unless an edit targets it. Generation writes it with every section and + tool in effect before the first request after a compaction or reset, and + omits the retained `pi.system` entries. Without `H`, the first `pi.system` + entry is the baseline unless an earlier entry carries a system message, in + its messages or an edit's replacement, even one an edit omits later: only + then did generation write it with every section and tool in effect, before + the first request or once an agent without any gets one. A run's input + is committed before generation renders the system prompt, so the context + otherwise starts with the input, or with the kept history, assistant + messages included. Providers treat only a leading system message as the + initial prompt and tool set; without one, a later tool change rewrites the + request's tool list and invalidates the whole prompt cache. Only the first + system message moves, so replay gives the same sections and tools. Any other + system message with an assistant message before it stays in place, since + moving it would change what it means; a retained system message of another + kind ahead of the baseline therefore keeps the baseline from leading. + Stored entries and `contributions` keep the committed order. Views carry raw active entries. UI reduction and model-context reduction are separate consumers. Older stored history is available through the owning diff --git a/packages/durable/src/harness/context.ts b/packages/durable/src/harness/context.ts index 95f00adec..c818b3cb5 100644 --- a/packages/durable/src/harness/context.ts +++ b/packages/durable/src/harness/context.ts @@ -1,5 +1,6 @@ import type { Context } from "@earendil-works/chord"; import type { AssistantMessage, Message, ToolCall, ToolResultMessage } from "@earendil-works/pi-ai"; +import { SystemEntry } from "../entries.ts"; import { idFromNumber } from "../ids.ts"; import type { SessionImpl } from "../session/session.ts"; import type { ContextEdit, ConversationId, Cursor, EntryId, EntryQuery, EntryRecord, Storage } from "../types.ts"; @@ -57,8 +58,13 @@ export type ContextRange = { readonly settled: readonly Message[]; /** Contributed messages from the last assistant message on, before tool result ordering. */ readonly open: readonly Message[]; + /** The baseline's system message; see `findBaseline()`. */ + readonly baseline: Baseline; }; +/** A baseline's system message, `null` when the range cannot have one, or `undefined` while a later entry may be one. */ +type Baseline = Message | null | undefined; + /** * `readContext()` that reuses `previous`, an earlier range of the same conversation: with the same head marker, only * entries after its tail are scanned. Returns the view and the range to pass to the next read. A range outlives the @@ -142,9 +148,10 @@ function deriveRange(bounds: ContextBounds, entries: readonly EntryRecord[]): Co const active = selectActive(bounds.head, entries); const contributions = active.map((entry) => contribute(entry, edits.get(entry.id))); const { settled, open } = settle([], contributions.flat()); - const messages = leadWithSystem([...settled, ...orderToolResults(open)]); + const baseline = findBaseline(bounds.head, active); + const messages = leadWithSystem([...settled, ...orderToolResults(open)], baseline); const view = { head: bounds.head, entries: active, contributions, messages }; - return { bounds, entries, view, edited: new Set(edits.keys()), settled, open }; + return { bounds, entries, view, edited: new Set(edits.keys()), settled, open, baseline }; } /** @@ -158,25 +165,58 @@ function extendRange(previous: ContextRange, bounds: ContextBounds, added: reado } const contributions = added.map((entry) => contribute(entry, undefined)); const { settled, open } = settle(previous.settled, [...previous.open, ...contributions.flat()]); + const baseline = previous.baseline !== undefined ? previous.baseline : findBaseline(bounds.head, added); const view = { head: bounds.head, entries: [...previous.view.entries, ...added], contributions: [...previous.view.contributions, ...contributions], - messages: leadWithSystem([...settled, ...orderToolResults(open)]), + messages: leadWithSystem([...settled, ...orderToolResults(open)], baseline), }; - return { bounds, entries, view, edited: previous.edited, settled, open }; + return { bounds, entries, view, edited: previous.edited, settled, open, baseline }; } /** - * Move a system message that only user messages precede to the front. A run's input is committed before generation - * renders the system prompt, so a transcript, or the range after a compaction or reset, starts with user messages - * followed by the baseline system message. Providers treat only a leading system message as the initial prompt and tool - * set; without it, a later tool change rewrites the request's tool list and invalidates the whole prompt cache. + * The baseline among `entries`, oldest first: the system message of the first `pi.system` entry after the head + * marker, or of the transcript without one, when generation wrote it with every section and tool in effect. After a + * head marker it always does, and omits the retained `pi.system` entries. Without one it does when it saw no system + * message before, which holds unless an earlier entry carries one, even if an edit omits it later. An edit that targets + * the baseline contributes other messages, or none, so none leads; a later entry does not take its place. */ -function leadWithSystem(messages: Message[]): Message[] { - const index = messages.findIndex((message) => message.role !== "user"); - if (index <= 0 || messages[index]!.role !== "system") return messages; - return [messages[index]!, ...messages.slice(0, index), ...messages.slice(index + 1)]; +function findBaseline(head: ContextBounds["head"], entries: readonly EntryRecord[]): Baseline { + for (const entry of entries) { + if (head !== undefined && entry.id <= head.id) continue; + if (SystemEntry.is(entry)) return entry.model?.find((message) => message.role === "system") ?? null; + if (head === undefined && carriesSystemMessage(entry)) return null; + } + return undefined; +} + +/** Whether an entry's own messages, or the replacement messages of its edits, include a system message. */ +function carriesSystemMessage(entry: EntryRecord): boolean { + const isSystem = (message: Message) => message.role === "system"; + return ( + (entry.model?.some(isSystem) ?? false) || + (entry.edits?.some((edit) => edit.action === "replace" && edit.messages.some(isSystem)) ?? false) + ); +} + +/** + * Move the first system message to the front when only user messages precede it, or when it is the baseline's. A run's + * input is committed before generation renders the system prompt, so a transcript starts with the input followed by + * the baseline, and the range after a compaction with the kept history, assistant messages included. Providers treat + * only a leading system message as the initial prompt and tool set; without it, a later tool change rewrites the + * request's tool list and invalidates the whole prompt cache. Only the first system message moves, so replaying + * system messages in order gives the same sections and tools. Any other system message, before an assistant message, + * stays where it was committed: moving it would change what it means to the model. + */ +function leadWithSystem(messages: Message[], baseline: Baseline): Message[] { + const index = messages.findIndex((message) => message.role === "system"); + if (index <= 0) return messages; + const system = messages[index]!; + if (system !== baseline && messages.slice(0, index).some((message) => message.role !== "user")) { + return messages; + } + return [system, ...messages.slice(0, index), ...messages.slice(index + 1)]; } /** One active entry's model messages after its edit and excluded stop reasons, before tool result ordering. */ diff --git a/packages/durable/test/harness-compaction.test.ts b/packages/durable/test/harness-compaction.test.ts index 9f081e64f..da5d21a51 100644 --- a/packages/durable/test/harness-compaction.test.ts +++ b/packages/durable/test/harness-compaction.test.ts @@ -418,9 +418,9 @@ describe("manual compaction", () => { expect(usage.input).toBeGreaterThan(usageBefore.input); await turn(chat, "next", "done"); const agent = chat.faux.agentRequests.at(-1)!.messages; - // The next request: the summary, the kept turn, the new input, then one complete system baseline. - expect(agent.map((message) => message.role)).toEqual(["user", "user", "assistant", "user", "system"]); - expect(agent[4]).toMatchObject({ sections: { preamble: "You are helpful." } }); + // The next request: one complete system baseline leading the summary, the kept turn, and the new input (#10904). + expect(agent.map((message) => message.role)).toEqual(["system", "user", "user", "assistant", "user"]); + expect(agent[0]).toMatchObject({ sections: { preamble: "You are helpful." } }); await chat.harness.close(context); }); @@ -484,9 +484,10 @@ describe("manual compaction", () => { await result(chat, await chat.root.compact(undefined, context)); gate.resolve(); expect((await input.wait(context)).status).toBe("done"); - // The continuation request starts with the summary. + // The continuation request starts with the new baseline, then the summary, ahead of the kept tool call (#10904). const continuation = chat.faux.agentRequests.at(-1)!.messages; - expect(userText(continuation[0])).toContain("\nSUMMARY\n"); + expect(continuation[0]?.role).toBe("system"); + expect(userText(continuation[1])).toContain("\nSUMMARY\n"); expect((await kinds(chat.root)).slice(-4)).toEqual([ "pi.tool-result", "pi.compaction", @@ -1246,7 +1247,8 @@ describe("overflow compaction", () => { const [compaction] = (await allEntries(chat.root)).filter((record) => record.kind === "pi.compaction"); expect(compaction).toMatchObject({ data: { reason: "overflow" } }); const retry = chat.faux.agentRequests.at(-1)!.messages; - expect(userText(retry[0])).toContain("SUMMARY"); + expect(retry[0]?.role).toBe("system"); + expect(userText(retry[1])).toContain("SUMMARY"); expect(retry.some((message) => message.role === "assistant" && message.stopReason === "error")).toBe(false); await chat.harness.close(context); }); diff --git a/packages/durable/test/harness-context.test.ts b/packages/durable/test/harness-context.test.ts index d6e9cc326..f3a9ada0e 100644 --- a/packages/durable/test/harness-context.test.ts +++ b/packages/durable/test/harness-context.test.ts @@ -191,6 +191,178 @@ describe("conversation context", () => { expect(view.messages.map(describeMessage)).toEqual(["system:preamble", "user:handoff"]); }); + // #10904 + it("leads with the baseline after a compaction that keeps assistant messages", async () => { + const { root, message, append } = await setup(); + await message(user("old")); + const kept = await message(user("kept")); + await message(assistant("calling", { calls: ["a"] })); + await message(toolResult("a")); + await message(assistant("answer")); + await append({ kind: "summary", head: kept.id, model: [user("summary")] }); + await message(user("next")); + await message(system({ preamble: "Baseline." }), "pi.system"); + let view = await root.context(context); + expect(view.messages.map(describeMessage)).toEqual([ + "system:preamble", + "user:summary", + "user:kept", + "assistant:calling", + "result:a:result a", + "assistant:answer", + "user:next", + ]); + + // Later system messages stay in place. + await message(assistant("done")); + await message(system({ cwd: "/repo" }), "pi.system"); + view = await root.context(context); + expect(view.messages.map(describeMessage).slice(0, 2)).toEqual(["system:preamble", "user:summary"]); + expect(view.messages.map(describeMessage).slice(-2)).toEqual(["assistant:done", "system:cwd"]); + }); + + // #10904 + it("leads with a baseline written after the first answers, when the agent had no section or tool before", async () => { + const { root, message } = await setup(); + await message(user("first"), "pi.user"); + await message(assistant("answer"), "pi.assistant"); + await message(user("next"), "pi.user"); + await message(system({ plan: "Plan." }), "pi.system"); + const view = await root.context(context); + expect(view.messages.map(describeMessage)).toEqual([ + "system:plan", + "user:first", + "assistant:answer", + "user:next", + ]); + }); + + // #10904 + it("keeps system messages other than the baseline in place once an assistant message precedes them", async () => { + const { root, message, append } = await setup(); + await message(user("first")); + await message(assistant("answer")); + // Another kind's system message is positional; the baseline after it is not the first system message. + await message(system({ note: "From here on." }), "host.note"); + await message(system({ preamble: "Baseline." }), "pi.system"); + let view = await root.context(context); + expect(view.messages.map(describeMessage)).toEqual([ + "user:first", + "assistant:answer", + "system:note", + "system:preamble", + ]); + + // After a compaction that keeps the other kind's system message, the new baseline does not move ahead of it. + const kept = await message(user("kept")); + await message(assistant("kept answer")); + await message(system({ mode: "plan" }), "host.note"); + await append({ kind: "summary", head: kept.id, model: [user("summary")] }); + await message(user("next")); + await message(system({ preamble: "Baseline." }), "pi.system"); + view = await root.context(context); + expect(view.messages.map(describeMessage)).toEqual([ + "user:summary", + "user:kept", + "assistant:kept answer", + "system:mode", + "user:next", + "system:preamble", + ]); + }); + + // #10904 + it("does not lead with a first system entry written after another system message, even once that is omitted", async () => { + const { root, message, append } = await setup(); + const host = await message(system({ legacy: "Use the legacy format." }), "host.system"); + await message(user("first")); + await message(assistant("answer")); + // Generation wrote this entry against the host's message: it only removes a section, so it is no baseline. + await message(system({ legacy: null }), "pi.system"); + await append({ kind: "edit", edits: [{ target: host.id, action: "omit" }] }); + let view = await root.context(context); + expect(view.messages.map(describeMessage)).toEqual(["user:first", "assistant:answer", "system:legacy"]); + + // The same holds for a system message an edit put before it. + const other = await setup(); + const first = await other.message(user("first")); + await other.append({ + kind: "edit", + edits: [{ target: first.id, action: "replace", messages: [user("first"), system({ note: "Note." })] }], + }); + await other.message(assistant("answer")); + await other.message(system({ note: null }), "pi.system"); + await other.append({ kind: "edit", edits: [{ target: first.id, action: "replace", messages: [user("first")] }] }); + view = await other.root.context(context); + expect(view.messages.map(describeMessage)).toEqual(["user:first", "assistant:answer", "system:note"]); + }); + + // #10904 + it("does not lead with a retained, edited, or later system entry in place of the baseline", async () => { + const { root, message, append } = await setup(); + const kept = await message(user("kept")); + await message(assistant("answer")); + const retained = await message(system({ cwd: "/repo" }), "pi.system"); + // Before generation writes the baseline after the head marker, a retained system entry stays in place. + await append({ kind: "summary", head: kept.id, model: [user("summary")] }); + await message(user("next")); + let view = await root.context(context); + expect(view.messages.map(describeMessage)).toEqual([ + "user:summary", + "user:kept", + "assistant:answer", + "system:cwd", + "user:next", + ]); + + // The baseline omits the retained entry, as generation writes it, and leads. + const baseline = await append({ + kind: "pi.system", + model: [system({ preamble: "Baseline." })], + edits: [{ target: retained.id, action: "omit" }], + }); + await message(assistant("done")); + await message(system({ preamble: null }), "pi.system"); + view = await root.context(context); + expect(view.messages.map(describeMessage)).toEqual([ + "system:preamble", + "user:summary", + "user:kept", + "assistant:answer", + "user:next", + "assistant:done", + "system:preamble", + ]); + + // An omitted baseline leaves no message to lead; the delta after it is not a baseline and stays in place. + await append({ kind: "edit", edits: [{ target: baseline.id, action: "omit" }] }); + view = await root.context(context); + expect(view.messages.map(describeMessage)).toEqual([ + "user:summary", + "user:kept", + "assistant:answer", + "user:next", + "assistant:done", + "system:preamble", + ]); + + // A replacement of the baseline is not one either. + await append({ + kind: "edit", + edits: [{ target: baseline.id, action: "replace", messages: [system({ other: "Other." })] }], + }); + view = await root.context(context); + expect(view.messages.map(describeMessage)).toEqual([ + "user:summary", + "user:kept", + "assistant:answer", + "user:next", + "system:other", + "assistant:done", + "system:preamble", + ]); + }); + it("synthesizes missing tool results after a fork and drops results cut from their call", async () => { const { root, message } = await setup(); await message(user("go")); @@ -436,8 +608,10 @@ describe("conversation context", () => { return { kind: "message", model: [toolResult(id)] }; } if (pick < 9) { - count("system"); - return { kind: "message", model: [system({ s: `v${next++}` })] }; + // `pi.system` entries can be the baseline that leads the context (#10904). + const kind = random(2) === 0 ? "pi.system" : "message"; + count(kind === "pi.system" ? "baseline system" : "system"); + return { kind, model: [system({ s: `v${next++}` })] }; } if (pick < 10) { count("note"); @@ -469,7 +643,17 @@ describe("conversation context", () => { }, taskContext); for (let step = 0; step < 150; step++) { await append(root.id); - expect(await runtime.context(root.id, taskContext)).toEqual(await root.context(taskContext)); + const whole = await root.context(taskContext); + expect(await runtime.context(root.id, taskContext)).toEqual(whole); + const contributed = whole.contributions.flat(); + const first = contributed.findIndex((message) => message.role === "system"); + if ( + first > 0 && + whole.messages[0] === contributed[first] && + contributed.slice(0, first).some((message) => message.role === "assistant") + ) { + count("baseline past assistant"); + } if (step % 10 === 9) { // Overlapping reads, one with an earlier cutoff, share the kept range. const at = written[random(written.length)]!.id; @@ -514,12 +698,98 @@ describe("conversation context", () => { "result", "stray result", "system", + "baseline system", + "baseline past assistant", "note", ]; for (const category of [...categories, "omit", "replace", "head"]) expect(seen.get(category)).toBeGreaterThan(0); await harness.close(context); }); + // #10904 + it("finds the baseline among the entries an extended read adds", async () => { + const { harness, registry, root, first, scanned } = await countingSetup(); + const Steps = defineTask, { phase: "run" }, null>({ + name: "test.context-baseline", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (_task, runtime, taskContext) => { + const append = (conversationId: typeof root.id, draft: EntryDraft) => + runtime.commit(async (tx) => { + await tx.appendEntry(conversationId, draft); + return undefined; + }, taskContext); + const read = async (conversation: typeof root) => { + const before = scanned.rows; + const view = await runtime.context(conversation.id, taskContext); + const rows = scanned.rows - before; + expect(view).toEqual(await conversation.context(taskContext)); + return { messages: view.messages.map(describeMessage), rows }; + }; + + // The cached range has answers and no system message; the baseline arrives in a later read. + expect((await read(root)).messages[0]).toBe("user:first"); + await append(root.id, { kind: "pi.system", model: [system({ plan: "Plan." })] }); + let read2 = await read(root); + // One bounds probe and the added entry: the range was extended, not derived again. + expect(read2.rows).toBe(2); + expect(read2.messages.slice(0, 3)).toEqual(["system:plan", "user:first", "assistant:old 0"]); + + // A system message the cached range already has keeps a later first `pi.system` entry in place. + const fork = await root.fork(first.id, { ownership: { kind: "ownerless" } }, taskContext); + await append(fork.id, { kind: "message", model: [assistant("answer")] }); + await append(fork.id, { kind: "host.system", model: [system({ note: "Note." })] }); + expect((await read(fork)).messages).toEqual(["user:first", "assistant:answer", "system:note"]); + await append(fork.id, { kind: "pi.system", model: [system({ note: null })] }); + read2 = await read(fork); + expect(read2.rows).toBe(2); + expect(read2.messages).toEqual(["user:first", "assistant:answer", "system:note", "system:note"]); + + // An omitted system message still rules out a later baseline; the extended read keeps that. + const omitted = await root.fork(first.id, { ownership: { kind: "ownerless" } }, taskContext); + await append(omitted.id, { kind: "message", model: [assistant("answer")] }); + let host: EntryRecord | undefined; + await runtime.commit(async (tx) => { + host = await tx.appendEntry(omitted.id, { + kind: "host.system", + model: [system({ legacy: "Legacy." })], + }); + return undefined; + }, taskContext); + await append(omitted.id, { kind: "edit", edits: [{ target: host!.id, action: "omit" }] }); + expect((await read(omitted)).messages).toEqual(["user:first", "assistant:answer"]); + await append(omitted.id, { kind: "pi.system", model: [system({ legacy: null })] }); + read2 = await read(omitted); + expect(read2.rows).toBe(2); + expect(read2.messages).toEqual(["user:first", "assistant:answer", "system:legacy"]); + + // A first `pi.system` entry without a system message leaves no baseline for a later one to take. + const empty = await root.fork(first.id, { ownership: { kind: "ownerless" } }, taskContext); + await append(empty.id, { kind: "pi.system" }); + await append(empty.id, { kind: "message", model: [assistant("answer")] }); + expect((await read(empty)).messages).toEqual(["user:first", "assistant:answer"]); + await append(empty.id, { kind: "pi.system", model: [system({ s: "Delta." })] }); + read2 = await read(empty); + expect(read2.rows).toBe(2); + expect(read2.messages).toEqual(["user:first", "assistant:answer", "system:s"]); + await runtime.commit(() => DONE, taskContext); + }, + }, + abort: async (_task, runtime, taskContext) => { + await runtime.commit(() => ABORTED, taskContext); + }, + }); + addTask(registry, Steps); + harness.resume(); + const id = await harness.commit( + (tx) => tx.createTask(Steps, {}, { ownership: { kind: "conversation" }, conversationId: root.id }), + context, + ); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual(DONE.outcome); + await harness.close(context); + }); + it("keeps returned views and kept ranges apart", async () => { const { harness, registry, root } = await countingSetup(); const Mutate = defineTask, { phase: "run" }, null>({