feat(agent): make run boundaries atomic

2026-11-05 23:37:25 +00:00 · 2026-08-26 17:34:56 +02:00 · 2026-08-26 17:34:56 +02:00 · 0f0ca0f6b7
commit 0f0ca0f6b7
parent c5a35eabbf
9 changed files with 705 additions and 495 deletions
--- a/packages/agent/docs/runtime-simplification.md
+++ b/packages/agent/docs/runtime-simplification.md
@ -41,7 +41,7 @@ M6 established structural execution on the simplified substrate. Its first coher
 - per-request structural intent and usage settlement, attempt retry/recovery, hook decisions, compaction/navigation publication, and terminal cleanup;
 - focused coverage for threshold/overflow entry, split-turn request accounting, generated and hook results, retry/cap/recovery, model absence, mid-request durable cancellation, navigation, and preparation corruption.

-R1a, R2, and R3 then moved queued input to the lane, replaced hydrated family outcomes with immutable operation records, and collapsed the family cross-product to 13 neutral leaves. At the reviewed R3 checkpoint the runtime is 5,722 lines; `runtime/drive` is 3,757 lines; and the structural module is 1,141 lines, down 319 lines (21.8%) from its 1,460-line foundation. The focused runtime set passes 189 tests across 15 files, with 44 memory/JSONL and 30 SQLite repository-conformance tests also passing. R1b boundary planning, M7 reconciliation/total dispatch, and M8 public surfaces have not started.
+R1a, R2, and R3 then moved queued input to the lane, replaced hydrated family outcomes with immutable operation records, and collapsed the family cross-product to 13 neutral leaves. R1b replaced the transitional checkpoint drains and marker fields with one shared atomic boundary planner, a transcript-derived threshold guard, and off-line finish-hook replanning. At the reviewed R1b checkpoint the runtime is 5,835 lines; `runtime/drive` is 3,870 lines; and the structural module is 1,170 lines, still down 290 lines (19.9%) from its 1,460-line foundation. The focused runtime set passes 194 tests across 15 files, with 44 memory/JSONL and 30 SQLite repository-conformance tests also passing. M7 reconciliation/total dispatch and M8 public surfaces have not started.

 Public drive remains disabled. Keep the visible durable procedure order:
 `prepare → publish intent → perform effect → publish outcome`.
--- a/packages/agent/docs/work-packages/05-direct-durable-drive.md
+++ b/packages/agent/docs/work-packages/05-direct-durable-drive.md
@ -1,6 +1,6 @@
 # WP05 — Direct durable drive

-**Status: M5 complete; M6 structural foundation committed; R1a lane-owned inbox relocation, R2 record-only outcomes, reviewed R3 family-neutral leaves, and R1b-1 structural boundary fusion implemented; R1b-2 next, then M7 (reconciliation), M8 (public surfaces), M9 (documentation reconciliation), and M10 (provider KV-cache identity review).**
+**Status: M5 complete; M6 structural foundation committed; R1a lane-owned inbox relocation, R2 record-only outcomes, reviewed R3 family-neutral leaves, and the reviewed R1b atomic boundary planner implemented; M7 reconciliation next, then M8 (public surfaces), M9 (documentation reconciliation), and M10 (provider KV-cache identity review).**

 WP06's Session/Branch/Lane separation is already part of the foundation. Public drive remains disabled until R1–R3, cancellation reconciliation, the total drive switch, every public execution surface, and the provider KV-cache identity review are complete.

@ -434,6 +434,17 @@ Order:

 `steer`/`followUp`/`nextRun` are tag sugar over one enqueue and always admit. `queue_update` and `LaneSnapshot.queues`/`pendingWrites` expose one identical shape: the tagged ordered inbox (clients group by tag).

