fix: avoid replaying superseded compaction summaries
This commit is contained in:
@@ -24,6 +24,17 @@ prompt, plan continuity, persistence, threshold, scheduling, and authorization
|
||||
lifecycle. It does not adopt Pi's coding summary or separate split-turn summary
|
||||
calls. A cut that drops no active history does not launch a summarizer.
|
||||
|
||||
When the active cutoff is applied, conversation requests carry the latest summary
|
||||
once, followed by the retained transcript without older compaction markers. When
|
||||
a subsequent compaction supplies that prior summary separately, it includes it
|
||||
once, plus the newly superseded transcript without summary markers. All markers
|
||||
and original messages remain stored. The existing full-history fallback for an
|
||||
unresolved cutoff is unchanged, including its summary text.
|
||||
|
||||
A provider response stopped at its output limit (`length`) is an incomplete
|
||||
summary, not a successful checkpoint. Compaction fails without changing the
|
||||
existing history or prior summary; a later attempt can retry normally.
|
||||
|
||||
## Conversation meter
|
||||
|
||||
The numerator is the latest qualifying completed conversation response's native
|
||||
|
||||
@@ -272,9 +272,9 @@ export function newCompactionMarker(summary: string, details: CompactionDetails)
|
||||
};
|
||||
}
|
||||
|
||||
/** The ACTIVE compaction marker: the newest one in the record (chained
|
||||
* compactions — the last one rules; older markers fall inside the
|
||||
* dropped span and never render). */
|
||||
/** The ACTIVE compaction marker: the newest one with a retained-message
|
||||
* id. Its summary supersedes older markers even when their stored
|
||||
* positions remain inside its retained tail. */
|
||||
export function activeCompactionOf(messages: readonly LhMessage[]): (LhMarkerMessage & { details: CompactionDetails }) | null {
|
||||
for (let i = messages.length - 1; i >= 0; i--) {
|
||||
const m = messages[i]!;
|
||||
|
||||
@@ -562,8 +562,9 @@ export function toLlmMessages(messages: readonly AgentMessage[]): Message[] {
|
||||
// THE COMPACTION CUT (owner round 14, 2026-10-05): the ACTIVE marker
|
||||
// (the newest — chained compactions rule last) names the first kept
|
||||
// message; everything before it is replaced by the marker's content
|
||||
// (header + summary) rendered ONCE, up top. The marker itself is then
|
||||
// skipped in place — it must never render twice. A cut naming a
|
||||
// (header + summary) rendered ONCE, up top. All compaction markers
|
||||
// in the retained tail are then skipped, including older summaries
|
||||
// whose stored positions still lie after the cut. A cut naming a
|
||||
// missing id degrades honestly (everything renders — never throw).
|
||||
const lhMessages = messages as readonly LhMessage[];
|
||||
const active = activeCompactionOf(lhMessages);
|
||||
@@ -604,11 +605,10 @@ export function toLlmMessages(messages: readonly AgentMessage[]): Message[] {
|
||||
// UI-only markers: plan blocks, and problem dismissals (the
|
||||
// synthesized conclude pair carries the model's face).
|
||||
if (marker.customType === "plan-block" || marker.customType === "problem-dismissed") continue;
|
||||
// The ACTIVE compaction marker already rendered as the head —
|
||||
// never twice. (Stale markers before the cut are unreachable; a
|
||||
// marker whose cut could not resolve has no head and renders as
|
||||
// its user text — honest degradation.)
|
||||
if (marker.customType === "compaction" && head !== null && marker.id === active!.id) continue;
|
||||
// The head supersedes every compaction marker, not only its own
|
||||
// stored position. Without a resolved head, preserve the existing
|
||||
// full-history fallback, including the summary text.
|
||||
if (marker.customType === "compaction" && head !== null) continue;
|
||||
out.push({
|
||||
role: "user",
|
||||
content: [{ type: "text", text: marker.content }],
|
||||
|
||||
@@ -69,7 +69,7 @@ async function subagentSummarizer(prompt: string, engine: PiEngine): Promise<{ s
|
||||
totalTokens: 0,
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
||||
};
|
||||
const { text } = await runSubagent([], {
|
||||
const { text, newMessages } = await runSubagent([], {
|
||||
// The engine override is always present (the chat's own conversation
|
||||
// engine — owner ruling), so the role label never resolves an
|
||||
// engine; "researcher" is the least-wrong existing ModelRole label
|
||||
@@ -83,6 +83,12 @@ async function subagentSummarizer(prompt: string, engine: PiEngine): Promise<{ s
|
||||
usage = u;
|
||||
},
|
||||
});
|
||||
// A provider-declared incomplete response is not a usable checkpoint,
|
||||
// even when it contains nonempty text. Match Pi's compaction failure
|
||||
// semantics: leave the existing history and prior summary untouched.
|
||||
if (newMessages.some((message) => message.role === "assistant" && message.stopReason === "length")) {
|
||||
throw new Error("Compaction summary hit the output limit and is incomplete");
|
||||
}
|
||||
return { summary: text, usage };
|
||||
}
|
||||
|
||||
@@ -201,7 +207,14 @@ async function compact(deps: CompactionDeps, chatId: string): Promise<void> {
|
||||
spanStart = prevKept;
|
||||
priorSummary = previous.content;
|
||||
}
|
||||
const spanText = serializeForSummary(messages.slice(spanStart, cut));
|
||||
const span = messages.slice(spanStart, cut);
|
||||
// The separately supplied prior summary replaces all summary markers
|
||||
// in this span. Without a resolved prior boundary, retain the existing
|
||||
// legacy fallback rather than silently discarding summary-only context.
|
||||
const summaryMessages = priorSummary
|
||||
? span.filter((m) => m.role !== "custom" || m.customType !== "compaction")
|
||||
: span;
|
||||
const spanText = serializeForSummary(summaryMessages);
|
||||
const fullSpan = priorSummary ? `[SYSTEM NOTE — prior compaction summary, still authoritative where not superseded]\n${priorSummary}\n\n${spanText}` : spanText;
|
||||
|
||||
const prompt = buildCompactionPrompt({
|
||||
|
||||
@@ -27,12 +27,14 @@ const {
|
||||
ensureCompacted,
|
||||
kickCompaction,
|
||||
awaitCompaction,
|
||||
compactionInFlight,
|
||||
effectiveThresholdTokens,
|
||||
projectedContextTokens,
|
||||
} = await import("../src/server/compaction.ts");
|
||||
import type { CompactionDeps } from "../src/server/compaction.ts";
|
||||
import { createFauxCore, fauxAssistantMessage } from "@earendil-works/pi-ai";
|
||||
const { subscribe } = await import("../src/core/events.ts");
|
||||
const { activeCompactionOf, newUserMessage, measuredContextTokens, COMPACTED_HISTORY_HEADER } = await import("../src/core/index.ts");
|
||||
const { activeCompactionOf, newCompactionMarker, newUserMessage, measuredContextTokens, toLlmMessages, COMPACTED_HISTORY_HEADER } = await import("../src/core/index.ts");
|
||||
import type { LhMessage } from "../src/core/index.ts";
|
||||
|
||||
after(() => {
|
||||
@@ -167,12 +169,120 @@ describe("the compaction service", () => {
|
||||
await ensureCompacted(deps, chatId);
|
||||
assert.ok(seenPrompt.includes("prior compaction summary"), "the prior summary rides the span");
|
||||
assert.ok(seenPrompt.includes(COMPACTED_HISTORY_HEADER), "as the marker's canonical text");
|
||||
assert.equal(seenPrompt.split(COMPACTED_HISTORY_HEADER).length - 1, 1, "the prior summary is supplied exactly once");
|
||||
const record = chatStore.getChat(chatId)!;
|
||||
const markers = record.messages.filter((m) => (m as { customType?: string }).customType === "compaction");
|
||||
assert.equal(markers.length, 2, "both markers persist (history is never rewritten)");
|
||||
assert.ok(activeCompactionOf(record.messages)!.content.includes("SECOND SUMMARY"), "the newest rules");
|
||||
});
|
||||
|
||||
it("ten successive compactions keep one summary on the wire and preserve the complete record", async () => {
|
||||
const chat = chatStore.createChat("repeated-cycle fixture");
|
||||
chatStore.saveMessages(chat.id, Array.from({ length: 20 }, (_, i) => big(`exchange-${i}`, 8)));
|
||||
let calls = 0;
|
||||
const deps: CompactionDeps = {
|
||||
...depsOf(chat.id, 150_000),
|
||||
summarize: async (prompt) => {
|
||||
assert.equal(prompt.split(COMPACTED_HISTORY_HEADER).length - 1, calls === 0 ? 0 : 1);
|
||||
if (calls > 0) assert.ok(prompt.includes(`SUMMARY-${calls}.`), "only the immediately prior summary rides forward");
|
||||
return {
|
||||
summary: `SUMMARY-${++calls}.`,
|
||||
usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, totalTokens: 2, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } },
|
||||
};
|
||||
},
|
||||
};
|
||||
for (let cycle = 1; cycle <= 10; cycle++) {
|
||||
const messages = [...chatStore.getChat(chat.id)!.messages];
|
||||
if (cycle > 1) messages.push(big(`growth-${cycle}`, 8));
|
||||
// Controlled provider readings trigger compaction while the visible
|
||||
// retained tail advances slowly enough to contain older markers.
|
||||
messages.push({ id: `usage-${cycle}`, role: "assistant", content: [{ type: "text", text: "measured response" }], stopReason: "stop", usage: { input: 150_001, output: 1, cacheRead: 0, cacheWrite: 0, totalTokens: 150_002 }, timestamp: cycle } as LhMessage);
|
||||
chatStore.saveMessages(chat.id, messages);
|
||||
await ensureCompacted(deps, chat.id);
|
||||
assert.equal(calls, cycle);
|
||||
const record = chatStore.getChat(chat.id)!;
|
||||
assert.deepEqual(record.messages.slice(0, -1), messages, "every original message and marker remains stored");
|
||||
assert.equal(record.messages.filter((m) => m.role === "custom" && m.customType === "compaction").length, cycle);
|
||||
const wire = JSON.stringify(toLlmMessages(record.messages));
|
||||
assert.equal(wire.split(COMPACTED_HISTORY_HEADER).length - 1, 1, `cycle ${cycle}: one summary on the wire`);
|
||||
assert.ok(wire.includes(`SUMMARY-${cycle}.`));
|
||||
assert.equal(measuredContextTokens(record.messages), null, "committing a marker invalidates pre-compaction readings");
|
||||
}
|
||||
});
|
||||
|
||||
it("a resolved prior summary replaces all summary markers in the newly superseded span", async () => {
|
||||
const chat = chatStore.createChat("summary span fixture");
|
||||
const kept = big("previously retained");
|
||||
const details = { firstKeptMessageId: kept.id, tokensBefore: 150_000, at: "fixture" };
|
||||
const steer: LhMessage = { id: "steer", role: "custom", customType: "subagent-steer", content: "RESEARCH DELIVERY", display: true, details: {}, timestamp: 1 };
|
||||
const messages = [big("discarded"), kept, newCompactionMarker("SUPERSEDED SUMMARY", details), steer, newCompactionMarker("LATEST SUMMARY", details), big("growth"), big("recent")];
|
||||
chatStore.saveMessages(chat.id, messages);
|
||||
let seenPrompt = "";
|
||||
await ensureCompacted({ ...depsOf(chat.id, 40_000), summarize: async (prompt, engine) => {
|
||||
seenPrompt = prompt;
|
||||
return depsOf(chat.id, 40_000).summarize!(prompt, engine);
|
||||
} }, chat.id);
|
||||
assert.ok(seenPrompt.includes("previously retained"));
|
||||
assert.ok(seenPrompt.includes("RESEARCH DELIVERY"), "non-compaction context markers still ride the span");
|
||||
assert.ok(!seenPrompt.includes("SUPERSEDED SUMMARY"));
|
||||
assert.equal(seenPrompt.split("LATEST SUMMARY").length - 1, 1);
|
||||
assert.equal(seenPrompt.split(COMPACTED_HISTORY_HEADER).length - 1, 1);
|
||||
assert.deepEqual(chatStore.getChat(chat.id)!.messages.slice(0, -1), messages);
|
||||
});
|
||||
|
||||
it("an unresolved cutoff preserves legacy summary text in the next compaction input", async () => {
|
||||
const chat = chatStore.createChat("legacy summary fixture");
|
||||
const legacy = newCompactionMarker("LEGACY-ONLY INFORMATION", { firstKeptMessageId: "missing", tokensBefore: 150_000, at: "fixture" });
|
||||
const messages = [big("old"), legacy, big("recent"), newUserMessage("latest")];
|
||||
chatStore.saveMessages(chat.id, messages);
|
||||
let seenPrompt = "";
|
||||
await ensureCompacted({ ...depsOf(chat.id, 40_000), summarize: async (prompt, engine) => {
|
||||
seenPrompt = prompt;
|
||||
return depsOf(chat.id, 40_000).summarize!(prompt, engine);
|
||||
} }, chat.id);
|
||||
assert.equal(seenPrompt.split("LEGACY-ONLY INFORMATION").length - 1, 1);
|
||||
assert.ok(!seenPrompt.includes("prior compaction summary"), "no separate prior summary exists for an unresolved cutoff");
|
||||
assert.deepEqual(chatStore.getChat(chat.id)!.messages.slice(0, -1), messages);
|
||||
});
|
||||
|
||||
it("the production summarizer rejects a length-stopped checkpoint and a complete retry can commit", async () => {
|
||||
const chatId = overThresholdChat("incomplete summary fixture");
|
||||
await ensureCompacted(depsOf(chatId, 40_000), chatId);
|
||||
chatStore.saveMessages(chatId, [...chatStore.getChat(chatId)!.messages, big("growth 1"), big("growth 2")]);
|
||||
const faux = createFauxCore({});
|
||||
const { summarize: _injected, ...base } = depsOf(chatId, 40_000);
|
||||
const deps: CompactionDeps = { ...base, engine: { model: faux.getModel(), streamFn: faux.streamSimple } };
|
||||
const before = chatStore.getChat(chatId)!;
|
||||
assert.ok(activeCompactionOf(before.messages), "the failed replacement must preserve an existing checkpoint too");
|
||||
const events: string[] = [];
|
||||
const off = subscribe((event) => {
|
||||
const { chatId: id, running } = event as { chatId?: string; running?: boolean };
|
||||
if (event.type === "compaction-changed" && id === chatId) events.push(`${running}`);
|
||||
});
|
||||
// Exercise the actual zero-tool summarizer with an offline provider;
|
||||
// NODE_ENV=test otherwise deliberately skips non-injected summaries.
|
||||
const env: Record<string, string | undefined> = process.env;
|
||||
const originalNodeEnv = env.NODE_ENV;
|
||||
env.NODE_ENV = "development";
|
||||
try {
|
||||
faux.setResponses([fauxAssistantMessage("INCOMPLETE SUMMARY", { stopReason: "length" })]);
|
||||
await ensureCompacted(deps, chatId);
|
||||
assert.deepEqual(chatStore.getChat(chatId), before, "an explicitly unfinished summary must not become a checkpoint");
|
||||
assert.equal(compactionInFlight(chatId), false, "the failed run releases its gate");
|
||||
assert.deepEqual(events, ["true", "false"]);
|
||||
faux.setResponses([fauxAssistantMessage("COMPLETE SUMMARY")]);
|
||||
await ensureCompacted(deps, chatId);
|
||||
const record = chatStore.getChat(chatId)!;
|
||||
assert.ok(activeCompactionOf(record.messages)!.content.includes("COMPLETE SUMMARY"));
|
||||
assert.deepEqual(record.messages.slice(0, -1), before.messages);
|
||||
assert.ok(!JSON.stringify(record.messages).includes("INCOMPLETE SUMMARY"));
|
||||
} finally {
|
||||
if (originalNodeEnv === undefined) delete env.NODE_ENV;
|
||||
else env.NODE_ENV = originalNodeEnv;
|
||||
off();
|
||||
}
|
||||
});
|
||||
|
||||
it("effectiveThresholdTokens: the setting rules big windows; the window rule binds small ones", () => {
|
||||
assert.equal(effectiveThresholdTokens(150_000, 200_000), 150_000);
|
||||
assert.equal(effectiveThresholdTokens(150_000, 100_000), 100_000 - 16_384);
|
||||
|
||||
@@ -256,6 +256,22 @@ describe("toLlmMessages honors the cut", () => {
|
||||
assert.equal(JSON.stringify(out).indexOf("COMPACTED]"), JSON.stringify(out).lastIndexOf("COMPACTED]"), "header appears exactly once");
|
||||
});
|
||||
|
||||
it("omits nine superseded summaries inside the retained tail without changing stored history", () => {
|
||||
const kept = newUserMessage("retained exchange");
|
||||
const details = { firstKeptMessageId: kept.id, tokensBefore: 200_000, at: "fixture" };
|
||||
const old = Array.from({ length: 9 }, (_, i) => newCompactionMarker(`SUPERSEDED-${i}`, details));
|
||||
const fresh = newCompactionMarker("LATEST SUMMARY", details);
|
||||
const messages = [newUserMessage("discarded"), kept, ...old, newUserMessage("latest reply"), fresh];
|
||||
const original = JSON.stringify(messages);
|
||||
const out = toLlmMessages(messages);
|
||||
const flat = JSON.stringify(out);
|
||||
assert.equal(out.length, 3, "latest summary + retained exchange + latest reply");
|
||||
assert.ok(flat.includes("LATEST SUMMARY"));
|
||||
assert.ok(!flat.includes("SUPERSEDED-"));
|
||||
assert.equal(flat.split(COMPACTED_HISTORY_HEADER).length - 1, 1);
|
||||
assert.equal(JSON.stringify(messages), original, "projection never rewrites the record");
|
||||
});
|
||||
|
||||
it("a pending (dangling) question call in the KEPT tail still gets the interrupted synthesis", () => {
|
||||
const kept = [assistantWithCall("a1", "call_1")];
|
||||
const marker = newCompactionMarker("S", { firstKeptMessageId: "a1", tokensBefore: 1, at: "t" });
|
||||
@@ -270,6 +286,22 @@ describe("toLlmMessages honors the cut", () => {
|
||||
assert.equal(out.length, 3, "no head swap — all three render (marker as its user text)");
|
||||
});
|
||||
|
||||
it("an unresolved newest cutoff preserves the full-history fallback, including summary text", () => {
|
||||
const kept = newUserMessage("retained exchange");
|
||||
const old = newCompactionMarker("OLD SUMMARY", { firstKeptMessageId: kept.id, tokensBefore: 1, at: "fixture" });
|
||||
const fresh = newCompactionMarker("LATEST SUMMARY", { firstKeptMessageId: "missing", tokensBefore: 2, at: "fixture" });
|
||||
const messages = [newUserMessage("earliest exchange"), kept, old, newUserMessage("latest reply"), fresh];
|
||||
const original = JSON.stringify(messages);
|
||||
const out = toLlmMessages(messages);
|
||||
assert.equal(out.length, messages.length);
|
||||
assert.equal(out[0]!.content, "earliest exchange", "do not silently apply the older cutoff");
|
||||
const flat = JSON.stringify(out);
|
||||
assert.ok(flat.includes("OLD SUMMARY"));
|
||||
assert.ok(flat.includes("LATEST SUMMARY"));
|
||||
assert.equal(flat.split(COMPACTED_HISTORY_HEADER).length - 1, 2);
|
||||
assert.equal(JSON.stringify(messages), original);
|
||||
});
|
||||
|
||||
it("chained compaction: the newest marker's cut wins; the older marker falls in the dropped span", () => {
|
||||
const oldMarker = newCompactionMarker("OLD", { firstKeptMessageId: "m1", tokensBefore: 1, at: "t" });
|
||||
const m1 = newUserMessage("x");
|
||||
@@ -303,6 +335,11 @@ describe("serializeForSummary + buildCompactionPrompt", () => {
|
||||
assert.ok(!text.toLowerCase().includes("plan-block"), "UI-only markers never serialize");
|
||||
});
|
||||
|
||||
it("preserves legacy summary text when the caller has no separately supplied prior summary", () => {
|
||||
const marker = newCompactionMarker("LEGACY SUMMARY", { firstKeptMessageId: "missing", tokensBefore: 1, at: "fixture" });
|
||||
assert.ok(serializeForSummary([marker]).includes("LEGACY SUMMARY"));
|
||||
});
|
||||
|
||||
it("the prompt demands the four sections in order and carries the plan tree verbatim", () => {
|
||||
const prompt = buildCompactionPrompt({ spanText: "TRANSCRIPT", planTreeText: "- [done] A (a1)\n- A → B", kind: "teach" });
|
||||
const order = ["The learner's understanding", "The teaching arc", "The plan dependency tree", "The current moment"].map(
|
||||
@@ -388,6 +425,11 @@ describe("toLlmMessages carries the leading system message (request chrome)", ()
|
||||
const record = (): LhMessage[] => [
|
||||
{ id: "old", role: "user", content: "dropped history", timestamp: 1 },
|
||||
{ id: "kept", role: "user", content: "kept tail", timestamp: 2 },
|
||||
newCompactionMarker("SUPERSEDED SUMMARY", {
|
||||
firstKeptMessageId: "kept",
|
||||
tokensBefore: 98,
|
||||
at: new Date(0).toISOString(),
|
||||
}),
|
||||
newCompactionMarker("SUMMARY", {
|
||||
firstKeptMessageId: "kept",
|
||||
tokensBefore: 99,
|
||||
@@ -407,6 +449,8 @@ describe("toLlmMessages carries the leading system message (request chrome)", ()
|
||||
: ((m.content as { text?: string }[])[0]?.text ?? "");
|
||||
assert.ok(out.slice(2).some((m) => m.role === "user" && textOf(m) === "kept tail"));
|
||||
assert.ok(!out.some((m) => m.role === "user" && textOf(m) === "dropped history"));
|
||||
assert.ok(!JSON.stringify(out).includes("SUPERSEDED SUMMARY"));
|
||||
assert.equal(JSON.stringify(out).split(COMPACTED_HISTORY_HEADER).length - 1, 1);
|
||||
});
|
||||
|
||||
it("without a cut the chrome renders EXACTLY ONCE (no duplicate)", () => {
|
||||
|
||||
Reference in New Issue
Block a user