From 0f0ca0f6b78019b11f494b1476b0263a03a59597 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 26 Aug 2026 17:34:56 +0200 Subject: [PATCH] feat(agent): make run boundaries atomic --- packages/agent/docs/runtime-simplification.md | 2 +- .../work-packages/05-direct-durable-drive.md | 24 +- .../src/harness/runtime/drive/boundary.ts | 276 +++++++++++++ .../src/harness/runtime/drive/checkpoint.ts | 367 +++++------------- .../src/harness/runtime/drive/response.ts | 1 - .../src/harness/runtime/drive/structural.ts | 284 +++++--------- packages/agent/src/harness/session/types.ts | 2 - .../harness/runtime/drive-generation.test.ts | 93 ++++- .../harness/runtime/drive-structural.test.ts | 151 ++++++- 9 files changed, 705 insertions(+), 495 deletions(-) create mode 100644 packages/agent/src/harness/runtime/drive/boundary.ts diff --git a/packages/agent/docs/runtime-simplification.md b/packages/agent/docs/runtime-simplification.md index 3859f1c37..5fc33a93d 100644 --- 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`. diff --git a/packages/agent/docs/work-packages/05-direct-durable-drive.md b/packages/agent/docs/work-packages/05-direct-durable-drive.md index 2d6220b98..43c4674c0 100644 --- 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` 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` — 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. diff --git a/packages/agent/src/harness/runtime/drive/boundary.ts b/packages/agent/src/harness/runtime/drive/boundary.ts new file mode 100644 index 000000000..5bd8202ce --- /dev/null +++ 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( + lane: Lane, +): NormalizedRetryPolicy { + const retry = lane.readConfig().retryPolicy; + return retry.enabled + ? { maxAttempts: retry.maxRetries + 1, baseDelayMs: retry.baseDelayMs } + : { maxAttempts: 1, baseDelayMs: retry.baseDelayMs }; +} + +export function assistantReadyAtBoundary( + lane: Lane, + 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( + lane: Lane, + drive: Drive, + state: LaneState, + scope: OperationScope, + reader: SessionReader, + tipId: string | null, + followUpWhenNoTrigger: boolean, +): Promise { + 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( + lane: Lane, + drive: Drive, + capability: TState, + continuation: Extract, + plannedEntryIds: readonly string[], + pendingEvents: HarnessEvent[] = [], +): Promise { + 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( + 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; +} diff --git a/packages/agent/src/harness/runtime/drive/checkpoint.ts b/packages/agent/src/harness/runtime/drive/checkpoint.ts index bf4ec2a1d..4a6940919 100644 --- 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; +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( @@ -93,283 +91,94 @@ export async function startRun( return result.kind === "cancel_requested" ? { kind: "continue" } : result.value; } -async function applyPendingWrites( - lane: Lane, - drive: Drive, - checkpoint: CheckpointOperation, -): Promise { - 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( - lane: Lane, - drive: Drive, - checkpoint: CheckpointOperation, - queue: "steer" | "followUp", -): Promise { - 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[] = 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( - lane: Lane, - drive: Drive, - checkpoint: CheckpointOperation, - continuation: FinishContinuation, -): Promise { - 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( - 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 = { - 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( lane: Lane, drive: Drive, run: CheckpointOperation, ): Promise { - 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( + 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); } diff --git a/packages/agent/src/harness/runtime/drive/response.ts b/packages/agent/src/harness/runtime/drive/response.ts index b5c992edb..67d5f4c3e 100644 --- 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( resumeAfter: { continuation: { kind: "need_assistant", overflowRecoveryUsed: true }, triggerEntryId: current.generationContext.triggerEntryId, - thresholdCheckedTriggerEntryId: current.generationContext.triggerEntryId, }, }, }, diff --git a/packages/agent/src/harness/runtime/drive/structural.ts b/packages/agent/src/harness/runtime/drive/structural.ts index 809f7f9c8..cd7662363 100644 --- 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(lane: Lane): 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( lane: Lane, drive: Drive, @@ -226,7 +221,7 @@ async function publishStructuralOutcome( : undefined; const published = await lane.continueOperation< SummaryDecidingOperation | SummaryReadyOperation | SummaryEffectPendingOperation, - ProcedureResult + StructuralPublication >( capability, async (state, current, meta, reader) => { @@ -309,99 +304,56 @@ async function publishStructuralOutcome( (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( 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( }, 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( @@ -1102,72 +1063,41 @@ export async function recoverStructuralGeneration( +/** Prepare threshold compaction only when no newer compaction already guards this trigger. */ +export async function prepareCompactionThreshold( lane: Lane, drive: Drive, checkpoint: CheckpointOperation, -): Promise { +): Promise> { 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. */ diff --git a/packages/agent/src/harness/session/types.ts b/packages/agent/src/harness/session/types.ts index 39c2eb77b..0c8eb19e8 100644 --- 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"; diff --git a/packages/agent/test/harness/runtime/drive-generation.test.ts b/packages/agent/test/harness/runtime/drive-generation.test.ts index 20ae321da..d758d85f1 100644 --- 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", () => { diff --git a/packages/agent/test/harness/runtime/drive-structural.test.ts b/packages/agent/test/harness/runtime/drive-structural.test.ts index a2a401270..e040372bf 100644 --- 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" } }, );