diff --git a/packages/agent/docs/harness.md b/packages/agent/docs/harness.md index 2b018a32b..533a05e13 100644 --- a/packages/agent/docs/harness.md +++ b/packages/agent/docs/harness.md @@ -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; @@ -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) diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index dcfef8700..18d289ccf 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -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; @@ -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 } & ( diff --git a/packages/agent/src/harness/config.ts b/packages/agent/src/harness/config.ts index 4c8b6aeec..fe0e19ec1 100644 --- a/packages/agent/src/harness/config.ts +++ b/packages/agent/src/harness/config.ts @@ -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) || diff --git a/packages/agent/src/harness/runtime/agent-harness-runtime.ts b/packages/agent/src/harness/runtime/agent-harness-runtime.ts index e467f8d1b..9b25c5152 100644 --- a/packages/agent/src/harness/runtime/agent-harness-runtime.ts +++ b/packages/agent/src/harness/runtime/agent-harness-runtime.ts @@ -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 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" }; } diff --git a/packages/agent/src/harness/runtime/assistant-procedure.ts b/packages/agent/src/harness/runtime/assistant-procedure.ts index b92e6bc3f..7bb35fb43 100644 --- a/packages/agent/src/harness/runtime/assistant-procedure.ts +++ b/packages/agent/src/harness/runtime/assistant-procedure.ts @@ -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( }), ); } - 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, diff --git a/packages/agent/src/harness/runtime/tool-batch-procedure.ts b/packages/agent/src/harness/runtime/tool-batch-procedure.ts index ba86cf4d0..80b1102f6 100644 --- a/packages/agent/src/harness/runtime/tool-batch-procedure.ts +++ b/packages/agent/src/harness/runtime/tool-batch-procedure.ts @@ -1000,7 +1000,7 @@ async function suspendToolBatchForMissingIdentities { - const missing = { tools, models: [] }; + const missing = { tools }; runtime.restoredSuspensions.set(lane.name, { ...suspensionBase(restored), reason: "missing_identities", diff --git a/packages/agent/src/harness/runtime/transitions.ts b/packages/agent/src/harness/runtime/transitions.ts index a21363077..8c78b96fd 100644 --- a/packages/agent/src/harness/runtime/transitions.ts +++ b/packages/agent/src/harness/runtime/transitions.ts @@ -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( - models: Models, - configuration: LaneConfiguration, - settings: RuntimeSettings, -): { 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( - configuration: LaneConfiguration, - settings: RuntimeSettings, -): 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 } }; } diff --git a/packages/agent/src/harness/runtime/types.ts b/packages/agent/src/harness/runtime/types.ts index 4d98dd814..45ba760f1 100644 --- a/packages/agent/src/harness/runtime/types.ts +++ b/packages/agent/src/harness/runtime/types.ts @@ -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; diff --git a/packages/agent/src/harness/runtime2/harness.ts b/packages/agent/src/harness/runtime2/harness.ts index 6b2260a83..714035890 100644 --- a/packages/agent/src/harness/runtime2/harness.ts +++ b/packages/agent/src/harness/runtime2/harness.ts @@ -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 extends Lane implement closePromise: Promise | undefined; faultError: HarnessFault | undefined; - constructor(options: AgentHarnessOptions, seed: LaneConfiguration, restored: Map) { + constructor( + options: AgentHarnessOptions, + seed: LaneConfiguration, + restored: Map, + suspensionsByLane: Map, + ) { const main = restored.get("main"); if (main === undefined) throw new SessionInvariantError("Session is missing main lane"); super( @@ -52,6 +65,7 @@ export class Harness 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 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 extends Lane implement async lanes(): Promise { 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 { @@ -208,7 +220,7 @@ export class Harness 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 extends Lane implement state, (cause) => this.fault(cause), (event) => this.events.emit(event), + suspension, ); } @@ -293,25 +306,74 @@ export async function createAgentHarness [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 { + 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 { await session.mutate("main", async (mutator) => { const [leaf, state, configuration, lastResult] = await Promise.all([ diff --git a/packages/agent/src/harness/runtime2/lane.ts b/packages/agent/src/harness/runtime2/lane.ts index cc16e6a10..5ffe79808 100644 --- a/packages/agent/src/harness/runtime2/lane.ts +++ b/packages/agent/src/harness/runtime2/lane.ts @@ -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 { @@ -78,7 +88,7 @@ export class Lane implements AgentLane { return this.state.leafId; } - async getLastResult() { + async getLastResult(): Promise { 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 { - throw new SliceNotImplemented("inspectExecution"); + async inspectExecution(): Promise { + 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 { diff --git a/packages/agent/test/harness/agent-harness-r2.test.ts b/packages/agent/test/harness/agent-harness-r2.test.ts index 284f67c70..01b56fb2f 100644 --- a/packages/agent/test/harness/agent-harness-r2.test.ts +++ b/packages/agent/test/harness/agent-harness-r2.test.ts @@ -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(); }); diff --git a/packages/agent/test/harness/agent-harness-r3.test.ts b/packages/agent/test/harness/agent-harness-r3.test.ts index 80113b417..a915bf19e 100644 --- a/packages/agent/test/harness/agent-harness-r3.test.ts +++ b/packages/agent/test/harness/agent-harness-r3.test.ts @@ -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); diff --git a/packages/agent/test/harness/agent-harness-r4.test.ts b/packages/agent/test/harness/agent-harness-r4.test.ts index 02b0e60df..3466baf53 100644 --- a/packages/agent/test/harness/agent-harness-r4.test.ts +++ b/packages/agent/test/harness/agent-harness-r4.test.ts @@ -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, diff --git a/packages/agent/test/harness/runtime2/harness.test.ts b/packages/agent/test/harness/runtime2/harness.test.ts index fe0033c59..14c829306 100644 --- a/packages/agent/test/harness/runtime2/harness.test.ts +++ b/packages/agent/test/harness/runtime2/harness.test.ts @@ -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 { return repo.create({}); } +async function configureMain(session: Session, configuration: LaneConfiguration = configuredMain): Promise { + 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 { 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 { + 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); diff --git a/packages/agent/test/harness/runtime2/test-utils.ts b/packages/agent/test/harness/runtime2/test-utils.ts index 7e49a9ffd..a0b6f9d86 100644 --- a/packages/agent/test/harness/runtime2/test-utils.ts +++ b/packages/agent/test/harness/runtime2/test-utils.ts @@ -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) | undefined;