sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix(durable): lead with the system baseline after a compaction that keeps assistant messages

Browse source 
Context derivation moved a system message to the front only when user messages
alone preceded it. After a compaction that keeps an assistant message, the
baseline written for the next run followed the kept history and the new input,
so Anthropic and OpenAI Codex requests had no initial prompt or tool set.

Derivation now identifies the harness baseline: the first pi.system entry after
the head marker, or of the transcript when no earlier entry carries a system
message, even one omitted later. Its message leads even past assistant
messages. Only the first system message ever moves, so replay is unchanged;
other system messages after an assistant message stay in place.

closes #10904
This commit is contained in:
Mario Zechner 2026-10-14 09:47:02 +00:00
parent 301617df7b
commit 5713c9507c
5 changed files with 353 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/durable/CHANGELOG.md
View file

@ -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

26
packages/durable/docs/spec.md
View file

@ -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

64
packages/durable/src/harness/context.ts
View file

@ -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. */

14
packages/durable/test/harness-compaction.test.ts
View file

@ -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("<summary>\nSUMMARY\n</summary>");
expect(continuation[0]?.role).toBe("system");
expect(userText(continuation[1])).toContain("<summary>\nSUMMARY\n</summary>");
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);
});

276
packages/durable/test/harness-context.test.ts
View file

@ -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<Record<string, never>, { 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<Record<string, never>, { phase: "run" }, null>({