+### Client replication surface
+
+Findings from a working RPC-shaped client replica (mini TUI exercise) fold in here. The goal is one testable contract: a remote client renders exclusively from a replicated `LaneSnapshot` and the event stream, with no side-channel getters.
+
+- **Normative reducer.** Export `reduceLaneSnapshot(snapshot, event): LaneSnapshot | { rebase: true }` from `packages/agent`. It is the single supported event fold: in-run `compaction_start`/`compaction_end` are segments and must not clear `operation` (the fold knows the open operation's kind, so no event field is needed); `run_suspend` keeps `operation` non-null with `deferred` set; only `run_end`, `navigation_end`, and `compaction_end` under a compaction-kind operation are operation-terminal. `navigation_end` returns `{ rebase: true }` because the moved tip may lie outside the replica. Conformance assertion: for every non-navigation flow, folding a `watch()` snapshot over its own event stream equals a later `watch()` snapshot — this equivalence test is what keeps the event vocabulary complete.
+- **Snapshot completeness.** `LaneSnapshot` gains `configuration: LaneConfiguration`, `lastResult?: OperationResultRecord` (replacing the bare `lastOperationId` field), a real `faulted` flag (the current hardcoded `false` is a defect), and session `stats: SessionStats` as the usage baseline (usage events already carry `totals`; the snapshot supplies the value before the first event).
+- **Replicable configuration events.** Every `config_update` variant whose value is data carries `value`/`previous` (`streamOptions`, `retryPolicy`, `compactionSettings`, `steeringMode`, `followUpMode`, plus the existing lane variants); `tools`/`resources` remain notification-only because registries are code — clients re-fetch names.
+- **Identity-based `setModel`.** `setModel` accepts `ModelIdentity` (`{ provider, modelId }`); a live `Model` object is a process-local registry concern, and an unregistered identity fails in-band at generation exactly like any registry absence.
+- **Re-basing.** `WatchHandle.resnapshot(context): Promise<LaneSnapshot>` captures a fresh snapshot on the mutation line, serialized against the same stream — the recovery path after `{ rebase: true }` with no teardown/re-subscribe choreography.
+- **Start-event timestamps.** `run_start`/`compaction_start`/`navigation_start` carry `startedAt` (operation starts: `meta.startedAt`; in-run segment starts: the commit timestamp), so folds never invent times.
+
 ### Convenience compositions and continuation

 - `prompt`/`skill`/`promptFromTemplate`: accept + drive; return `Result<OperationResultRecord | SuspendedRun, …>` — the record for terminal outcomes, `SuspendedRun` when the run defers.
@ -446,7 +457,7 @@ Order:

 ### Focused validation

-One install and same-id joins; stale-id isolation; record lookup for old ids; caller cancellation before/after installation; close/fault during cooperative and non-cooperative effects; accept/drive vs convenience equivalence including continuation via the public empty-prompt request; steer-only, followUp-only, and nextRun-only continuation each start run B; both continuation race orders; abort drain-and-return including the both-orders abort/settlement race; enqueue during every state through public surfaces; full crash matrix across all 13 leaves; no unhandled detached rejection; no `SliceNotImplemented` except `watchSession`.
+One install and same-id joins; stale-id isolation; record lookup for old ids; caller cancellation before/after installation; close/fault during cooperative and non-cooperative effects; accept/drive vs convenience equivalence including continuation via the public empty-prompt request; steer-only, followUp-only, and nextRun-only continuation each start run B; both continuation race orders; abort drain-and-return including the both-orders abort/settlement race; enqueue during every state through public surfaces; full crash matrix across all 13 leaves; no unhandled detached rejection; no `SliceNotImplemented` except `watchSession`; reducer fold-equivalence across every non-navigation flow (including in-run compaction segments, suspend/resume, retry waits, and reattach mid-stream); `navigation_end` fold returns rebase and `resnapshot` restores equivalence; snapshot `configuration`/`lastResult`/`faulted`/`stats` replicate through events alone.

 ## 11. M9 — Documentation reconciliation

@ -465,7 +476,10 @@ Update `docs/harness.md` to the implemented state; it must again be normative on
 - §4.6 abort: drain-and-return, no drained control, and the client-visible consequence — drained steer/followUp payloads exist only in the returned result, so a crash or lost response in that window loses their content permanently;
 - §5.1 lane surface and all result types; §5.5 `queue_update` and end-event payloads (`compaction_end` as a segment event with `aborted`);
 - Part 9: invariants 12–16, 21, 26 (result-record lifetime, observation via records, continuation equivalence), the new terminal-control invariant (a non-aborted terminal status implies running terminal control), and the race catalog rows touching drained items, `cancelQueued`, and continuation;
- Appendix A glossary (Inbox, Result record, Continuation run, Boundary pass; remove Drained).
+- Appendix A glossary (Inbox, Result record, Continuation run, Boundary pass; remove Drained);
+- §5.5 event taxonomy: state explicitly which events are operation-terminal (`run_end`, `navigation_end`, compaction-kind `compaction_end`) versus segment brackets (in-run `compaction_start`/`compaction_end`) versus non-terminal lifecycle (`run_suspend` leaves the operation durably open), and name `reduceLaneSnapshot` as the normative fold;
+- §2.5 branch-scan sharp edge: `stopAtType` applies after ordering, so `oldestFirst` + `stopAtType: "compaction"` returns the oldest segment; document the canonical context read as `newestFirst` + reverse;
+- serving-layer boundary statement: tree browsing, forks, labels inventory, and session listing are deliberately not `AgentLane` surface; an RPC facade composes a Session/repository read service beside the lane (protocol work, P1).

 Also update `docs/runtime-simplification.md` status and remove every remaining promotion reference in the repository's docs. Grep gate:

@ -535,7 +549,9 @@ Do not introduce:
 - external finalization;
 - storage rereads of authoritative control state;
 - read caches, read budgets, or generic `getValues` batching;
- compatibility aliases for any pre-redesign durable shape.
+- compatibility aliases for any pre-redesign durable shape;
+- a `phase`/segment discriminator field on structural events — the reducer derives segment-vs-terminal from the open operation's kind;
+- tree, fork, label-inventory, or repository methods on `AgentLane` — that surface lives beside the lane, not on it.

 Procedure-specific writes, effect admission, settlement classification, and event construction remain visible at their call sites.

--- a/packages/agent/src/harness/runtime/drive/boundary.ts
+++ b/packages/agent/src/harness/runtime/drive/boundary.ts
@ -0,0 +1,276 @@
+import type { AgentMessage } from "../../../types.ts";
+import type { HarnessEvent } from "../../agent-harness.ts";
+import { insertEntry } from "../../session/commit.ts";
+import { SessionInvariantError } from "../../session/session.ts";
+import type {
+	AssistantReadyOperation,
+	CheckpointOperation,
+	CommitResult,
+	NewEntry,
+	NormalizedRetryPolicy,
+	OperationScope,
+	SessionReader,
+	SummaryDecidingOperation,
+	SummaryEffectPendingOperation,
+	SummaryReadyOperation,
+	Write,
+} from "../../session/types.ts";
+import { branchTip, deleteValue, pendingEntry, setValue } from "../../session/values.ts";
+import type { Lane } from "../lane.ts";
+import { committedEntryEvents, entryLifecycleEvents, readBoundedContext, readPendingMessages } from "../transcript.ts";
+import type { Drive, LaneState, ProcedureResult } from "../types.ts";
+import { operationCleanupWrites, operationResultRecord } from "./terminal.ts";
+
+interface QueueSnapshot {
+	steer: Array<{ entryId: string; message: AgentMessage }>;
+	followUp: Array<{ entryId: string; message: AgentMessage }>;
+	nextRun: Array<{ entryId: string; message: AgentMessage }>;
+}
+
+export interface BoundaryFinishPending {
+	kind: "finish_pending";
+	entryIds: string[];
+}
+
+export interface BoundaryPlacement {
+	entries: NewEntry[];
+	writes: Write[];
+	tipId: string | null;
+	inbox: LaneState["inbox"];
+	triggerEntryId?: string;
+	queue?: QueueSnapshot;
+}
+
+type FinishBoundaryOperation =
+	| CheckpointOperation
+	| SummaryDecidingOperation
+	| SummaryReadyOperation
+	| SummaryEffectPendingOperation;
+
+export function normalizedRetryPolicy<TContext extends object | undefined>(
+	lane: Lane<TContext>,
+): NormalizedRetryPolicy {
+	const retry = lane.readConfig().retryPolicy;
+	return retry.enabled
+		? { maxAttempts: retry.maxRetries + 1, baseDelayMs: retry.baseDelayMs }
+		: { maxAttempts: 1, baseDelayMs: retry.baseDelayMs };
+}
+
+export function assistantReadyAtBoundary<TContext extends object | undefined>(
+	lane: Lane<TContext>,
+	state: LaneState,
+	scope: OperationScope,
+	triggerEntryId: string,
+	overflowRecoveryUsed: boolean,
+): AssistantReadyOperation {
+	const config = lane.readConfig();
+	return {
+		...scope,
+		at: "assistant.ready",
+		generationContext: {
+			stepId: lane.session.idGenerator.next(),
+			triggerEntryId,
+			configuration: state.configuration,
+			streamOptions: config.streamOptions,
+			retryPolicy: normalizedRetryPolicy(lane),
+			overflowRecoveryUsed,
+		},
+		nextAttempt: 1,
+	};
+}
+
+/** Select and materialize one boundary's lane-owned input without committing it. */
+export async function planBoundaryInbox<TContext extends object | undefined>(
+	lane: Lane<TContext>,
+	drive: Drive,
+	state: LaneState,
+	scope: OperationScope,
+	reader: SessionReader,
+	tipId: string | null,
+	followUpWhenNoTrigger: boolean,
+): Promise<BoundaryPlacement> {
+	const steer = state.inbox.filter((item) => item.kind === "steer");
+	const selectedSteer = scope.settings.steeringMode === "all" ? steer : steer.slice(0, 1);
+	let selected = state.inbox.filter(
+		(item) => item.kind === "write" || selectedSteer.some((candidate) => candidate.entryId === item.entryId),
+	);
+	const load = (items: typeof selected) =>
+		Promise.all(
+			items.map(async (item) => {
+				const stored = await reader.getValue(pendingEntry(item.entryId), drive.context);
+				if (stored === undefined) {
+					throw new SessionInvariantError(`Pending ${item.kind} entry ${item.entryId} is missing its payload`);
+				}
+				if (item.kind !== "write" && stored.value.type !== "message") {
+					throw new SessionInvariantError(`Queued ${item.kind} entry ${item.entryId} is not a message`);
+				}
+				return { item, pending: stored.value };
+			}),
+		);
+	let pending = await load(selected);
+	const projects = (value: (typeof pending)[number]["pending"]): boolean =>
+		value.type === "message" || lane.readConfig().entryProjectors[value.customType] !== undefined;
+	if (followUpWhenNoTrigger && !pending.some(({ pending: value }) => projects(value))) {
+		const followUp = state.inbox.filter((item) => item.kind === "followUp");
+		const selectedFollowUp = scope.settings.followUpMode === "all" ? followUp : followUp.slice(0, 1);
+		selected = [...selected, ...selectedFollowUp].sort((a, b) => state.inbox.indexOf(a) - state.inbox.indexOf(b));
+		pending = await load(selected);
+	}
+
+	let parentId = tipId;
+	let triggerEntryId: string | undefined;
+	const entries: NewEntry[] = pending.map(({ item, pending: value }) => {
+		const entry: NewEntry =
+			value.type === "message"
+				? { id: item.entryId, parentId, type: "message", message: value.payload }
+				: {
+						id: item.entryId,
+						parentId,
+						type: "custom",
+						customType: value.customType,
+						...(value.payload === undefined ? {} : { data: value.payload }),
+					};
+		parentId = item.entryId;
+		if (projects(value)) triggerEntryId = item.entryId;
+		return entry;
+	});
+	const selectedIds = new Set(selected.map((item) => item.entryId));
+	const inbox = state.inbox.filter((item) => !selectedIds.has(item.entryId));
+	const queue = selected.some((item) => item.kind === "steer" || item.kind === "followUp")
+		? await (async () => {
+				const remaining = (kind: "steer" | "followUp" | "nextRun") =>
+					inbox.filter((item) => item.kind === kind).map((item) => item.entryId);
+				const [remainingSteer, remainingFollowUp, nextRun] = await Promise.all([
+					readPendingMessages(reader, remaining("steer"), "Steer entry", drive.context),
+					readPendingMessages(reader, remaining("followUp"), "Follow-up entry", drive.context),
+					readPendingMessages(reader, remaining("nextRun"), "Pending next-run entry", drive.context),
+				]);
+				return { steer: remainingSteer, followUp: remainingFollowUp, nextRun };
+			})()
+		: undefined;
+	return {
+		entries,
+		writes: [
+			...entries.map((entry) => insertEntry(entry)),
+			...selected.map((item) => deleteValue(pendingEntry(item.entryId))),
+			...(entries.length === 0 ? [] : [setValue(branchTip(lane.name), parentId)]),
+		],
+		tipId: parentId,
+		inbox,
+		...(triggerEntryId === undefined ? {} : { triggerEntryId }),
+		...(queue === undefined ? {} : { queue }),
+	};
+}
+
+export function boundaryPlacementEvents(
+	placement: BoundaryPlacement,
+	commit: CommitResult,
+	firstWriteIndex: number,
+	lane: string,
+	runId: string,
+): HarnessEvent[] {
+	return [
+		...committedEntryEvents(placement.entries, commit, lane, runId, firstWriteIndex),
+		...(placement.queue === undefined ? [] : [{ type: "queue_update" as const, lane, ...placement.queue }]),
+	];
+}
+
+/** Replan after before_run_end and commit either renewed work or the terminal run result. */
+export async function finishRunBoundary<TContext extends object | undefined, TState extends FinishBoundaryOperation>(
+	lane: Lane<TContext>,
+	drive: Drive,
+	capability: TState,
+	continuation: Extract<CheckpointOperation["continuation"], { kind: "may_finish" }>,
+	plannedEntryIds: readonly string[],
+	pendingEvents: HarnessEvent[] = [],
+): Promise<ProcedureResult> {
+	const context = await readBoundedContext(lane, drive, capability);
+	if (context.kind === "cancel_requested") return { kind: "continue" };
+	const hook = await lane.hooks.runWithGate(
+		"before_run_end",
+		{ lane: lane.name, runId: drive.operationId, messages: context.value },
+		drive.gate,
+		drive.context,
+	);
+	const followUp =
+		hook?.followUp === undefined
+			? undefined
+			: {
+					id: lane.session.idGenerator.next(),
+					message: { role: "user" as const, content: hook.followUp, timestamp: Date.now() },
+				};
+	const result = await lane.continueOperation<TState, ProcedureResult>(
+		capability,
+		async (state, current, meta, reader) => {
+			const placement = await planBoundaryInbox(lane, drive, state, current, reader, state.tipId, true);
+			if (placement.triggerEntryId !== undefined) {
+				return {
+					kind: "commit",
+					writes: placement.writes,
+					operationState: assistantReadyAtBoundary(lane, state, current, placement.triggerEntryId, false),
+					lane: { tipId: placement.tipId, inbox: placement.inbox },
+					materialize: () => ({ kind: "continue" }) as const,
+					events: (commit) => [
+						...pendingEvents,
+						...boundaryPlacementEvents(placement, commit, 0, lane.name, drive.operationId),
+					],
+				};
+			}
+			const hookPlanIsCurrent =
+				placement.entries.length === plannedEntryIds.length &&
+				placement.entries.every((entry, index) => entry.id === plannedEntryIds[index]);
+			if (hookPlanIsCurrent && followUp !== undefined) {
+				const entry: NewEntry = {
+					id: followUp.id,
+					parentId: placement.tipId,
+					type: "message",
+					message: followUp.message,
+				};
+				const entryWriteIndex = placement.writes.length;
+				return {
+					kind: "commit",
+					writes: [...placement.writes, insertEntry(entry), setValue(branchTip(lane.name), followUp.id)],
+					operationState: assistantReadyAtBoundary(lane, state, current, followUp.id, false),
+					lane: { tipId: followUp.id, inbox: placement.inbox },
+					materialize: () => ({ kind: "continue" }) as const,
+					events: (commit) => [
+						...pendingEvents,
+						...boundaryPlacementEvents(placement, commit, 0, lane.name, drive.operationId),
+						...entryLifecycleEvents(
+							{ ...entry, seq: commit.seqs[entryWriteIndex]!, timestamp: commit.timestamp },
+							lane.name,
+							drive.operationId,
+						),
+					],
+				};
+			}
+			if (placement.tipId === null) throw new SessionInvariantError("Completed run has no tip");
+			if (continuation.includeFinalAssistant && current.latestAssistantEntryId === null) {
+				throw new SessionInvariantError("Completed run is missing its final assistant");
+			}
+			const record = operationResultRecord(meta, "completed", placement.tipId);
+			const cleanup = await operationCleanupWrites(reader, drive.operationId, current, drive.context);
+			return {
+				kind: "finish",
+				writes: [...placement.writes, ...cleanup],
+				record,
+				lane: { tipId: placement.tipId, inbox: placement.inbox },
+				materialize: () => ({ kind: "settled", outcome: record }) as const,
+				events: (commit) => [
+					...pendingEvents,
+					...boundaryPlacementEvents(placement, commit, 0, lane.name, drive.operationId),
+					{
+						type: "run_end",
+						lane: lane.name,
+						runId: drive.operationId,
+						status: "completed",
+						fromTipId: meta.sourceTipId,
+						tipId: placement.tipId,
+					},
+				],
+			};
+		},
+		drive.context,
+	);
+	return result.kind === "cancel_requested" ? { kind: "continue" } : result.value;
+}
--- a/packages/agent/src/harness/runtime/drive/checkpoint.ts
+++ b/packages/agent/src/harness/runtime/drive/checkpoint.ts
@ -1,27 +1,25 @@
 import { insertEntry } from "../../session/commit.ts";
 import { SessionInvariantError } from "../../session/session.ts";
 import {
-	type AssistantReadyOperation,
 	type CheckpointOperation,
 	type MessageEntry,
 	type NewEntry,
 	operationScopeOf,
 	type StartingOperation,
+	type SummaryDecidingOperation,
 } from "../../session/types.ts";
-import { branchTip, deleteValue, pendingEntry, setValue } from "../../session/values.ts";
+import { branchTip, operationPreparation, setValue } from "../../session/values.ts";
 import type { Lane } from "../lane.ts";
-import {
-	chainEntries,
-	committedEntryEvents,
-	entryLifecycleEvents,
-	readBoundedContext,
-	readPendingMessages,
-} from "../transcript.ts";
+import { chainEntries, committedEntryEvents } from "../transcript.ts";
 import type { Drive, ProcedureResult } from "../types.ts";
-import { runCompactionThreshold } from "./structural.ts";
-import { operationCleanupWrites, operationResultRecord } from "./terminal.ts";
-
-type FinishContinuation = Extract<CheckpointOperation["continuation"], { kind: "may_finish" }>;
+import {
+	assistantReadyAtBoundary,
+	type BoundaryFinishPending,
+	boundaryPlacementEvents,
+	finishRunBoundary,
+	planBoundaryInbox,
+} from "./boundary.ts";
+import { prepareCompactionThreshold } from "./structural.ts";

 /** Consume before_run and commit the initial checkpoint. */
 export async function startRun<TContext extends object | undefined>(
@ -93,283 +91,94 @@ export async function startRun<TContext extends object | undefined>(
 	return result.kind === "cancel_requested" ? { kind: "continue" } : result.value;
 }

-async function applyPendingWrites<TContext extends object | undefined>(
-	lane: Lane<TContext>,
-	drive: Drive,
-	checkpoint: CheckpointOperation,
-): Promise<ProcedureResult> {
-	const result = await lane.continueOperation(
-		checkpoint,
-		async (state, current, _meta, reader) => {
-			const items = state.inbox.filter((item) => item.kind === "write");
-			const ids = items.map((item) => item.entryId);
-			if (ids.length === 0) return { kind: "return", result: { kind: "continue" } as const };
-			const pending = await Promise.all(
-				ids.map(async (id) => {
-					const value = await reader.getValue(pendingEntry(id), drive.context);
-					if (value === undefined) throw new SessionInvariantError(`Pending write ${id} is missing its payload`);
-					return { id, pending: value.value };
-				}),
-			);
-			let parentId = state.tipId;
-			let triggerEntryId: string | undefined;
-			const entries: NewEntry[] = pending.map(({ id, pending: item }) => {
-				const entry: NewEntry =
-					item.type === "message"
-						? { id, parentId, type: "message", message: item.payload }
-						: {
-								id,
-								parentId,
-								type: "custom",
-								customType: item.customType,
-								...(item.payload === undefined ? {} : { data: item.payload }),
-							};
-				parentId = id;
-				if (item.type === "message" || lane.readConfig().entryProjectors[item.customType] !== undefined) {
-					triggerEntryId = id;
-				}
-				return entry;
-			});
-			const inbox = state.inbox.filter((item) => item.kind !== "write");
-			const nextState: CheckpointOperation =
-				triggerEntryId === undefined
-					? current
-					: {
-							...operationScopeOf(current),
-							at: "checkpoint",
-							continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
-							triggerEntryId,
-							skipInboxOnce: true,
-						};
-			return {
-				kind: "commit",
-				writes: [
-					...entries.map((entry) => insertEntry(entry)),
-					...ids.map((id) => deleteValue(pendingEntry(id))),
-					setValue(branchTip(lane.name), parentId),
-				],
-				operationState: nextState,
-				lane: { tipId: parentId, inbox },
-				materialize: () => ({ kind: "continue" }) as const,
-				events: (commit) => committedEntryEvents(entries, commit, lane.name, drive.operationId),
-			};
-		},
-		drive.context,
-	);
-	return result.kind === "cancel_requested" ? { kind: "continue" } : result.value;
-}
-
-async function consumeQueuedMessages<TContext extends object | undefined>(
-	lane: Lane<TContext>,
-	drive: Drive,
-	checkpoint: CheckpointOperation,
-	queue: "steer" | "followUp",
-): Promise<ProcedureResult> {
-	const result = await lane.continueOperation(
-		checkpoint,
-		async (state, current, _meta, reader) => {
-			const candidates = state.inbox.filter((item) => item.kind === queue);
-			const mode = queue === "steer" ? current.settings.steeringMode : current.settings.followUpMode;
-			const selected = mode === "all" ? candidates : candidates.slice(0, 1);
-			const ids = selected.map((item) => item.entryId);
-			if (ids.length === 0) return { kind: "return", result: { kind: "continue" } as const };
-			const messages = (await readPendingMessages(reader, ids, `Queued ${queue} entry`, drive.context)).map(
-				({ entryId, message }) => ({ id: entryId, message }),
-			);
-			const selectedIds = new Set(ids);
-			const inbox = state.inbox.filter((item) => !selectedIds.has(item.entryId));
-			const remainingIds = (kind: "steer" | "followUp" | "nextRun") =>
-				inbox.filter((item) => item.kind === kind).map((item) => item.entryId);
-			const [remainingSteer, remainingFollowUp, nextRun] = await Promise.all([
-				readPendingMessages(reader, remainingIds("steer"), "Steer entry", drive.context),
-				readPendingMessages(reader, remainingIds("followUp"), "Follow-up entry", drive.context),
-				readPendingMessages(reader, remainingIds("nextRun"), "Pending next-run entry", drive.context),
-			]);
-			const entries: NewEntry<MessageEntry>[] = chainEntries(
-				state.tipId,
-				messages.map(({ id, message }) => ({ id, type: "message" as const, message })),
-			);
-			const triggerEntryId = messages[messages.length - 1]!.id;
-			const nextState: CheckpointOperation = {
-				...operationScopeOf(current),
-				at: "checkpoint",
-				continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
-				triggerEntryId,
-				skipInboxOnce: true,
-			};
-			return {
-				kind: "commit",
-				writes: [
-					...entries.map((entry) => insertEntry(entry)),
-					...ids.map((id) => deleteValue(pendingEntry(id))),
-					setValue(branchTip(lane.name), triggerEntryId),
-				],
-				operationState: nextState,
-				lane: { tipId: triggerEntryId, inbox },
-				materialize: () => ({ kind: "continue" }) as const,
-				events: (commit) => [
-					...committedEntryEvents(entries, commit, lane.name, drive.operationId),
-					{
-						type: "queue_update",
-						lane: lane.name,
-						steer: remainingSteer,
-						followUp: remainingFollowUp,
-						nextRun,
-					},
-				],
-			};
-		},
-		drive.context,
-	);
-	return result.kind === "cancel_requested" ? { kind: "continue" } : result.value;
-}
-
-async function finishRun<TContext extends object | undefined>(
-	lane: Lane<TContext>,
-	drive: Drive,
-	checkpoint: CheckpointOperation,
-	continuation: FinishContinuation,
-): Promise<ProcedureResult> {
-	const context = await readBoundedContext(lane, drive, checkpoint);
-	if (context.kind === "cancel_requested") return { kind: "continue" };
-	const hook = await lane.hooks.runWithGate(
-		"before_run_end",
-		{ lane: lane.name, runId: drive.operationId, messages: context.value },
-		drive.gate,
-		drive.context,
-	);
-	const followUp =
-		hook?.followUp === undefined
-			? undefined
-			: {
-					id: lane.session.idGenerator.next(),
-					message: { role: "user" as const, content: hook.followUp, timestamp: Date.now() },
-				};
-
-	const result = await lane.continueOperation<CheckpointOperation, ProcedureResult>(
-		checkpoint,
-		async (state, current, meta, reader) => {
-			if (state.inbox.some((item) => item.kind !== "nextRun")) {
-				return { kind: "return", result: { kind: "continue" } as const };
-			}
-			if (followUp !== undefined) {
-				const nextState: CheckpointOperation = {
-					...operationScopeOf(current),
-					at: "checkpoint",
-					continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
-					triggerEntryId: followUp.id,
-				};
-				const entry: NewEntry<MessageEntry> = {
-					id: followUp.id,
-					parentId: state.tipId,
-					type: "message",
-					message: followUp.message,
-				};
-				return {
-					kind: "commit",
-					writes: [insertEntry(entry), setValue(branchTip(lane.name), followUp.id)],
-					operationState: nextState,
-					lane: { tipId: followUp.id },
-					materialize: () => ({ kind: "continue" }) as const,
-					events: (commit) =>
-						entryLifecycleEvents(
-							{ ...entry, seq: commit.seqs[0]!, timestamp: commit.timestamp },
-							lane.name,
-							drive.operationId,
-						),
-				};
-			}
-
-			if (state.tipId === null) throw new SessionInvariantError("Completed run has no tip");
-			if (continuation.includeFinalAssistant && current.latestAssistantEntryId === null) {
-				throw new SessionInvariantError("Completed run is missing its final assistant");
-			}
-			const record = operationResultRecord(meta, "completed", state.tipId);
-			const cleanup = await operationCleanupWrites(reader, drive.operationId, current, drive.context);
-			return {
-				kind: "finish",
-				writes: cleanup,
-				record,
-				materialize: () => ({ kind: "settled", outcome: record }) as const,
-				events: () => [
-					{
-						type: "run_end",
-						lane: lane.name,
-						runId: drive.operationId,
-						status: "completed",
-						fromTipId: meta.sourceTipId,
-						tipId: state.tipId,
-					},
-				],
-			};
-		},
-		drive.context,
-	);
-	return result.kind === "cancel_requested" ? { kind: "continue" } : result.value;
-}
-
-/** Advance one durable run checkpoint by one visible transition. */
+/** Advance one durable run boundary with at most one commit. */
 export async function runCheckpoint<TContext extends object | undefined>(
 	lane: Lane<TContext>,
 	drive: Drive,
 	run: CheckpointOperation,
 ): Promise<ProcedureResult> {
-	if (run.skipInboxOnce !== true && lane.state.inbox.some((item) => item.kind === "write")) {
-		return applyPendingWrites(lane, drive, run);
-	}
-	if (run.skipInboxOnce !== true && lane.state.inbox.some((item) => item.kind === "steer")) {
-		return consumeQueuedMessages(lane, drive, run, "steer");
-	}
-	if (run.thresholdCheckedTriggerEntryId !== run.triggerEntryId) {
-		const result = await runCompactionThreshold(lane, drive, run);
-		if (result.kind !== "continue") return result;
-		const current = lane.state.operation?.state;
-		if (current?.control.status === "cancel_requested" || current?.at !== "checkpoint") return result;
-		return runCheckpoint(lane, drive, current);
-	}
-	const continuation = run.continuation;
-	if (continuation.kind === "need_assistant") {
-		const overflowRecoveryUsed = continuation.overflowRecoveryUsed;
-		const config = lane.readConfig();
-		const retryPolicy = config.retryPolicy.enabled
-			? { maxAttempts: config.retryPolicy.maxRetries + 1, baseDelayMs: config.retryPolicy.baseDelayMs }
-			: { maxAttempts: 1, baseDelayMs: config.retryPolicy.baseDelayMs };
-		const stepId = lane.session.idGenerator.next();
-		const result = await lane.continueOperation(
-			run,
-			(state, current) => {
-				if (
-					current.skipInboxOnce !== true &&
-					state.inbox.some((item) => item.kind === "write" || item.kind === "steer")
-				) {
-					return { kind: "return", result: { kind: "continue" } as const };
-				}
-				const nextState: AssistantReadyOperation = {
+	const threshold = await prepareCompactionThreshold(lane, drive, run);
+	if (threshold.kind === "cancel_requested") return { kind: "continue" };
+	const planned = await lane.continueOperation<CheckpointOperation, ProcedureResult | BoundaryFinishPending>(
+		run,
+		async (state, current, _meta, reader) => {
+			const placement = await planBoundaryInbox(
+				lane,
+				drive,
+				state,
+				current,
+				reader,
+				state.tipId,
+				threshold.value === undefined && current.continuation.kind === "may_finish",
+			);
+			if (placement.triggerEntryId !== undefined) {
+				return {
+					kind: "commit",
+					writes: placement.writes,
+					operationState: assistantReadyAtBoundary(lane, state, current, placement.triggerEntryId, false),
+					lane: { tipId: placement.tipId, inbox: placement.inbox },
+					materialize: () => ({ kind: "continue" }) as const,
+					events: (commit) => boundaryPlacementEvents(placement, commit, 0, lane.name, drive.operationId),
+				};
+			}
+			if (threshold.value !== undefined) {
+				const structural: SummaryDecidingOperation = {
 					...operationScopeOf(current),
-					at: "assistant.ready",
-					generationContext: {
-						stepId,
-						triggerEntryId: current.triggerEntryId,
-						configuration: state.configuration,
-						streamOptions: config.streamOptions,
-						retryPolicy,
-						overflowRecoveryUsed,
+					at: "summary.deciding",
+					task: {
+						taskId: threshold.value.taskId,
+						reason: "threshold",
+						boundary: {
+							kind: "resume_checkpoint",
+							resumeAfter: { continuation: current.continuation, triggerEntryId: current.triggerEntryId },
+						},
 					},
-					nextAttempt: 1,
 				};
 				return {
 					kind: "commit",
-					writes: [],
-					operationState: nextState,
+					writes: [
+						...placement.writes,
+						setValue(
+							operationPreparation(drive.operationId, threshold.value.taskId),
+							threshold.value.preparation,
+						),
+					],
+					operationState: structural,
+					lane: { tipId: placement.tipId, inbox: placement.inbox },
 					materialize: () => ({ kind: "continue" }) as const,
+					events: (commit) => [
+						...boundaryPlacementEvents(placement, commit, 0, lane.name, drive.operationId),
+						{ type: "compaction_start", lane: lane.name, runId: drive.operationId, reason: "threshold" },
+					],
 				};
-			},
-			drive.context,
-		);
-		return result.kind === "cancel_requested" ? { kind: "continue" } : result.value;
+			}
+			if (current.continuation.kind === "need_assistant") {
+				return {
+					kind: "commit",
+					writes: placement.writes,
+					operationState: assistantReadyAtBoundary(
+						lane,
+						state,
+						current,
+						current.triggerEntryId,
+						current.continuation.overflowRecoveryUsed,
+					),
+					lane: { tipId: placement.tipId, inbox: placement.inbox },
+					materialize: () => ({ kind: "continue" }) as const,
+					events: (commit) => boundaryPlacementEvents(placement, commit, 0, lane.name, drive.operationId),
+				};
+			}
+			return {
+				kind: "return",
+				result: { kind: "finish_pending", entryIds: placement.entries.map((entry) => entry.id) } as const,
+			};
+		},
+		drive.context,
+	);
+	if (planned.kind === "cancel_requested") return { kind: "continue" };
+	if (planned.value.kind !== "finish_pending") return planned.value;
+	if (run.continuation.kind !== "may_finish") {
+		throw new SessionInvariantError("Checkpoint finish mediation requires a finish continuation");
 	}
-	if (lane.state.inbox.some((item) => item.kind === "followUp")) {
-		return consumeQueuedMessages(lane, drive, run, "followUp");
-	}
-	return finishRun(lane, drive, run, continuation);
+	return finishRunBoundary(lane, drive, run, run.continuation, planned.value.entryIds);
 }
--- a/packages/agent/src/harness/runtime/drive/response.ts
+++ b/packages/agent/src/harness/runtime/drive/response.ts
@ -238,7 +238,6 @@ export async function publishResponse<TContext extends object | undefined>(
 								resumeAfter: {
 									continuation: { kind: "need_assistant", overflowRecoveryUsed: true },
 									triggerEntryId: current.generationContext.triggerEntryId,
-									thresholdCheckedTriggerEntryId: current.generationContext.triggerEntryId,
 								},
 							},
 						},
--- a/packages/agent/src/harness/runtime/drive/structural.ts
+++ b/packages/agent/src/harness/runtime/drive/structural.ts
@ -28,7 +28,6 @@ import {
 	type LaneConfiguration,
 	type NavigationReadyToCommitOperation,
 	type NewEntry,
-	type NormalizedRetryPolicy,
 	type OperationError,
 	operationScopeOf,
 	type ResultBoundary,
@ -41,18 +40,19 @@ import {
 	type UsageRow,
 	type Write,
 } from "../../session/types.ts";
-import {
-	branchTip,
-	deleteValue,
-	entryLabel,
-	operationPreparation,
-	pendingEntry,
-	setValue,
-} from "../../session/values.ts";
+import { branchTip, entryLabel, operationPreparation, setValue } from "../../session/values.ts";
 import type { AgentHarnessStreamOptions } from "../../types.ts";
 import type { Lane } from "../lane.ts";
-import { committedEntryEvents, readBoundedEntries, readPendingMessages } from "../transcript.ts";
+import { committedEntryEvents, readBoundedEntries } from "../transcript.ts";
 import type { ContinueOperationResult, Drive, ProcedureResult } from "../types.ts";
+import {
+	assistantReadyAtBoundary,
+	type BoundaryFinishPending,
+	boundaryPlacementEvents,
+	finishRunBoundary,
+	normalizedRetryPolicy,
+	planBoundaryInbox,
+} from "./boundary.ts";
 import { retryDelay, retryNotBefore, waitUntil } from "./retry.ts";
 import { operationCleanupWrites, operationResultRecord } from "./terminal.ts";

@ -63,13 +63,6 @@ class StructuralCancelled extends Error {
 	}
 }

-function normalizedRetryPolicy<TContext extends object | undefined>(lane: Lane<TContext>): NormalizedRetryPolicy {
-	const retry = lane.readConfig().retryPolicy;
-	return retry.enabled
-		? { maxAttempts: retry.maxRetries + 1, baseDelayMs: retry.baseDelayMs }
-		: { maxAttempts: 1, baseDelayMs: retry.baseDelayMs };
-}
-
 function durableFileOperations(fileOps: CompactionPreparation["fileOps"]): DurableFileOperations {
 	return {
 		read: [...fileOps.read],
@ -212,6 +205,8 @@ type StructuralOutcome =
 	| { kind: "declined" }
 	| { kind: "failed"; error: OperationError };

+type StructuralPublication = ProcedureResult | BoundaryFinishPending;
+
 async function publishStructuralOutcome<TContext extends object | undefined>(
 	lane: Lane<TContext>,
 	drive: Drive,
@ -226,7 +221,7 @@ async function publishStructuralOutcome<TContext extends object | undefined>(
 			: undefined;
 	const published = await lane.continueOperation<
 		SummaryDecidingOperation | SummaryReadyOperation | SummaryEffectPendingOperation,
-		ProcedureResult
+		StructuralPublication
 	>(
 		capability,
 		async (state, current, meta, reader) => {
@ -309,99 +304,56 @@ async function publishStructuralOutcome<TContext extends object | undefined>(
 						(outcome.kind === "declined" && current.task.reason === "threshold")
 					) {
 						if (terminalTipId === null) throw new SessionInvariantError("Run compaction has no Branch tip");
-						const steer = state.inbox.filter((item) => item.kind === "steer");
-						const selectedSteer = new Set(
-							(current.settings.steeringMode === "all" ? steer : steer.slice(0, 1)).map((item) => item.entryId),
-						);
-						const selected = state.inbox.filter(
-							(item) => item.kind === "write" || (item.kind === "steer" && selectedSteer.has(item.entryId)),
-						);
-						const pending = await Promise.all(
-							selected.map(async (item) => {
-								const stored = await reader.getValue(pendingEntry(item.entryId), drive.context);
-								if (stored === undefined) {
-									throw new SessionInvariantError(
-										`Pending ${item.kind} entry ${item.entryId} is missing its payload`,
-									);
-								}
-								if (item.kind === "steer" && stored.value.type !== "message") {
-									throw new SessionInvariantError(`Queued steer entry ${item.entryId} is not a message`);
-								}
-								return { item, pending: stored.value };
-							}),
-						);
-						let parentId = terminalTipId;
-						let triggerEntryId: string | undefined;
-						const entries: NewEntry[] = pending.map(({ item, pending: value }) => {
-							const entry: NewEntry =
-								value.type === "message"
-									? { id: item.entryId, parentId, type: "message", message: value.payload }
-									: {
-											id: item.entryId,
-											parentId,
-											type: "custom",
-											customType: value.customType,
-											...(value.payload === undefined ? {} : { data: value.payload }),
-										};
-							parentId = item.entryId;
-							if (
-								value.type === "message" ||
-								lane.readConfig().entryProjectors[value.customType] !== undefined
-							) {
-								triggerEntryId = item.entryId;
-							}
-							return entry;
-						});
-						const selectedIds = new Set(selected.map((item) => item.entryId));
-						const inbox = state.inbox.filter((item) => !selectedIds.has(item.entryId));
-						const queueUpdate =
-							selectedSteer.size === 0
-								? undefined
-								: await Promise.all(
-										(["steer", "followUp", "nextRun"] as const).map((kind) =>
-											readPendingMessages(
-												reader,
-												inbox.filter((item) => item.kind === kind).map((item) => item.entryId),
-												`Pending ${kind} entry`,
-												drive.context,
-											),
-										),
-									);
-						const entryWriteIndex = writes.length;
-						writes.push(
-							...entries.map((entry) => insertEntry(entry)),
-							...selected.map((item) => deleteValue(pendingEntry(item.entryId))),
-							...(entries.length === 0 ? [] : [setValue(branchTip(lane.name), parentId)]),
-						);
 						const continuation = current.task.boundary.resumeAfter.continuation;
-						const operationState: AssistantReadyOperation | CheckpointOperation =
-							triggerEntryId !== undefined || continuation.kind === "need_assistant"
-								? {
-										...operationScopeOf(current),
-										at: "assistant.ready",
-										generationContext: {
-											stepId: lane.session.idGenerator.next(),
-											triggerEntryId: triggerEntryId ?? current.task.boundary.resumeAfter.triggerEntryId,
-											configuration: state.configuration,
-											streamOptions: lane.readConfig().streamOptions,
-											retryPolicy: normalizedRetryPolicy(lane),
-											overflowRecoveryUsed:
-												triggerEntryId === undefined && continuation.kind === "need_assistant"
-													? continuation.overflowRecoveryUsed
-													: false,
-										},
-										nextAttempt: 1,
-									}
-								: {
-										...operationScopeOf(current),
-										at: "checkpoint",
-										...current.task.boundary.resumeAfter,
-									};
+						const placement = await planBoundaryInbox(
+							lane,
+							drive,
+							state,
+							current,
+							reader,
+							terminalTipId,
+							outcome.kind === "declined" && continuation.kind === "may_finish",
+						);
+						if (
+							outcome.kind === "declined" &&
+							placement.triggerEntryId === undefined &&
+							continuation.kind === "may_finish"
+						) {
+							return {
+								kind: "return",
+								result: {
+									kind: "finish_pending",
+									entryIds: placement.entries.map((entry) => entry.id),
+								} as const,
+							};
+						}
+						const placementWriteIndex = writes.length;
+						writes.push(...placement.writes);
+						let operationState: AssistantReadyOperation | CheckpointOperation;
+						if (placement.triggerEntryId !== undefined || continuation.kind === "need_assistant") {
+							const overflowRecoveryUsed =
+								placement.triggerEntryId === undefined && continuation.kind === "need_assistant"
+									? continuation.overflowRecoveryUsed
+									: false;
+							operationState = assistantReadyAtBoundary(
+								lane,
+								state,
+								current,
+								placement.triggerEntryId ?? current.task.boundary.resumeAfter.triggerEntryId,
+								overflowRecoveryUsed,
+							);
+						} else {
+							operationState = {
+								...operationScopeOf(current),
+								at: "checkpoint",
+								...current.task.boundary.resumeAfter,
+							};
+						}
 						return {
 							kind: "commit",
 							writes,
 							operationState,
-							lane: { tipId: parentId, inbox },
+							lane: { tipId: placement.tipId, inbox: placement.inbox },
 							materialize: () => ({ kind: "continue" }) as const,
 							events: (commit) => [
 								...(outcome.kind === "compaction"
@ -415,18 +367,13 @@ async function publishStructuralOutcome<TContext extends object | undefined>(
 												status: "declined" as const,
 											},
 										]),
-								...committedEntryEvents(entries, commit, lane.name, drive.operationId, entryWriteIndex),
-								...(queueUpdate === undefined
-									? []
-									: [
-											{
-												type: "queue_update" as const,
-												lane: lane.name,
-												steer: queueUpdate[0],
-												followUp: queueUpdate[1],
-												nextRun: queueUpdate[2],
-											},
-										]),
+								...boundaryPlacementEvents(
+									placement,
+									commit,
+									placementWriteIndex,
+									lane.name,
+									drive.operationId,
+								),
 							],
 						};
 					}
@ -563,7 +510,21 @@ async function publishStructuralOutcome<TContext extends object | undefined>(
 		},
 		drive.context,
 	);
-	return published.kind === "cancel_requested" ? { kind: "continue" } : published.value;
+	if (published.kind === "cancel_requested") return { kind: "continue" };
+	if (published.value.kind !== "finish_pending") return published.value;
+	const boundary = capability.task.boundary;
+	if (boundary.kind !== "resume_checkpoint" || boundary.resumeAfter.continuation.kind !== "may_finish") {
+		throw new SessionInvariantError("Structural finish mediation requires a resumable finish boundary");
+	}
+	return finishRunBoundary(lane, drive, capability, boundary.resumeAfter.continuation, published.value.entryIds, [
+		{
+			type: "compaction_end",
+			lane: lane.name,
+			runId: drive.operationId,
+			reason: "threshold",
+			status: "declined",
+		},
+	]);
 }

 async function publishStructuralReady<TContext extends object | undefined>(
@ -1102,72 +1063,41 @@ export async function recoverStructuralGeneration<TContext extends object | unde
 	return published.kind === "cancel_requested" ? { kind: "continue" } : published.value;
 }

-/** Check and durably mark one run checkpoint's threshold boundary. */
-export async function runCompactionThreshold<TContext extends object | undefined>(
+/** Prepare threshold compaction only when no newer compaction already guards this trigger. */
+export async function prepareCompactionThreshold<TContext extends object | undefined>(
 	lane: Lane<TContext>,
 	drive: Drive,
 	checkpoint: CheckpointOperation,
-): Promise<ProcedureResult> {
+): Promise<ContinueOperationResult<{ taskId: string; preparation: DurableStructuralPreparation } | undefined>> {
 	const settings = checkpoint.settings.compaction;
 	const identity = lane.state.configuration.model;
 	const model = lane.models.getModel(identity.provider, identity.modelId);
-	let preparation: CompactionPreparation | undefined;
-	if (settings.enabled && model !== undefined) {
-		const entries = await readBoundedEntries(lane, drive, checkpoint);
-		if (entries.kind === "cancel_requested") return { kind: "continue" };
-		const prepared = prepareCompaction(entries.value, settings);
-		if (!prepared.ok) throw prepared.error;
-		if (prepared.value !== undefined && shouldCompact(prepared.value.tokensBefore, model.contextWindow, settings)) {
-			preparation = prepared.value;
+	if (!settings.enabled || model === undefined) return { kind: "result", value: undefined };
+	const path = await readBoundedEntries(lane, drive, checkpoint);
+	if (path.kind === "cancel_requested") return path;
+	const triggerIndex = path.value.findIndex((entry) => entry.id === checkpoint.triggerEntryId);
+	let newestCompactionIndex = -1;
+	for (let index = path.value.length - 1; index >= 0; index--) {
+		if (path.value[index]!.type === "compaction") {
+			newestCompactionIndex = index;
+			break;
 		}
 	}
-	const taskId = preparation === undefined ? undefined : lane.session.idGenerator.next();
-	const published = await lane.continueOperation(
-		checkpoint,
-		(_state, current) => {
-			const marked: CheckpointOperation = {
-				...current,
-				thresholdCheckedTriggerEntryId: current.triggerEntryId,
-			};
-			if (preparation === undefined || taskId === undefined) {
-				return {
-					kind: "commit",
-					writes: [],
-					operationState: marked,
-					materialize: () => ({ kind: "continue" }) as const,
-				};
-			}
-			const resumeAfter = {
-				continuation: marked.continuation,
-				triggerEntryId: marked.triggerEntryId,
-				thresholdCheckedTriggerEntryId: marked.thresholdCheckedTriggerEntryId,
-				...(marked.skipInboxOnce === undefined ? {} : { skipInboxOnce: marked.skipInboxOnce }),
-			};
-			const deciding: SummaryDecidingOperation = {
-				...operationScopeOf(current),
-				at: "summary.deciding",
-				task: { taskId, reason: "threshold", boundary: { kind: "resume_checkpoint", resumeAfter } },
-			};
-			return {
-				kind: "commit",
-				writes: [
-					setValue(operationPreparation(drive.operationId, taskId), durableCompactionPreparation(preparation)),
-				],
-				operationState: deciding,
-				materialize: () => ({ kind: "continue" }) as const,
-				events: () => [
-					{
-						type: "compaction_start",
-						lane: lane.name,
-						runId: drive.operationId,
-						reason: "threshold",
-					},
-				],
-			};
-		},
-		drive.context,
-	);
-	return published.kind === "cancel_requested" ? { kind: "continue" } : published.value;
+	if (newestCompactionIndex >= triggerIndex && newestCompactionIndex !== -1) {
+		return { kind: "result", value: undefined };
+	}
+	if (triggerIndex === -1) {
+		throw new SessionInvariantError(`Checkpoint trigger ${checkpoint.triggerEntryId} is missing from its Branch`);
+	}
+	const prepared = prepareCompaction(path.value, settings);
+	if (!prepared.ok) throw prepared.error;
+	if (prepared.value === undefined || !shouldCompact(prepared.value.tokensBefore, model.contextWindow, settings)) {
+		return { kind: "result", value: undefined };
+	}
+	return {
+		kind: "result",
+		value: { taskId: lane.session.idGenerator.next(), preparation: durableCompactionPreparation(prepared.value) },
+	};
 }

 /** Prepare one overflow compaction before the response settlement transaction. */
--- a/packages/agent/src/harness/session/types.ts
+++ b/packages/agent/src/harness/session/types.ts
@ -125,8 +125,6 @@ export type Continuation =
 export interface CheckpointData {
 	continuation: Continuation;
 	triggerEntryId: string;
-	thresholdCheckedTriggerEntryId?: string;
-	skipInboxOnce?: boolean;
 }

 export type InboxItemKind = "steer" | "followUp" | "nextRun" | "write";
--- a/packages/agent/test/harness/runtime/drive-generation.test.ts
+++ b/packages/agent/test/harness/runtime/drive-generation.test.ts
@ -260,16 +260,17 @@ describe("runtime generation checkpoint", () => {
 		}, BACKGROUND_CONTEXT);
 		const run = currentRun(fixture.lane);
 		if (run.at !== "checkpoint") throw new Error("missing checkpoint");
+		fixture.storage.clearCommitAttempts();

 		await runCheckpoint(fixture.lane, fixture.drive, run);

-		const drained = currentRun(fixture.lane);
-		if (drained.at !== "checkpoint") throw new Error("missing drained checkpoint");
+		expect(fixture.storage.getCommitAttempts()).toHaveLength(1);
+		const routed = currentRun(fixture.lane);
+		if (routed.at !== "assistant.ready") throw new Error("pending writes did not route to generation");
 		expect(fixture.lane.state.tipId).toBe(customId);
-		expect(drained.triggerEntryId).toBe(messageId);
-		expect(drained).toMatchObject({
-			continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
-			skipInboxOnce: true,
+		expect(routed.generationContext).toMatchObject({
+			triggerEntryId: messageId,
+			overflowRecoveryUsed: false,
 		});
 		expect(fixture.lane.state.inbox).toEqual([]);
 		await expectProjectionRestores(fixture);
@ -284,8 +285,6 @@ describe("runtime generation checkpoint", () => {
 		const checkpoint: CheckpointOperation = {
 			...current,
 			continuation: { kind: "may_finish", includeFinalAssistant: false },
-			thresholdCheckedTriggerEntryId: current.triggerEntryId,
-			skipInboxOnce: false,
 		};
 		await fixture.lane.command((state) => {
 			const operation = state.operation;
@ -310,22 +309,82 @@ describe("runtime generation checkpoint", () => {
 				materialize: () => undefined,
 			};
 		}, BACKGROUND_CONTEXT);
+		fixture.hooks.on("before_run_end", () => ({ followUp: "continue after custom write" }));

-		await runCheckpoint(fixture.lane, fixture.drive, checkpoint);
+		expect(await runCheckpoint(fixture.lane, fixture.drive, checkpoint)).toEqual({ kind: "continue" });

-		const drained = currentRun(fixture.lane);
-		if (drained.at !== "checkpoint") throw new Error("missing drained checkpoint");
-		expect(fixture.lane.state.tipId).toBe(customId);
-		expect(drained).toMatchObject({
-			continuation: checkpoint.continuation,
-			triggerEntryId: checkpoint.triggerEntryId,
-			thresholdCheckedTriggerEntryId: checkpoint.triggerEntryId,
-			skipInboxOnce: false,
+		const ready = currentRun(fixture.lane);
+		if (ready.at !== "assistant.ready") throw new Error("hook follow-up did not route to generation");
+		const followUpId = ready.generationContext.triggerEntryId;
+		expect(followUpId).toBe(fixture.lane.state.tipId);
+		expect(await fixture.session.getEntry(customId, BACKGROUND_CONTEXT)).toMatchObject({ type: "custom" });
+		expect(await fixture.session.getEntry(followUpId, BACKGROUND_CONTEXT)).toMatchObject({
+			parentId: customId,
+			type: "message",
+			message: { role: "user", content: "continue after custom write" },
 		});
 		expect(fixture.lane.state.inbox).toEqual([]);
 		expect(await fixture.session.getValue(storedValues.pendingEntry(customId), BACKGROUND_CONTEXT)).toBeUndefined();
 		await expectProjectionRestores(fixture);
 	});
+
+	it("drops a stale finish-hook follow-up when input arrives during the hook", async () => {
+		const fixture = await createFixture();
+		await startFixtureRun(fixture);
+		const current = currentRun(fixture.lane);
+		if (current.at !== "checkpoint") throw new Error("missing checkpoint");
+		const checkpoint: CheckpointOperation = {
+			...current,
+			continuation: { kind: "may_finish", includeFinalAssistant: false },
+		};
+		await fixture.lane.command((state) => {
+			const operation = state.operation;
+			if (operation === null) throw new Error("missing operation");
+			return {
+				kind: "commit",
+				writes: [storedValues.setValue(storedValues.operationState(operationId), checkpoint)],
+				next: { ...state, operation: { meta: operation.meta, state: checkpoint } },
+				materialize: () => undefined,
+			};
+		}, BACKGROUND_CONTEXT);
+		const hookStarted = deferred();
+		const releaseHook = deferred();
+		fixture.hooks.on("before_run_end", async () => {
+			hookStarted.resolve();
+			await releaseHook.promise;
+			return { followUp: "stale hook follow-up" };
+		});
+
+		const running = runCheckpoint(fixture.lane, fixture.drive, checkpoint);
+		await hookStarted.promise;
+		await fixture.lane.command((state) => {
+			const inbox = [...state.inbox, { entryId: "steer-during-finish", kind: "steer" as const }];
+			return {
+				kind: "commit",
+				writes: [
+					storedValues.setValue(storedValues.pendingEntry("steer-during-finish"), {
+						type: "message",
+						payload: { role: "user", content: "new input", timestamp: 4 },
+					}),
+					storedValues.setValue(storedValues.laneState("main"), {
+						currentOperationId: operationId,
+						lastOperationId: null,
+						inbox,
+					}),
+				],
+				next: { ...state, inbox },
+				materialize: () => undefined,
+			};
+		}, BACKGROUND_CONTEXT);
+		releaseHook.resolve();
+
+		expect(await running).toEqual({ kind: "continue" });
+		const ready = currentRun(fixture.lane);
+		if (ready.at !== "assistant.ready") throw new Error("new input did not replace stale finish decision");
+		expect(ready.generationContext.triggerEntryId).toBe("steer-during-finish");
+		expect(fixture.lane.state.tipId).toBe("steer-during-finish");
+		expect((await fixture.session.getEntry("steer-during-finish", BACKGROUND_CONTEXT))?.type).toBe("message");
+	});
 });

 describe("runtime assistant generation", () => {
--- a/packages/agent/test/harness/runtime/drive-structural.test.ts
+++ b/packages/agent/test/harness/runtime/drive-structural.test.ts
@ -9,7 +9,6 @@ import { runGeneration } from "../../../src/harness/runtime/drive/generation.ts"
 import {
 	commitNavigation,
 	recoverStructuralGeneration,
-	runCompactionThreshold,
 	runStructuralDecision,
 	runStructuralGeneration,
 	runStructuralRetryWait,
@ -317,11 +316,10 @@ describe("runtime structural drive", () => {
 			},
 		);

-		expect(await runCompactionThreshold(fixture.lane, fixture.drive, checkpoint)).toEqual({ kind: "continue" });
+		expect(await runCheckpoint(fixture.lane, fixture.drive, checkpoint)).toEqual({ kind: "continue" });
 		const deciding = currentState(fixture);
 		if (deciding.at !== "summary.deciding") throw new Error("threshold did not enter compaction");
 		if (deciding.task.boundary.kind !== "resume_checkpoint") throw new Error("threshold has wrong boundary");
-		expect(deciding.task.boundary.resumeAfter.thresholdCheckedTriggerEntryId).toBe("assistant");
 		expect(
 			await fixture.session.getValue(
 				storedValues.operationPreparation(operationId, deciding.task.taskId),
@ -341,6 +339,136 @@ describe("runtime structural drive", () => {
 		expect(fixture.events.filter((event) => event.type === "compaction_end")).toHaveLength(1);
 	});

+	it("uses a newer compaction entry as the durable threshold guard", async () => {
+		const fixture = await createFixture();
+		const model = fixture.faux.getModel();
+		const checkpoint: CheckpointOperation = {
+			...runScope({ enabled: true, reserveTokens: model.contextWindow, keepRecentTokens: 1 }),
+			at: "checkpoint",
+			continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
+			triggerEntryId: "trigger",
+		};
+		await installOperation(
+			fixture,
+			checkpoint,
+			{ kind: "run", promptEntryIds: ["trigger"] },
+			{
+				entries: [
+					{ id: "trigger", parentId: null, type: "message", message: user("history") },
+					{
+						id: "compacted",
+						parentId: "trigger",
+						type: "compaction",
+						summary: "already compacted",
+						retainedTail: [],
+						tokensBefore: model.contextWindow,
+						fromHook: false,
+					},
+				],
+			},
+		);
+
+		expect(await runCheckpoint(fixture.lane, fixture.drive, checkpoint)).toEqual({ kind: "continue" });
+		const ready = currentState(fixture);
+		if (ready.at !== "assistant.ready") throw new Error("newer compaction did not guard threshold re-entry");
+		expect(ready.generationContext.triggerEntryId).toBe("trigger");
+		expect(fixture.events.some((event) => event.type === "compaction_start")).toBe(false);
+	});
+
+	it("finishes a may-finish run directly after threshold decline", async () => {
+		const fixture = await createFixture();
+		const deciding = {
+			...runScope(),
+			at: "summary.deciding",
+			task: runCompactionTask("threshold", {
+				continuation: { kind: "may_finish", includeFinalAssistant: false },
+				triggerEntryId: "tip",
+			}),
+		} as const;
+		await installOperation(
+			fixture,
+			deciding,
+			{ kind: "run", promptEntryIds: ["tip"] },
+			{
+				entries: [{ id: "tip", parentId: null, type: "message", message: user("history") }],
+				preparation: { taskId: "task", value: compactionPreparation() },
+			},
+		);
+		let finishHooks = 0;
+		fixture.hooks.on("before_compaction", () => ({ decline: true }));
+		fixture.hooks.on("before_run_end", () => {
+			finishHooks++;
+			return undefined;
+		});
+
+		expect(await runStructuralDecision(fixture.lane, fixture.drive, deciding)).toMatchObject({
+			kind: "settled",
+			outcome: { operationId, kind: "run", status: "completed", tipId: "tip" },
+		});
+		expect(fixture.storage.getCommitAttempts()).toHaveLength(1);
+		expect(finishHooks).toBe(1);
+		expect(fixture.lane.state.operation).toBeNull();
+		expect(fixture.events.slice(-2)).toMatchObject([
+			{ type: "compaction_end", reason: "threshold", status: "declined" },
+			{ type: "run_end", status: "completed" },
+		]);
+	});
+
+	it("routes queued follow-up before before_run_end after threshold decline", async () => {
+		const fixture = await createFixture();
+		const deciding = {
+			...runScope(),
+			at: "summary.deciding",
+			task: runCompactionTask("threshold", {
+				continuation: { kind: "may_finish", includeFinalAssistant: false },
+				triggerEntryId: "tip",
+			}),
+		} as const;
+		await installOperation(
+			fixture,
+			deciding,
+			{ kind: "run", promptEntryIds: ["tip"] },
+			{
+				entries: [{ id: "tip", parentId: null, type: "message", message: user("history") }],
+				preparation: { taskId: "task", value: compactionPreparation() },
+			},
+		);
+		await fixture.lane.command((projection) => {
+			const inbox = [...projection.inbox, { entryId: "follow-up", kind: "followUp" as const }];
+			return {
+				kind: "commit",
+				writes: [
+					storedValues.setValue(storedValues.pendingEntry("follow-up"), {
+						type: "message",
+						payload: user("continue"),
+					}),
+					storedValues.setValue(storedValues.laneState("main"), {
+						currentOperationId: operationId,
+						lastOperationId: null,
+						inbox,
+					}),
+				],
+				next: { ...projection, inbox },
+				materialize: () => undefined,
+			};
+		}, BACKGROUND_CONTEXT);
+		fixture.storage.clearCommitAttempts();
+		let finishHooks = 0;
+		fixture.hooks.on("before_compaction", () => ({ decline: true }));
+		fixture.hooks.on("before_run_end", () => {
+			finishHooks++;
+			return undefined;
+		});
+
+		expect(await runStructuralDecision(fixture.lane, fixture.drive, deciding)).toEqual({ kind: "continue" });
+		expect(fixture.storage.getCommitAttempts()).toHaveLength(1);
+		expect(finishHooks).toBe(0);
+		const ready = currentState(fixture);
+		if (ready.at !== "assistant.ready") throw new Error("follow-up did not route to generation");
+		expect(ready.generationContext.triggerEntryId).toBe("follow-up");
+		expect(fixture.events.map((event) => event.type)).toContain("compaction_end");
+	});
+
 	it.each(["steer", "followUp"] as const)(
 		"continues to an assistant turn when %s arrives during in-run compaction",
 		async (queue) => {
@ -351,7 +479,6 @@ describe("runtime structural drive", () => {
 				task: runCompactionTask("threshold", {
 					continuation: { kind: "may_finish", includeFinalAssistant: true },
 					triggerEntryId: "tip",
-					thresholdCheckedTriggerEntryId: "tip",
 				}),
 			} as const;
 			await installOperation(
@ -420,12 +547,13 @@ describe("runtime structural drive", () => {
 				if (routed.at !== "checkpoint") throw new Error("follow-up did not reach the finish checkpoint");
 				expect(fixture.lane.state.inbox).toEqual([{ entryId: "queued", kind: "followUp" }]);
 				expect(await runCheckpoint(fixture.lane, fixture.drive, routed)).toEqual({ kind: "continue" });
-				const checkpoint = currentState(fixture);
-				if (checkpoint.at !== "checkpoint") throw new Error("follow-up did not reach checkpoint");
-				expect(checkpoint.continuation).toEqual({ kind: "need_assistant", overflowRecoveryUsed: false });
+				const ready = currentState(fixture);
+				if (ready.at !== "assistant.ready") throw new Error("follow-up did not route directly to generation");
+				expect(ready.generationContext).toMatchObject({
+					triggerEntryId: "queued",
+					overflowRecoveryUsed: false,
+				});
 				expect(fixture.lane.state.inbox).toEqual([]);
-				expect(await runCheckpoint(fixture.lane, fixture.drive, checkpoint)).toEqual({ kind: "continue" });
-				expect(currentState(fixture).at).toBe("assistant.ready");
 			}
 		},
 	);
@ -440,7 +568,6 @@ describe("runtime structural drive", () => {
 			task: runCompactionTask("threshold", {
 				continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
 				triggerEntryId: "tip",
-				thresholdCheckedTriggerEntryId: "tip",
 			}),
 		} as const;
 		await installOperation(
@ -579,7 +706,6 @@ describe("runtime structural drive", () => {
 		expect(deciding.task.boundary.resumeAfter).toMatchObject({
 			continuation: { kind: "need_assistant", overflowRecoveryUsed: true },
 			triggerEntryId: "tip",
-			thresholdCheckedTriggerEntryId: "tip",
 		});
 		expect(
 			await fixture.session.getValue(
@ -765,14 +891,12 @@ describe("runtime structural drive", () => {
 			at: "checkpoint",
 			continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
 			triggerEntryId: "tip",
-			thresholdCheckedTriggerEntryId: "tip",
 		};
 		const ready = summaryReady(
 			runScope(),
 			runCompactionTask("overflow", {
 				continuation: { kind: "need_assistant", overflowRecoveryUsed: true },
 				triggerEntryId: resumeAfter.triggerEntryId,
-				thresholdCheckedTriggerEntryId: resumeAfter.thresholdCheckedTriggerEntryId,
 			}),
 			fixture.configuration,
 		);
@ -856,7 +980,6 @@ describe("runtime structural drive", () => {
 			runCompactionTask("threshold", {
 				continuation: { kind: "need_assistant", overflowRecoveryUsed: false },
 				triggerEntryId: "tip",
-				thresholdCheckedTriggerEntryId: "tip",
 			}),
 			{ ...fixture.configuration, model: { provider: "missing", modelId: "missing" } },
 		);