feat(durable): compaction and overflow (Package 20)

Manual compact(), background and blocking threshold compaction, and overflow compaction with one retry. Summaries are headed pi.compaction entries placed through the inbox, or appended by a compaction the generation waits on; the stale rule decides between concurrent summaries.
2026-11-05 23:37:25 +00:00 · 2026-09-30 18:23:17 +02:00 · 2026-09-30 18:23:17 +02:00 · ed0d6b91b6
commit ed0d6b91b6
parent b72cf98ec7
19 changed files with 3285 additions and 110 deletions
--- a/packages/durable/CHANGELOG.md
+++ b/packages/durable/CHANGELOG.md
@ -37,6 +37,8 @@
 - Added `Conversation.abort(context, { background: true })`, which also aborts background work and waits for it.
 - Conversation and Harness idle waits now include work in owned conversations and stop at background tasks.
 - A tool task whose `execute()` throws, or that is interrupted by a restart without a safe rerun, now ends `failed` (still with its error result entry), which aborts the conversations the call owns.
+- Added compaction: the `pi.compaction` task (`CompactionTask`) summarizes an older prefix of the model context and places a `pi.compaction` entry (`CompactionEntry`) that heads the first kept entry. `Conversation.compact()` starts one manually; generation starts one in the background above a soft threshold, waits for one above `contextWindow - reserveTokens`, and compacts and retries once after a context overflow. Configure it with `get/setCompaction()`; `beforeCompact` hooks can decline or supply the summary; `pi.live.compactions`, `compaction_start`/`compaction_end` events, and `pi.usage` report it.
+- `ContextView` now includes `contributions`, each active entry's model messages after edits.
 - Added `TaskRuntime.conversation()` and `ToolExecutionApi.conversation()`: invocation-bound `ConversationHandle`s for submitting to, aborting, and waiting on existing conversations, such as the ones a task owns.

 ### Fixed
--- a/packages/durable/README.md
+++ b/packages/durable/README.md
@ -18,6 +18,7 @@ Built on [`@earendil-works/pi-ai`](../ai/README.md) for model access and `@earen
 - [Watching a Conversation](#watching-a-conversation)
 - [Busy Conversations](#busy-conversations)
 - [Reset and Handoff](#reset-and-handoff)
+- [Compaction](#compaction)
 - [Agent Events (Experimental)](#agent-events-experimental)
 - [Hooks](#hooks)
 - [More Conversations and Forks](#more-conversations-and-forks)
@ -234,6 +235,36 @@ await root.reset("We were fixing the flaky login test. Continue.", context); //

 While busy, the reset is queued like a write. When it is placed during a tool round, the current run ends. A tool can request the same with `control: { handoff: "..." }`.

+## Compaction
+
+Compaction shrinks what the model sees: it summarizes older entries and appends a `pi.compaction` entry that holds the summary and heads the first entry it keeps. Older entries stay in storage.
+
+```typescript
+const id = await root.compact("Keep the failing test names", context); // manual, with optional instructions
+const { outcome } = (await harness.waitForTask(id, context)).state;
+if (outcome.status === "completed" && outcome.result.submissionId !== undefined) {
+	const placed = await (await harness.submission(outcome.result.submissionId, context))!.wait(context);
+	console.log(placed.status); // "done", or "unanswered" with reason "stale"
+}
+```
+
+The conversation keeps working while the summary is made. The summary is placed at once when the conversation is idle, otherwise at the next turn boundary. Esc (`abort()`) cancels a manual compaction.
+
+Generation also compacts on its own, controlled per conversation:
+
+```typescript
+await root.setCompaction({
+	enabled: true, // automatic compaction; manual compact() always works
+	reserveTokens: 16384, // above contextWindow - reserveTokens, the next request waits for a compaction
+	keepRecentTokens: 20000, // roughly how much recent context stays verbatim
+	backgroundTokens: 32768, // this far below that, a compaction starts in the background; 0 disables it
+}, context);
+```
+
+When a provider rejects a request because the context is too long, generation compacts and retries once. A summary that would cut before the start of the current context settles as `stale` when it is placed, so when several are in flight, the furthest cut stays in effect. Summarization spend counts in `pi.usage`. A `beforeCompact` hook on `CompactionTask` can decline or supply its own summary.
+
+Running compactions are listed in `docs["pi.live"].compactions` with their reason, attempt, and retry backoff. The agent events add `compaction_start` and `compaction_end`, and a `compactions` field in the snapshot.
+
 ## Agent Events (Experimental)

 For consumers that want coding-agent style events (`message_start`, `message_update`, `tool_execution_start`, ...) instead of structural state:
@ -242,7 +273,7 @@ For consumers that want coding-agent style events (`message_start`, `message_upd
 import { watchEvents } from "@earendil-works/pi-durable";

 const stream = await watchEvents(harness, root.id, context);
-initialize(stream.snapshot); // entries, run, in-flight generation, tools, inbox, config, usage
+initialize(stream.snapshot); // entries, run, in-flight generation, tools, compactions, inbox, config, usage
 stream.start(async (events) => {
 	for (const event of events) console.log(JSON.stringify(event));
 });
@ -426,6 +457,7 @@ node --conditions=source --experimental-strip-types test/examples/14-chat.ts
 | [22-subagent-foreground](test/examples/22-subagent-foreground.ts) | A replay-safe subagent tool whose child the call owns, with the child's events under the call |
 | [23-subagent-background](test/examples/23-subagent-background.ts) | Persistent subagents: spawn, steer, stop, list, answers reported back, restart-safe |
 | [24-child-tasks](test/examples/24-child-tasks.ts) | A checkout that owns and waits for four payments: failFast, abort, restart |
+| [25-compaction](test/examples/25-compaction.ts) | A long chat compacted in the background, manually, and after a context overflow |
 | [00](test/examples/00-conversation.ts)–[13](test/examples/13-recovery.ts) | The layers underneath: sessions, documents, forks, watches, tasks, recovery |

 Examples that call OpenAI need `OPENAI_API_KEY`; most use the faux provider otherwise.
--- a/packages/durable/src/entries.ts
+++ b/packages/durable/src/entries.ts
@ -1,5 +1,5 @@
 import type { JsonValue } from "@earendil-works/chord";
-import type { ToolDiagnostic } from "./harness/types.ts";
+import type { CompactionReason, ToolDiagnostic } from "./harness/types.ts";
 import type { Entry, EntryRecord, TypedEntry } from "./types.ts";

 /** Define a typed entry kind whose `is()` guard narrows by `EntryRecord.kind`. */
