From 9f666bfbbbf413c0abc4ad04fae8b2c3e8e7b8fb Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Mon, 24 Aug 2026 20:46:18 +0200 Subject: [PATCH] feat(agent): add durable terminal mechanics --- .../work-packages/05-direct-durable-drive.md | 32 +- .../src/harness/runtime2/drive/terminal.ts | 248 ++++++++++++ packages/agent/src/harness/runtime2/lane.ts | 80 +++- .../agent/src/harness/runtime2/progress.ts | 83 ++-- packages/agent/src/harness/runtime2/types.ts | 9 +- packages/agent/src/harness/session/session.ts | 7 +- .../harness/runtime2/drive-terminal.test.ts | 367 ++++++++++++++++++ .../agent/test/harness/runtime2/lane.test.ts | 82 ++++ .../test/harness/runtime2/progress.test.ts | 139 ++++++- packages/agent/test/harness/types.test.ts | 19 +- 10 files changed, 990 insertions(+), 76 deletions(-) create mode 100644 packages/agent/src/harness/runtime2/drive/terminal.ts create mode 100644 packages/agent/test/harness/runtime2/drive-terminal.test.ts 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 1b7e2c1a7..7ea712903 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: implementation-ready.** M1 is complete. Independent Opus and GPT-5.6-Sol reviews approved the authoritative-`Lane.state`, straight-line procedure correction below with no blockers. +**Status: M2 complete; M3 next.** Independent Opus and GPT-5.6-Sol reviews approved the authoritative-`Lane.state`, straight-line procedure correction and M2 implementation with no blockers. Removes breakpoints/manual drive, then implements the complete durable execution graph. Public drive stays disabled until every reachable phase and reconciliation path exists. @@ -22,7 +22,7 @@ Read completely, in this order, before editing anything. 3. `packages/agent/src/harness/session/types.ts` — durable types (`OperationState`, `RunState`, `RunPhase`, `Generation`, `Deferred`, `ToolBatch`, `ToolCall`, `StructuralDecision`, `NavigationState`, `LaneLastResult`, `SessionReader`, `Write`, `CommitResult`). 4. `packages/agent/src/harness/session/values.ts` — every typed address and write helper. 5. `packages/agent/src/harness/runtime2/lane.ts` — `Lane`, `Lane.command`, `LaneCommand`, `accept`. -6. `packages/agent/src/harness/runtime2/types.ts` — `Config`, `LaneState`, `LaneCommand`, `Drive`, the current `StepResult` that M2 simplifies to `ProcedureResult`, and `SliceNotImplemented`. +6. `packages/agent/src/harness/runtime2/types.ts` — `Config`, `LaneState`, `LaneCommand`, `Drive`, `ProcedureResult`, and `SliceNotImplemented`. 7. `packages/agent/src/harness/runtime2/harness.ts` — `Harness`, config store, `fault`, `close`. 8. `packages/agent/src/harness/runtime2/restore.ts` — projection restore. 9. `packages/agent/src/harness/execution/effect-gate.ts`, `execution/assistant.ts`, `execution/tools.ts`. @@ -131,7 +131,7 @@ private removeDrive(drive: Drive): void { Every late transition and progress write checks the current operation in the authoritative `Lane.state` supplied by the lane mutation line and exact process identity. Exact identity is checked again synchronously at commit admission, immediately before `mutator.commit()`, with no `await` between the check and invocation. A planner-side check alone is insufficient because abandonment may replace the owner after an asynchronous planner returns. -This prevents ABA writes: after invocation abandonment, a replacement `Drive` may own the same durable operation id; the old object still fails exact identity and cannot write or remove the replacement. The small package-internal `Lane.commandOwned(drive, plan, context)` variant performs only this admission check; it otherwise has the same visible `LaneCommand` decision as `command` and introduces no planner or transition framework. +This prevents ABA writes: after invocation abandonment, a replacement `Drive` may own the same durable operation id; the old object still fails exact identity and cannot write or remove the replacement. The small package-internal `Lane.commandDriveOwned(drive, plan, context)` variant performs only this admission check; it otherwise has the same visible `LaneCommand` decision as `command` and introduces no planner or transition framework. `startDrive` is ordinary Lane code: @@ -376,7 +376,7 @@ private mismatch(expected: string, currentOperationId: string | null, last: Lane ### 2.4 Drive switch contract — approved -**Compile ordering.** `Drive` lives in `runtime2/types.ts` from M1; M2 replaces the initial `StepResult` with `ProcedureResult`. `runtime2/drive.ts` is created in M7, once every procedure module it imports exists. M2–M6 call procedures directly from tests with a concrete `Lane` and `Drive`; public drive stays guarded until M8. +**Compile ordering.** `Drive` lives in `runtime2/types.ts` from M1; M2 replaced the initial `StepResult` with `ProcedureResult`. `runtime2/drive.ts` is created in M7, once every procedure module it imports exists. M2–M6 call procedures directly from tests with a concrete `Lane` and `Drive`; public drive stays guarded until M8. ```ts // packages/agent/src/harness/runtime2/types.ts @@ -462,7 +462,7 @@ Drive-owned commands use one small `Lane` variant so exact-object ownership is c ```ts // packages/agent/src/harness/runtime2/lane.ts -async commandOwned( +async commandDriveOwned( drive: Drive, plan: (state: LaneState, reader: SessionReader) => LaneCommand | Promise>, @@ -475,12 +475,12 @@ async commandOwned( } ``` -`commandOwned` is not a scheduler, planner, transaction wrapper, or capability facade. It is `Lane.command` plus the exact-Drive admission fence. If its initial operation/id check or final exact-object check fails, it returns `lost_ownership` and commits nothing. Close/fault is checked first at final admission and propagates its lifecycle error rather than being mislabeled as ownership loss. Ordinary public lane commands continue to use `command`. +`commandDriveOwned` is not a scheduler, planner, transaction wrapper, or capability facade. It is `Lane.command` plus the exact-Drive admission fence. If its initial operation/id check or final exact-object check fails, it returns `lost_ownership` and commits nothing. Close/fault is checked first at final admission and propagates its lifecycle error rather than being mislabeled as ownership loss. This extra final open check is drive-specific: exact owner abandonment may occur outside the lane line while an asynchronous planner runs. An ordinary `command` is admitted when its mutation callback starts and may finish that active commit after close begins. Ordinary public lane commands continue to use `command`. A representative transition stays direct: ```ts -const intent = await lane.commandOwned(drive, (state) => { +const intent = await lane.commandDriveOwned(drive, (state) => { const operation = state.operation; if (operation === null || operation.state.kind !== "run") { throw new SessionInvariantError("generation intent reached a non-run operation"); @@ -597,7 +597,7 @@ function openProgress( return { write(item) { if (sealed) return; - const write = lane.commandOwned(drive, (state) => { + const write = lane.commandDriveOwned(drive, (state) => { if (!stillOwns(state)) return { kind: "return", result: undefined }; return { kind: "commit", writes: [commitWrite(item)], next: state, materialize: () => undefined }; @@ -612,11 +612,11 @@ function openProgress( } ``` -`openFrameProgress(lane, drive, responseEntryId)` and `openToolProgress(lane, drive, invocationId)` inspect the current projection's phase/id and use `commandOwned` for the final exact-Drive admission fence. They perform no durable control reads. The provider/tool callback enqueues synchronously, settlement seals then drains, and the settling transaction includes `clearWrite()`. +`openFrameProgress(lane, drive, responseEntryId)` and `openToolProgress(lane, drive, invocationId)` inspect the current projection's phase/id and use `commandDriveOwned` for the final exact-Drive admission fence. They perform no durable control reads. The provider/tool callback enqueues synchronously, settlement seals then drains, and the settling transaction includes `clearWrite()`. ### 2.8 Tool batch — approved shape, illustrative bodies -The tool procedure is one ordinary async routine. It keeps its process-local started-promise map, but every status patch is computed from the current `Lane.state` supplied to that specific `commandOwned` job. It never patches a batch snapshot captured before the job. +The tool procedure is one ordinary async routine. It keeps its process-local started-promise map, but every status patch is computed from the current `Lane.state` supplied to that specific `commandDriveOwned` job. It never patches a batch snapshot captured before the job. Each child uses module-local results only: @@ -777,7 +777,7 @@ Staging is completion-ordered (`outcome_ready`); `materializeReadyPrefix` places ### 2.9 Terminal transactions stay in their procedures — approved -The procedure that owns a terminal boundary also owns its terminal `Lane.commandOwned`. An authorized in-process external finalizer is the one exception: it owns an ordinary `Lane.command` because it may finalize an unowned operation. Its `materialize` continuation records `finalizedOutcome` and closes any matching live Drive gate synchronously after projection publication. The finalizer settles that Drive only after `Lane.command` returns, so terminal event delivery still precedes public completion. The finalizer must still verify that the current projection names the exact operation it intends to end; a terminal winner observed first commits nothing. Each procedure already knows its exact semantic state, local result, publication writes, public outcome, and terminal event. There is no generic `commitTerminal`, `TerminalRequest`, terminal validator callback, or terminal planner. +The procedure that owns a terminal boundary also owns its terminal `Lane.commandDriveOwned`. An authorized in-process external finalizer is the one exception: it owns an ordinary `Lane.command` because it may finalize an unowned operation. Its `materialize` continuation records `finalizedOutcome` and closes any matching live Drive gate synchronously after projection publication. The finalizer settles that Drive only after `Lane.command` returns, so terminal event delivery still precedes public completion. The finalizer must still verify that the current projection names the exact operation it intends to end; a terminal winner observed first commits nothing. Each procedure already knows its exact semantic state, local result, publication writes, public outcome, and terminal event. There is no generic `commitTerminal`, `TerminalRequest`, terminal validator callback, or terminal planner. Inside that one command, the procedure: @@ -792,7 +792,7 @@ Inside that one command, the procedure: A normal completed-run finish is representative. Its terminal check is visibly part of the checkpoint procedure: ```ts -return lane.commandOwned(drive, async (state, reader) => { +return lane.commandDriveOwned(drive, async (state, reader) => { const operation = state.operation; if (operation === null || operation.state.kind !== "run") { throw new SessionInvariantError("run finish reached a non-run operation"); @@ -1008,7 +1008,7 @@ async function runCheckpoint( ): Promise { /* ... */ } ``` -`Lane` already exposes the package-internal members procedures need: `state`, `session`, `models`, `hooks`, `command`, `emitBatch`, `readConfig`, `mismatch`, and `isDriveActive`; M2 adds only `commandOwned` for exact-Drive commit admission. The package root exports only `AgentLane` / `AgentHarness`, so these members do not widen public API. Procedure imports of `Lane` are type-only and erase; `lane.ts` may value-import `drive.ts` in M8 without a runtime cycle. +`Lane` already exposes the package-internal members procedures need: `state`, `session`, `models`, `hooks`, `command`, `emitBatch`, `readConfig`, `mismatch`, and `isDriveActive`; M2 adds only `commandDriveOwned` for exact-Drive commit admission. The package root exports only `AgentLane` / `AgentHarness`, so these members do not widen public API. Procedure imports of `Lane` are type-only and erase; `lane.ts` may value-import `drive.ts` in M8 without a runtime cycle. Rules: - `TContext extends object | undefined` everywhere; no other bound. @@ -1274,9 +1274,9 @@ Only `packages/coding-agent/src/core/cache-stats.ts` (comment) and vendored `hig **Modify** | Path | Change | |---|---| -| `src/harness/runtime2/types.ts` | replace global `StepResult.advance/reload` with `ProcedureResult.continue`; retain waiting, settled, and lost ownership (§2.4) | -| `src/harness/runtime2/lane.ts` | add `commandOwned`, with exact-Drive identity checked immediately before commit admission (§2.5); document `state` as the live control authority | -| `src/harness/runtime2/progress.ts` | use the planner's current `LaneState` and `commandOwned`; remove durable lane/operation control reads (§2.7) | +| `src/harness/runtime2/types.ts` | replaced global `StepResult.advance/reload` with `ProcedureResult.continue`; retained waiting, settled, and lost ownership (§2.4) | +| `src/harness/runtime2/lane.ts` | add `commandDriveOwned`, with exact-Drive identity checked immediately before commit admission (§2.5); document `state` as the live control authority | +| `src/harness/runtime2/progress.ts` | use the planner's current `LaneState` and `commandDriveOwned`; remove durable lane/operation control reads (§2.7) | | `src/harness/session/session.ts` | document retained mutators as forbidden while a harness owns the Session; no runtime compatibility layer | **Behavior/invariants landed.** A live `Lane.state` is the sole control-flow authority. Process loss restores it from durable values. Runtime reads dereference payloads only. Exact-Drive replacement between planner return and commit admission declines the commit. Cleanup enumerates every operation-owned address through typed prefix scans and concrete deletes; `pendingNextRun` is not cleanup-owned. Hydration reads only the bounded latest-result value and entries it directly names. diff --git a/packages/agent/src/harness/runtime2/drive/terminal.ts b/packages/agent/src/harness/runtime2/drive/terminal.ts new file mode 100644 index 000000000..0489ce156 --- /dev/null +++ b/packages/agent/src/harness/runtime2/drive/terminal.ts @@ -0,0 +1,248 @@ +import type { AgentMessage } from "../../../types.ts"; +import type { TerminalOperationOutcome } from "../../agent-harness.ts"; +import type { Context } from "../../context.ts"; +import { SessionInvariantError } from "../../session/session.ts"; +import type { + BranchSummaryEntry, + CompactionEntry, + LaneLastResult, + MessageEntry, + OperationState, + SessionReader, + SettledAssistantMessage, + Write, +} from "../../session/types.ts"; +import { + deleteList, + deleteValue, + operationMeta, + operationPreparationPrefix, + operationState as operationStateValue, + operationToolArgsPrefix, + operationToolMemoPrefix, + pendingAssistantFrames, + pendingEntry, + pendingToolOutputPrefix, +} from "../../session/values.ts"; + +/** Build the mechanical operation-owned suffix used by an owning procedure's terminal transaction. */ +export async function operationCleanupWrites( + reader: SessionReader, + operationId: string, + state: OperationState, + context: Context, +): Promise { + const [toolArguments, toolMemos, preparations, toolOutputs] = await Promise.all([ + reader.scanValues(operationToolArgsPrefix(operationId), context), + reader.scanValues(operationToolMemoPrefix(operationId), context), + reader.scanValues(operationPreparationPrefix(operationId), context), + reader.scanValues(pendingToolOutputPrefix(operationId), context), + ]); + + const pendingIds = new Set(); + if (state.kind === "run") { + for (const id of state.inbox.steer) pendingIds.add(id); + for (const id of state.inbox.followUp) pendingIds.add(id); + for (const id of state.inbox.writes) pendingIds.add(id); + if (state.phase.kind === "tools") { + for (const call of state.phase.batch.calls) { + if (call.status === "outcome_ready") pendingIds.add(call.resultEntryId); + } + } + } + if (state.control.status === "cancel_requested") { + for (const id of state.control.drainedSteer) pendingIds.add(id); + for (const id of state.control.drainedFollowUp) pendingIds.add(id); + } + + let frameDelete: Write | undefined; + if (state.kind === "run" && state.phase.kind === "assistant") { + const generation = state.phase.generation; + if (generation.status === "effect_pending") { + frameDelete = deleteList(pendingAssistantFrames(operationId, generation.responseEntryId)); + } + } else if (state.kind === "run" && state.phase.kind === "deferred") { + const deferred = state.phase.deferred; + if (deferred.status === "effect_pending") { + frameDelete = deleteList(pendingAssistantFrames(operationId, deferred.responseEntryId)); + } + } + + return [ + deleteValue(operationMeta(operationId)), + deleteValue(operationStateValue(operationId)), + ...toolArguments.map(({ address }) => deleteValue(address)), + ...toolMemos.map(({ address }) => deleteValue(address)), + ...preparations.map(({ address }) => deleteValue(address)), + ...toolOutputs.map(({ address }) => deleteValue(address)), + ...(frameDelete === undefined ? [] : [frameDelete]), + ...[...pendingIds].map((id) => deleteValue(pendingEntry(id))), + ]; +} + +function isSettledAssistantMessage(message: AgentMessage): message is SettledAssistantMessage { + return message.role === "assistant" && message.stopReason !== "pending"; +} + +async function requireMessageEntry( + reader: SessionReader, + entryId: string, + context: Context, +): Promise { + const entry = (await reader.getEntries([entryId], context)).get(entryId); + if (entry?.type !== "message") { + throw new SessionInvariantError(`Terminal result references invalid assistant ${entryId}`); + } + const message = entry.message; + if (!isSettledAssistantMessage(message)) { + throw new SessionInvariantError(`Terminal result references invalid assistant ${entryId}`); + } + return { ...entry, message }; +} + +async function requireCompactionEntry( + reader: SessionReader, + entryId: string, + context: Context, +): Promise { + const entry = (await reader.getEntries([entryId], context)).get(entryId); + if (entry?.type !== "compaction") { + throw new SessionInvariantError(`Terminal result references invalid compaction ${entryId}`); + } + return entry; +} + +async function requireSummaryEntry( + reader: SessionReader, + entryId: string, + context: Context, +): Promise { + const entry = (await reader.getEntries([entryId], context)).get(entryId); + if (entry?.type !== "branch_summary") { + throw new SessionInvariantError(`Terminal result references invalid branch summary ${entryId}`); + } + return entry; +} + +/** Hydrate one bounded public terminal outcome from the latest-result value's direct references. */ +export async function hydrateTerminalOutcome( + reader: SessionReader, + lastResult: LaneLastResult, + context: Context, +): Promise { + if (lastResult.kind === "run") { + if (lastResult.outcome === "completed") { + if (lastResult.runCompletion === "terminated_tools") { + if (lastResult.finalAssistantEntryId !== undefined) { + throw new SessionInvariantError("Terminated-tools result cannot reference a final assistant"); + } + return { + operation: "run", + runId: lastResult.operationId, + kind: "completed", + leafId: lastResult.leafId, + }; + } + if (lastResult.finalAssistantEntryId === undefined) { + throw new SessionInvariantError("Assistant-completed result is missing its final assistant"); + } + const entry = await requireMessageEntry(reader, lastResult.finalAssistantEntryId, context); + return { + operation: "run", + runId: lastResult.operationId, + kind: "completed", + leafId: lastResult.leafId, + finalEntryId: entry.id, + finalMessage: entry.message, + }; + } + + const final = + lastResult.finalAssistantEntryId === undefined + ? {} + : await requireMessageEntry(reader, lastResult.finalAssistantEntryId, context).then((entry) => ({ + finalEntryId: entry.id, + finalMessage: entry.message, + })); + return lastResult.outcome === "failed" + ? { + operation: "run", + runId: lastResult.operationId, + kind: "failed", + leafId: lastResult.leafId, + error: lastResult.error, + ...final, + } + : { + operation: "run", + runId: lastResult.operationId, + kind: "aborted", + leafId: lastResult.leafId, + ...final, + }; + } + + if (lastResult.kind === "compaction") { + if (lastResult.outcome === "completed") { + const entry = await requireCompactionEntry(reader, lastResult.leafId, context); + return { + operation: "compaction", + runId: lastResult.operationId, + kind: "completed", + leafId: lastResult.leafId, + entry, + }; + } + if (lastResult.outcome === "failed") { + return { + operation: "compaction", + runId: lastResult.operationId, + kind: "failed", + leafId: lastResult.leafId, + error: lastResult.error, + }; + } + return { + operation: "compaction", + runId: lastResult.operationId, + kind: lastResult.outcome, + leafId: lastResult.leafId, + }; + } + + if (lastResult.outcome === "completed") { + const summaryEntry = + lastResult.summaryEntryId === undefined + ? undefined + : await requireSummaryEntry(reader, lastResult.summaryEntryId, context); + if ( + summaryEntry !== undefined && + (summaryEntry.id !== lastResult.leafId || summaryEntry.fromId !== lastResult.oldLeafId) + ) { + throw new SessionInvariantError("Completed navigation summary contradicts its terminal result"); + } + return { + operation: "navigation", + runId: lastResult.operationId, + kind: "completed", + oldLeafId: lastResult.oldLeafId, + newLeafId: lastResult.leafId, + ...(summaryEntry === undefined ? {} : { summaryEntry }), + }; + } + if (lastResult.outcome === "failed") { + return { + operation: "navigation", + runId: lastResult.operationId, + kind: "failed", + leafId: lastResult.leafId, + error: lastResult.error, + }; + } + return { + operation: "navigation", + runId: lastResult.operationId, + kind: lastResult.outcome, + leafId: lastResult.leafId, + }; +} diff --git a/packages/agent/src/harness/runtime2/lane.ts b/packages/agent/src/harness/runtime2/lane.ts index 7e18838d9..97d7f51f5 100644 --- a/packages/agent/src/harness/runtime2/lane.ts +++ b/packages/agent/src/harness/runtime2/lane.ts @@ -60,7 +60,14 @@ import { setValue, } from "../session/values.ts"; import { formatSkillInvocation } from "../skills.ts"; -import { type Config, type Drive, type LaneCommand, type LaneState, SliceNotImplemented } from "./types.ts"; +import { + type Config, + type Drive, + type LaneCommand, + type LaneState, + type LostOwnership, + SliceNotImplemented, +} from "./types.ts"; type EmitBatch = (events: readonly HarnessEvent[], context: Context) => Promise; type WatchHandler = (snapshot: T, filter: (event: HarnessEvent) => boolean, context: Context) => WatchHandle; @@ -115,6 +122,7 @@ export class Lane implements AgentLane { private readonly config: () => Config; /** Package-internal drive owner. Public only because deterministic procedure tests install exact owners directly. */ activeDrive: Drive | undefined; + /** Authoritative live control projection while this harness owns the Session. */ state: LaneState; closedError: Error | undefined; @@ -193,8 +201,10 @@ export class Lane implements AgentLane { * - `reject` rejects outside the mutation/fault boundary as an expected caller error without a commit. * * Planner, commit, and materialization errors fault the harness before releasing the lane line. Close/fault gates - * are checked both before queueing and when the callback starts: close-first rejects, while an admitted successful - * commit always publishes memory and resolves without another open check. Never invoke providers, tools, hooks, + * are checked both before queueing and when the callback starts: close-first rejects, while a callback admitted + * before close may finish its commit, publish memory, and resolve without another open check. Drive-owned commands + * additionally need a final fence because their exact owner may be removed outside the lane line. Never invoke + * providers, tools, hooks, * timers, event handlers, or wait for task completion here; perform those after `command()` returns. */ async command( @@ -243,6 +253,70 @@ export class Lane implements AgentLane { return outcome.result; } + /** + * Run one drive-owned command with an exact-object fence at commit admission. + * Durable control state comes from the authoritative projection; storage reads are only for referenced payloads. + * Unlike an ordinary command, owner abandonment can occur outside the lane line while the planner awaits. The final + * open check preserves close/fault as lifecycle errors before the exact-owner check can report lost ownership. + */ + async commandDriveOwned( + drive: Drive, + plan: (state: LaneState, reader: SessionReader) => LaneCommand | Promise>, + context: Context, + ): Promise { + this.assertOpen(); + let outcome: LaneCommandOutcome; + try { + outcome = await this.session.mutate( + this.name, + async (mutator) => { + this.assertOpen(); + const operation = this.state.operation; + if (operation?.meta.operationId !== drive.operationId || this.activeDrive !== drive) { + return { kind: "return", result: { kind: "lost_ownership" } as const }; + } + try { + const decision = await plan(this.state, mutator); + switch (decision.kind) { + case "return": + return { kind: "return", result: decision.result }; + case "reject": + return { kind: "reject", error: decision.error }; + case "commit": { + // Keep these checks adjacent to commit admission. Owner abandonment may occur + // outside the line; close/fault must still win classification over ownership loss. + this.assertOpen(); + if (this.activeDrive !== drive) { + return { kind: "return", result: { kind: "lost_ownership" } as const }; + } + const commitPromise = mutator.commit(decision.writes, context); + const commit = await commitPromise; + this.state = decision.next; + const result = decision.materialize(commit); + if (isPromiseLike(result)) { + throw new TypeError("Lane command materialize() must be synchronous"); + } + const events = decision.events?.(commit) ?? []; + const delivery = events.length === 0 ? undefined : this.emitBatch(events, context); + return { kind: "return", result, ...(delivery === undefined ? {} : { delivery }) }; + } + } + } catch (error) { + if (this.closedError !== undefined) throw this.closedError; + throw this.onFault(error, context); + } + }, + context, + ); + } catch (error) { + if (this.closedError !== undefined) throw this.closedError; + throw error; + } + if (outcome.kind === "reject") throw outcome.error; + await outcome.delivery; + return outcome.result; + } + async accept(request: OperationRequest, context: Context): Promise { if (this.closedError instanceof HarnessClosed) { return Result.err(new Closed({ message: this.closedError.message })); diff --git a/packages/agent/src/harness/runtime2/progress.ts b/packages/agent/src/harness/runtime2/progress.ts index 72671d542..567fb4a16 100644 --- a/packages/agent/src/harness/runtime2/progress.ts +++ b/packages/agent/src/harness/runtime2/progress.ts @@ -1,19 +1,16 @@ import type { AssistantMessageFrame } from "@earendil-works/pi-ai"; import type { AgentToolResult } from "../../types.ts"; -import { SessionInvariantError } from "../session/session.ts"; -import type { RunState, SessionReader, Write } from "../session/types.ts"; +import type { Write } from "../session/types.ts"; import { appendList, deleteList, deleteValue, - laneState as laneStateValue, - operationState as operationStateValue, pendingAssistantFrames, pendingToolOutput, setValue, } from "../session/values.ts"; import type { Lane } from "./lane.ts"; -import type { Drive } from "./types.ts"; +import type { Drive, LaneState } from "./types.ts"; export interface ProgressChannel { write(item: T): void; @@ -27,22 +24,28 @@ function openProgress( drive: Drive, commitWrite: (item: T) => Write, clear: () => Write, - stillOwns: (reader: SessionReader) => Promise, + stillOwns: (state: LaneState) => boolean, ): ProgressChannel { let sealed = false; let latest: Promise = Promise.resolve(); return { write(item) { if (sealed) return; - const write = lane.command(async (projection, reader) => { - if (!(await stillOwns(reader))) return { kind: "return", result: undefined }; - return { - kind: "commit", - writes: [commitWrite(item)], - next: projection, - materialize: () => undefined, - }; - }, drive.context); + const write = lane + .commandDriveOwned( + drive, + (projection) => { + if (!stillOwns(projection)) return { kind: "return", result: undefined }; + return { + kind: "commit", + writes: [commitWrite(item)], + next: projection, + materialize: () => undefined, + }; + }, + drive.context, + ) + .then(() => undefined); latest = write; void write.catch(() => {}); }, @@ -56,27 +59,6 @@ function openProgress( }; } -async function readOwnedRun( - lane: Lane, - drive: Drive, - reader: SessionReader, -): Promise { - const durableLane = await reader.getValue(laneStateValue(lane.name), drive.context); - if (durableLane === undefined) { - throw new SessionInvariantError(`Lane ${JSON.stringify(lane.name)} has no lane state`); - } - if (durableLane.value.currentOperationId !== drive.operationId) return undefined; - const durableState = await reader.getValue(operationStateValue(drive.operationId), drive.context); - if (durableState === undefined) { - throw new SessionInvariantError(`Operation ${drive.operationId} has no operation state`); - } - if (durableState.value.kind !== "run") { - throw new SessionInvariantError(`Operation ${drive.operationId} is ${durableState.value.kind}, expected run`); - } - if (!lane.isDriveActive(drive)) return undefined; - return durableState.value; -} - export function openFrameProgress( lane: Lane, drive: Drive, @@ -88,15 +70,16 @@ export function openFrameProgress( drive, (frame) => appendList(address, frame), () => deleteList(address), - async (reader) => { - const state = await readOwnedRun(lane, drive, reader); - if (state === undefined) return false; - if (state.phase.kind === "assistant") { - const generation = state.phase.generation; + (state) => { + const operation = state.operation; + if (operation?.meta.operationId !== drive.operationId || operation.state.kind !== "run") return false; + const run = operation.state; + if (run.phase.kind === "assistant") { + const generation = run.phase.generation; return generation.status === "effect_pending" && generation.responseEntryId === responseEntryId; } - if (state.phase.kind === "deferred") { - const deferred = state.phase.deferred; + if (run.phase.kind === "deferred") { + const deferred = run.phase.deferred; return deferred.status === "effect_pending" && deferred.responseEntryId === responseEntryId; } return false; @@ -115,10 +98,16 @@ export function openToolProgress( drive, (snapshot) => setValue(address, snapshot), () => deleteValue(address), - async (reader) => { - const state = await readOwnedRun(lane, drive, reader); - if (state?.phase.kind !== "tools") return false; - return state.phase.batch.calls.some( + (state) => { + const operation = state.operation; + if ( + operation?.meta.operationId !== drive.operationId || + operation.state.kind !== "run" || + operation.state.phase.kind !== "tools" + ) { + return false; + } + return operation.state.phase.batch.calls.some( (call) => call.resultEntryId === invocationId && call.status === "effect_pending", ); }, diff --git a/packages/agent/src/harness/runtime2/types.ts b/packages/agent/src/harness/runtime2/types.ts index 7fcd35faa..eb779b891 100644 --- a/packages/agent/src/harness/runtime2/types.ts +++ b/packages/agent/src/harness/runtime2/types.ts @@ -121,9 +121,10 @@ export class Drive { } } -export type StepResult = - | { kind: "advance" } - | { kind: "reload" } +export type LostOwnership = { kind: "lost_ownership" }; + +export type ProcedureResult = + | { kind: "continue" } | { kind: "waiting"; outcome: DriveOutcome } | { kind: "settled"; outcome: TerminalOperationOutcome } - | { kind: "lost_ownership" }; + | LostOwnership; diff --git a/packages/agent/src/harness/session/session.ts b/packages/agent/src/harness/session/session.ts index 7fb2b849a..a9452d184 100644 --- a/packages/agent/src/harness/session/session.ts +++ b/packages/agent/src/harness/session/session.ts @@ -180,7 +180,12 @@ class StorageBackedSessionMutator implements SessionMutator { } } -/** Package-internal typed boundary shared by concrete session repositories. */ +/** + * Package-internal typed boundary shared by concrete session repositories. + * After this Session is passed to AgentHarness.create(), retained Session and SessionTree + * facades are read-only until harness close; mutating through them would bypass the owning + * Lane's authoritative projection and is a trusted-programming defect. + */ export class StorageBackedSession implements Session { readonly metadata: TMetadata; readonly idGenerator: IdGenerator; diff --git a/packages/agent/test/harness/runtime2/drive-terminal.test.ts b/packages/agent/test/harness/runtime2/drive-terminal.test.ts new file mode 100644 index 000000000..90fb42af9 --- /dev/null +++ b/packages/agent/test/harness/runtime2/drive-terminal.test.ts @@ -0,0 +1,367 @@ +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { BACKGROUND_CONTEXT } from "../../../src/harness/context.ts"; +import { hydrateTerminalOutcome, operationCleanupWrites } from "../../../src/harness/runtime2/drive/terminal.ts"; +import { insertEntry } from "../../../src/harness/session/commit.ts"; +import { MemoryStorage } from "../../../src/harness/session/memory.ts"; +import { StorageBackedSession } from "../../../src/harness/session/session.ts"; +import type { + CompactionState, + LaneConfiguration, + LaneLastResult, + NavigationState, + OperationMeta, + OperationState, + RunState, + Session, + Write, +} from "../../../src/harness/session/types.ts"; +import * as storedValues from "../../../src/harness/session/values.ts"; + +const sessions: Session[] = []; +const configuration: LaneConfiguration = { + model: { provider: "test", modelId: "model" }, + thinkingLevel: "off", + activeToolNames: [], +}; + +function runState(phase: RunState["phase"]): RunState { + return { + kind: "run", + control: { status: "running" }, + settings: { + compaction: { enabled: true, reserveTokens: 1_000, keepRecentTokens: 2_000 }, + steeringMode: "all", + followUpMode: "all", + toolExecution: "parallel", + }, + phase, + inbox: { steer: [], followUp: [], writes: [] }, + latestAssistantEntryId: null, + }; +} + +function meta(operationId: string, state: OperationState): OperationMeta { + const intent: OperationMeta["intent"] = + state.kind === "run" + ? { kind: "run", promptEntryIds: [] } + : state.kind === "compaction" + ? { kind: "compaction" } + : { kind: "navigation", targetId: state.targetId, summarize: state.summarize }; + return { operationId, lane: "main", sourceLeafId: null, startedAt: 1, intent }; +} + +async function createSession(): Promise<{ session: Session; storage: MemoryStorage }> { + const storage = new MemoryStorage(); + const session = new StorageBackedSession( + { id: `terminal-${sessions.length}`, createdAt: 1, storageVersion: 1 }, + storage, + ); + sessions.push(session); + return { session, storage }; +} + +async function commit(session: Session, writes: Write[]): Promise { + await session.mutate( + "main", + async (mutator) => { + await mutator.commit(writes, BACKGROUND_CONTEXT); + }, + BACKGROUND_CONTEXT, + ); +} + +function address(write: Write): string { + if (write.kind === "entry") return `entry:${write.entry.id}`; + if (write.kind === "usage") return `usage:${write.row.id}`; + return `${write.kind}:${write.op}:${write.namespace}:${write.key}`; +} + +async function seedLeftovers(session: Session, operationId: string, state: OperationState): Promise { + await commit(session, [ + storedValues.setValue(storedValues.operationMeta(operationId), meta(operationId, state)), + storedValues.setValue(storedValues.operationState(operationId), state), + storedValues.setValue(storedValues.operationToolArgs(operationId, "step", 0), { value: true }), + storedValues.setValue(storedValues.operationToolMemo(operationId, "invocation", "memo"), { value: true }), + storedValues.setValue(storedValues.operationPreparation(operationId, "task"), { + kind: "branch_summary", + messages: [], + fileOps: { read: [], written: [], edited: [] }, + totalTokens: 0, + }), + storedValues.setValue(storedValues.pendingToolOutput(operationId, "invocation"), { + content: [{ type: "text", text: "partial" }], + details: {}, + }), + ]); +} + +afterEach(async () => { + for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); +}); + +describe("runtime2 terminal cleanup mechanics", () => { + it("deletes every run-owned family and exact live frame list but preserves pendingNextRun", async () => { + const { session } = await createSession(); + const operationId = "run"; + const responseEntryId = "response"; + const state: RunState = { + ...runState({ + kind: "assistant", + generation: { + status: "effect_pending", + context: { + stepId: "step", + triggerEntryId: "trigger", + configuration, + streamOptions: {}, + retryPolicy: { maxAttempts: 2, baseDelayMs: 1 }, + overflowRecoveryUsed: false, + }, + attempt: 1, + responseEntryId, + usageId: "usage", + intendedOutputLimit: 100, + contextWindow: 1_000, + }, + }), + control: { + status: "cancel_requested", + requestedAt: 2, + drainedSteer: ["drained-steer"], + drainedFollowUp: ["drained-follow"], + }, + inbox: { steer: ["steer"], followUp: ["follow"], writes: ["write"] }, + }; + await seedLeftovers(session, operationId, state); + await commit(session, [ + ...["steer", "follow", "write", "drained-steer", "drained-follow", "next"].map((id) => + storedValues.setValue(storedValues.pendingEntry(id), { + type: "custom", + customType: "test", + payload: { id }, + }), + ), + storedValues.appendList(storedValues.pendingAssistantFrames(operationId, responseEntryId), { + type: "text_delta", + contentIndex: 0, + delta: "partial", + }), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: operationId, + pendingNextRun: ["next"], + }), + ]); + + const writes = await operationCleanupWrites(session, operationId, state, BACKGROUND_CONTEXT); + + expect(writes.map(address)).toEqual([ + "value:delete:pi.op.meta:run", + "value:delete:pi.op.state:run", + "value:delete:pi.op.tool_args:run:step:0", + "value:delete:pi.op.tool_memo:run:invocation:memo", + "value:delete:pi.op.preparation:run:task", + "value:delete:pi.pending.tool_output:run:invocation", + "list:delete:pi.pending.assistant_frame:run:response", + "value:delete:pi.pending.entry:steer", + "value:delete:pi.pending.entry:follow", + "value:delete:pi.pending.entry:write", + "value:delete:pi.pending.entry:drained-steer", + "value:delete:pi.pending.entry:drained-follow", + ]); + await commit(session, writes); + expect(await session.getValue(storedValues.pendingEntry("next"), BACKGROUND_CONTEXT)).toBeDefined(); + expect( + (await session.getValue(storedValues.laneState("main"), BACKGROUND_CONTEXT))?.value.pendingNextRun, + ).toEqual(["next"]); + expect( + await session.readList( + storedValues.pendingAssistantFrames(operationId, responseEntryId), + undefined, + BACKGROUND_CONTEXT, + ), + ).toEqual([]); + }); + + it("deletes staged tool outcomes and leaves completed results alone", async () => { + const { session } = await createSession(); + const operationId = "tools"; + const state = runState({ + kind: "tools", + batch: { + assistantEntryId: "assistant", + configuration, + turnId: "step", + calls: [ + { status: "outcome_ready", sourceIndex: 0, resultEntryId: "staged", terminate: false }, + { status: "completed", sourceIndex: 1, resultEntryId: "placed", terminate: false }, + ], + }, + }); + await seedLeftovers(session, operationId, state); + await commit(session, [ + storedValues.setValue(storedValues.pendingEntry("staged"), { + type: "message", + payload: { + role: "toolResult", + toolCallId: "call", + toolName: "tool", + content: [], + isError: false, + timestamp: 1, + }, + }), + ]); + + const writes = await operationCleanupWrites(session, operationId, state, BACKGROUND_CONTEXT); + + expect(writes.map(address)).toContain("value:delete:pi.pending.entry:staged"); + expect(writes.map(address)).not.toContain("value:delete:pi.pending.entry:placed"); + }); + + it.each([ + [ + "compaction", + { + kind: "compaction", + control: { status: "running" }, + structural: { status: "deciding", taskId: "task" }, + } satisfies CompactionState, + ], + [ + "navigation", + { + kind: "navigation", + control: { status: "running" }, + targetId: null, + summarize: false, + phase: { kind: "ready_to_commit" }, + } satisfies NavigationState, + ], + ] as const)("defensively deletes leftover %s operation families", async (operationId, state) => { + const { session } = await createSession(); + await seedLeftovers(session, operationId, state); + + const writes = await operationCleanupWrites(session, operationId, state, BACKGROUND_CONTEXT); + + expect(writes.map(address)).toEqual([ + `value:delete:pi.op.meta:${operationId}`, + `value:delete:pi.op.state:${operationId}`, + `value:delete:pi.op.tool_args:${operationId}:step:0`, + `value:delete:pi.op.tool_memo:${operationId}:invocation:memo`, + `value:delete:pi.op.preparation:${operationId}:task`, + `value:delete:pi.pending.tool_output:${operationId}:invocation`, + ]); + }); +}); + +describe("runtime2 terminal outcome hydration", () => { + it("hydrates run, compaction, and navigation outcomes from only their direct entry references", async () => { + const { session, storage } = await createSession(); + const assistant = fauxAssistantMessage([{ type: "text", text: "done" }]); + await commit(session, [ + insertEntry({ id: "assistant", parentId: null, type: "message", message: assistant }), + insertEntry({ + id: "compaction", + parentId: null, + type: "compaction", + summary: "summary", + retainedTail: [], + tokensBefore: 10, + fromHook: false, + }), + insertEntry({ + id: "summary", + parentId: null, + type: "branch_summary", + fromId: "old", + summary: "branch", + fromHook: false, + }), + ]); + const getEntries = vi.spyOn(storage, "getEntries"); + const results: LaneLastResult[] = [ + { + operationId: "run", + kind: "run", + outcome: "completed", + leafId: "assistant", + finalAssistantEntryId: "assistant", + runCompletion: "assistant", + }, + { operationId: "compact", kind: "compaction", outcome: "completed", leafId: "compaction" }, + { + operationId: "navigate", + kind: "navigation", + outcome: "completed", + oldLeafId: "old", + leafId: "summary", + summaryEntryId: "summary", + }, + ]; + + const outcomes = await Promise.all( + results.map((result) => hydrateTerminalOutcome(session, result, BACKGROUND_CONTEXT)), + ); + + expect(outcomes).toEqual([ + { + operation: "run", + runId: "run", + kind: "completed", + leafId: "assistant", + finalEntryId: "assistant", + finalMessage: assistant, + }, + { + operation: "compaction", + runId: "compact", + kind: "completed", + leafId: "compaction", + entry: expect.objectContaining({ id: "compaction", type: "compaction", summary: "summary" }), + }, + { + operation: "navigation", + runId: "navigate", + kind: "completed", + oldLeafId: "old", + newLeafId: "summary", + summaryEntry: expect.objectContaining({ id: "summary", type: "branch_summary", summary: "branch" }), + }, + ]); + expect(getEntries.mock.calls.map(([ids]) => ids)).toEqual([["assistant"], ["compaction"], ["summary"]]); + }); + + it("hydrates entry-free outcomes without storage reads", async () => { + const { session, storage } = await createSession(); + const getEntries = vi.spyOn(storage, "getEntries"); + const results: LaneLastResult[] = [ + { + operationId: "run", + kind: "run", + outcome: "completed", + leafId: "tool-result", + runCompletion: "terminated_tools", + }, + { operationId: "compact", kind: "compaction", outcome: "declined", leafId: "leaf" }, + { + operationId: "navigate", + kind: "navigation", + outcome: "aborted", + oldLeafId: "old", + leafId: "old", + }, + ]; + + const outcomes = await Promise.all( + results.map((result) => hydrateTerminalOutcome(session, result, BACKGROUND_CONTEXT)), + ); + + expect(outcomes).toEqual([ + { operation: "run", runId: "run", kind: "completed", leafId: "tool-result" }, + { operation: "compaction", runId: "compact", kind: "declined", leafId: "leaf" }, + { operation: "navigation", runId: "navigate", kind: "aborted", leafId: "old" }, + ]); + expect(getEntries).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/agent/test/harness/runtime2/lane.test.ts b/packages/agent/test/harness/runtime2/lane.test.ts index b358cbad7..f8357e659 100644 --- a/packages/agent/test/harness/runtime2/lane.test.ts +++ b/packages/agent/test/harness/runtime2/lane.test.ts @@ -8,6 +8,7 @@ import { HookRegistry } from "../../../src/harness/hooks.ts"; import { convertToLlm } from "../../../src/harness/messages.ts"; import { Lane } from "../../../src/harness/runtime2/lane.ts"; import { restoreLane } from "../../../src/harness/runtime2/restore.ts"; +import { Drive } from "../../../src/harness/runtime2/types.ts"; import { StorageBackedSession } from "../../../src/harness/session/session.ts"; import type { LaneConfiguration, Session } from "../../../src/harness/session/types.ts"; import * as storedValues from "../../../src/harness/session/values.ts"; @@ -247,6 +248,87 @@ describe("runtime2 Lane commands", () => { ); }); + it("declines an owned commit when its exact drive is replaced while planning", async () => { + const { lane, session } = await createLane(); + const admission = await lane.accept({ kind: "prompt", prompt: "start" }, BACKGROUND_CONTEXT); + if (!admission.ok) throw admission.error; + const drive = new Drive({ operationId: admission.value.operationId }, BACKGROUND_CONTEXT); + lane.activeDrive = drive; + const plannerStarted = deferred(); + const releasePlanner = deferred(); + + const command = lane.commandDriveOwned( + drive, + async (state) => { + plannerStarted.resolve(); + await releasePlanner.promise; + const configuration: LaneConfiguration = { ...state.configuration, thinkingLevel: "high" }; + return { + kind: "commit", + writes: [storedValues.setValue(storedValues.laneConfig("main"), configuration)], + next: { ...state, configuration }, + materialize: () => "committed" as const, + }; + }, + BACKGROUND_CONTEXT, + ); + await plannerStarted.promise; + lane.activeDrive = new Drive({ operationId: admission.value.operationId }, BACKGROUND_CONTEXT); + releasePlanner.resolve(); + + expect(await command).toEqual({ kind: "lost_ownership" }); + expect(lane.state.configuration.thinkingLevel).toBe("off"); + expect((await session.getValue(storedValues.laneConfig("main"), BACKGROUND_CONTEXT))?.value.thinkingLevel).toBe( + "off", + ); + }); + + it("finishes a commit admitted before exact drive replacement", async () => { + const { lane, session, storage } = await createLane(); + const admission = await lane.accept({ kind: "prompt", prompt: "start" }, BACKGROUND_CONTEXT); + if (!admission.ok) throw admission.error; + const drive = new Drive({ operationId: admission.value.operationId }, BACKGROUND_CONTEXT); + lane.activeDrive = drive; + const commitStarted = deferred(); + const releaseCommit = deferred(); + storage.beforeNextCommit = async () => { + commitStarted.resolve(); + await releaseCommit.promise; + }; + + const command = lane.commandDriveOwned( + drive, + (state) => { + const configuration: LaneConfiguration = { ...state.configuration, thinkingLevel: "high" }; + return { + kind: "commit", + writes: [storedValues.setValue(storedValues.laneConfig("main"), configuration)], + next: { ...state, configuration }, + materialize: () => "committed" as const, + }; + }, + BACKGROUND_CONTEXT, + ); + await commitStarted.promise; + const replacement = new Drive({ operationId: admission.value.operationId }, BACKGROUND_CONTEXT); + let publishedBeforeReplacement = false; + const replacing = lane.command((state) => { + publishedBeforeReplacement = state === lane.state && state.configuration.thinkingLevel === "high"; + lane.activeDrive = replacement; + return { kind: "return", result: undefined }; + }, BACKGROUND_CONTEXT); + releaseCommit.resolve(); + + expect(await command).toBe("committed"); + await replacing; + expect(publishedBeforeReplacement).toBe(true); + expect(lane.activeDrive).toBe(replacement); + expect(lane.state.configuration.thinkingLevel).toBe("high"); + expect((await session.getValue(storedValues.laneConfig("main"), BACKGROUND_CONTEXT))?.value.thinkingLevel).toBe( + "high", + ); + }); + it("rejects work after sealing while an admitted commit finishes", async () => { const { lane, storage } = await createLane(); const commitStarted = deferred(); diff --git a/packages/agent/test/harness/runtime2/progress.test.ts b/packages/agent/test/harness/runtime2/progress.test.ts index ce78e800f..1a98dc10a 100644 --- a/packages/agent/test/harness/runtime2/progress.test.ts +++ b/packages/agent/test/harness/runtime2/progress.test.ts @@ -1,5 +1,5 @@ import { type AssistantMessageFrame, createModels } from "@earendil-works/pi-ai"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import type { WatchHandle } from "../../../src/harness/agent-harness.ts"; import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; import { BACKGROUND_CONTEXT, type Context, withAbortSignal } from "../../../src/harness/context.ts"; @@ -11,6 +11,7 @@ import { MemoryStorage } from "../../../src/harness/session/memory.ts"; import { StorageBackedSession } from "../../../src/harness/session/session.ts"; import type { RunState, Session, Write } from "../../../src/harness/session/types.ts"; import * as storedValues from "../../../src/harness/session/values.ts"; +import { deferred } from "./test-utils.ts"; const sessions: Session[] = []; const operationId = "operation"; @@ -22,12 +23,16 @@ const configuration = { class FailingStorage extends MemoryStorage { failure: Error | undefined; + beforeNextCommit: (() => Promise) | undefined; - override commit(writes: Write[], context: Context) { + override async commit(writes: Write[], context: Context) { + const beforeCommit = this.beforeNextCommit; + this.beforeNextCommit = undefined; + await beforeCommit?.(); if (this.failure === undefined) return super.commit(writes, context); const failure = this.failure; this.failure = undefined; - return Promise.reject(failure); + throw failure; } } @@ -211,6 +216,134 @@ describe("runtime2 progress channels", () => { ); }); + it("declines a queued frame after the authoritative projection leaves its phase without control reads", async () => { + const responseEntryId = "response"; + const { lane, drive, storage } = await createFixture( + runState({ + kind: "assistant", + generation: { + status: "effect_pending", + context: { + stepId: "step", + triggerEntryId: "trigger", + configuration, + streamOptions: {}, + retryPolicy: { maxAttempts: 2, baseDelayMs: 1 }, + overflowRecoveryUsed: false, + }, + attempt: 1, + responseEntryId, + usageId: "usage", + intendedOutputLimit: 100, + contextWindow: 1_000, + }, + }), + ); + const progress = openFrameProgress(lane, drive, responseEntryId); + const commitStarted = deferred(); + const releaseCommit = deferred(); + storage.beforeNextCommit = async () => { + commitStarted.resolve(); + await releaseCommit.promise; + }; + const getValue = vi.spyOn(storage, "getValue"); + const moving = lane.command((state) => { + const operation = state.operation; + if (operation?.state.kind !== "run") throw new Error("missing run"); + const nextState: RunState = { + ...operation.state, + phase: { + kind: "checkpoint", + continuation: { kind: "may_finish", includeFinalAssistant: true }, + triggerEntryId: responseEntryId, + }, + }; + return { + kind: "commit", + writes: [storedValues.setValue(storedValues.operationState(operationId), nextState)], + next: { ...state, operation: { meta: operation.meta, state: nextState } }, + materialize: () => undefined, + }; + }, BACKGROUND_CONTEXT); + await commitStarted.promise; + progress.write({ type: "text_delta", contentIndex: 0, delta: "late" }); + releaseCommit.resolve(); + await moving; + await progress.drain(); + + expect( + await lane.session.readList( + storedValues.pendingAssistantFrames(operationId, responseEntryId), + undefined, + BACKGROUND_CONTEXT, + ), + ).toEqual([]); + expect(getValue).not.toHaveBeenCalled(); + }); + + it("declines a queued frame after terminal projection publication", async () => { + const responseEntryId = "response"; + const { lane, drive, storage } = await createFixture( + runState({ + kind: "assistant", + generation: { + status: "effect_pending", + context: { + stepId: "step", + triggerEntryId: "trigger", + configuration, + streamOptions: {}, + retryPolicy: { maxAttempts: 2, baseDelayMs: 1 }, + overflowRecoveryUsed: false, + }, + attempt: 1, + responseEntryId, + usageId: "usage", + intendedOutputLimit: 100, + contextWindow: 1_000, + }, + }), + ); + const progress = openFrameProgress(lane, drive, responseEntryId); + const commitStarted = deferred(); + const releaseCommit = deferred(); + storage.beforeNextCommit = async () => { + commitStarted.resolve(); + await releaseCommit.promise; + }; + const getValue = vi.spyOn(storage, "getValue"); + const ending = lane.command( + (state) => ({ + kind: "commit", + writes: [ + storedValues.deleteValue(storedValues.operationMeta(operationId)), + storedValues.deleteValue(storedValues.operationState(operationId)), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: null, + pendingNextRun: state.pendingNextRun, + }), + ], + next: { ...state, operation: null }, + materialize: () => undefined, + }), + BACKGROUND_CONTEXT, + ); + await commitStarted.promise; + progress.write({ type: "text_delta", contentIndex: 0, delta: "late" }); + releaseCommit.resolve(); + await ending; + await progress.drain(); + + expect( + await lane.session.readList( + storedValues.pendingAssistantFrames(operationId, responseEntryId), + undefined, + BACKGROUND_CONTEXT, + ), + ).toEqual([]); + expect(getValue).not.toHaveBeenCalled(); + }); + it("replaces tool checkpoints and fences writes after owner identity changes", async () => { const invocationId = "result"; const { lane, drive } = await createFixture( diff --git a/packages/agent/test/harness/types.test.ts b/packages/agent/test/harness/types.test.ts index a90d6d279..f2f90f9da 100644 --- a/packages/agent/test/harness/types.test.ts +++ b/packages/agent/test/harness/types.test.ts @@ -477,13 +477,28 @@ it("covers Part 5 results, events, hooks, snapshots, tools, and stream options", expectTypeOf< Extract< keyof AgentLane, - "session" | "models" | "hooks" | "activeDrive" | "command" | "readConfig" | "isDriveActive" | "mismatch" + | "session" + | "models" + | "hooks" + | "activeDrive" + | "command" + | "commandDriveOwned" + | "readConfig" + | "isDriveActive" + | "mismatch" > >().toEqualTypeOf(); expectTypeOf< Extract< keyof AgentHarness, - "session" | "models" | "activeDrive" | "command" | "readConfig" | "isDriveActive" | "mismatch" + | "session" + | "models" + | "activeDrive" + | "command" + | "commandDriveOwned" + | "readConfig" + | "isDriveActive" + | "mismatch" > >().toEqualTypeOf(); expectTypeOf().toEqualTypeOf<