sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(agent): inspect restored runtime2 operations

Browse source
This commit is contained in:
Mario Zechner 2026-08-16 18:00:17 +02:00
parent 497faecc6d
commit 03f576f548
16 changed files with 517 additions and 173 deletions
Download patch file Download diff file Expand all files Collapse all files

18
packages/agent/docs/harness.md
View file

@ -1979,6 +1979,8 @@ Then run §3.3's bounded semantic checks over exactly that set. Every named comm
Restore never reads register history, folds anything, scans tables, builds provider context, audits completed operations, or infers a transition from an absent planned value. It installs no `ActiveOperation` and invokes no breakpoint. The application or serving layer explicitly activates an open operation with `drive({ operationId })`; `resume()` is the convenience composition that inspects and drives the current operation with one deferred-poll permit. Cancellation is durably admitted through `requestAbort(operationId)` or its `abort()` convenience wrapper.
Harness creation does not write a durable `suspended` state. An open operation remains at its committed phase with control `running` or `cancel_requested`. `suspended` is the public observation that no process-local `drive()` call currently owns that operation. A matching `drive({ operationId })` installs the local owner and dispatches from the committed phase; process loss removes the owner without changing durable control or phase.
### Worked example — crash in the uncertain window
The process died after committing an assistant intent:
@ -2274,7 +2276,7 @@ type OptionalFinalAssistant =
type MissingIdentitySuspension = {
kind: "suspended"; reason: "missing_identities";
missing: { tools: string[]; models: string[] };
missing: { tools: string[]; model?: string };
};
type RunOutcome =
@ -2312,7 +2314,7 @@ Expected errors use the existing `TaggedError` implementation in `harness/result
|---|---|
| `LaneBusy` | `lane`, `operationId`, `operationKind` |
| `OperationMismatch` | `lane`, `expectedOperationId`, optional `currentOperationId` and `lastOperationId` |
| `MissingIdentities` | `lane`, `tools`, `models` |
| `MissingIdentities` | `lane`, `tools`, `model` |
| `NoActiveRun`, `NoActiveOperation`, `NothingToResume`, `NothingToCompact` | `lane` |
| `InvalidMessage`, `InvalidNavigation` | `lane`, `reason` |
| `UnknownSkill`, `UnknownTemplate` | `name` |
@ -2339,7 +2341,7 @@ type DriveOutcome =
| { kind: "waiting"; operationId: string; reason: "deferred";
deferred: DeferredHandle }
| { kind: "waiting"; operationId: string; reason: "missing_identities";
missing: { tools: string[]; models: string[] } }
missing: { tools: string[]; model?: string } }
| { kind: "yielded"; operationId: string };
type DriveResult = Result<DriveOutcome, OperationMismatch | Closed>;
@ -2435,9 +2437,9 @@ type SuspendedOperation = {
} & (
| { reason: "deferred"; deferred: DeferredHandle; missing?: never }
| { reason: "missing_identities";
missing: { tools: string[]; models: string[] }; deferred?: never }
missing: { tools: string[]; model?: string }; deferred?: never }
| { reason: "crash"; deferred?: DeferredHandle;
missing?: { tools: string[]; models: string[] } }
missing?: { tools: string[]; model?: string } }
);
// QueueMode, RetryPolicy, and CompactionSettings use the source types named in §0.7.
@ -2620,11 +2622,11 @@ type HarnessEventPayload =
| { type: "run_suspend"; runId: string; reason: "deferred";
deferred: DeferredHandle }
| { type: "run_suspend"; runId: string; reason: "missing_identities";
missing: { tools: string[]; models: string[] } }
missing: { tools: string[]; model?: string } }
| { type: "compaction_suspend"; runId: string; reason: "missing_identities";
missing: { tools: string[]; models: string[] } }
missing: { tools: string[]; model?: string } }
| { type: "navigation_suspend"; runId: string; reason: "missing_identities";
missing: { tools: string[]; models: string[] } }
missing: { tools: string[]; model?: string } }
| { type: "run_abort"; runId: string; steer: AgentMessage[]; followUp: AgentMessage[] }
| ({ type: "run_end"; runId: string; leafId: string | null } & (
| ({ outcome: "completed" | "aborted" } & OptionalFinalAssistant)

26
packages/agent/src/harness/agent-harness.ts
View file

@ -53,12 +53,14 @@ export class OperationMismatch extends TaggedError("OperationMismatch")<{
lastOperationId?: string;
message: string;
}> {}
export class MissingIdentities extends TaggedError("MissingIdentities")<{
lane: string;
export interface MissingIdentityInfo {
tools: string[];
models: string[];
message: string;
}> {}
model?: string;
}
export class MissingIdentities extends TaggedError("MissingIdentities")<
{ lane: string; message: string } & MissingIdentityInfo
> {}
export class NoActiveRun extends TaggedError("NoActiveRun")<{ lane: string; message: string }> {}
export class NoActiveOperation extends TaggedError("NoActiveOperation")<{ lane: string; message: string }> {}
export class NothingToResume extends TaggedError("NothingToResume")<{ lane: string; message: string }> {}
@ -108,7 +110,7 @@ export type OptionalFinalAssistant =
export type MissingIdentitySuspension = {
kind: "suspended";
reason: "missing_identities";
missing: { tools: string[]; models: string[] };
missing: MissingIdentityInfo;
};
export type RunOutcome =
@ -236,7 +238,7 @@ export type DriveOutcome =
kind: "waiting";
operationId: string;
reason: "missing_identities";
missing: { tools: string[]; models: string[] };
missing: MissingIdentityInfo;
}
| { kind: "yielded"; operationId: string };
export type DriveResult = Result<DriveOutcome, OperationMismatch | Closed>;
@ -284,13 +286,13 @@ export type SuspendedOperation = {
| { reason: "deferred"; deferred: DeferredHandle; missing?: never }
| {
reason: "missing_identities";
missing: { tools: string[]; models: string[] };
missing: MissingIdentityInfo;
deferred?: never;
}
| {
reason: "crash";
deferred?: DeferredHandle;
missing?: { tools: string[]; models: string[] };
missing?: MissingIdentityInfo;
}
);
@ -342,19 +344,19 @@ export type HarnessEventPayload =
type: "run_suspend";
runId: string;
reason: "missing_identities";
missing: { tools: string[]; models: string[] };
missing: MissingIdentityInfo;
}
| {
type: "compaction_suspend";
runId: string;
reason: "missing_identities";
missing: { tools: string[]; models: string[] };
missing: MissingIdentityInfo;
}
| {
type: "navigation_suspend";
runId: string;
reason: "missing_identities";
missing: { tools: string[]; models: string[] };
missing: MissingIdentityInfo;
}
| { type: "run_abort"; runId: string; steer: AgentMessage[]; followUp: AgentMessage[] }
| ({ type: "run_end"; runId: string; leafId: string | null } & (

25
packages/agent/src/harness/config.ts
View file

@ -1,5 +1,7 @@
import type { RetryPolicy } from "@earendil-works/pi-ai";
import type { Models, RetryPolicy } from "@earendil-works/pi-ai";
import type { MissingIdentityInfo } from "./agent-harness.ts";
import type { CompactionSettings } from "./compaction/compaction.ts";
import type { LaneConfiguration } from "./session/types.ts";
export const DEFAULT_RETRY_POLICY: RetryPolicy = { enabled: true, maxRetries: 3, baseDelayMs: 1_000 };
@ -23,6 +25,27 @@ export function validateRetryPolicy(policy: RetryPolicy): void {
}
}
export function hasMissingIdentities(info: MissingIdentityInfo): boolean {
return info.tools.length !== 0 || info.model !== undefined;
}
export function missingIdentities(
models: Models,
configuration: LaneConfiguration,
tools: readonly { name: string }[],
): MissingIdentityInfo {
const model = models.getModel(configuration.model.provider, configuration.model.modelId);
return {
tools: missingToolIdentities(configuration, tools),
...(model === undefined ? { model: `${configuration.model.provider}/${configuration.model.modelId}` } : {}),
};
}
export function missingToolIdentities(configuration: LaneConfiguration, tools: readonly { name: string }[]): string[] {
const availableTools = new Set(tools.map((tool) => tool.name));
return configuration.activeToolNames.filter((name) => !availableTools.has(name));
}
export function validateCompactionSettings(settings: CompactionSettings): void {
if (
!Number.isSafeInteger(settings.reserveTokens) ||

20
packages/agent/src/harness/runtime/agent-harness-runtime.ts
View file

@ -38,7 +38,15 @@ import {
type WatchHandle,
} from "../agent-harness.ts";
import { type CompactionSettings, DEFAULT_COMPACTION_SETTINGS } from "../compaction/compaction.ts";
import { DEFAULT_RETRY_POLICY, validateCompactionSettings, validateRetryPolicy, validateToolNames } from "../config.ts";
import {
DEFAULT_RETRY_POLICY,
hasMissingIdentities,
missingIdentities,
missingToolIdentities,
validateCompactionSettings,
validateRetryPolicy,
validateToolNames,
} from "../config.ts";
import { HarnessEventBus } from "../events.ts";
import { HookRegistry } from "../hooks.ts";
import { convertToLlm } from "../messages.ts";
@ -61,7 +69,7 @@ import type {
} from "../session/types.ts";
import type { AgentHarnessTool, AgentHarnessToolContextSource } from "../types.ts";
import { AgentLaneRuntime, createPublicSessionView } from "./lane-runtime.ts";
import { cloneConfiguration, missingIdentities, missingToolIdentities, suspensionBase } from "./transitions.ts";
import { cloneConfiguration, suspensionBase } from "./transitions.ts";
import {
type ActiveOperation,
type AdmissionReservation,
@ -448,15 +456,15 @@ class AgentHarnessRuntime<TContext extends object | undefined> implements AgentH
const missing = missingIdentities(
this.models,
current.state.phase.generation.context.configuration,
this.settings,
this.settings.tools,
);
if (missing.tools.length !== 0 || missing.models.length !== 0) {
if (hasMissingIdentities(missing)) {
return { ...base, reason: "crash", missing };
}
}
if (current.state.kind === "run" && current.state.phase.kind === "tools") {
const missing = missingToolIdentities(current.state.phase.batch.configuration, this.settings);
if (missing.length !== 0) return { ...base, reason: "crash", missing: { tools: missing, models: [] } };
const missing = missingToolIdentities(current.state.phase.batch.configuration, this.settings.tools);
if (missing.length !== 0) return { ...base, reason: "crash", missing: { tools: missing } };
}
return { ...base, reason: "crash" };
}

6
packages/agent/src/harness/runtime/assistant-procedure.ts
View file

@ -16,6 +16,7 @@ import {
HarnessFault,
type SuspendedOperation,
} from "../agent-harness.ts";
import { hasMissingIdentities, missingIdentities } from "../config.ts";
import { streamHarnessAssistant } from "../execution/assistant.ts";
import { applyStreamOptionsPatch } from "../hooks.ts";
import { type RestoredLane, restoreLane } from "../restore.ts";
@ -32,7 +33,6 @@ import {
cloneConfiguration,
cloneUsage,
deadlineReached,
missingIdentities,
normalizeInvalidDeferredResponse,
normalizeRetryPolicy,
predictAssistantStepOutcome,
@ -563,8 +563,8 @@ export async function executeAssistantGeneration<TContext extends object | undef
const ready = state.phase.generation;
const context = ready.context;
const settings = await runtime.snapshotSettings();
const missing = missingIdentities(runtime.models, context.configuration, settings);
if (missing.tools.length !== 0 || missing.models.length !== 0) {
const missing = missingIdentities(runtime.models, context.configuration, settings.tools);
if (hasMissingIdentities(missing)) {
const descriptor: SuspendedOperation = {
...suspensionBase(restored),
reason: "missing_identities",

7
packages/agent/src/harness/runtime/lane-runtime.ts
View file

@ -32,6 +32,7 @@ import {
UnknownTemplate,
type WatchHandle,
} from "../agent-harness.ts";
import { hasMissingIdentities, missingIdentities } from "../config.ts";
import { BreakpointBarrier } from "../execution/breakpoint.ts";
import { formatPromptTemplateInvocation } from "../prompt-templates.ts";
import { type RestoredLane, restoreLane } from "../restore.ts";
@ -50,7 +51,7 @@ import type {
} from "../session/types.ts";
import { formatSkillInvocation } from "../skills.ts";
import { deferredValue, driveLane } from "./operation-task.ts";
import { isPendingAssistant, missingIdentities } from "./transitions.ts";
import { isPendingAssistant } from "./transitions.ts";
import type {
AcceptancePublication,
AdmissionReservation,
@ -186,8 +187,8 @@ export async function acceptLane<TContext extends object | undefined>(
}),
);
}
const missing = missingIdentities(runtime.models, restored.configuration, settings);
if (missing.tools.length !== 0 || missing.models.length !== 0) {
const missing = missingIdentities(runtime.models, restored.configuration, settings.tools);
if (hasMissingIdentities(missing)) {
return Result.err(
new MissingIdentities({
lane: lane.name,

2
packages/agent/src/harness/runtime/tool-batch-procedure.ts
View file

@ -1000,7 +1000,7 @@ async function suspendToolBatchForMissingIdentities<TContext extends object | un
tools: string[],
eventOrigin: ToolEventOrigin,
): Promise<ToolBatchExecutionResult> {
const missing = { tools, models: [] };
const missing = { tools };
runtime.restoredSuspensions.set(lane.name, {
...suspensionBase(restored),
reason: "missing_identities",

25
packages/agent/src/harness/runtime/transitions.ts
View file

@ -1,4 +1,4 @@
import type { Api, AssistantMessage, Models, RetryPolicy, Usage } from "@earendil-works/pi-ai";
import type { Api, AssistantMessage, RetryPolicy, Usage } from "@earendil-works/pi-ai";
import { isContextOverflow, isRecoverableLength, isRetryableAssistantError } from "@earendil-works/pi-ai";
import type { AgentMessage, AgentToolCall } from "../../types.ts";
import type { DriveOptions, SuspendedOperation, TerminalOperationOutcome } from "../agent-harness.ts";
@ -11,7 +11,7 @@ import type {
SessionMutator,
SettledAssistantMessage,
} from "../session/types.ts";
import type { AssistantSettlementDecision, RuntimeSettings } from "./types.ts";
import type { AssistantSettlementDecision } from "./types.ts";
import { RuntimeSliceNotImplemented } from "./types.ts";
export function uuidV7Timestamp(id: string): number {
@ -226,27 +226,6 @@ export function suspensionBase(restored: RestoredLane): Omit<SuspendedOperation,
};
}
export function missingIdentities<TContext extends object | undefined>(
models: Models,
configuration: LaneConfiguration,
settings: RuntimeSettings<TContext>,
): { tools: string[]; models: string[] } {
const model = models.getModel(configuration.model.provider, configuration.model.modelId);
const availableTools = new Set(settings.tools.map((tool) => tool.name));
return {
tools: configuration.activeToolNames.filter((name) => !availableTools.has(name)),
models: model === undefined ? [`${configuration.model.provider}/${configuration.model.modelId}`] : [],
};
}
export function missingToolIdentities<TContext extends object | undefined>(
configuration: LaneConfiguration,
settings: RuntimeSettings<TContext>,
): string[] {
const availableTools = new Set(settings.tools.map((tool) => tool.name));
return configuration.activeToolNames.filter((name) => !availableTools.has(name));
}
export function cloneUsage(usage: Usage): Usage {
return { ...usage, cost: { ...usage.cost } };
}

3
packages/agent/src/harness/runtime/types.ts
View file

@ -7,6 +7,7 @@ import type {
DriveResult,
HarnessClosed,
HarnessFault,
MissingIdentityInfo,
Resources,
SuspendedOperation,
} from "../agent-harness.ts";
@ -112,7 +113,7 @@ export interface CommittedAssistantSettlement {
export type AssistantExecutionResult =
| { kind: "advanced" }
| { kind: "yielded" }
| { kind: "missing_identities"; missing: { tools: string[]; models: string[] } };
| { kind: "missing_identities"; missing: MissingIdentityInfo };
export type ToolBatchExecutionResult = AssistantExecutionResult;

124
packages/agent/src/harness/runtime2/harness.ts
View file

@ -1,4 +1,4 @@
import type { RetryPolicy } from "@earendil-works/pi-ai";
import type { Models, RetryPolicy } from "@earendil-works/pi-ai";
import { NOOP_TELEMETRY_CONTEXT } from "@earendil-works/pi-telemetry";
import type { QueueMode } from "../../types.ts";
import type {
@ -13,7 +13,15 @@ import type {
} from "../agent-harness.ts";
import { Closed, HarnessClosed, HarnessFault, InvalidLane, LaneExists, UnknownTarget } from "../agent-harness.ts";
import { type CompactionSettings, DEFAULT_COMPACTION_SETTINGS } from "../compaction/compaction.ts";
import { DEFAULT_RETRY_POLICY, validateCompactionSettings, validateRetryPolicy, validateToolNames } from "../config.ts";
import {
DEFAULT_RETRY_POLICY,
hasMissingIdentities,
missingIdentities,
missingToolIdentities,
validateCompactionSettings,
validateRetryPolicy,
validateToolNames,
} from "../config.ts";
import { HarnessEventBus } from "../events.ts";
import { HookRegistry } from "../hooks.ts";
import { convertToLlm } from "../messages.ts";
@ -42,7 +50,12 @@ export class Harness<TContext extends object | undefined> extends Lane implement
closePromise: Promise<void> | undefined;
faultError: HarnessFault | undefined;
constructor(options: AgentHarnessOptions<TContext>, seed: LaneConfiguration, restored: Map<string, LaneState>) {
constructor(
options: AgentHarnessOptions<TContext>,
seed: LaneConfiguration,
restored: Map<string, LaneState>,
suspensionsByLane: Map<string, SuspendedOperation>,
) {
const main = restored.get("main");
if (main === undefined) throw new SessionInvariantError("Session is missing main lane");
super(
@ -52,6 +65,7 @@ export class Harness<TContext extends object | undefined> extends Lane implement
main,
(cause) => this.fault(cause),
(event) => this.events.emit(event),
suspensionsByLane.get("main"),
);
this.seed = seed;
this.config = {
@ -83,7 +97,7 @@ export class Harness<TContext extends object | undefined> extends Lane implement
);
this.lanesByName.set("main", this);
for (const [name, state] of restored) {
if (name !== "main") this.lanesByName.set(name, this.buildLane(name, state));
if (name !== "main") this.lanesByName.set(name, this.buildLane(name, state, suspensionsByLane.get(name)));
}
}
@ -94,21 +108,19 @@ export class Harness<TContext extends object | undefined> extends Lane implement
async lanes(): Promise<LaneInfo[]> {
this.assertOpen();
return [...this.lanesByName.values()].map((lane) => {
const operation = lane.state.operation;
return {
name: lane.name,
leafId: lane.state.leafId,
operation:
operation === null
? null
: {
id: operation.meta.operationId,
kind: operation.meta.intent.kind,
status: operation.state.control.status === "cancel_requested" ? "aborting" : "suspended",
},
};
});
const executions = await Promise.all([...this.lanesByName.values()].map((lane) => lane.inspectExecution()));
return executions.map((execution) => ({
name: execution.lane,
leafId: execution.leafId,
operation:
execution.current === null
? null
: {
id: execution.current.id,
kind: execution.current.kind,
status: execution.current.status,
},
}));
}
async createLane(name: string, at: string | null): Promise<CreateLaneResult> {
@ -208,7 +220,7 @@ export class Harness<TContext extends object | undefined> extends Lane implement
throw new SliceNotImplemented("watchSession");
}
private buildLane(name: string, state: LaneState): Lane {
private buildLane(name: string, state: LaneState, suspension?: SuspendedOperation): Lane {
return new Lane(
name,
this.session,
@ -216,6 +228,7 @@ export class Harness<TContext extends object | undefined> extends Lane implement
state,
(cause) => this.fault(cause),
(event) => this.events.emit(event),
suspension,
);
}
@ -293,25 +306,74 @@ export async function createAgentHarness<TContext extends object | undefined = o
try {
await seedMain(options.session, seed);
const restored = await restoreSession(options.session);
const harness = new Harness(options, seed, restored);
const suspended: SuspendedOperation[] = [];
for (const [lane, state] of restored) {
const operation = state.operation;
if (operation === null) continue;
suspended.push({
lane,
operationId: operation.meta.operationId,
kind: operation.meta.intent.kind,
startedAt: operation.meta.startedAt,
reason: "crash",
});
if (state.operation !== null) {
suspended.push(await describeSuspension(options.session, options.models, tools, lane, state));
}
}
return { harness, suspended };
const suspensionsByLane = new Map(suspended.map((descriptor) => [descriptor.lane, descriptor]));
return { harness: new Harness(options, seed, restored, suspensionsByLane), suspended };
} catch (error) {
throw new HarnessFault("AgentHarness storage or invariant fault", error);
}
}
async function describeSuspension(
session: Session,
models: Models,
tools: readonly { name: string }[],
lane: string,
state: LaneState,
): Promise<SuspendedOperation> {
const operation = state.operation;
if (operation === null) throw new SessionInvariantError(`Lane ${JSON.stringify(lane)} is not suspended`);
const promptIds = operation.meta.intent.kind === "run" ? operation.meta.intent.promptEntryIds : [];
const deferredSourceId =
operation.state.kind === "run" && operation.state.phase.kind === "deferred"
? operation.state.phase.deferred.sourceEntryId
: undefined;
const entries = await session.getEntries([
...new Set([...promptIds, ...(deferredSourceId === undefined ? [] : [deferredSourceId])]),
]);
const prompt = promptIds.map((id) => {
const entry = entries.get(id);
if (entry?.type !== "message") throw new SessionInvariantError(`Prompt entry ${id} is missing`);
return entry.message;
});
const base = {
lane,
operationId: operation.meta.operationId,
kind: operation.meta.intent.kind,
startedAt: operation.meta.startedAt,
...(operation.meta.intent.kind === "run" ? { prompt } : {}),
};
if (deferredSourceId !== undefined) {
const entry = entries.get(deferredSourceId);
if (entry?.type !== "message" || entry.message.role !== "assistant" || entry.message.deferred === undefined) {
throw new SessionInvariantError("Deferred suspension source is invalid");
}
return { ...base, reason: "deferred", deferred: entry.message.deferred };
}
if (
operation.state.kind === "run" &&
operation.state.phase.kind === "assistant" &&
operation.state.phase.generation.status === "ready"
) {
const missing = missingIdentities(models, operation.state.phase.generation.context.configuration, tools);
if (hasMissingIdentities(missing)) {
return { ...base, reason: "crash", missing };
}
}
if (operation.state.kind === "run" && operation.state.phase.kind === "tools") {
const missing = missingToolIdentities(operation.state.phase.batch.configuration, tools);
if (missing.length !== 0) {
return { ...base, reason: "crash", missing: { tools: missing } };
}
}
return { ...base, reason: "crash" };
}
async function seedMain(session: Session, seed: LaneConfiguration): Promise<void> {
await session.mutate("main", async (mutator) => {
const [leaf, state, configuration, lastResult] = await Promise.all([

47
packages/agent/src/harness/runtime2/lane.ts
View file

@ -1,11 +1,18 @@
import type { Api, Model, Models } from "@earendil-works/pi-ai";
import type { ThinkingLevel } from "../../types.ts";
import type { AgentLane, HarnessEvent, LaneConfigEventPayload } from "../agent-harness.ts";
import type {
AgentLane,
HarnessEvent,
LaneConfigEventPayload,
LaneExecutionInfo,
SuspendedOperation,
} from "../agent-harness.ts";
import { SessionPendingAssistantMessageError } from "../session/session.ts";
import type {
BranchScan,
CommitResult,
Entry,
LaneLastResult,
PendingEntry,
Session,
SessionReader,
@ -44,6 +51,7 @@ export class Lane implements AgentLane {
protected readonly models: Models;
private readonly onFault: FaultHandler;
private readonly onEvent: EventHandler;
private suspension: SuspendedOperation | undefined;
state: LaneState;
closedError: Error | undefined;
@ -54,6 +62,7 @@ export class Lane implements AgentLane {
state: LaneState,
onFault: FaultHandler,
onEvent: EventHandler,
suspension?: SuspendedOperation,
) {
this.session = session;
this.models = models;
@ -71,6 +80,7 @@ export class Lane implements AgentLane {
this.state = state;
this.onFault = onFault;
this.onEvent = onEvent;
this.suspension = suspension;
}
async getLeafId(): Promise<string | null> {
@ -78,7 +88,7 @@ export class Lane implements AgentLane {
return this.state.leafId;
}
async getLastResult() {
async getLastResult(): Promise<LaneLastResult | undefined> {
this.assertOpen();
return this.state.lastResult;
}
@ -118,6 +128,9 @@ export class Lane implements AgentLane {
case "commit": {
const commit = await mutator.commit(decision.transaction);
this.state = decision.next;
if (this.suspension?.operationId !== decision.next.operation?.meta.operationId) {
this.suspension = undefined;
}
const result = decision.materialize(commit);
if (isPromiseLike(result)) throw new TypeError("Lane command materialize() must be synchronous");
return { kind: "return", result };
@ -148,8 +161,34 @@ export class Lane implements AgentLane {
throw new SliceNotImplemented("requestAbort");
}
async inspectExecution(): Promise<never> {
throw new SliceNotImplemented("inspectExecution");
async inspectExecution(): Promise<LaneExecutionInfo> {
this.assertOpen();
const operation = this.state.operation;
if (operation === null) {
return {
lane: this.name,
leafId: this.state.leafId,
current: null,
...(this.state.lastResult === undefined ? {} : { lastResult: this.state.lastResult }),
};
}
const status = operation.state.control.status === "cancel_requested" ? "aborting" : "suspended";
const suspended =
status === "suspended" && this.suspension?.operationId === operation.meta.operationId
? this.suspension
: undefined;
return {
lane: this.name,
leafId: this.state.leafId,
current: {
id: operation.meta.operationId,
kind: operation.meta.intent.kind,
status,
startedAt: operation.meta.startedAt,
...(suspended === undefined ? {} : { suspended }),
},
...(this.state.lastResult === undefined ? {} : { lastResult: this.state.lastResult }),
};
}
async prompt(): Promise<never> {

4
packages/agent/test/harness/agent-harness-r2.test.ts
View file

@ -239,7 +239,7 @@ describe("AgentHarness R2 minimal run", () => {
await modelFixture.harness.setModel({ ...modelFixture.faux.getModel(), id: "missing-model" });
expect(await modelFixture.harness.accept({ kind: "prompt", prompt: "hello" })).toMatchObject({
ok: false,
error: { _tag: "MissingIdentities", models: [expect.stringContaining("missing-model")], tools: [] },
error: { _tag: "MissingIdentities", model: expect.stringContaining("missing-model"), tools: [] },
});
expect((await modelFixture.harness.inspectExecution()).current).toBeNull();
@ -247,7 +247,7 @@ describe("AgentHarness R2 minimal run", () => {
await toolFixture.harness.setActiveTools(["missing-tool"]);
expect(await toolFixture.harness.accept({ kind: "prompt", prompt: "hello" })).toMatchObject({
ok: false,
error: { _tag: "MissingIdentities", models: [], tools: ["missing-tool"] },
error: { _tag: "MissingIdentities", tools: ["missing-tool"] },
});
expect((await toolFixture.harness.inspectExecution()).current).toBeNull();
});

2
packages/agent/test/harness/agent-harness-r3.test.ts
View file

@ -772,7 +772,7 @@ describe("AgentHarness R3 generation recovery and retry", () => {
value: {
kind: "waiting",
reason: "missing_identities",
missing: { models: [expect.stringContaining("faux-1")], tools: ["echo"] },
missing: { model: expect.stringContaining("faux-1"), tools: ["echo"] },
},
});
expect(fixture.faux.state.callCount).toBe(0);

6
packages/agent/test/harness/agent-harness-r4.test.ts
View file

@ -470,7 +470,7 @@ describe("AgentHarness R4 tools", () => {
await fixture.harness.executeAction();
expect(await firstDrive).toMatchObject({
ok: true,
value: { kind: "waiting", reason: "missing_identities", missing: { tools: ["echo"], models: [] } },
value: { kind: "waiting", reason: "missing_identities", missing: { tools: ["echo"] } },
});
expect(executions).toBe(0);
expect((await fixture.session.getRegister("op.state", accepted.value.operationId))?.value).toMatchObject({
@ -808,7 +808,7 @@ describe("AgentHarness R4 tools", () => {
expect(await reopened.harness.drive({ operationId: accepted.value.operationId })).toMatchObject({
ok: true,
value: { kind: "waiting", reason: "missing_identities", missing: { tools: ["echo"], models: [] } },
value: { kind: "waiting", reason: "missing_identities", missing: { tools: ["echo"] } },
});
expect(executions).toBe(1);
const branch = await reopened.harness.sessionTree.findEntriesOnBranch({ order: "oldestFirst" });
@ -1272,7 +1272,7 @@ describe("AgentHarness R4 tools", () => {
expect(await reopened.harness.drive({ operationId: accepted.value.operationId })).toMatchObject({
ok: true,
value: { kind: "waiting", reason: "missing_identities", missing: { tools: ["echo"], models: [] } },
value: { kind: "waiting", reason: "missing_identities", missing: { tools: ["echo"] } },
});
expect((await reopened.fixture.session.getRegister("op.state", accepted.value.operationId))?.value).toEqual(
before,

365
packages/agent/test/harness/runtime2/harness.test.ts
View file

@ -1,4 +1,4 @@
import { createModels, fauxProvider } from "@earendil-works/pi-ai";
import { createModels, fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai";
import { Type } from "typebox";
import { afterEach, describe, expect, it, vi } from "vitest";
import { HarnessClosed, HarnessFault } from "../../../src/harness/agent-harness.ts";
@ -6,29 +6,18 @@ import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/com
import type { Result } from "../../../src/harness/result.ts";
import { createAgentHarness, Harness } from "../../../src/harness/runtime2/harness.ts";
import { Lane } from "../../../src/harness/runtime2/lane.ts";
import { SliceNotImplemented } from "../../../src/harness/runtime2/types.ts";
import { MemorySessionRepo, MemoryStorage } from "../../../src/harness/session/memory.ts";
import { MemorySessionRepo, type MemoryStorage } from "../../../src/harness/session/memory.ts";
import { StorageBackedSession } from "../../../src/harness/session/session.ts";
import type {
LaneConfiguration,
NavigationState,
OperationMeta,
OperationState,
RunState,
Session,
Transaction,
} from "../../../src/harness/session/types.ts";
import type { AgentHarnessTool } from "../../../src/harness/types.ts";
import { ControlledMemoryStorage, deferred } from "./test-utils.ts";
class FailingMemoryStorage extends MemoryStorage {
failure: Error | undefined;
override commit(transaction: Transaction) {
const failure = this.failure;
this.failure = undefined;
return failure === undefined ? super.commit(transaction) : Promise.reject(failure);
}
}
import { ControlledMemoryStorage, deferred, FailingMemoryStorage } from "./test-utils.ts";
const repos: MemorySessionRepo[] = [];
const standaloneSessions: Session[] = [];
@ -56,6 +45,14 @@ async function createSession(): Promise<Session> {
return repo.create({});
}
async function configureMain(session: Session, configuration: LaneConfiguration = configuredMain): Promise<void> {
await session.mutate("main", (mutator) =>
mutator.commit({
writes: [{ kind: "register", op: "set", namespace: "lane.config", key: "main", value: configuration }],
}),
);
}
async function createStorageSession(storage: MemoryStorage): Promise<Session> {
const session = new StorageBackedSession(
{ id: `runtime2-storage-${standaloneSessions.length}`, createdAt: 1, storageVersion: 1 },
@ -96,6 +93,28 @@ function modelOptions(session: Session) {
return { session, models, model: faux.getModel() };
}
function operationMeta(session: Session, intent: OperationMeta["intent"], startedAt = 1, lane = "main"): OperationMeta {
return { operationId: session.idGenerator.next(), lane, sourceLeafId: null, startedAt, intent };
}
async function installOperation(session: Session, meta: OperationMeta, state: OperationState): Promise<void> {
await session.mutate(meta.lane, (mutator) =>
mutator.commit({
writes: [
{ kind: "register", op: "set", namespace: "op.meta", key: meta.operationId, value: meta },
{ kind: "register", op: "set", namespace: "op.state", key: meta.operationId, value: state },
{
kind: "register",
op: "set",
namespace: "lane.state",
key: meta.lane,
value: { currentOperationId: meta.operationId, pendingNextRun: [] },
},
],
}),
);
}
function runState(triggerEntryId: string): RunState {
return {
kind: "run",
@ -443,43 +462,263 @@ describe("runtime2 AgentHarness", () => {
await expect(worker.getLeafId()).rejects.toBeInstanceOf(HarnessClosed);
});
it("classifies assistant-ready suspensions from captured identities", async () => {
const session = await createSession();
const options = modelOptions(session);
await configureMain(session);
const prompt = { role: "user" as const, content: "hello", timestamp: 1 };
const promptId = await session.appendMessage(prompt);
const meta = operationMeta(session, { kind: "run", promptEntryIds: [promptId] }, 2);
const { operationId } = meta;
const captured: LaneConfiguration = {
model: { provider: "missing-provider", modelId: "missing-model" },
thinkingLevel: "off",
activeToolNames: ["missing-tool"],
};
await installOperation(session, meta, {
...runState(promptId),
phase: {
kind: "assistant",
generation: {
status: "ready",
context: {
stepId: session.idGenerator.next(),
triggerEntryId: promptId,
configuration: captured,
streamOptions: {},
retryPolicy: { maxAttempts: 1, baseDelayMs: 0 },
overflowRecoveryUsed: false,
},
nextAttempt: 1,
},
},
});
const { harness, suspended } = await createAgentHarness({ ...options, tools: [tool("available")] });
const expected = {
lane: "main",
operationId,
kind: "run" as const,
startedAt: 2,
prompt: [prompt],
reason: "crash" as const,
missing: { tools: ["missing-tool"], model: "missing-provider/missing-model" },
};
expect(suspended).toEqual([expected]);
expect((await harness.inspectExecution()).current?.suspended).toBe(suspended[0]);
const getEntries = vi.spyOn(session, "getEntries");
await harness.inspectExecution();
await harness.lanes();
expect(getEntries).not.toHaveBeenCalled();
});
it("classifies tool-phase suspensions from captured tools", async () => {
const session = await createSession();
const options = modelOptions(session);
await configureMain(session);
const meta = operationMeta(session, { kind: "run", promptEntryIds: [] });
const { operationId } = meta;
await installOperation(session, meta, {
...runState(session.idGenerator.next()),
phase: {
kind: "tools",
batch: {
assistantEntryId: session.idGenerator.next(),
configuration: { ...configuredMain, activeToolNames: ["missing-tool"] },
turnId: session.idGenerator.next(),
calls: [],
},
},
});
const { harness, suspended } = await createAgentHarness(options);
expect(suspended).toEqual([
{
lane: "main",
operationId,
kind: "run",
startedAt: 1,
prompt: [],
reason: "crash",
missing: { tools: ["missing-tool"] },
},
]);
expect((await harness.inspectExecution()).current?.suspended).toBe(suspended[0]);
});
it("restores deferred suspension handles from their source entries", async () => {
const session = await createSession();
const options = modelOptions(session);
await configureMain(session);
const prompt = { role: "user" as const, content: "defer", timestamp: 1 };
const promptId = await session.appendMessage(prompt);
const handle = {
provider: options.model.provider,
modelId: options.model.id,
api: options.model.api,
id: "deferred-1",
};
const sourceId = await session.appendMessage(
fauxAssistantMessage([], { stopReason: "deferred", deferred: handle }),
);
const meta = operationMeta(session, { kind: "run", promptEntryIds: [promptId] }, 2);
const { operationId } = meta;
await installOperation(session, meta, {
...runState(promptId),
phase: {
kind: "deferred",
deferred: {
status: "suspended",
stepId: session.idGenerator.next(),
sourceEntryId: sourceId,
poll: 0,
configuration: configuredMain,
streamOptions: {},
},
},
});
const { harness, suspended } = await createAgentHarness(options);
expect(suspended).toEqual([
{
lane: "main",
operationId,
kind: "run",
startedAt: 2,
prompt: [prompt],
reason: "deferred",
deferred: handle,
},
]);
expect((await harness.inspectExecution()).current?.suspended).toBe(suspended[0]);
});
it("faults harness creation when restored suspension payloads are invalid", async () => {
const session = await createSession();
await configureMain(session);
const meta = operationMeta(session, { kind: "run", promptEntryIds: [] });
await installOperation(session, meta, {
...runState(session.idGenerator.next()),
phase: {
kind: "deferred",
deferred: {
status: "suspended",
stepId: session.idGenerator.next(),
sourceEntryId: "missing-source",
poll: 0,
configuration: configuredMain,
streamOptions: {},
},
},
});
await expect(createAgentHarness(modelOptions(session))).rejects.toMatchObject({
name: "HarnessFault",
cause: { name: "SessionInvariantError", message: "Deferred suspension source is invalid" },
});
const missingPromptSession = await createSession();
await configureMain(missingPromptSession);
await installOperation(
missingPromptSession,
operationMeta(missingPromptSession, { kind: "run", promptEntryIds: ["missing-prompt"] }),
runState(missingPromptSession.idGenerator.next()),
);
await expect(createAgentHarness(modelOptions(missingPromptSession))).rejects.toMatchObject({
name: "HarnessFault",
cause: { name: "SessionInvariantError", message: "Prompt entry missing-prompt is missing" },
});
});
it("reports restored cancellation as aborting without a suspension descriptor", async () => {
const session = await createSession();
await configureMain(session);
const meta = operationMeta(session, { kind: "run", promptEntryIds: [] });
const { operationId } = meta;
await installOperation(session, meta, {
...runState(session.idGenerator.next()),
control: { status: "cancel_requested", requestedAt: 2, drainedSteer: [], drainedFollowUp: [] },
});
const { harness, suspended } = await createAgentHarness(modelOptions(session));
expect(suspended).toHaveLength(1);
expect(await harness.inspectExecution()).toEqual({
lane: "main",
leafId: null,
current: { id: operationId, kind: "run", status: "aborting", startedAt: 1 },
});
expect(await harness.lanes()).toEqual([
{ name: "main", leafId: null, operation: { id: operationId, kind: "run", status: "aborting" } },
]);
});
it("inspects idle last results from owned state", async () => {
const session = await createSession();
const operationId = session.idGenerator.next();
const lastResult = {
operationId,
kind: "navigation" as const,
outcome: "completed" as const,
oldLeafId: null,
leafId: null,
};
await session.mutate("main", (mutator) =>
mutator.commit({
writes: [
{ kind: "register", op: "set", namespace: "lane.config", key: "main", value: configuredMain },
{ kind: "register", op: "set", namespace: "lane.lastResult", key: "main", value: lastResult },
],
}),
);
const { harness } = await createAgentHarness(modelOptions(session));
expect(await harness.inspectExecution()).toEqual({ lane: "main", leafId: null, current: null, lastResult });
});
it("persists custom facts through idle and active lane facades", async () => {
const session = await createSession();
await configureMain(session);
const meta = operationMeta(session, { kind: "run", promptEntryIds: [] });
const { operationId } = meta;
await installOperation(session, meta, runState(session.idGenerator.next()));
const { harness } = await createAgentHarness(modelOptions(session));
const worker = unwrap(await harness.createLane("worker", null));
await worker.sessionTree.setName("idle-worker");
await harness.sessionTree.setName("active-main");
await harness.sessionTree.setLabel("target", "label");
await harness.sessionTree.setCustomFact("key", { durable: true });
expect(await session.getName()).toBe("active-main");
expect(await session.getLabel("target")).toBe("label");
expect(await session.getCustomFact("key")).toEqual({ durable: true });
expect((await harness.inspectExecution()).current?.id).toBe(operationId);
});
it("reports restored open operations without activating them", async () => {
const repo = new MemorySessionRepo();
repos.push(repo);
const session = await repo.create({});
await session.mutate("main", (mutator) =>
mutator.commit({
writes: [{ kind: "register", op: "set", namespace: "lane.config", key: "main", value: configuredMain }],
}),
);
const operationId = session.idGenerator.next();
const meta: OperationMeta = {
operationId,
lane: "main",
sourceLeafId: null,
startedAt: 1,
intent: { kind: "run", promptEntryIds: [] },
};
await configureMain(session);
const meta = operationMeta(session, { kind: "run", promptEntryIds: [] });
const { operationId } = meta;
const state = runState(session.idGenerator.next());
await session.mutate("main", (mutator) =>
mutator.commit({
writes: [
{ kind: "register", op: "set", namespace: "op.meta", key: operationId, value: meta },
{ kind: "register", op: "set", namespace: "op.state", key: operationId, value: state },
{
kind: "register",
op: "set",
namespace: "lane.state",
key: "main",
value: { currentOperationId: operationId, pendingNextRun: [] },
},
],
}),
);
await installOperation(session, meta, state);
const { harness, suspended } = await createAgentHarness(modelOptions(session));
expect(suspended).toEqual([{ lane: "main", operationId, kind: "run", startedAt: 1, reason: "crash" }]);
expect(suspended).toEqual([
{ lane: "main", operationId, kind: "run", startedAt: 1, prompt: [], reason: "crash" },
]);
expect(await harness.inspectExecution()).toEqual({
lane: "main",
leafId: null,
current: { id: operationId, kind: "run", status: "suspended", startedAt: 1, suspended: suspended[0] },
});
expect(await harness.lanes()).toEqual([
{ name: "main", leafId: null, operation: { id: operationId, kind: "run", status: "suspended" } },
]);
@ -494,12 +733,12 @@ describe("runtime2 AgentHarness", () => {
payload: { text: "queued" },
});
expect((await session.getRegister("lane.leaf", "main"))?.value).toBeNull();
await expect(harness.inspectExecution()).rejects.toBeInstanceOf(SliceNotImplemented);
const closing = harness.close();
expect(harness.close()).toBe(closing);
await closing;
await expect(harness.lanes()).rejects.toBeInstanceOf(HarnessClosed);
await expect(harness.inspectExecution()).rejects.toBeInstanceOf(HarnessClosed);
const reopened = await repo.open(session.metadata);
expect((await reopened.getRegister("lane.state", "main"))?.value.currentOperationId).toBe(operationId);
expect((await reopened.getRegister("op.meta", operationId))?.value).toEqual(meta);
@ -541,19 +780,9 @@ describe("runtime2 AgentHarness", () => {
it("rejects append during a structural operation without faulting", async () => {
const session = await createSession();
await session.mutate("main", (mutator) =>
mutator.commit({
writes: [{ kind: "register", op: "set", namespace: "lane.config", key: "main", value: configuredMain }],
}),
);
const operationId = session.idGenerator.next();
const meta: OperationMeta = {
operationId,
lane: "main",
sourceLeafId: null,
startedAt: 1,
intent: { kind: "navigation", targetId: null, summarize: false },
};
await configureMain(session);
const meta = operationMeta(session, { kind: "navigation", targetId: null, summarize: false });
const { operationId } = meta;
const state: NavigationState = {
kind: "navigation",
control: { status: "running" },
@ -561,23 +790,10 @@ describe("runtime2 AgentHarness", () => {
summarize: false,
phase: { kind: "ready_to_commit" },
};
await session.mutate("main", (mutator) =>
mutator.commit({
writes: [
{ kind: "register", op: "set", namespace: "op.meta", key: operationId, value: meta },
{ kind: "register", op: "set", namespace: "op.state", key: operationId, value: state },
{
kind: "register",
op: "set",
namespace: "lane.state",
key: "main",
value: { currentOperationId: operationId, pendingNextRun: [] },
},
],
}),
);
const { harness } = await createAgentHarness(modelOptions(session));
await installOperation(session, meta, state);
const { harness, suspended } = await createAgentHarness(modelOptions(session));
expect(suspended).toEqual([{ lane: "main", operationId, kind: "navigation", startedAt: 1, reason: "crash" }]);
await expect(harness.sessionTree.appendMessage({ role: "user", content: "later", timestamp: 2 })).rejects.toThrow(
`Cannot append while structural operation ${operationId} is active`,
);
@ -616,6 +832,7 @@ describe("runtime2 AgentHarness", () => {
await expect(queued).rejects.toBe(rejected);
await expect(harness.getLeafId()).rejects.toBe(rejected);
await expect(worker.getThinkingLevel()).rejects.toBe(rejected);
await expect(worker.inspectExecution()).rejects.toBe(rejected);
await expect(harness.getTools()).rejects.toBe(rejected);
await expect(harness.setResources({})).rejects.toBe(rejected);
expect(harness.fault(new Error("later"))).toBe(rejected);

10
packages/agent/test/harness/runtime2/test-utils.ts
View file

@ -1,6 +1,16 @@
import { MemoryStorage } from "../../../src/harness/session/memory.ts";
import type { CommitResult, Transaction } from "../../../src/harness/session/types.ts";
export class FailingMemoryStorage extends MemoryStorage {
failure: Error | undefined;
override commit(transaction: Transaction) {
const failure = this.failure;
this.failure = undefined;
return failure === undefined ? super.commit(transaction) : Promise.reject(failure);
}
}
export class ControlledMemoryStorage extends MemoryStorage {
beforeNextCommit: (() => Promise<void>) | undefined;