@ -27,3 +27,8 @@ export const ToolResultEntry = defineEntry<{ diagnostics: ToolDiagnostic[] }>("p
 * handoff text. Written by `Conversation.reset()` and the `handoff` tool control.
 */
 export const ResetEntry = defineEntry("pi.reset");
+/**
+ * Compaction summary: `model` is `[UserMessage]` with the wrapped summary, `head` the first kept entry. Written by
+ * compaction tasks, directly or through a write submission.
+ */
+export const CompactionEntry = defineEntry<{ reason: CompactionReason }>("pi.compaction");
--- a/packages/durable/src/harness/compaction.ts
+++ b/packages/durable/src/harness/compaction.ts
@ -0,0 +1,456 @@
+import type { Context, Draft } from "@earendil-works/chord";
+import type { AssistantMessage, Message, ModelThinkingLevel, SimpleStreamOptions } from "@earendil-works/pi-ai";
+import { calculateContextTokens, estimateMessageTokens } from "@earendil-works/pi-ai/utils/estimate";
+import { isRetryableAssistantError, retryDelayMs } from "@earendil-works/pi-ai/utils/retry";
+import { CompactionEntry } from "../entries.ts";
+import type { Transaction } from "../session/transaction.ts";
+import { defineTask } from "../tasks.ts";
+import type {
+	ConversationId,
+	EntryDraft,
+	EntryId,
+	NextTaskState,
+	RunningTask,
+	TaskId,
+	TaskRuntime,
+	Tx,
+} from "../types.ts";
+import { ConversationConfig, DEFAULT_COMPACTION_POLICY, DEFAULT_RETRY_POLICY } from "./config.ts";
+import { orderToolResults } from "./context.ts";
+import { addCompactionStatus, compactionStatus, LiveDoc, type LiveState, removeCompactionStatus } from "./live.ts";
+import { admitSubmission } from "./submissions.ts";
+import type {
+	CompactionHooks,
+	CompactionReason,
+	CompactionResult,
+	ContextView,
+	ConversationStreamOptions,
+	ModelRef,
+} from "./types.ts";
+import { recordUsage } from "./usage.ts";
+
+export type CompactionInput = { reason: CompactionReason; instructions?: string };
+
+/** The pinned summarization request. */
+export type SummaryRequest = {
+	attempt: number;
+	model: ModelRef;
+	thinkingLevel: ModelThinkingLevel;
+	streamOptions: ConversationStreamOptions;
+	maxTokens: number;
+	/** Newest entry of the context the range was selected from. */
+	tail: EntryId;
+	/** First entry kept verbatim; the summary's `head`. */
+	firstKept: EntryId;
+};
+
+export type CompactionCheckpoint =
+	| { phase: "select" }
+	| ({ phase: "summarize" } & SummaryRequest)
+	| ({ phase: "retry"; until: number } & SummaryRequest);
+
+type Runtime = TaskRuntime<CompactionInput, CompactionCheckpoint, CompactionResult, CompactionHooks>;
+type Next = NextTaskState<CompactionCheckpoint, CompactionResult>;
+
+/** Longest tool result text a serialized summary source keeps. */
+const TOOL_RESULT_MAX_CHARS = 2000;
+
+const SUMMARY_PREFIX =
+	"The conversation history before this point was compacted into the following summary:\n\n<summary>\n";
+const SUMMARY_SUFFIX = "\n</summary>";
+
+const SUMMARIZATION_SYSTEM_PROMPT = `You are a context summarization assistant. Your task is to read a conversation between a user and an AI assistant, then produce a structured summary following the exact format specified.
+
+Do NOT continue the conversation. Do NOT respond to any questions in the conversation. ONLY output the structured summary.`;
+
+const SUMMARIZATION_PROMPT = `The messages above are a conversation to summarize. Create a structured context checkpoint summary that another LLM will use to continue the work. If the conversation starts with an earlier summary, preserve its information and fold the newer messages into it.
+
+Use this EXACT format:
+
+## Goal
+[What is the user trying to accomplish? Can be multiple items if the session covers different tasks.]
+
+## Constraints & Preferences
+- [Any constraints, preferences, or requirements mentioned by user]
+- [Or "(none)" if none were mentioned]
+
+## Progress
+### Done
+- [x] [Completed tasks/changes]
+
+### In Progress
+- [ ] [Current work]
+
+### Blocked
+- [Issues preventing progress, if any]
+
+## Key Decisions
+- **[Decision]**: [Brief rationale]
+
+## Next Steps
+1. [Ordered list of what should happen next]
+
+## Critical Context
+- [Any data, examples, or references needed to continue]
+- [Or "(none)" if not applicable]
+
+Keep each section concise. Preserve exact file paths, function names, and error messages.`;
+
+/**
+ * Built-in compaction task (spec §8.7): select an old prefix of the model context, summarize it, and place a summary
+ * entry whose `head` is the first kept entry. A compaction the generation owns blocks it and appends directly; a
+ * conversation-owned one places its summary through a write submission.
+ */
+export const CompactionTask = defineTask<CompactionInput, CompactionCheckpoint, CompactionResult, CompactionHooks>({
+	name: "pi.compaction",
+	version: 1,
+	initial: () => ({ phase: "select" }),
+	phases: {
+		select: async (task, runtime, context) => {
+			const { conversationId } = runtime;
+			const config =
+				(await runtime.snapshot(ConversationConfig, conversationId, context)) ??
+				ConversationConfig.definition.initial();
+			const ref = config.model;
+			const model = ref === undefined ? undefined : runtime.models.getModel(ref.provider, ref.modelId);
+			if (ref === undefined || model === undefined) return failNoModel(runtime, ref, context);
+			const policy = config.compaction ?? DEFAULT_COMPACTION_POLICY;
+			const view = await runtime.context(conversationId, context);
+			const cut = selectCut(view, policy.keepRecentTokens);
+			if (cut === undefined) return complete(runtime, context);
+			const firstKept = view.entries[cut]!.id;
+			const { reason, instructions } = task.input;
+			const compaction = {
+				reason,
+				entries: view.entries.slice(0, cut),
+				messages: summarizedMessages(view, cut),
+				firstKept,
+				...(instructions === undefined ? {} : { instructions }),
+			};
+			let decision: { readonly decline: true } | { readonly summary: string } | undefined;
+			await runtime.hooks.each("beforeCompact", async (hook) => {
+				if (decision === undefined) decision = await hook(compaction, runtime, context);
+			});
+			if (decision !== undefined && "decline" in decision) return complete(runtime, context);
+			if (decision !== undefined) return place(runtime, firstKept, decision.summary, context);
+			const request: SummaryRequest = {
+				attempt: 1,
+				model: ref,
+				thinkingLevel: config.thinkingLevel,
+				streamOptions: config.streamOptions ?? {},
+				maxTokens: Math.min(
+					Math.floor(0.8 * policy.reserveTokens),
+					model.maxTokens > 0 ? model.maxTokens : Number.POSITIVE_INFINITY,
+				),
+				tail: view.entries.reduce((tail, entry) => (entry.id > tail ? entry.id : tail), firstKept),
+				firstKept,
+			};
+			await runtime.commit(() => ({ status: "running", checkpoint: { phase: "summarize", ...request } }), context);
+		},
+		summarize: async (task, runtime, context) => {
+			const { phase: _, ...request } = task.state.checkpoint;
+			const { model: ref, thinkingLevel, streamOptions, maxTokens, tail, firstKept, attempt } = request;
+			const model = runtime.models.getModel(ref.provider, ref.modelId);
+			if (model === undefined) return failNoModel(runtime, ref, context);
+			// The context at `tail` is immutable, so this is the range `select` chose.
+			const view = await runtime.context(runtime.conversationId, context, tail);
+			const cut = view.entries.findIndex((entry) => entry.id === firstKept);
+			const now = runtime.now();
+			const messages: Message[] = [
+				{ role: "system", content: SUMMARIZATION_SYSTEM_PROMPT, timestamp: now },
+				{
+					role: "user",
+					content: [{ type: "text", text: summaryPrompt(summarizedMessages(view, cut), task.input.instructions) }],
+					timestamp: now,
+				},
+			];
+			const { deferred: _deferred, ...forwarded } = streamOptions;
+			const options: SimpleStreamOptions = {
+				...forwarded,
+				cacheRetention: "none",
+				maxTokens,
+				signal: runtime.signal,
+				...(thinkingLevel === "off" ? {} : { reasoning: thinkingLevel }),
+			};
+			const message = await runtime.models.completeSimple(model, { messages }, options);
+			// An abort mark or close: the abort invocation or the reopened task handles the committed state.
+			runtime.signal.throwIfAborted();
+			const summary = summaryText(message);
+			const policy =
+				(await runtime.snapshot(ConversationConfig, runtime.conversationId, context))?.retry ??
+				DEFAULT_RETRY_POLICY;
+			const retry =
+				message.stopReason === "error" &&
+				isRetryableAssistantError(message) &&
+				policy.enabled &&
+				attempt <= policy.maxRetries;
+			const until = retry ? runtime.now() + retryDelayMs(policy, attempt) : 0;
+			await runtime.commit(async (tx, current): Promise<Next> => {
+				await recordUsage(
+					tx,
+					runtime.conversationId,
+					"models",
+					`${message.provider}/${message.model}`,
+					message.usage,
+				);
+				const live = await tx.doc(LiveDoc, runtime.conversationId);
+				if (summary !== undefined) return placeSummary(tx, runtime, current, live, firstKept, summary);
+				if (retry) {
+					const status = compactionStatus(live, runtime.taskId);
+					if (status !== undefined) status.retry = { at: until, error: message.errorMessage ?? "" };
+					return { status: "running", checkpoint: { phase: "retry", ...request, until } };
+				}
+				removeCompactionStatus(live, runtime.taskId);
+				const text = summaryFailure(message);
+				return {
+					status: "terminal",
+					outcome: { status: "failed", error: { message: text, detail: { reason: "model_error" } } },
+				};
+			}, context);
+		},
+		retry: async (task, runtime, context) => {
+			const { phase: _, until, ...request } = task.state.checkpoint;
+			await runtime.sleep(until, context);
+			const attempt = request.attempt + 1;
+			await runtime.commit(async (tx) => {
+				const status = compactionStatus(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId);
+				if (status !== undefined) {
+					status.attempt = attempt;
+					delete status.retry;
+				}
+				return { status: "running", checkpoint: { phase: "summarize", ...request, attempt } };
+			}, context);
+		},
+	},
+	abort: async (_task, runtime, context) => {
+		await runtime.commit(async (tx) => {
+			removeCompactionStatus(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId);
+			return { status: "terminal", outcome: { status: "aborted" } };
+		}, context);
+	},
+});
+
+/**
+ * Create a compaction task with its status in this commit. `owner` is the generation that waits for it (a blocking
+ * compaction); without one it is conversation-owned, and `background` unless it is manual.
+ */
+export async function createCompaction(
+	tx: Tx,
+	conversationId: ConversationId,
+	input: CompactionInput,
+	owner?: TaskId,
+): Promise<TaskId<CompactionResult>> {
+	const ownership =
+		owner === undefined ? ({ kind: "conversation" } as const) : { kind: "task" as const, taskId: owner };
+	const background = owner === undefined && input.reason !== "manual";
+	const taskId = await tx.createTask(CompactionTask, input, { ownership, conversationId, background });
+	const status = { taskId, reason: input.reason, blocking: owner !== undefined, attempt: 1 };
+	addCompactionStatus(await tx.doc(LiveDoc, conversationId), status);
+	return taskId;
+}
+
+/**
+ * Index in `view.entries` of the first entry a summary keeps, or `undefined` when there is nothing to compact
+ * (spec §8.7). Walks back from the tail until `keepRecentTokens` are kept, then cuts at the first candidate at or after
+ * that entry: an entry whose contribution starts with a user or assistant message, never a tool result, and never a
+ * user entry that a result of the preceding assistant's calls still follows.
+ */
+export function selectCut(view: ContextView, keepRecentTokens: number): number | undefined {
+	const { contributions } = view;
+	const start = view.head === undefined ? 0 : 1;
+	const candidates: number[] = [];
+	for (let index = start; index < contributions.length; index++) {
+		if (isCandidate(contributions, index)) candidates.push(index);
+	}
+	let kept = 0;
+	let cut: number | undefined;
+	for (let index = contributions.length - 1; index >= start; index--) {
+		for (const message of contributions[index]!) kept += estimateMessageTokens(message);
+		if (kept < keepRecentTokens) continue;
+		cut = candidates.find((candidate) => candidate >= index) ?? candidates.at(-1);
+		break;
+	}
+	if (cut === undefined) return undefined;
+	for (let index = start; index < cut; index++) if (contributions[index]!.length > 0) return cut;
+	return undefined;
+}
+
+function isCandidate(contributions: ContextView["contributions"], index: number): boolean {
+	const first = contributions[index]![0];
+	if (first?.role === "assistant") return true;
+	if (first?.role !== "user") return false;
+	// A result of the preceding assistant's calls that follows this entry, before the next assistant, belongs before it.
+	let calls: ReadonlySet<string> = new Set();
+	for (let before = index - 1; before >= 0; before--) {
+		const assistant = contributions[before]!.findLast((message) => message.role === "assistant");
+		if (assistant === undefined) continue;
+		calls = new Set(assistant.content.flatMap((content) => (content.type === "toolCall" ? [content.id] : [])));
+		break;
+	}
+	if (calls.size === 0) return true;
+	for (let after = index; after < contributions.length; after++) {
+		for (const [position, message] of contributions[after]!.entries()) {
+			if (message.role === "assistant" && (after > index || position > 0)) return true;
+			if (message.role === "toolResult" && calls.has(message.toolCallId)) return false;
+		}
+	}
+	return true;
+}
+
+/** Model messages of the entries before `cut`: the head marker first, ordered like model context (spec §2.1). */
+export function summarizedMessages(view: ContextView, cut: number): Message[] {
+	return orderToolResults(view.contributions.slice(0, cut).flat());
+}
+
+/**
+ * Size of a request over `view` followed by `extra` (spec §8.3): the usage of the newest assistant appended after the
+ * head marker, whose request included the marker, plus estimates of the messages after it; without one, estimates of
+ * every message.
+ */
+export function estimateContext(view: ContextView, extra: readonly Message[]): number {
+	let measured: AssistantMessage | undefined;
+	const after = view.head?.id ?? Number.NEGATIVE_INFINITY;
+	for (let index = view.entries.length - 1; index >= 0 && measured === undefined; index--) {
+		if (view.entries[index]!.id <= after) continue;
+		measured = view.contributions[index]!.findLast(
+			(message): message is AssistantMessage =>
+				message.role === "assistant" && calculateContextTokens(message.usage) > 0,
+		);
+	}
+	const from = measured === undefined ? 0 : view.messages.lastIndexOf(measured) + 1;
+	let tokens = measured === undefined ? 0 : calculateContextTokens(measured.usage);
+	for (const message of view.messages.slice(from)) tokens += estimateMessageTokens(message);
+	for (const message of extra) tokens += estimateMessageTokens(message);
+	return tokens;
+}
+
+/** The summary of a clean `stop` with text and no tool call; anything else is not a summary. */
+function summaryText(message: AssistantMessage): string | undefined {
+	if (message.stopReason !== "stop" || message.content.some((content) => content.type === "toolCall"))
+		return undefined;
+	const text = message.content
+		.flatMap((content) => (content.type === "text" ? [content.text] : []))
+		.join("\n")
+		.trim();
+	return text.length === 0 ? undefined : text;
+}
+
+function summaryFailure(message: AssistantMessage): string {
+	if (message.stopReason === "error" || message.stopReason === "aborted") {
+		return `Summarization failed: ${message.errorMessage ?? message.stopReason}`;
+	}
+	if (message.stopReason === "length") return "Summarization hit the token limit; the summary is incomplete";
+	if (message.content.some((content) => content.type === "toolCall")) return "Summarization attempted to call a tool";
+	return "Summarization produced no text";
+}
+
+/** The summarizer's user message: the serialized conversation, the prompt, and any instructions. */
+function summaryPrompt(messages: readonly Message[], instructions: string | undefined): string {
+	const focus = instructions === undefined ? "" : `\n\nAdditional focus: ${instructions}`;
+	return `<conversation>\n${serializeConversation(messages)}\n</conversation>\n\n${SUMMARIZATION_PROMPT}${focus}`;
+}
+
+/** Messages as plain text, so the summarizer reads a transcript instead of continuing it. System messages are omitted. */
+export function serializeConversation(messages: readonly Message[]): string {
+	const parts: string[] = [];
+	for (const message of messages) {
+		if (message.role === "user") {
+			const text = contentText(message.content);
+			if (text.length > 0) parts.push(`[User]: ${text}`);
+		} else if (message.role === "assistant") {
+			const thinking = message.content.flatMap((content) => (content.type === "thinking" ? [content.thinking] : []));
+			const text = message.content.flatMap((content) => (content.type === "text" ? [content.text] : []));
+			const calls = message.content.flatMap((content) =>
+				content.type === "toolCall"
+					? [
+							`${content.name}(${Object.entries(content.arguments)
+								.map(([key, value]) => `${key}=${JSON.stringify(value)}`)
+								.join(", ")})`,
+						]
+					: [],
+			);
+			if (thinking.length > 0) parts.push(`[Assistant thinking]: ${thinking.join("\n")}`);
+			if (text.length > 0) parts.push(`[Assistant]: ${text.join("\n")}`);
+			if (calls.length > 0) parts.push(`[Assistant tool calls]: ${calls.join("; ")}`);
+		} else if (message.role === "toolResult") {
+			const text = contentText(message.content);
+			if (text.length > 0) parts.push(`[Tool result]: ${truncate(text, TOOL_RESULT_MAX_CHARS)}`);
+		}
+	}
+	return parts.join("\n\n");
+}
+
+function contentText(content: string | readonly { readonly type: string; readonly text?: string }[]): string {
+	if (typeof content === "string") return content;
+	return content
+		.flatMap((block) => (block.type === "text" && block.text !== undefined ? [block.text] : []))
+		.join("\n");
+}
+
+function truncate(text: string, maxChars: number): string {
+	if (text.length <= maxChars) return text;
+	return `${text.slice(0, maxChars)}\n\n[... ${text.length - maxChars} more characters truncated]`;
+}
+
+/** Place a summary supplied by a hook in its own commit. */
+async function place(runtime: Runtime, firstKept: EntryId, summary: string, context: Context): Promise<void> {
+	await runtime.commit(
+		async (tx, current) =>
+			placeSummary(tx, runtime, current, await tx.doc(LiveDoc, runtime.conversationId), firstKept, summary),
+		context,
+	);
+}
+
+/**
+ * Place the summary entry and complete (spec §8.7). A blocking compaction, owned by its generation, appends it: the
+ * generation holds the run and waits. A conversation-owned one admits it as a write submission, placed at once when
+ * idle, otherwise at the next boundary, or settled `stale`.
+ * REMINDER: nothing else may append to a busy conversation, so every non-blocking summary goes through admission.
+ */
+async function placeSummary(
+	tx: Tx,
+	runtime: Runtime,
+	current: RunningTask<CompactionInput, CompactionCheckpoint, CompactionResult>,
+	live: Draft<LiveState>,
+	firstKept: EntryId,
+	summary: string,
+): Promise<Next> {
+	removeCompactionStatus(live, runtime.taskId);
+	const text = `${SUMMARY_PREFIX}${summary}${SUMMARY_SUFFIX}`;
+	const entry = {
+		kind: CompactionEntry.kind,
+		head: firstKept,
+		model: [{ role: "user", content: [{ type: "text", text }], timestamp: runtime.now() }],
+		data: { reason: current.input.reason },
+	} satisfies EntryDraft;
+	const result: CompactionResult =
+		current.owner === undefined
+			? {
+					// The scheduler commits runtime changes through a Session transaction, which admission needs.
+					submissionId: await admitSubmission(
+						tx as Transaction,
+						runtime.conversationId,
+						{ type: "write", requestId: `compaction:${runtime.taskId}`, entry },
+						runtime.now(),
+					),
+				}
+			: { entryId: (await tx.appendEntry(runtime.conversationId, entry)).id };
+	return { status: "terminal", outcome: { status: "completed", result } };
+}
+
+/** Remove the status and complete without a summary. */
+async function complete(runtime: Runtime, context: Context): Promise<void> {
+	await runtime.commit(async (tx) => {
+		removeCompactionStatus(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId);
+		return { status: "terminal", outcome: { status: "completed", result: {} } };
+	}, context);
+}
+
+async function failNoModel(runtime: Runtime, ref: ModelRef | undefined, context: Context): Promise<void> {
+	const message =
+		ref === undefined ? "No model is configured" : `Model ${ref.provider}/${ref.modelId} is not available`;
+	await runtime.commit(async (tx) => {
+		removeCompactionStatus(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId);
+		return { status: "terminal", outcome: { status: "failed", error: { message, detail: { reason: "no_model" } } } };
+	}, context);
+}
--- a/packages/durable/src/harness/config.ts
+++ b/packages/durable/src/harness/config.ts
@ -1,6 +1,12 @@
 import type { ModelThinkingLevel } from "@earendil-works/pi-ai";
 import { defineDoc } from "../documents.ts";
-import type { ConversationRetryPolicy, ConversationStreamOptions, QueueMode, ToolExecutionMode } from "./types.ts";
+import type {
+	CompactionPolicy,
+	ConversationRetryPolicy,
+	ConversationStreamOptions,
+	QueueMode,
+	ToolExecutionMode,
+} from "./types.ts";

 /** Durable per-conversation model, thinking level, request options, and desired tool loadout. */
 export type ConversationConfigState = {
@ -18,6 +24,8 @@ export type ConversationConfigState = {
 	steeringMode?: QueueMode;
 	/** How many queued follow-ups a final boundary places; absent means `one-at-a-time`. */
 	followUpMode?: QueueMode;
+	/** Automatic compaction thresholds; absent uses `DEFAULT_COMPACTION_POLICY`. */
+	compaction?: CompactionPolicy;
 };

 export const DEFAULT_RETRY_POLICY: ConversationRetryPolicy = {
@ -27,6 +35,13 @@ export const DEFAULT_RETRY_POLICY: ConversationRetryPolicy = {
 	maxAgentDelayMs: 60000,
 };

+export const DEFAULT_COMPACTION_POLICY: CompactionPolicy = {
+	enabled: true,
+	reserveTokens: 16384,
+	keepRecentTokens: 20000,
+	backgroundTokens: 32768,
+};
+
 /** Built-in configuration document; rewindable so forks start from the configuration at their fork entry. */
 export const ConversationConfig = defineDoc<ConversationConfigState>({
 	kind: "pi.conversation.config",
--- a/packages/durable/src/harness/context.ts
+++ b/packages/durable/src/harness/context.ts
@ -62,24 +62,23 @@ export async function deriveContext(
 	bounds: ContextBounds | undefined,
 	context: Context,
 ): Promise<ContextView> {
-	if (bounds === undefined) return { head: undefined, entries: [], messages: [] };
+	if (bounds === undefined) return { head: undefined, entries: [], contributions: [], messages: [] };
 	const head = bounds.head;
 	const range = await scanRange(storage, conversationId, bounds, context);
 	const edits = new Map<EntryId, ContextEdit>();
+	// Edits of every entry in the range count, including older head markers that `selectActive()` drops.
 	for (const entry of range) for (const edit of entry.edits ?? []) edits.set(edit.target, edit);

 	const entries = selectActive(head, range);
-	const messages: Message[] = [];
-	for (const entry of entries) {
+	const contributions = entries.map((entry): Message[] => {
 		const edit = edits.get(entry.id);
-		if (edit?.action === "omit") continue;
+		if (edit?.action === "omit") return [];
 		const contributed = edit?.action === "replace" ? edit.messages : (entry.model ?? []);
-		for (const message of contributed) {
-			if (message.role === "assistant" && EXCLUDED_STOP_REASONS.has(message.stopReason)) continue;
-			messages.push(message);
-		}
-	}
-	return { head, entries, messages: orderToolResults(messages) };
+		return contributed.filter(
+			(message) => message.role !== "assistant" || !EXCLUDED_STOP_REASONS.has(message.stopReason),
+		);
+	});
+	return { head, entries, contributions, messages: orderToolResults(contributions.flat()) };
 }

 /** The raw active entries within captured bounds, without deriving model context. */
@ -127,7 +126,7 @@ function selectActive(head: ContextBounds["head"], range: EntryRecord[]): EntryR
 * Place each assistant's tool results directly after it in call order. Results are taken from the messages before
 * the next assistant; a missing result is synthesized and unmatched results are dropped.
 */
-function orderToolResults(messages: readonly Message[]): Message[] {
+export function orderToolResults(messages: readonly Message[]): Message[] {
 	const ordered: Message[] = [];
 	for (let index = 0; index < messages.length; index++) {
 		const message = messages[index]!;
--- a/packages/durable/src/harness/events.ts
+++ b/packages/durable/src/harness/events.ts
@ -15,8 +15,8 @@ import type {
 } from "../types.ts";
 import { ConversationConfig, type ConversationConfigState } from "./config.ts";
 import type { InboxItem, InboxState } from "./inbox.ts";
-import type { LiveState, ToolSlot } from "./live.ts";
-import type { Harness, ToolDiagnostic } from "./types.ts";
+import type { CompactionStatus, LiveState, ToolSlot } from "./live.ts";
+import type { CompactionReason, Harness, ToolDiagnostic } from "./types.ts";
 import { UsageDoc, type UsageState } from "./usage.ts";
 import { scanAll } from "./util.ts";
 import { type ConversationView, conversationViews } from "./view.ts";
@ -44,6 +44,8 @@ export type SnapshotEvent = {
 		deferred?: { pollAt: number };
 	};
 	tools: readonly ToolSlot[];
+	/** `pi.live.compactions`: live compactions with their attempt and retry backoff. */
+	compactions: readonly CompactionStatus[];
 	inbox: readonly QueuedItem[];
 	config: ConversationConfigState;
 	usage: UsageState;
@ -80,7 +82,10 @@ export type AgentEvent =
 	| { type: "entry_appended"; entry: EntryRecord }
 	| { type: "config_changed"; config: ConversationConfigState }
 	| { type: "usage_changed"; usage: UsageState }
-	| { type: "task_failed"; taskId: TaskId; kind: string; message: string };
+	| { type: "task_failed"; taskId: TaskId; kind: string; message: string }
+	| { type: "compaction_start"; taskId: TaskId; reason: CompactionReason; blocking: boolean }
+	/** The task's receipt tells whether it produced a summary; the summary entry has its own events. */
+	| { type: "compaction_end"; taskId: TaskId; reason: CompactionReason };

 /** Serialized stream of one conversation's event batches, one per commit. */
 export interface AgentEventStream {
@ -116,6 +121,7 @@ function snapshotOf(view: ConversationView): SnapshotEvent {
 		...(live.run === undefined ? {} : { run: { inputs: live.run.inputs } }),
 		...(live.generation === undefined ? {} : { generation: live.generation as SnapshotEvent["generation"] }),
 		tools: live.tools ?? [],
+		compactions: live.compactions ?? [],
 		inbox: queued(inbox),
 		config: config ?? ConversationConfig.definition.initial(),
 		usage: usage ?? UsageDoc.definition.initial(),
@ -287,8 +293,16 @@ function translate(
 	// Ends without a result entry: a faulted or orphaned tool, or one whose run ended.
 	events.push(...toolEnds.filter((end) => end.entry === undefined));

-	// Task failures, then turn and run ends. A generation's turn ends when its outcome is committed: at a `completing`
-	// hold or at terminal, whichever comes first, so a successor created at the hold starts after it.
+	// Compaction ends, task failures, then turn and run ends.
+	const compactionsBefore = was.live.compactions ?? [];
+	const compactions = now.live.compactions ?? [];
+	for (const { taskId, reason } of compactionsBefore) {
+		if (!compactions.some((status) => status.taskId === taskId)) {
+			events.push({ type: "compaction_end", taskId, reason });
+		}
+	}
+	// A generation's turn ends when its outcome is committed: at a `completing` hold or at terminal, whichever comes
+	// first, so a successor created at the hold starts after it.
 	let turnEnded = false;
 	for (const task of tasks.values()) {
 		const status = task.state.status;
@ -319,6 +333,11 @@ function translate(
 	}
 	if (now.usage !== was.usage)
 		events.push({ type: "usage_changed", usage: now.usage ?? UsageDoc.definition.initial() });
+	for (const { taskId, reason, blocking } of compactions) {
+		if (!compactionsBefore.some((status) => status.taskId === taskId)) {
+			events.push({ type: "compaction_start", taskId, reason, blocking });
+		}
+	}
 	if (run !== undefined && runChanged) events.push({ type: "run_start", inputs: run.inputs });
 	if (run !== undefined && run.taskId !== runBefore?.taskId && tasks.get(run.taskId)?.kind === "pi.generation") {
 		events.push({ type: "turn_start" });
--- a/packages/durable/src/harness/generation.ts
+++ b/packages/durable/src/harness/generation.ts
@ -9,6 +9,7 @@ import type {
 	SimpleStreamOptions,
 	ToolCall,
 } from "@earendil-works/pi-ai";
+import { isContextOverflow } from "@earendil-works/pi-ai/utils/overflow";
 import { isRetryableAssistantError, retryDelayMs } from "@earendil-works/pi-ai/utils/retry";
 import { getCurrentTools } from "@earendil-works/pi-ai/utils/transcript";
 import { AssistantEntry, ResetEntry, SystemEntry, UserEntry } from "../entries.ts";
@ -23,13 +24,17 @@ import type {
 	Tx,
 	TypedEntry,
 } from "../types.ts";
-import { ConversationConfig, DEFAULT_RETRY_POLICY } from "./config.ts";
+import { createCompaction, estimateContext, selectCut } from "./compaction.ts";
+import { ConversationConfig, DEFAULT_COMPACTION_POLICY, DEFAULT_RETRY_POLICY } from "./config.ts";
 import { applyBoundary, prepareBoundary } from "./inbox.ts";
 import { assignJson } from "./json.ts";
 import { endRun, LiveDoc, type LiveState, type ToolSlot } from "./live.ts";
 import { desiredTools, planSystemEntries, renderSections, replaySections } from "./prompt.ts";
 import { appendToolResult, harnessError, ToolTask, type ToolTaskResult } from "./tool.ts";
 import type {
+	CompactionPolicy,
+	CompactionResult,
+	ContextView,
 	ConversationStreamOptions,
 	GenerationHooks,
 	ModelRef,
@ -44,10 +49,18 @@ import { recordUsage } from "./usage.ts";
 export type GenerationInput = Record<string, never>;

 export type GenerationCheckpoint =
-	| { phase: "prepare"; attempt: number }
+	| {
+			phase: "prepare";
+			attempt: number;
+			/** The blocking compaction this generation waited for; it starts no other compaction (spec §8.3). */
+			compacted?: TaskId<CompactionResult>;
+			/** Error text of the overflow that started `compacted`; checked once when `prepare` resumes. */
+			overflow?: string;
+	  }
 	| {
 			phase: "request";
 			attempt: number;
+			compacted?: TaskId<CompactionResult>;
 			model: ModelRef;
 			thinkingLevel: ModelThinkingLevel;
 			/** Configured request options when preparation committed; a resend after recovery uses them unchanged. */
@ -56,10 +69,11 @@ export type GenerationCheckpoint =
 			/** Newest entry included in the request. */
 			cutoff: EntryId;
 	  }
-	| { phase: "retry"; attempt: number; until: number }
+	| { phase: "retry"; attempt: number; compacted?: TaskId<CompactionResult>; until: number }
 	| {
 			phase: "poll";
 			attempt: number;
+			compacted?: TaskId<CompactionResult>;
 			model: ModelRef;
 			toolExecution: ToolExecutionMode;
 			cutoff: EntryId;
@ -85,6 +99,7 @@ type Next = NextTaskState<GenerationCheckpoint, GenerationResult>;
 /** What classification needs from the request that produced a message. */
 type Request = {
 	readonly attempt: number;
+	readonly compacted: TaskId<CompactionResult> | undefined;
 	readonly model: ModelRef;
 	readonly toolExecution: ToolExecutionMode;
 	readonly cutoff: EntryId;
@ -118,8 +133,14 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
 				(await runtime.snapshot(ConversationConfig, conversationId, context)) ??
 				ConversationConfig.definition.initial();
 			const { model, thinkingLevel, streamOptions } = config;
-			if (model === undefined || runtime.models.getModel(model.provider, model.modelId) === undefined) {
-				return failNoModel(runtime, model, context);
+			const resolved = model === undefined ? undefined : runtime.models.getModel(model.provider, model.modelId);
+			if (model === undefined || resolved === undefined) return failNoModel(runtime, model, context);
+			const { attempt, compacted, overflow } = task.state.checkpoint;
+			if (compacted !== undefined && overflow !== undefined) {
+				const [outcome] = await runtime.outcomes([compacted], context);
+				if (outcome?.status !== "completed" || outcome.result.entryId === undefined) {
+					return failModelError(runtime, overflow, context);
+				}
 			}
 			const view = await runtime.context(conversationId, context);
 			const shown = replaySections(view.messages);
@ -135,12 +156,35 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
 			const report = (error: unknown) => runtime.report(error);
 			const desired = await renderSections(registry.sections(), input, shown, report, context);
 			const entries = planSystemEntries(view, desired, tools, runtime.now());
+			const threshold =
+				compacted === undefined
+					? thresholdCompaction(
+							view,
+							entries,
+							resolved.contextWindow,
+							config.compaction ?? DEFAULT_COMPACTION_POLICY,
+						)
+					: undefined;
+			if (threshold === "blocking") {
+				// Compact first and prepare again; the transcript is unchanged until the compaction appends.
+				await runtime.commit(async (tx): Promise<Next> => {
+					const child = await createCompaction(tx, conversationId, { reason: "threshold" }, runtime.taskId);
+					const checkpoint = { phase: "prepare", attempt, compacted: child } as const;
+					return { status: "waiting", checkpoint, on: [child], policy: "allSettled" };
+				}, context);
+				return;
+			}
 			await runtime.commit(async (tx) => {
 				let cutoff = (await tx.scanEntries({ conversationId }, 1)).items[0]?.id;
 				for (const entry of entries) cutoff = (await tx.appendEntry(SystemEntry, conversationId, entry)).id;
 				if (cutoff === undefined) throw new Error(`Conversation ${conversationId} has no entries to send`);
+				// Checked in this commit, so a compaction admitted during preparation counts.
+				if (threshold === "background" && (await tx.doc(LiveDoc, conversationId)).compactions === undefined) {
+					await createCompaction(tx, conversationId, { reason: "threshold" });
+				}
 				const request = {
-					attempt: task.state.checkpoint.attempt,
+					attempt,
+					...(compacted === undefined ? {} : { compacted }),
 					model,
 					thinkingLevel,
 					streamOptions: streamOptions ?? {},
@ -151,7 +195,15 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
 			}, context);
 		},
 		request: async (task, runtime, context) => {
-			const { attempt, model: ref, thinkingLevel, streamOptions, toolExecution, cutoff } = task.state.checkpoint;
+			const {
+				attempt,
+				compacted,
+				model: ref,
+				thinkingLevel,
+				streamOptions,
+				toolExecution,
+				cutoff,
+			} = task.state.checkpoint;
 			const conversationId = runtime.conversationId;
 			await runtime.commit(async (tx) => {
 				const live = await tx.doc(LiveDoc, conversationId);
@ -173,24 +225,30 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
 				...(thinkingLevel === "off" ? {} : { reasoning: thinkingLevel }),
 			};
 			const message = await streamResponse(runtime, model, messages, options, attempt, context);
-			const request = { attempt, model: ref, toolExecution, cutoff, messages: view.messages };
+			const request = { attempt, compacted, model: ref, toolExecution, cutoff, messages: view.messages };
 			await classify(runtime, request, message, context);
 		},
 		retry: async (task, runtime, context) => {
-			const { attempt, until } = task.state.checkpoint;
+			const { attempt, compacted, until } = task.state.checkpoint;
 			await runtime.sleep(until, context);
 			await runtime.commit(async (tx) => {
 				(await tx.doc(LiveDoc, runtime.conversationId)).generation = { attempt: attempt + 1 };
-				return { status: "running", checkpoint: { phase: "prepare", attempt: attempt + 1 } };
+				const checkpoint: GenerationCheckpoint = {
+					phase: "prepare",
+					attempt: attempt + 1,
+					...(compacted === undefined ? {} : { compacted }),
+				};
+				return { status: "running", checkpoint };
 			}, context);
 		},
 		poll: async (task, runtime, context) => {
-			const { attempt, model: ref, toolExecution, cutoff, handle, pollAt } = task.state.checkpoint;
+			const { attempt, compacted, model: ref, toolExecution, cutoff, handle, pollAt } = task.state.checkpoint;
 			const model = runtime.models.getModel(ref.provider, ref.modelId);
 			if (model === undefined) return failNoModel(runtime, ref, context);
 			await runtime.sleep(pollAt, context);
 			const message = await runtime.models.fetchDeferred(model, handle, { signal: runtime.signal });
-			await classify(runtime, { attempt, model: ref, toolExecution, cutoff, pollAt }, message, context);
+			const request = { attempt, compacted, model: ref, toolExecution, cutoff, pollAt };
+			await classify(runtime, request, message, context);
 		},
 		tools: async (task, runtime, context) => {
 			const { assistant, tools, pending } = task.state.checkpoint;
@ -260,6 +318,42 @@ function createToolTask(tx: Tx, runtime: Runtime, assistant: EntryId, callId: st
 	return tx.createTask(ToolTask, { assistant, callId }, { ownership: { kind: "task", taskId: runtime.taskId } });
 }

+/**
+ * Which threshold compaction preparation starts before its request (spec §8.3): `blocking` above
+ * `contextWindow - reserveTokens`, `background` above the background threshold, and only when range selection finds a
+ * cut. The caller starts a background one only while no compaction is listed.
+ */
+function thresholdCompaction(
+	view: ContextView,
+	planned: readonly { readonly model?: readonly Message[] }[],
+	contextWindow: number,
+	policy: CompactionPolicy,
+): "blocking" | "background" | undefined {
+	if (!policy.enabled || contextWindow <= 0) return undefined;
+	const tokens = estimateContext(
+		view,
+		planned.flatMap((entry) => entry.model ?? []),
+	);
+	const blocking = contextWindow - policy.reserveTokens;
+	const background = blocking - policy.backgroundTokens;
+	const over =
+		tokens > blocking ? "blocking" : policy.backgroundTokens > 0 && tokens > background ? "background" : undefined;
+	if (over === undefined || selectCut(view, policy.keepRecentTokens) === undefined) return undefined;
+	return over;
+}
+
+/** Settle the run's inputs `unanswered` with `model_error` and fail with `text`. */
+async function failModelError(runtime: Runtime, text: string, context: Context): Promise<void> {
+	await runtime.commit(async (tx) => {
+		const live = await tx.doc(LiveDoc, runtime.conversationId);
+		endRun(tx, live, runtime.taskId, { status: "unanswered", reason: "model_error", detail: text });
+		return {
+			status: "terminal",
+			outcome: { status: "failed", error: { message: text, detail: { reason: "model_error" } } },
+		};
+	}, context);
+}
+
 /** Settle the run's inputs `unanswered` with `no_model` and fail. */
 async function failNoModel(runtime: Runtime, ref: ModelRef | undefined, context: Context): Promise<void> {
 	const message =
@ -349,7 +443,7 @@ async function classify(
 	// An abort mark or close: the abort invocation or the reopened run handles the committed state.
 	runtime.signal.throwIfAborted();
 	const conversationId = runtime.conversationId;
-	const { attempt, model: ref, toolExecution, cutoff } = request;
+	const { attempt, compacted, model: ref, toolExecution, cutoff } = request;
 	if (message.stopReason === "deferred" && message.deferred !== undefined) {
 		const handle = message.deferred;
 		const pollAt = Math.max(
@ -358,7 +452,16 @@ async function classify(
 		);
 		await runtime.commit(async (tx) => {
 			(await tx.doc(LiveDoc, conversationId)).generation = { attempt, deferred: { pollAt } };
-			const checkpoint = { phase: "poll", attempt, model: ref, toolExecution, cutoff, handle, pollAt } as const;
+			const checkpoint = {
+				phase: "poll",
+				attempt,
+				...(compacted === undefined ? {} : { compacted }),
+				model: ref,
+				toolExecution,
+				cutoff,
+				handle,
+				pollAt,
+			} as const;
 			return { status: "running", checkpoint };
 		}, context);
 		return;
@ -371,10 +474,30 @@ async function classify(
 	if (message.stopReason === "stop" || message.stopReason === "length" || message.stopReason === "toolUse") {
 		return answer(runtime, message, context);
 	}
-	// The retry policy governs the next attempt, so it is read now rather than pinned at preparation.
-	const policy = (await runtime.snapshot(ConversationConfig, conversationId, context))?.retry ?? DEFAULT_RETRY_POLICY;
+	// The retry and compaction policies govern the next attempt, so they are read now rather than pinned at preparation.
+	const config = await runtime.snapshot(ConversationConfig, conversationId, context);
+	const overflow = message.stopReason === "error" && isContextOverflow(message);
+	if (overflow && compacted === undefined && (config?.compaction ?? DEFAULT_COMPACTION_POLICY).enabled) {
+		const policy = config?.compaction ?? DEFAULT_COMPACTION_POLICY;
+		const view = await runtime.context(conversationId, context, cutoff);
+		if (selectCut(view, policy.keepRecentTokens) !== undefined) {
+			const text = message.errorMessage ?? "Context overflow";
+			await runtime.commit(async (tx): Promise<Next> => {
+				const live = await tx.doc(LiveDoc, conversationId);
+				await appendAssistant(tx, conversationId, message);
+				delete live.generation;
+				const child = await createCompaction(tx, conversationId, { reason: "overflow" }, runtime.taskId);
+				const checkpoint = { phase: "prepare", attempt, compacted: child, overflow: text } as const;
+				return { status: "waiting", checkpoint, on: [child], policy: "allSettled" };
+			}, context);
+			return;
+		}
+	}
+	const policy = config?.retry ?? DEFAULT_RETRY_POLICY;
+	// An overflow is never retried: only a compaction can make the next request fit.
 	const retry =
 		message.stopReason === "error" &&
+		!overflow &&
 		isRetryableAssistantError(message) &&
 		policy.enabled &&
 		attempt <= policy.maxRetries;
@ -384,7 +507,13 @@ async function classify(
 		await appendAssistant(tx, conversationId, message);
 		if (retry) {
 			live.generation = { attempt, retry: { at: until, error: message.errorMessage ?? "" } };
-			return { status: "running", checkpoint: { phase: "retry", attempt, until } };
+			const checkpoint = {
+				phase: "retry",
+				attempt,
+				...(compacted === undefined ? {} : { compacted }),
+				until,
+			} as const;
+			return { status: "running", checkpoint };
 		}
 		const text = message.errorMessage ?? `Model response ended with stop reason ${message.stopReason}`;
 		endRun(tx, live, runtime.taskId, { status: "unanswered", reason: "model_error", detail: text });
--- a/packages/durable/src/harness/harness.ts
+++ b/packages/durable/src/harness/harness.ts
@ -21,7 +21,13 @@ import type {
 	WatchHandle,
 } from "../types.ts";
 import { ROOT_CONVERSATION_ID } from "../types.ts";
-import { ConversationConfig, type ConversationConfigState, DEFAULT_RETRY_POLICY } from "./config.ts";
+import { createCompaction } from "./compaction.ts";
+import {
+	ConversationConfig,
+	type ConversationConfigState,
+	DEFAULT_COMPACTION_POLICY,
+	DEFAULT_RETRY_POLICY,
+} from "./config.ts";
 import { readContext } from "./context.ts";
 import { withdrawQueuedInputs } from "./inbox.ts";
 import { settleSchedulerOutcome } from "./live.ts";
@ -29,6 +35,8 @@ import { BUILTIN_SETUP_KEY, BUILTIN_TASKS } from "./registry.ts";
 import { type InvocationBinding, TaskScheduler } from "./scheduler.ts";
 import { Submissions } from "./submissions.ts";
 import type {
+	CompactionPolicy,
+	CompactionResult,
 	ContextView,
 	Conversation,
 	ConversationAbortOptions,
@ -174,10 +182,27 @@ class ConversationImpl<Tool extends ToolRegistration> implements Conversation {
 		}, context);
 	}

+	async getCompaction(context: Context): Promise<CompactionPolicy> {
+		return (await this.#config(context)).compaction ?? { ...DEFAULT_COMPACTION_POLICY };
+	}
+
+	setCompaction(policy: CompactionPolicy | undefined, context: Context): Promise<void> {
+		return this.#editConfig((config) => {
+			if (policy === undefined) delete config.compaction;
+			else config.compaction = policy;
+		}, context);
+	}
+
 	submit(submission: SubmissionDraft, context: Context): Promise<Submission> {
 		return this.#host.submissions.submit(this.id, submission, context);
 	}

+	compact(instructions: string | undefined, context: Context): Promise<TaskId<CompactionResult>> {
+		this.#host.tasks.resume();
+		const input = { reason: "manual", ...(instructions === undefined ? {} : { instructions }) } as const;
+		return this.#host.harness.commitWith((tx) => createCompaction(tx, this.id, input), context);
+	}
+
 	async reset(handoff: string | undefined, context: Context): Promise<void> {
 		const model =
 			handoff === undefined
--- a/packages/durable/src/harness/live.ts
+++ b/packages/durable/src/harness/live.ts
@ -5,7 +5,7 @@ import type { Transaction } from "../session/transaction.ts";
 import type { EntryId, SubmissionId, SubmissionSettlement, TaskId, TaskRecord, Tx } from "../types.ts";
 import { convertPartial } from "./generation.ts";
 import type { SchedulerOutcome } from "./scheduler.ts";
-import type { ToolDiagnostic } from "./types.ts";
+import type { CompactionReason, CompactionResult, ToolDiagnostic } from "./types.ts";

 /** Presentation of one tool call of the current round. */
 export type ToolSlot = {
@ -29,6 +29,17 @@ export type ToolSlot = {
 	entry?: EntryId;
 };

+/** Presentation of one live compaction task (spec §8.7). */
+export type CompactionStatus = {
+	taskId: TaskId<CompactionResult>;
+	reason: CompactionReason;
+	/** Whether a generation waits for it: a compaction the generation owns. */
+	blocking: boolean;
+	attempt: number;
+	/** Durable backoff before the next summarization attempt. */
+	retry?: { at: number; error: string };
+};
+
 /** Built-in live conversation state: run control and presentation of the current generation and tool round. */
 export type LiveState = {
 	/** Run control: the task that settles the run's inputs, and those inputs; present exactly while busy. */
@ -45,6 +56,8 @@ export type LiveState = {
 	};
 	/** The current tool round in call order, from the tool-calling answer until the generation's `tools` phase ends it. */
 	tools?: ToolSlot[];
+	/** Live compaction tasks in task ID order; absent when none. */
+	compactions?: CompactionStatus[];
 };

 export const LiveDoc = defineDoc<LiveState>({
@ -65,6 +78,7 @@ export const LiveDoc = defineDoc<LiveState>({
 /** Built-in task kinds that can own `pi.live.run`. */
 const RUN_TASK_KINDS: ReadonlySet<string> = new Set(["pi.generation"]);
 const TOOL_TASK_KIND = "pi.tool";
+const COMPACTION_TASK_KIND = "pi.compaction";

 /**
 * End the run owned by `taskId`: settle each of its inputs and remove `run`. Always removes `generation` and `tools`,
@ -79,6 +93,26 @@ export function endRun(tx: Tx, live: Draft<LiveState>, taskId: TaskId, settlemen
 	delete live.tools;
 }

+/** Add the status of a compaction task created in this commit; statuses stay in task ID order. */
+export function addCompactionStatus(live: Draft<LiveState>, status: CompactionStatus): void {
+	live.compactions ??= [];
+	live.compactions.push(status);
+}
+
+/** The status of compaction task `taskId`, if listed. */
+export function compactionStatus(live: Draft<LiveState>, taskId: TaskId): Draft<CompactionStatus> | undefined {
+	return live.compactions?.find((status) => status.taskId === taskId);
+}
+
+/** Remove the status of compaction task `taskId`, and the list once empty. */
+export function removeCompactionStatus(live: Draft<LiveState>, taskId: TaskId): void {
+	const statuses = live.compactions;
+	if (statuses === undefined) return;
+	const index = statuses.findIndex((status) => status.taskId === taskId);
+	if (index >= 0) statuses.splice(index, 1);
+	if (statuses.length === 0) delete live.compactions;
+}
+
 /** The slot of tool task `taskId` in the current round, if the round still lists it. */
 export function toolSlot(live: Draft<LiveState>, taskId: TaskId): Draft<ToolSlot> | undefined {
 	return live.tools?.find((slot) => slot.taskId === taskId);
@ -102,7 +136,8 @@ export function clearProgress(slot: Draft<ToolSlot>): void {

 /**
 * Harness cleanup for a terminal outcome the scheduler writes itself (`faulted` or `orphaned`). A run task ends its
- * run; a tool task's slot is marked done without an entry, and context derivation synthesizes the missing result.
+ * run; a tool task's slot is marked done without an entry, and context derivation synthesizes the missing result; a
+ * compaction task's status is removed.
 * Ignores other kinds so it never creates `pi.live` elsewhere. The scheduler calls this without knowing task kinds;
 * the Harness passes it in (spec §5.4).
 * REMINDER: a committed generation partial becomes an aborted assistant entry here, exactly as in the generation abort
@ -121,6 +156,10 @@ export async function settleSchedulerOutcome(
 		if (slot !== undefined) finishSlot(slot, undefined);
 		return;
 	}
+	if (record.kind === COMPACTION_TASK_KIND) {
+		removeCompactionStatus(await tx.doc(LiveDoc, record.conversationId), record.id);
+		return;
+	}
 	if (!RUN_TASK_KINDS.has(record.kind)) return;
 	const live = await tx.doc(LiveDoc, record.conversationId);
 	if (live.run?.taskId !== record.id) return;
--- a/packages/durable/src/harness/registry.ts
+++ b/packages/durable/src/harness/registry.ts
@ -1,4 +1,5 @@
 import type { ConversationRecord, Tx } from "../types.ts";
+import { CompactionTask } from "./compaction.ts";
 import { ConversationConfig } from "./config.ts";
 import { GenerationTask } from "./generation.ts";
 import { InboxDoc } from "./inbox.ts";
@ -24,7 +25,7 @@ import { UsageDoc } from "./usage.ts";
 const SECTION_KEY = /^[a-z][a-z0-9_-]*$/;

 /** Built-in task definitions every registry starts with; they cannot be disposed or replaced. */
-export const BUILTIN_TASKS: readonly AnyTask[] = [GenerationTask, ToolTask];
+export const BUILTIN_TASKS: readonly AnyTask[] = [GenerationTask, ToolTask, CompactionTask];

 export const BUILTIN_SETUP_KEY = "pi";

--- a/packages/durable/src/harness/submissions.ts
+++ b/packages/durable/src/harness/submissions.ts
@ -2,6 +2,7 @@ import { type Context, copyJson, type JsonRepresentation } from "@earendil-works
 import { UserEntry } from "../entries.ts";
 import { ConversationBusy } from "../errors.ts";
 import type { SessionImpl } from "../session/session.ts";
+import type { Transaction } from "../session/transaction.ts";
 import type {
 	CommitPublication,
 	ConversationId,
@ -40,70 +41,13 @@ export class Submissions {
 		});
 	}

-	/**
-	 * Admit a submission in one commit (spec §6). A known request ID returns its existing submission without writing.
-	 * A busy conversation queues it in `pi.inbox`, or rejects `whenBusy: "reject"` input with `ConversationBusy`
-	 * without writing. An idle conversation with queued items queues it behind them and runs a final boundary. Otherwise
-	 * idle input places a user entry and starts a run, and an idle write appends its entry and settles `done`.
-	 */
+	/** Admit a submission in one commit; see `admitSubmission()`. */
 	async submit(conversationId: ConversationId, draft: SubmissionDraft, context: Context): Promise<Submission> {
 		this.#resume();
-		const id = await this.#session.commitWith(async (tx) => {
-			if (draft.requestId !== undefined) {
-				const existing = await tx.submissionByRequest(conversationId, draft.requestId);
-				if (existing !== undefined) {
-					if (existing.type !== draft.type) {
-						throw new Error(
-							`Request ${draft.requestId} already identifies a submission of type ${existing.type}`,
-						);
-					}
-					return existing.id;
-				}
-			}
-			const live = await tx.doc(LiveDoc, conversationId);
-			const busy = live.run !== undefined;
-			if (busy && draft.type === "input" && draft.whenBusy === "reject") throw new ConversationBusy(conversationId);
-			const requestId = draft.requestId === undefined ? {} : { requestId: draft.requestId };
-			// A boundary reads the table, so it is prepared before the first table write; a busy one needs none.
-			const boundary = busy ? undefined : await prepareBoundary(tx, conversationId);
-			if (boundary === undefined || boundary.inbox.items.length > 0) {
-				const { id } = await tx.createSubmission({
-					conversationId,
-					...requestId,
-					type: draft.type,
-					status: "queued",
-				});
-				// Hosts may leave optional fields `undefined`; drafts take strict JSON.
-				const value = copyJson(draft.type === "write" ? draft.entry : draft.content, {
-					omitUndefinedProperties: true,
-				});
-				const items = (boundary?.inbox ?? (await tx.doc(InboxDoc, conversationId))).items;
-				if (draft.type === "write") items.push({ id, mode: "write", entry: value as JsonObject });
-				else {
-					const mode = draft.whenBusy === "steer" ? "steer" : "followUp";
-					items.push({ id, mode, content: value as JsonRepresentation<UserInput> });
-				}
-				if (boundary === undefined) return id;
-				const { users } = await applyBoundary(tx, boundary, "final", this.#now());
-				if (users.length > 0) await startRun(tx, conversationId, live, users);
-				return id;
-			}
-			if (draft.type === "write") {
-				if (isStale(boundary, draft.entry)) {
-					const stale = { status: "unanswered", reason: "stale" } as const;
-					return (await tx.createSubmission({ conversationId, ...requestId, type: "write", ...stale })).id;
-				}
-				const entry = await tx.appendEntry(conversationId, draft.entry);
-				const write = { conversationId, ...requestId, type: "write", status: "done", entry: entry.id } as const;
-				return (await tx.createSubmission(write)).id;
-			}
-			const message = { role: "user", content: draft.content, timestamp: this.#now() } as const;
-			const entry = await tx.appendEntry(UserEntry, conversationId, { model: [message] });
-			const input = { conversationId, ...requestId, type: "input", status: "placed", entry: entry.id } as const;
-			const { id } = await tx.createSubmission(input);
-			await startRun(tx, conversationId, live, [id]);
-			return id;
-		}, context);
+		const id = await this.#session.commitWith(
+			(tx) => admitSubmission(tx, conversationId, draft, this.#now()),
+			context,
+		);
 		return new SubmissionHandle(id, this);
 	}

@ -184,3 +128,70 @@ class SubmissionHandle implements Submission {
 function isSettled(record: SubmissionRecord): record is SettledSubmissionRecord {
 	return record.status === "done" || record.status === "unanswered";
 }
+
+/**
+ * Admit a submission inside a commit (spec §6); `Conversation.submit()` and conversation-owned compactions share it. A
+ * known request ID returns its existing submission without writing. A busy conversation queues it in `pi.inbox`, or
+ * rejects `whenBusy: "reject"` input with `ConversationBusy`. An idle conversation with queued items queues it behind
+ * them and runs a final boundary. Otherwise idle input places a user entry and starts a run, and an idle write appends
+ * its entry and settles `done`, or `stale` when its head reaches before the active range.
+ */
+export async function admitSubmission(
+	tx: Transaction,
+	conversationId: ConversationId,
+	draft: SubmissionDraft,
+	now: number,
+): Promise<SubmissionId> {
+	if (draft.requestId !== undefined) {
+		const existing = await tx.submissionByRequest(conversationId, draft.requestId);
+		if (existing !== undefined) {
+			if (existing.type !== draft.type) {
+				throw new Error(`Request ${draft.requestId} already identifies a submission of type ${existing.type}`);
+			}
+			return existing.id;
+		}
+	}
+	const live = await tx.doc(LiveDoc, conversationId);
+	const busy = live.run !== undefined;
+	if (busy && draft.type === "input" && draft.whenBusy === "reject") throw new ConversationBusy(conversationId);
+	const requestId = draft.requestId === undefined ? {} : { requestId: draft.requestId };
+	// A boundary reads the table, so it is prepared before the first table write; a busy one needs none.
+	const boundary = busy ? undefined : await prepareBoundary(tx, conversationId);
+	if (boundary === undefined || boundary.inbox.items.length > 0) {
+		const { id } = await tx.createSubmission({
+			conversationId,
+			...requestId,
+			type: draft.type,
+			status: "queued",
+		});
+		// Hosts may leave optional fields `undefined`; drafts take strict JSON.
+		const value = copyJson(draft.type === "write" ? draft.entry : draft.content, {
+			omitUndefinedProperties: true,
+		});
+		const items = (boundary?.inbox ?? (await tx.doc(InboxDoc, conversationId))).items;
+		if (draft.type === "write") items.push({ id, mode: "write", entry: value as JsonObject });
+		else {
+			const mode = draft.whenBusy === "steer" ? "steer" : "followUp";
+			items.push({ id, mode, content: value as JsonRepresentation<UserInput> });
+		}
+		if (boundary === undefined) return id;
+		const { users } = await applyBoundary(tx, boundary, "final", now);
+		if (users.length > 0) await startRun(tx, conversationId, live, users);
+		return id;
+	}
+	if (draft.type === "write") {
+		if (isStale(boundary, draft.entry)) {
+			const stale = { status: "unanswered", reason: "stale" } as const;
+			return (await tx.createSubmission({ conversationId, ...requestId, type: "write", ...stale })).id;
+		}
+		const entry = await tx.appendEntry(conversationId, draft.entry);
+		const write = { conversationId, ...requestId, type: "write", status: "done", entry: entry.id } as const;
+		return (await tx.createSubmission(write)).id;
+	}
+	const message = { role: "user", content: draft.content, timestamp: now } as const;
+	const entry = await tx.appendEntry(UserEntry, conversationId, { model: [message] });
+	const input = { conversationId, ...requestId, type: "input", status: "placed", entry: entry.id } as const;
+	const { id } = await tx.createSubmission(input);
+	await startRun(tx, conversationId, live, [id]);
+	return id;
+}
--- a/packages/durable/src/harness/types.ts
+++ b/packages/durable/src/harness/types.ts
@ -364,6 +364,27 @@ export type ConversationRetryPolicy = {
 	maxAgentDelayMs?: number;
 };

+/** Automatic compaction thresholds (spec §8.7); manual compaction ignores `enabled`. */
+export type CompactionPolicy = {
+	/** Threshold and overflow compaction. */
+	enabled: boolean;
+	/** Room kept free for the answer: generation blocks to compact above `contextWindow - reserveTokens`. */
+	reserveTokens: number;
+	/** Approximate size of the recent context a summary keeps verbatim. */
+	keepRecentTokens: number;
+	/** Background compaction starts `backgroundTokens` below the blocking threshold; `0` disables it. */
+	backgroundTokens: number;
+};
+
+/** Why a compaction runs: `compact()`, a threshold in generation preparation, or a context overflow. */
+export type CompactionReason = "manual" | "threshold" | "overflow";
+
+/**
+ * `entryId` of a blocking compaction's summary, or the `submissionId` of a conversation-owned compaction's summary
+ * write; both absent when nothing was compacted.
+ */
+export type CompactionResult = { entryId?: EntryId; submissionId?: SubmissionId };
+
 export type HarnessOptions<Tool extends ToolRegistration = ToolRegistration> = {
 	/** pi-ai model access used by generation. */
 	readonly models: Models;
@ -413,6 +434,8 @@ export type ContextView = {
 	readonly head: EntryRecord | undefined;
 	/** Raw active entries: the head marker followed by non-head entries from its head through the tail. */
 	readonly entries: readonly EntryRecord[];
+	/** Per entry of `entries`, its model messages after edits and excluded stop reasons, before tool result ordering. */
+	readonly contributions: readonly (readonly Message[])[];
 	/** Model context for the next provider request. */
 	readonly messages: readonly Message[];
 };
@ -446,6 +469,10 @@ export interface Conversation {
 	getFollowUpMode(context: Context): Promise<QueueMode>;
 	/** `undefined` removes the configured mode. */
 	setFollowUpMode(mode: QueueMode | undefined, context: Context): Promise<void>;
+	/** `DEFAULT_COMPACTION_POLICY` when unset. */
+	getCompaction(context: Context): Promise<CompactionPolicy>;
+	/** `undefined` removes the configured policy. */
+	setCompaction(policy: CompactionPolicy | undefined, context: Context): Promise<void>;

 	/**
 	 * Durably admit user input or a passive entry write. A busy conversation, or one with queued items, queues it in
@ -457,6 +484,11 @@ export interface Conversation {
 	 * Resolves after admission; while busy, it is placed at the next boundary.
 	 */
 	reset(handoff: string | undefined, context: Context): Promise<void>;
+	/**
+	 * Admit a manual compaction task and return its ID. It summarizes while the conversation keeps working and places its
+	 * summary through a write submission: at once when idle, otherwise at the next boundary (spec §8.7).
+	 */
+	compact(instructions: string | undefined, context: Context): Promise<TaskId<CompactionResult>>;

 	/** Session commit whose `tx.createTask()` defaults to this conversation. */
 	commit<T>(change: (tx: Tx) => T | Promise<T>, context: Context): Promise<T>;
@ -566,3 +598,23 @@ export interface ToolHooks {
 		context: Context,
 	): HookResult<ToolExecutionResult>;
 }
+
+/** Hooks of the built-in compaction task. */
+export interface CompactionHooks {
+	/**
+	 * After range selection, before summarizing; the first decision wins. `entries` are the active entries the summary
+	 * replaces, the head marker first, and `messages` their model context, the summarizer's source; `firstKept` is the
+	 * first entry kept verbatim.
+	 */
+	beforeCompact(
+		compaction: {
+			readonly reason: CompactionReason;
+			readonly entries: readonly EntryRecord[];
+			readonly messages: readonly Message[];
+			readonly firstKept: EntryId;
+			readonly instructions?: string;
+		},
+		api: HookApi,
+		context: Context,
+	): HookResult<{ readonly decline: true } | { readonly summary: string }>;
+}
--- a/packages/durable/src/harness/usage.ts
+++ b/packages/durable/src/harness/usage.ts
@ -3,9 +3,9 @@ import type { Usage } from "@earendil-works/pi-ai";
 import { defineDoc } from "../documents.ts";
 import type { ConversationId, Tx } from "../types.ts";

-/** Spend recorded by one conversation's own entries; the entries stay authoritative. */
+/** Ledger of one conversation's own spend: its entries, and compaction summarization attempts, which have none. */
 export type UsageState = {
-	/** Assistant entries, keyed `provider/modelId`. */
+	/** Assistant entries and summarization attempts, keyed `provider/modelId`. */
 	models: Record<string, JsonRepresentation<Usage>>;
 	/** Tool results, keyed by tool name; their usage has no model identity. */
 	tools: Record<string, JsonRepresentation<Usage>>;
@ -21,7 +21,7 @@ export const UsageDoc = defineDoc<UsageState>({
 	checkpointWhen: () => true,
 });

-/** Add `usage` to one bucket of the conversation's `pi.usage`, in the commit that appends its entry. */
+/** Add `usage` to one bucket of the conversation's `pi.usage`, in the commit that records the response. */
 export async function recordUsage(
 	tx: Tx,
 	conversationId: ConversationId,
--- a/packages/durable/src/index.ts
+++ b/packages/durable/src/index.ts
@ -1,6 +1,20 @@
 export { defineDoc, defineDocFamily } from "./documents.ts";
-export { AssistantEntry, defineEntry, ResetEntry, SystemEntry, ToolResultEntry, UserEntry } from "./entries.ts";
+export {
+	AssistantEntry,
+	CompactionEntry,
+	defineEntry,
+	ResetEntry,
+	SystemEntry,
+	ToolResultEntry,
+	UserEntry,
+} from "./entries.ts";
 export { ConversationBusy, ReadAfterWrite, StorageRejected } from "./errors.ts";
+export {
+	type CompactionCheckpoint,
+	type CompactionInput,
+	CompactionTask,
+	type SummaryRequest,
+} from "./harness/compaction.ts";
 export { ConversationConfig, type ConversationConfigState } from "./harness/config.ts";
 export {
 	type AgentEvent,
@ -17,11 +31,15 @@ export {
 } from "./harness/generation.ts";
 export { Harness } from "./harness/harness.ts";
 export { InboxDoc, type InboxItem, type InboxState } from "./harness/inbox.ts";
-export { LiveDoc, type LiveState, type ToolSlot } from "./harness/live.ts";
+export { type CompactionStatus, LiveDoc, type LiveState, type ToolSlot } from "./harness/live.ts";
 export { createRegistry } from "./harness/registry.ts";
 export { ToolTask, type ToolTaskCheckpoint, type ToolTaskInput, type ToolTaskResult } from "./harness/tool.ts";
 export type {
 	AnyTask,
+	CompactionHooks,
+	CompactionPolicy,
+	CompactionReason,
+	CompactionResult,
 	ContextView,
 	Conversation,
 	ConversationAbortOptions,
--- a/packages/durable/test/examples/25-compaction.ts
+++ b/packages/durable/test/examples/25-compaction.ts
@ -0,0 +1,118 @@
+// Compaction: a long trip-planning chat whose older messages are summarized so the model context stays small.
+// Run from packages/durable:
+//   node --conditions=source --experimental-strip-types test/examples/25-compaction.ts
+import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
+import type { AssistantMessage, Message, TranscriptContext } from "@earendil-works/pi-ai";
+import { createModels } from "@earendil-works/pi-ai/models";
+import { fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai/providers/faux";
+import {
+	CompactionEntry,
+	type Conversation,
+	createRegistry,
+	Harness,
+	LiveDoc,
+	MemoryStorage,
+} from "../../src/index.ts";
+
+const context = BACKGROUND_CONTEXT;
+
+// A fake model with a tiny 3000-token window. It answers chat messages, writes summaries when asked to summarize,
+// and once rejects a request as too long, the way real providers report a context overflow.
+let overflowOnce = false;
+let summaries = 0;
+// While set, the next chat answer waits for it, which keeps the conversation busy.
+let hold: Promise<void> | undefined;
+async function respond(transcript: TranscriptContext): Promise<AssistantMessage> {
+	const first = transcript.messages[0];
+	if (first?.role === "system" && typeof first.content === "string" && first.content.includes("summarization")) {
+		summaries++;
+		return fauxAssistantMessage(`## Goal\nPlan a week in Lisbon (summary #${summaries}).`);
+	}
+	if (overflowOnce) {
+		overflowOnce = false;
+		return fauxAssistantMessage("", { stopReason: "error", errorMessage: "prompt is too long" });
+	}
+	const held = hold;
+	hold = undefined;
+	await held;
+	const question = transcript.messages.findLast((message) => message.role === "user");
+	return fauxAssistantMessage(`A detailed answer to "${text(question)}": ${"details ".repeat(200)}`);
+}
+const faux = fauxProvider({ models: [{ id: "tiny", contextWindow: 3000, maxTokens: 1000 }] });
+faux.setResponses(Array.from({ length: 100 }, () => respond));
+const models = createModels();
+models.setProvider(faux.provider);
+
+const harness = await Harness.open(new MemoryStorage(), { models, registry: createRegistry() }, context);
+const root = await harness.root(context);
+await root.setModel({ provider: "faux", modelId: "tiny" }, context);
+// Generation blocks to compact above 3000 - 1000 = 2000 tokens and starts a background compaction above 2000 - 800.
+await root.setCompaction({ enabled: true, reserveTokens: 1000, keepRecentTokens: 400, backgroundTokens: 800 }, context);
+
+async function ask(question: string): Promise<void> {
+	const submission = await root.submit({ type: "input", content: question }, context);
+	const record = await submission.wait(context);
+	// Let a background compaction started by this turn finish, so its summary shows below.
+	const running = (await harness.snapshot(LiveDoc, root.id, context))?.compactions ?? [];
+	for (const { taskId } of running) await harness.waitForTask(taskId, context);
+	await show(root, `after "${question}" (${record.status === "done" ? "answered" : record.reason})`);
+}
+
+// 1. A long chat: once the context crosses the background threshold, a compaction runs while the chat goes on, and
+// its summary lands at once when the conversation is idle, otherwise at the next turn boundary.
+for (const question of [
+	"Where should we stay?",
+	"What should we eat?",
+	"Which day trips?",
+	"Any museums?",
+	"Nightlife?",
+]) {
+	await ask(question);
+}
+
+// 2. A manual compaction while an answer is still being written. The summary is ready first, waits in the inbox,
+// and is placed right after the answer.
+let finishAnswer!: () => void;
+hold = new Promise((resolve) => {
+	finishAnswer = resolve;
+});
+const busy = await root.submit({ type: "input", content: "How do we get around?" }, context);
+const manual = await root.compact("Keep the hotel shortlist", context);
+const { state } = await harness.waitForTask(manual, context);
+const placement = state.outcome.status === "completed" ? state.outcome.result.submissionId : undefined;
+const summary = (await harness.submission(placement!, context))!;
+console.log("\nmanual compaction finished; its summary is", (await summary.status(context)).status);
+finishAnswer();
+await busy.wait(context);
+console.log("after the answer, the summary is", (await summary.wait(context)).status);
+await show(root, "after compact()");
+
+// 3. The provider rejects a request as too long: generation compacts and retries it once. Background compaction is
+// turned off so the summary below is the overflow one.
+await root.setCompaction({ enabled: true, reserveTokens: 1000, keepRecentTokens: 400, backgroundTokens: 0 }, context);
+await ask("What should we pack?");
+overflowOnce = true;
+await ask("Summarize the plan for my partner");
+
+await harness.close(context);
+
+/** Print the model context: one line per message, and the number of stored entries behind it. */
+async function show(conversation: Conversation, label: string): Promise<void> {
+	const view = await conversation.context(context);
+	const stored = (await conversation.entries({}, 1000, undefined, context)).items.length;
+	console.log(`\n${label}: ${view.messages.length} messages in context, ${stored} entries stored`);
+	const reason = CompactionEntry.is(view.head) ? view.head.data.reason : undefined;
+	if (reason !== undefined) console.log(`  (${reason} compaction summary first)`);
+	for (const message of view.messages) {
+		console.log(
+			`  ${message.role.padEnd(9)} ${message.role === "system" ? "(system prompt)" : text(message).slice(0, 70)}`,
+		);
+	}
+}
+
+function text(message: Message | undefined): string {
+	if (message === undefined || message.role === "system") return "";
+	if (typeof message.content === "string") return message.content;
+	const block = message.content.find((content) => content.type === "text");
+	return block?.type === "text" ? block.text.replaceAll("\n", " ") : "";
+}
--- a/packages/durable/test/harness-compaction.test.ts
+++ b/packages/durable/test/harness-compaction.test.ts
--- a/packages/durable/test/harness-registry.test.ts
+++ b/packages/durable/test/harness-registry.test.ts
@ -1,6 +1,7 @@
 import type { JsonValue } from "@earendil-works/chord";
 import { Type } from "@earendil-works/pi-ai";
 import {
+	CompactionTask,
 	createRegistry,
 	defineTask,
 	GenerationTask,
@ -208,7 +209,7 @@ describe("registry", () => {
 		const worker = task("worker");
 		registry.tasks.add(worker);
 		expect(() => registry.tasks.add(task("worker"))).toThrow("Task worker is already registered");
-		expect(registry.tasks.list()).toEqual([GenerationTask, ToolTask, worker]);
+		expect(registry.tasks.list()).toEqual([GenerationTask, ToolTask, CompactionTask, worker]);

 		const first = { beforeRun: () => {} };
 		const second = { beforeRun: () => {} };
@ -230,7 +231,7 @@ describe("registry", () => {

 	it("starts with undisposable, non-overridable built-in tasks", () => {
 		const registry = createRegistry();
-		expect(registry.tasks.list()).toEqual([GenerationTask, ToolTask]);
+		expect(registry.tasks.list()).toEqual([GenerationTask, ToolTask, CompactionTask]);
 		expect(registry.snapshot().task("pi.generation")).toBe(GenerationTask);
 		expect(() => registry.tasks.add({ definition: { ...GenerationTask.definition } })).toThrow(
 			"Task pi.generation is already registered",
--- a/scripts/check-entry-graphs.mjs
+++ b/scripts/check-entry-graphs.mjs
@ -44,7 +44,7 @@ const BUDGETS = {
 	"packages/durable": {
 		".": {
 			// The built-in tool task validates arguments with pi-ai's TypeBox-based validation, so TypeBox is allowed.
-			maxFiles: 56,
+			maxFiles: 58,
 			forbid: ["packages/ai/src/index.ts", "packages/ai/src/utils/typebox-helpers.ts"],
 		},
 	},