mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(agent): add durable terminal mechanics
This commit is contained in:
Mario Zechner
parent
6bb8252099
commit
9f666bfbbb
10 changed files with 990 additions and 76 deletions
|
|
@ -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<TResult>(
|
||||
async commandDriveOwned<TResult>(
|
||||
drive: Drive,
|
||||
plan: (state: LaneState, reader: SessionReader) =>
|
||||
LaneCommand<TResult> | Promise<LaneCommand<TResult>>,
|
||||
|
|
@ -475,12 +475,12 @@ async commandOwned<TResult>(
|
|||
}
|
||||
```
|
||||
|
||||
`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<TContext extends object | undefined, T>(
|
|||
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<TContext extends object | undefined, T>(
|
|||
}
|
||||
```
|
||||
|
||||
`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<TContext extends object | undefined>(
|
|||
): Promise<ProcedureResult> { /* ... */ }
|
||||
```
|
||||
|
||||
`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.
|
||||
|
|
|
|||
248
packages/agent/src/harness/runtime2/drive/terminal.ts
Normal file
248
packages/agent/src/harness/runtime2/drive/terminal.ts
Normal file
|
|
@ -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<Write[]> {
|
||||
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<string>();
|
||||
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<MessageEntry & { message: SettledAssistantMessage }> {
|
||||
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<CompactionEntry> {
|
||||
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<BranchSummaryEntry> {
|
||||
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<TerminalOperationOutcome> {
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
|
@ -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<void>;
|
||||
type WatchHandler = <T>(snapshot: T, filter: (event: HarnessEvent) => boolean, context: Context) => WatchHandle<T>;
|
||||
|
|
@ -115,6 +122,7 @@ export class Lane<TContext extends object | undefined> implements AgentLane {
|
|||
private readonly config: () => Config<TContext>;
|
||||
/** 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<TContext extends object | undefined> 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<TResult>(
|
||||
|
|
@ -243,6 +253,70 @@ export class Lane<TContext extends object | undefined> 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<TResult>(
|
||||
drive: Drive,
|
||||
plan: (state: LaneState, reader: SessionReader) => LaneCommand<TResult> | Promise<LaneCommand<TResult>>,
|
||||
context: Context,
|
||||
): Promise<TResult | LostOwnership> {
|
||||
this.assertOpen();
|
||||
let outcome: LaneCommandOutcome<TResult | LostOwnership>;
|
||||
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<OperationAdmissionResult> {
|
||||
if (this.closedError instanceof HarnessClosed) {
|
||||
return Result.err(new Closed({ message: this.closedError.message }));
|
||||
|
|
|
|||
|
|
@ -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<T> {
|
||||
write(item: T): void;
|
||||
|
|
@ -27,22 +24,28 @@ function openProgress<TContext extends object | undefined, T>(
|
|||
drive: Drive,
|
||||
commitWrite: (item: T) => Write,
|
||||
clear: () => Write,
|
||||
stillOwns: (reader: SessionReader) => Promise<boolean>,
|
||||
stillOwns: (state: LaneState) => boolean,
|
||||
): ProgressChannel<T> {
|
||||
let sealed = false;
|
||||
let latest: Promise<void> = 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<TContext extends object | undefined, T>(
|
|||
};
|
||||
}
|
||||
|
||||
async function readOwnedRun<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
reader: SessionReader,
|
||||
): Promise<RunState | undefined> {
|
||||
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<TContext extends object | undefined>(
|
||||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
|
|
@ -88,15 +70,16 @@ export function openFrameProgress<TContext extends object | undefined>(
|
|||
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<TContext extends object | undefined>(
|
|||
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",
|
||||
);
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<TMetadata extends SessionMetadata = SessionMetadata> implements Session<TMetadata> {
|
||||
readonly metadata: TMetadata;
|
||||
readonly idGenerator: IdGenerator;
|
||||
|
|
|
|||
367
packages/agent/test/harness/runtime2/drive-terminal.test.ts
Normal file
367
packages/agent/test/harness/runtime2/drive-terminal.test.ts
Normal file
|
|
@ -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<void> {
|
||||
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<void> {
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<void>) | 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(
|
||||
|
|
|
|||
|
|
@ -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<never>();
|
||||
expectTypeOf<
|
||||
Extract<
|
||||
keyof AgentHarness,
|
||||
"session" | "models" | "activeDrive" | "command" | "readConfig" | "isDriveActive" | "mismatch"
|
||||
| "session"
|
||||
| "models"
|
||||
| "activeDrive"
|
||||
| "command"
|
||||
| "commandDriveOwned"
|
||||
| "readConfig"
|
||||
| "isDriveActive"
|
||||
| "mismatch"
|
||||
>
|
||||
>().toEqualTypeOf<never>();
|
||||
expectTypeOf<OperationRequest["kind"]>().toEqualTypeOf<
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue