From 1d882a7eca12d07fec9468db0192d13d922bdb31 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Fri, 14 Aug 2026 16:18:33 +0200 Subject: [PATCH] feat(agent): add AgentHarness R2 minimal run --- packages/agent/CHANGELOG.md | 1 + .../src/harness/agent-harness-runtime.ts | 1427 ++++++++++++++++- packages/agent/src/harness/hooks.ts | 6 +- packages/agent/src/harness/restore.ts | 170 +- .../test/harness/agent-harness-r2.test.ts | 549 +++++++ .../harness/agent-harness-runtime.test.ts | 5 +- packages/agent/test/harness/restore.test.ts | 110 ++ packages/agent/test/harness/scratch.ts | 48 + packages/agent/test/utils/wait-for-tick.ts | 3 + 9 files changed, 2265 insertions(+), 54 deletions(-) create mode 100644 packages/agent/test/harness/agent-harness-r2.test.ts create mode 100644 packages/agent/test/harness/scratch.ts create mode 100644 packages/agent/test/utils/wait-for-tick.ts diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index d95b7d564..59e9a83a3 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -11,6 +11,7 @@ - Added scheduler-facing harness execution primitives (`accept`, fenced `drive`, `requestAbort`, and `inspectExecution`) plus replay-stable tool invocation identities. - Added isolated harness execution primitives for assistant streaming and durable-boundary tool phases. +- Added durable `AgentHarness` prompt acceptance and minimal no-tool execution with recovery, lifecycle events, usage settlement, and terminal cleanup. ## [0.84.2] - 2026-08-14 diff --git a/packages/agent/src/harness/agent-harness-runtime.ts b/packages/agent/src/harness/agent-harness-runtime.ts index 0231f1259..ae69f69f0 100644 --- a/packages/agent/src/harness/agent-harness-runtime.ts +++ b/packages/agent/src/harness/agent-harness-runtime.ts @@ -1,5 +1,18 @@ -import type { Api, AssistantMessage, ImageContent, Model, RetryPolicy, Usage } from "@earendil-works/pi-ai"; -import type { AgentMessage, QueueMode, ThinkingLevel } from "../types.ts"; +import type { + Api, + AssistantMessage, + AssistantMessageEventStream, + Context, + ImageContent, + Message, + Model, + RetryPolicy, + SimpleStreamOptions, + Usage, +} from "@earendil-works/pi-ai"; +import { createAssistantMessageEventStream } from "@earendil-works/pi-ai"; +import { NOOP_TELEMETRY_CONTEXT, type TelemetryContext } from "@earendil-works/pi-telemetry"; +import type { AgentMessage, AgentTool, QueueMode, ThinkingLevel } from "../types.ts"; import { type AbortRequestResult, type AbortResult, @@ -18,10 +31,13 @@ import { type HarnessEvent, HarnessFault, InvalidLane, + InvalidMessage, + LaneBusy, type LaneExecutionInfo, LaneExists, type LaneInfo, type LaneSnapshot, + MissingIdentities, type NavigateOptions, type NavigationResult, type NextRunResult, @@ -36,32 +52,48 @@ import { type SessionSnapshot, type SuspendedOperation, type TerminalOperationOutcome, + UnknownSkill, UnknownTarget, + UnknownTemplate, type WatchHandle, } from "./agent-harness.ts"; import { type CompactionSettings, DEFAULT_COMPACTION_SETTINGS } from "./compaction/compaction.ts"; import { HarnessEventBus } from "./events.ts"; +import { streamHarnessAssistant } from "./execution/assistant.ts"; import { BreakpointBarrier } from "./execution/breakpoint.ts"; import { OperationEffectGate } from "./execution/effect-gate.ts"; -import { HookRegistry } from "./hooks.ts"; +import { applyStreamOptionsPatch, HookRegistry } from "./hooks.ts"; +import { convertToLlm } from "./messages.ts"; +import { formatPromptTemplateInvocation } from "./prompt-templates.ts"; import { type RestoredLane, restoreLane } from "./restore.ts"; import { Result } from "./result.ts"; +import { buildSessionContext } from "./session/context.ts"; import { LaneMutationLine } from "./session/lane-mutations.ts"; import { SessionInvalidLaneError, SessionInvariantError, SessionLaneExistsError, + SessionPendingAssistantMessageError, SessionUnknownTargetError, } from "./session/session.ts"; import type { + Entry, + EntryProjector, JsonValue, LaneConfiguration, LaneLastResult, + NewEntry, + Operation, + OperationError, + PendingEntry, + RunState, Session, SessionMutator, SessionTree, } from "./session/types.ts"; -import type { AgentHarnessStreamOptions, AgentHarnessTool } from "./types.ts"; +import { formatSkillInvocation } from "./skills.ts"; +import { startHarnessSpan } from "./telemetry.ts"; +import type { AgentHarnessStreamOptions, AgentHarnessTool, AgentHarnessToolContextSource } from "./types.ts"; const DEFAULT_RETRY_POLICY: RetryPolicy = { enabled: true, maxRetries: 3, baseDelayMs: 1_000 }; @@ -80,6 +112,7 @@ interface RuntimeSettings { compaction: CompactionSettings; steeringMode: QueueMode; followUpMode: QueueMode; + toolExecution: "sequential" | "parallel"; } interface DeferredValue { @@ -90,6 +123,7 @@ interface DeferredValue { interface ActiveOperation { operationId: string; + operationKind: "run" | "compaction" | "navigation"; completion: Promise; resolve: (result: DriveResult) => void; reject: (error: unknown) => void; @@ -97,11 +131,31 @@ interface ActiveOperation { task?: Promise; } +interface AdmissionReservation { + operationId: string; + operationKind: "run" | "compaction" | "navigation"; + completion: Promise; + resolve(): void; +} + type DriveArbitration = | { kind: "result"; result: DriveResult } | { kind: "join"; completion: Promise } | { kind: "installed"; active: ActiveOperation }; +interface NormalizedRunRequest { + operationId: string; + startedAt: number; + messages: AgentMessage[]; + resources: Resources; +} + +interface AcceptancePublication { + admission: { operationId: string; kind: "run"; startedAt: number }; + entries: Entry[]; + capturedNextRun: boolean; +} + function deferredValue(): DeferredValue { let resolvePromise: ((value: T) => void) | undefined; let rejectPromise: ((error: unknown) => void) | undefined; @@ -141,9 +195,16 @@ class AgentHarnessRuntime implements AgentH readonly models: AgentHarnessOptions["models"]; readonly driveMode: "automatic" | "manual"; readonly seed: LaneConfiguration; + readonly toolContext: AgentHarnessToolContextSource | undefined; + readonly systemPromptSource: AgentHarnessOptions["systemPrompt"]; + readonly toProviderMessages: (messages: AgentMessage[]) => Message[] | Promise; + readonly entryProjectors: Readonly>; + readonly telemetryContext: TelemetryContext; readonly settingsLine = new LaneMutationLine(); readonly laneRuntimes = new Map>(); readonly activeOperations = new Map(); + readonly admissionReservations = new Map(); + readonly attachedOperationIds = new Set(); readonly restoredSuspensions = new Map(); settings: RuntimeSettings; settingsRevision = 0; @@ -157,6 +218,11 @@ class AgentHarnessRuntime implements AgentH this.sessionStorage = options.session; this.models = options.models; this.driveMode = options.drive ?? "automatic"; + this.toolContext = options.toolContext; + this.systemPromptSource = options.systemPrompt; + this.toProviderMessages = options.toProviderMessages ?? convertToLlm; + this.entryProjectors = options.entryProjectors ?? {}; + this.telemetryContext = options.telemetryContext ?? NOOP_TELEMETRY_CONTEXT; const tools = [...(options.tools ?? [])]; this.seed = { model: { provider: options.model.provider, modelId: options.model.id }, @@ -171,6 +237,7 @@ class AgentHarnessRuntime implements AgentH compaction: options.compaction ?? DEFAULT_COMPACTION_SETTINGS, steeringMode: options.steeringMode ?? "all", followUpMode: options.followUpMode ?? "all", + toolExecution: options.toolExecution ?? "parallel", }; this.events = new HarnessEventBus(); this.hooks = new HookRegistry((error, hook, lane) => @@ -183,7 +250,7 @@ class AgentHarnessRuntime implements AgentH lane, }), ); - this.session = options.session.view("main"); + this.session = this.createPublicSessionView("main"); } async initialize(): Promise { @@ -350,6 +417,293 @@ class AgentHarnessRuntime implements AgentH throw new RuntimeSliceNotImplemented("watchSession"); } + createPublicSessionView(lane: string): SessionTree { + const delegate = this.sessionStorage.view(lane); + return { + getLeafId: () => delegate.getLeafId(), + getEntry: (id) => delegate.getEntry(id), + getStats: () => delegate.getStats(), + getName: () => delegate.getName(), + setName: (name) => delegate.setName(name), + getLabel: (targetId) => delegate.getLabel(targetId), + setLabel: (targetId, label) => delegate.setLabel(targetId, label), + getCustomFact: (key) => delegate.getCustomFact(key), + setCustomFact: (key, value) => delegate.setCustomFact(key, value), + findEntries: (query) => delegate.findEntries(query), + findEntry: (query) => delegate.findEntry(query), + findEntriesOnBranch: (query) => delegate.findEntriesOnBranch(query), + findEntryOnBranch: (query) => delegate.findEntryOnBranch(query), + appendMessage: (message) => this.appendPublicEntry(lane, { type: "message", payload: message }), + appendCustomEntry: (customType, data) => + this.appendPublicEntry(lane, { + type: "custom", + customType, + ...(data === undefined ? {} : { payload: data }), + }), + }; + } + + async acceptLane(lane: AgentLaneRuntime, request: OperationRequest): Promise { + const closed = this.resultClosedError(); + if (closed !== undefined) return Result.err(closed); + if (request.kind === "compaction" || request.kind === "navigation") { + throw new RuntimeSliceNotImplemented(`accept(${request.kind})`); + } + + const resources = await this.readSettings((settings) => settings.resources); + const normalized = this.normalizeRunRequest(request, resources); + if (!normalized.ok) return normalized; + const provisional = normalized.value; + if (provisional.messages.some(isPendingAssistant)) { + return Result.err( + new InvalidMessage({ + lane: lane.name, + reason: "pending_assistant", + message: "A pending assistant message cannot be accepted", + }), + ); + } + const reserved = deferredValue(); + const reservation: AdmissionReservation = { + operationId: provisional.operationId, + operationKind: "run", + completion: reserved.promise, + resolve: () => reserved.resolve(undefined), + }; + + try { + const busy = await this.sessionStorage.mutate(lane.name, async (reader) => { + const restored = await restoreLane(reader, lane.name); + const existingReservation = this.admissionReservations.get(lane.name); + if (existingReservation !== undefined) return this.busy(lane.name, existingReservation); + const active = this.activeOperations.get(lane.name); + if (active !== undefined) return this.busy(lane.name, active); + if (restored.current !== undefined) { + return new LaneBusy({ + lane: lane.name, + operationId: restored.current.operation.operationId, + operationKind: restored.current.operation.intent.kind, + message: `Lane ${JSON.stringify(lane.name)} already has an active operation`, + }); + } + this.admissionReservations.set(lane.name, reservation); + return undefined; + }); + if (busy !== undefined) return Result.err(busy); + + let systemPrompt = ""; + let hookMessages: AgentMessage[] = []; + let systemPromptOverride: string | undefined; + let resumeData: Record | undefined; + if (this.hooks.has("before_run")) { + systemPrompt = (await this.resolveSystemPrompt()) ?? ""; + await lane.breakpoint.hit({ + kind: "hook.before_run", + description: "Run pre-acceptance hooks", + details: { operationId: provisional.operationId }, + }); + this.assertOpen(); + const aggregate = await this.hooks.runBeforeAcceptanceWithResumeData( + { + lane: lane.name, + runId: provisional.operationId, + prompt: provisional.messages, + systemPrompt, + resources: provisional.resources, + }, + () => this.assertOpen(), + ); + hookMessages = aggregate.result?.messages ?? []; + systemPromptOverride = aggregate.result?.systemPrompt; + if (Object.keys(aggregate.resumeData).length !== 0) resumeData = aggregate.resumeData; + } + const settings = await this.snapshotSettings(); + const publication = await this.sessionStorage.mutate(lane.name, async (mutator) => { + if (this.admissionReservations.get(lane.name) !== reservation) { + throw new SessionInvariantError(`Lane ${JSON.stringify(lane.name)} lost its admission reservation`); + } + const restored = await restoreLane(mutator, lane.name); + if (restored.current !== undefined || this.activeOperations.has(lane.name)) { + const owner = restored.current?.operation; + return Result.err( + owner === undefined + ? this.busy(lane.name, this.activeOperations.get(lane.name)!) + : new LaneBusy({ + lane: lane.name, + operationId: owner.operationId, + operationKind: owner.intent.kind, + message: `Lane ${JSON.stringify(lane.name)} already has an active operation`, + }), + ); + } + const missing = await this.missingIdentities(restored.configuration, settings); + if (missing.tools.length !== 0 || missing.models.length !== 0) { + return Result.err( + new MissingIdentities({ + lane: lane.name, + ...missing, + message: `Lane ${JSON.stringify(lane.name)} has unresolved model or tool identities`, + }), + ); + } + + const pendingIds = [...restored.laneState.pendingNextRun]; + const pendingRegisters = await Promise.all( + pendingIds.map((id) => mutator.getRegister("pending.entry", id)), + ); + for (let index = 0; index < pendingIds.length; index++) { + if (pendingRegisters[index] === undefined) { + throw new SessionInvariantError(`Pending next-run entry ${pendingIds[index]} is missing`); + } + } + const callerIds = provisional.messages.map(() => this.sessionStorage.idGenerator.next()); + const hookIds = hookMessages.map(() => this.sessionStorage.idGenerator.next()); + const placements: Array<{ id: string; pending: NonNullable<(typeof pendingRegisters)[number]>["value"] }> = + []; + for (let index = 0; index < pendingIds.length; index++) { + placements.push({ id: pendingIds[index]!, pending: pendingRegisters[index]!.value }); + } + for (let index = 0; index < provisional.messages.length; index++) { + placements.push({ + id: callerIds[index]!, + pending: { type: "message", payload: provisional.messages[index]! }, + }); + } + for (let index = 0; index < hookMessages.length; index++) { + placements.push({ id: hookIds[index]!, pending: { type: "message", payload: hookMessages[index]! } }); + } + if ( + placements.some( + (placement) => placement.pending.type === "message" && isPendingAssistant(placement.pending.payload), + ) + ) { + return Result.err( + new InvalidMessage({ + lane: lane.name, + reason: "pending_assistant", + message: "A pending assistant message cannot be accepted", + }), + ); + } + if (placements.length === 0) { + return Result.err( + new InvalidMessage({ + lane: lane.name, + reason: "empty", + message: "A run must place at least one message or pending entry", + }), + ); + } + let parentId = restored.leafId; + const entryWrites: Array<{ kind: "entry"; entry: NewEntry }> = []; + for (const placement of placements) { + const pending = placement.pending; + const entry: NewEntry = + pending.type === "message" + ? { id: placement.id, parentId, type: "message", message: pending.payload } + : { + id: placement.id, + parentId, + type: "custom", + customType: pending.customType, + ...(pending.payload === undefined ? {} : { data: pending.payload }), + }; + entryWrites.push({ kind: "entry", entry }); + parentId = placement.id; + } + const triggerEntryId = parentId!; + const operation: Operation = { + operationId: provisional.operationId, + lane: lane.name, + sourceLeafId: restored.leafId, + startedAt: provisional.startedAt, + intent: { + kind: "run", + promptEntryIds: callerIds, + ...(systemPromptOverride === undefined ? {} : { systemPromptOverride }), + ...(resumeData === undefined ? {} : { resumeData }), + }, + }; + const state: RunState = { + kind: "run", + control: { status: "running" }, + settings: { + compaction: { ...settings.compaction }, + steeringMode: settings.steeringMode, + followUpMode: settings.followUpMode, + toolExecution: settings.toolExecution, + }, + phase: { + kind: "checkpoint", + continuation: { kind: "need_assistant", overflowRecoveryUsed: false }, + triggerEntryId, + skipInboxOnce: true, + }, + inbox: { steer: [], followUp: [], writes: [] }, + latestAssistantEntryId: null, + }; + await mutator.commit({ + writes: [ + ...entryWrites, + ...pendingIds.map( + (id) => ({ kind: "register", op: "delete", namespace: "pending.entry", key: id }) as const, + ), + { kind: "register", op: "set", namespace: "lane.leaf", key: lane.name, value: triggerEntryId }, + { kind: "register", op: "set", namespace: "op.meta", key: provisional.operationId, value: operation }, + { kind: "register", op: "set", namespace: "op.state", key: provisional.operationId, value: state }, + { + kind: "register", + op: "set", + namespace: "lane.state", + key: lane.name, + value: { ...restored.laneState, currentOperationId: provisional.operationId, pendingNextRun: [] }, + }, + ], + }); + const entries = await mutator.getEntries(placements.map((placement) => placement.id)); + return Result.ok({ + admission: { operationId: provisional.operationId, kind: "run", startedAt: provisional.startedAt }, + entries: placements.map((placement) => entries.get(placement.id)!), + capturedNextRun: pendingIds.length !== 0, + }); + }); + if (!publication.ok) return publication; + this.attachedOperationIds.add(provisional.operationId); + await this.events.emit({ type: "run_start", runId: provisional.operationId, lane: lane.name }); + for (const entry of publication.value.entries) { + if (entry.type === "message") { + await this.events.emit({ + type: "message_start", + runId: provisional.operationId, + message: entry.message, + lane: lane.name, + }); + await this.events.emit({ + type: "message_end", + runId: provisional.operationId, + message: entry.message, + entryId: entry.id, + lane: lane.name, + }); + } + await this.events.emit({ type: "entry_added", entry, lane: lane.name }); + } + if (publication.value.capturedNextRun) { + await this.events.emit({ type: "queue_update", steer: [], followUp: [], nextRun: [], lane: lane.name }); + } + return Result.ok(publication.value.admission); + } catch (error) { + if (error instanceof HarnessClosed) return Result.err(this.closedError()); + if (error instanceof HarnessFault) throw error; + throw this.fault(error); + } finally { + if (this.admissionReservations.get(lane.name) === reservation) { + this.admissionReservations.delete(lane.name); + } + reservation.resolve(); + } + } + close(): Promise { if (this.closePromise !== undefined) return this.closePromise; this.state = "closing"; @@ -468,6 +822,12 @@ class AgentHarnessRuntime implements AgentH async driveLane(lane: AgentLaneRuntime, options: DriveOptions): Promise { const closed = this.resultClosedError(); if (closed !== undefined) return Result.err(closed); + const reservation = this.admissionReservations.get(lane.name); + if (reservation?.operationId === options.operationId) { + await reservation.completion; + const afterReservation = this.resultClosedError(); + if (afterReservation !== undefined) return Result.err(afterReservation); + } let arbitration: DriveArbitration; try { arbitration = await this.sessionStorage.mutate(lane.name, async (reader): Promise => { @@ -496,9 +856,12 @@ class AgentHarnessRuntime implements AgentH } return { kind: "join", completion: existing.completion }; } + if (restored.current === undefined) + throw new SessionInvariantError("Current operation metadata is missing"); const deferred = deferredValue(); const active: ActiveOperation = { operationId: options.operationId, + operationKind: restored.current.operation.intent.kind, completion: deferred.promise, resolve: deferred.resolve, reject: deferred.reject, @@ -612,6 +975,162 @@ class AgentHarnessRuntime implements AgentH return fault; } + private async appendPublicEntry(lane: string, pending: PendingEntry): Promise { + this.assertOpen(); + if ( + pending.type === "message" && + pending.payload.role === "assistant" && + pending.payload.stopReason === "pending" + ) { + throw new SessionPendingAssistantMessageError(); + } + const id = this.sessionStorage.idGenerator.next(); + while (true) { + const disposition = await this.sessionStorage.mutate(lane, async (mutator) => { + const reservation = this.admissionReservations.get(lane); + if (reservation !== undefined) return { kind: "wait" as const, completion: reservation.completion }; + const [leaf, laneState] = await Promise.all([ + mutator.getRegister("lane.leaf", lane), + mutator.getRegister("lane.state", lane), + ]); + if (leaf === undefined || laneState === undefined) throw new SessionInvariantError(`Unknown lane: ${lane}`); + const operationId = laneState.value.currentOperationId; + if (operationId === null) { + await mutator.commit({ + writes: [ + { + kind: "entry", + entry: + pending.type === "message" + ? { id, parentId: leaf.value, type: "message", message: pending.payload } + : { + id, + parentId: leaf.value, + type: "custom", + customType: pending.customType, + ...(pending.payload === undefined ? {} : { data: pending.payload }), + }, + }, + { kind: "register", op: "set", namespace: "lane.leaf", key: lane, value: id }, + ], + }); + return { kind: "done" as const }; + } + const [operation, state] = await Promise.all([ + mutator.getRegister("op.meta", operationId), + mutator.getRegister("op.state", operationId), + ]); + if (operation === undefined || state === undefined || operation.value.lane !== lane) { + throw new SessionInvariantError(`Lane ${JSON.stringify(lane)} has incomplete operation state`); + } + if (state.value.kind !== "run" || operation.value.intent.kind !== "run") { + throw new RuntimeSliceNotImplemented("tree write during structural operation"); + } + await mutator.commit({ + writes: [ + { kind: "register", op: "set", namespace: "pending.entry", key: id, value: pending }, + { + kind: "register", + op: "set", + namespace: "op.state", + key: operationId, + value: { + ...state.value, + inbox: { ...state.value.inbox, writes: [...state.value.inbox.writes, id] }, + }, + }, + ], + }); + return { kind: "done" as const }; + }); + if (disposition.kind === "done") return id; + await disposition.completion; + this.assertOpen(); + } + } + + private normalizeRunRequest( + request: Extract, + resources: Resources, + ) { + const operationId = request.operationId ?? this.sessionStorage.idGenerator.next(); + const startedAt = Date.now(); + let messages: AgentMessage[]; + if (request.kind === "prompt") { + if (typeof request.prompt === "string") { + const content = [ + ...(request.prompt.length === 0 ? [] : [{ type: "text" as const, text: request.prompt }]), + ...(request.images ?? []), + ]; + messages = content.length === 0 ? [] : [{ role: "user", content, timestamp: startedAt }]; + } else { + messages = Array.isArray(request.prompt) ? [...request.prompt] : [request.prompt]; + } + } else if (request.kind === "skill") { + const skill = resources.skills?.find((candidate) => candidate.name === request.name); + if (skill === undefined) { + return Result.err(new UnknownSkill({ name: request.name, message: `Unknown skill: ${request.name}` })); + } + messages = [ + { + role: "user", + content: formatSkillInvocation(skill, request.additionalInstructions), + timestamp: startedAt, + }, + ]; + } else { + const template = resources.promptTemplates?.find((candidate) => candidate.name === request.name); + if (template === undefined) { + return Result.err( + new UnknownTemplate({ name: request.name, message: `Unknown prompt template: ${request.name}` }), + ); + } + messages = [ + { role: "user", content: formatPromptTemplateInvocation(template, request.args), timestamp: startedAt }, + ]; + } + return Result.ok({ operationId, startedAt, messages, resources }); + } + + private async snapshotSettings(): Promise> { + return this.readSettings((settings) => ({ + ...settings, + tools: [...settings.tools], + streamOptions: { ...settings.streamOptions }, + retryPolicy: { ...settings.retryPolicy }, + compaction: { ...settings.compaction }, + })); + } + + private async missingIdentities( + configuration: LaneConfiguration, + settings: RuntimeSettings, + ): Promise<{ tools: string[]; models: string[] }> { + const model = this.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}`] : [], + }; + } + + private busy(lane: string, owner: Pick): LaneBusy { + return new LaneBusy({ + lane, + operationId: owner.operationId, + operationKind: owner.operationKind, + message: `Lane ${JSON.stringify(lane)} already has an active operation`, + }); + } + + private async resolveSystemPrompt(): Promise { + const source = this.systemPromptSource; + if (source === undefined || typeof source === "string") return source; + const contextSource = this.toolContext; + const context = typeof contextSource === "function" ? await contextSource() : contextSource; + return source(context as TContext); + } + private mainLane(): AgentLaneRuntime { const lane = this.laneRuntimes.get("main"); if (lane === undefined) throw new SessionInvariantError("AgentHarness main lane is not initialized"); @@ -655,7 +1174,13 @@ class AgentHarnessRuntime implements AgentH active.resolve(result); } catch (error) { await this.removeActiveOperation(lane.name, active); - active.reject(error); + active.reject( + error instanceof HarnessClosed || + error instanceof HarnessFault || + error instanceof RuntimeSliceNotImplemented + ? error + : this.fault(error), + ); } })(); } @@ -665,43 +1190,752 @@ class AgentHarnessRuntime implements AgentH active: ActiveOperation, options: DriveOptions, ): Promise { - let restored: RestoredLane; - try { - restored = await this.sessionStorage.mutate(lane.name, (reader) => restoreLane(reader, lane.name)); - } catch (error) { - throw this.fault(error); - } - if (restored.current === undefined || restored.current.operation.operationId !== active.operationId) { - let inspected: RestoredLane; - try { - inspected = await this.sessionStorage.mutate(lane.name, (reader) => - restoreLane(reader, lane.name, { includeLastResult: true }), - ); - } catch (error) { - throw this.fault(error); - } - if (inspected.lastResult?.operationId === active.operationId) { - let outcome: TerminalOperationOutcome; - try { - outcome = await this.sessionStorage.mutate(lane.name, (reader) => - hydrateTerminalOutcome(reader, inspected.lastResult!), - ); - } catch (error) { - throw this.fault(error); - } - return Result.ok({ kind: "settled", operationId: active.operationId, outcome }); - } - return Result.err(this.mismatch(lane.name, active.operationId, inspected)); - } + const initial = await this.loadExpected(lane.name, active.operationId, false); + if (initial.current === undefined) return this.settledOrMismatch(lane.name, active.operationId, initial); if (deadlineReached(options)) return Result.ok({ kind: "yielded", operationId: active.operationId }); await lane.breakpoint.hit({ kind: "runtime.dispatch", description: "Advance durable operation", - details: { operationId: active.operationId, operationKind: restored.current.state.kind }, + details: { operationId: active.operationId, operationKind: initial.current.state.kind }, }); if (deadlineReached(options)) return Result.ok({ kind: "yielded", operationId: active.operationId }); - throw new RuntimeSliceNotImplemented(`drive(${restored.current.state.kind})`); + if (initial.current.state.kind !== "run") { + throw new RuntimeSliceNotImplemented(`drive(${initial.current.state.kind})`); + } + + const recovery = !this.attachedOperationIds.has(active.operationId); + return startHarnessSpan( + this.telemetryContext, + "pi.harness.run", + { + "pi.session.id": this.sessionStorage.metadata.id, + "pi.lane.name": lane.name, + "pi.operation.id": active.operationId, + "pi.operation.recovery": recovery, + "pi.operation.kind": "run", + }, + async (runSpan) => { + if (recovery) { + const resumed = await this.resumeRun(lane, active, initial, options); + if (!resumed) return Result.ok({ kind: "yielded", operationId: active.operationId }); + this.attachedOperationIds.add(active.operationId); + } + while (true) { + const restored = await this.loadExpected(lane.name, active.operationId, true); + if (restored.current === undefined) { + const terminal = await this.settledOrMismatch(lane.name, active.operationId, restored); + if (terminal.ok && terminal.value.kind === "settled" && terminal.value.outcome.operation === "run") { + runSpan.setAttributes({ "pi.operation.outcome": terminal.value.outcome.kind }); + } + return terminal; + } + const state = restored.current.state; + if (state.kind !== "run") throw new SessionInvariantError("Run operation changed state kind"); + if (state.control.status !== "running") { + throw new RuntimeSliceNotImplemented("drive(cancel_requested)"); + } + if (deadlineReached(options)) return Result.ok({ kind: "yielded", operationId: active.operationId }); + if (state.phase.kind === "checkpoint") { + if (state.phase.continuation.kind === "need_assistant") { + const advanced = await this.startGeneration(lane, active, state, runSpan, options); + if (!advanced) return Result.ok({ kind: "yielded", operationId: active.operationId }); + continue; + } + const finished = await this.finishRun(lane, active, undefined, options); + if (finished === undefined) return Result.ok({ kind: "yielded", operationId: active.operationId }); + runSpan.setAttributes({ "pi.operation.outcome": "completed" }); + return Result.ok({ kind: "settled", operationId: active.operationId, outcome: finished }); + } + if (state.phase.kind === "assistant") { + if (state.phase.generation.status !== "ready") { + throw new RuntimeSliceNotImplemented(`drive(assistant.${state.phase.generation.status})`); + } + const settled = await this.executeAssistantGeneration( + lane, + active, + restored, + state, + runSpan, + options, + ); + if (!settled) return Result.ok({ kind: "yielded", operationId: active.operationId }); + continue; + } + if (state.phase.kind === "failure_drain") { + const finished = await this.finishRun(lane, active, state.phase.error, options); + if (finished === undefined) return Result.ok({ kind: "yielded", operationId: active.operationId }); + runSpan.setAttributes({ "pi.operation.outcome": "failed", "pi.error.code": state.phase.error.code }); + return Result.ok({ kind: "settled", operationId: active.operationId, outcome: finished }); + } + throw new RuntimeSliceNotImplemented(`drive(run.${state.phase.kind})`); + } + }, + ); + } + + private async loadExpected(lane: string, operationId: string, includeLastResult: boolean): Promise { + try { + return await this.sessionStorage.mutate(lane, async (reader) => { + const restored = await restoreLane(reader, lane, { includeLastResult }); + const currentId = restored.laneState.currentOperationId; + if (currentId !== null && currentId !== operationId) { + throw new SessionInvariantError(`Lane ${JSON.stringify(lane)} changed operation while a drive owns it`); + } + return restored; + }); + } catch (error) { + if (error instanceof HarnessClosed || error instanceof HarnessFault) throw error; + throw this.fault(error); + } + } + + private async settledOrMismatch(lane: string, operationId: string, restored: RestoredLane): Promise { + if (restored.lastResult?.operationId !== operationId) { + return Result.err(this.mismatch(lane, operationId, restored)); + } + try { + const outcome = await this.sessionStorage.mutate(lane, (reader) => + hydrateTerminalOutcome(reader, restored.lastResult!), + ); + return Result.ok({ kind: "settled", operationId, outcome }); + } catch (error) { + throw this.fault(error); + } + } + + private async resumeRun( + lane: AgentLaneRuntime, + active: ActiveOperation, + restored: RestoredLane, + options: DriveOptions, + ): Promise { + const current = restored.current; + if (current === undefined || current.operation.intent.kind !== "run") { + throw new SessionInvariantError("Run resume is missing run metadata"); + } + await this.events.emit({ type: "run_resume", runId: active.operationId, lane: lane.name, recovery: true }); + if (!this.hooks.has("before_resume")) return true; + await lane.breakpoint.hit({ + kind: "hook.before_resume", + description: "Run resume hooks", + details: { operationId: active.operationId }, + }); + if (deadlineReached(options)) return false; + const prompt = current.operation.intent.promptEntryIds.map((id) => { + const entry = current.entries.get(id); + if (entry?.type !== "message") throw new SessionInvariantError(`Prompt entry ${id} is missing`); + return entry.message; + }); + await this.hooks.runBeforeResumeWithGate( + { + kind: "run", + lane: lane.name, + runId: active.operationId, + prompt, + ...(current.operation.intent.systemPromptOverride === undefined + ? {} + : { systemPromptOverride: current.operation.intent.systemPromptOverride }), + }, + current.operation.intent.resumeData ?? {}, + active.effectGate, + ); + return true; + } + + private async startGeneration( + lane: AgentLaneRuntime, + active: ActiveOperation, + state: RunState, + runTelemetry: TelemetryContext, + options: DriveOptions, + ): Promise { + if (state.phase.kind !== "checkpoint" || state.phase.continuation.kind !== "need_assistant") { + throw new SessionInvariantError("Generation start is not at a need-assistant checkpoint"); + } + const expectedTriggerEntryId = state.phase.triggerEntryId; + const settings = await this.snapshotSettings(); + await lane.breakpoint.hit({ + kind: "run.generation_ready", + description: "Prepare an assistant generation", + details: { operationId: active.operationId }, + }); + if (deadlineReached(options)) return false; + const stepId = this.sessionStorage.idGenerator.next(); + await startHarnessSpan( + runTelemetry, + "pi.harness.checkpoint", + { + "pi.lane.name": lane.name, + "pi.operation.id": active.operationId, + "pi.checkpoint.kind": "normal", + }, + async () => { + try { + await this.sessionStorage.mutate(lane.name, async (mutator) => { + const restored = await restoreLane(mutator, lane.name); + const current = restored.current; + if ( + current === undefined || + current.operation.operationId !== active.operationId || + current.state.kind !== "run" + ) { + throw new SessionInvariantError("Generation start lost run ownership"); + } + const latest = current.state; + if ( + latest.phase.kind !== "checkpoint" || + latest.phase.continuation.kind !== "need_assistant" || + latest.phase.triggerEntryId !== expectedTriggerEntryId || + latest.control.status !== "running" + ) { + throw new SessionInvariantError("Generation start found another run phase"); + } + const context = { + stepId, + triggerEntryId: latest.phase.triggerEntryId, + configuration: cloneConfiguration(restored.configuration), + streamOptions: { ...settings.streamOptions }, + retryPolicy: normalizeRetryPolicy(settings.retryPolicy), + overflowRecoveryUsed: latest.phase.continuation.overflowRecoveryUsed, + }; + await mutator.commit({ + writes: [ + { + kind: "register", + op: "set", + namespace: "op.state", + key: active.operationId, + value: { + ...latest, + phase: { kind: "assistant", generation: { status: "ready", context, nextAttempt: 1 } }, + }, + }, + ], + }); + }); + } catch (error) { + if (error instanceof HarnessClosed || error instanceof HarnessFault) throw error; + throw this.fault(error); + } + }, + ); + return true; + } + + private async executeAssistantGeneration( + lane: AgentLaneRuntime, + active: ActiveOperation, + restored: RestoredLane, + state: RunState, + runTelemetry: TelemetryContext, + options: DriveOptions, + ): Promise { + if (state.phase.kind !== "assistant" || state.phase.generation.status !== "ready") { + throw new SessionInvariantError("Assistant generation is not ready"); + } + const ready = state.phase.generation; + const context = ready.context; + const model = this.models.getModel(context.configuration.model.provider, context.configuration.model.modelId); + if (model === undefined) throw new RuntimeSliceNotImplemented("drive(missing model suspension)"); + const settings = await this.snapshotSettings(); + const missingTools = context.configuration.activeToolNames.filter( + (name) => !settings.tools.some((tool) => tool.name === name), + ); + if (missingTools.length !== 0) throw new RuntimeSliceNotImplemented("drive(missing tool suspension)"); + + let streamOptions = { ...context.streamOptions }; + if (this.hooks.has("before_request")) { + await lane.breakpoint.hit({ + kind: "hook.before_request", + description: "Transform assistant request options", + details: { operationId: active.operationId, attempt: ready.nextAttempt }, + }); + if (deadlineReached(options)) return false; + const result = await this.hooks.runWithGate( + "before_request", + { + lane: lane.name, + runId: active.operationId, + model, + step: "assistant", + attempt: ready.nextAttempt, + streamOptions, + }, + active.effectGate, + ); + if (result?.streamOptions !== undefined) { + streamOptions = applyStreamOptionsPatch(streamOptions, result.streamOptions); + } + if (deadlineReached(options)) return false; + } + const operation = restored.current?.operation; + if (operation?.intent.kind !== "run") + throw new SessionInvariantError("Assistant generation is missing run metadata"); + const systemPrompt = operation.intent.systemPromptOverride ?? (await this.resolveSystemPrompt()); + const responseEntryId = this.sessionStorage.idGenerator.next(); + const usageId = this.sessionStorage.idGenerator.next(); + await lane.breakpoint.hit({ + kind: "assistant.intent", + description: "Commit assistant request intent", + details: { operationId: active.operationId, stepId: context.stepId, attempt: ready.nextAttempt }, + }); + if (deadlineReached(options)) return false; + try { + await this.sessionStorage.mutate(lane.name, async (mutator) => { + const latest = await restoreLane(mutator, lane.name); + const current = latest.current; + if ( + current === undefined || + current.operation.operationId !== active.operationId || + current.state.kind !== "run" + ) { + throw new SessionInvariantError("Assistant intent lost run ownership"); + } + const phase = current.state.phase; + if ( + phase.kind !== "assistant" || + phase.generation.status !== "ready" || + phase.generation.context.stepId !== context.stepId || + phase.generation.nextAttempt !== ready.nextAttempt + ) { + throw new SessionInvariantError("Assistant intent found another restart point"); + } + await mutator.commit({ + writes: [ + { + kind: "register", + op: "set", + namespace: "op.state", + key: active.operationId, + value: { + ...current.state, + phase: { + kind: "assistant", + generation: { + status: "effect_pending", + context, + attempt: ready.nextAttempt, + responseEntryId, + usageId, + intendedOutputLimit: model.maxTokens, + contextWindow: model.contextWindow, + }, + }, + }, + }, + ], + }); + }); + } catch (error) { + if (error instanceof HarnessClosed || error instanceof HarnessFault) throw error; + throw this.fault(error); + } + + const newestFirst = await lane.session.findEntriesOnBranch({ order: "newestFirst", stopAtType: "compaction" }); + const messages = await buildSessionContext([...newestFirst].reverse(), { entryProjectors: this.entryProjectors }); + await this.events.emit({ + type: "turn_start", + runId: active.operationId, + turnId: context.stepId, + lane: lane.name, + }); + const providerTools = context.configuration.activeToolNames.map( + (name) => settings.tools.find((tool) => tool.name === name)! as unknown as AgentTool, + ); + const message = await startHarnessSpan( + runTelemetry, + "pi.harness.turn", + { + "pi.lane.name": lane.name, + "pi.operation.id": active.operationId, + "pi.turn.id": context.stepId, + }, + (turnSpan) => + startHarnessSpan( + turnSpan, + "pi.harness.step", + { + "pi.lane.name": lane.name, + "pi.operation.id": active.operationId, + "pi.step.kind": "assistant", + "pi.step.attempt": ready.nextAttempt, + }, + async (stepSpan) => { + const settled = await streamHarnessAssistant(messages, { + model, + ...(systemPrompt === undefined ? {} : { systemPrompt }), + ...(providerTools.length === 0 ? {} : { tools: providerTools }), + thinkingLevel: context.configuration.thinkingLevel, + streamOptions, + transformContext: this.hooks.has("transform_context") + ? async (input) => { + await lane.breakpoint.hit({ + kind: "hook.transform_context", + description: "Transform assistant context", + details: { operationId: active.operationId, stepId: context.stepId }, + }); + const result = await this.hooks.runWithGate( + "transform_context", + { lane: lane.name, runId: active.operationId, messages: input }, + active.effectGate, + ); + return result?.messages ?? input; + } + : undefined, + toProviderMessages: this.toProviderMessages, + beforePayload: this.hooks.has("before_payload") + ? async (payload, requestModel) => { + await lane.breakpoint.hit({ + kind: "hook.before_payload", + description: "Transform provider payload", + details: { operationId: active.operationId, stepId: context.stepId }, + }); + return ( + await this.hooks.runWithGate( + "before_payload", + { lane: lane.name, runId: active.operationId, model: requestModel, payload }, + active.effectGate, + ) + )?.payload; + } + : undefined, + afterResponse: this.hooks.has("after_response") + ? async (settledMessage, metadata) => { + await lane.breakpoint.hit({ + kind: "hook.after_response", + description: "Transform assistant response", + details: { operationId: active.operationId, stepId: context.stepId }, + }); + const result = await this.hooks.runWithGate( + "after_response", + { + lane: lane.name, + runId: active.operationId, + ...metadata, + message: settledMessage, + }, + active.effectGate, + ); + return result?.message ?? settledMessage; + } + : undefined, + request: async ( + providerContext: Context, + providerOptions: SimpleStreamOptions, + ): Promise => { + await lane.breakpoint.hit({ + kind: "assistant_request", + description: "Request assistant response", + details: { + operationId: active.operationId, + stepId: context.stepId, + attempt: ready.nextAttempt, + }, + }); + const requestModel = this.models.getModel( + context.configuration.model.provider, + context.configuration.model.modelId, + ); + active.effectGate.assertOpen(); + return requestModel === undefined + ? createMissingModelStream(model) + : this.models.streamSimple(requestModel, providerContext, providerOptions); + }, + observer: { + start: (draft) => + this.events.emit({ + type: "message_start", + runId: active.operationId, + message: draft, + lane: lane.name, + }), + update: (draft, event) => + this.events.emit({ + type: "message_update", + runId: active.operationId, + message: draft, + event, + lane: lane.name, + }), + end: (finalMessage) => + this.events.emit({ + type: "message_end", + runId: active.operationId, + message: finalMessage, + entryId: responseEntryId, + lane: lane.name, + }), + }, + telemetryContext: stepSpan, + signal: active.effectGate.signal, + }); + stepSpan.setAttributes({ + "pi.step.outcome": + settled.stopReason === "stop" || settled.stopReason === "length" ? "succeeded" : "failed", + }); + return settled; + }, + ), + ); + if (message.role !== "assistant") { + throw this.fault(new SessionInvariantError("after_response returned an invalid assistant message")); + } + this.assertOpen(); + await lane.breakpoint.hit({ + kind: "assistant.settlement", + description: "Commit assistant response", + details: { operationId: active.operationId, stepId: context.stepId, attempt: ready.nextAttempt }, + }); + const settled = await this.commitAssistantSettlement( + lane.name, + active.operationId, + context.stepId, + responseEntryId, + usageId, + message, + ); + await this.events.emit({ type: "entry_added", entry: settled.entry, lane: lane.name }); + await this.events.emit({ type: "usage", lane: lane.name, row: settled.row, totals: settled.totals }); + await this.events.emit({ + type: "turn_end", + runId: active.operationId, + turnId: context.stepId, + message, + toolResults: [], + lane: lane.name, + }); + return true; + } + + private async commitAssistantSettlement( + lane: string, + operationId: string, + stepId: string, + responseEntryId: string, + usageId: string, + message: AssistantMessage, + ): Promise<{ + entry: Entry; + row: { id: string; seq: number; usage: Usage; entryId: string; adjustment: false }; + totals: Usage; + }> { + try { + const committed = await this.sessionStorage.mutate(lane, async (mutator) => { + const restored = await restoreLane(mutator, lane); + const current = restored.current; + if ( + current === undefined || + current.operation.operationId !== operationId || + current.state.kind !== "run" + ) { + throw new SessionInvariantError("Assistant settlement lost run ownership"); + } + const phase = current.state.phase; + if ( + phase.kind !== "assistant" || + phase.generation.status !== "effect_pending" || + phase.generation.context.stepId !== stepId || + phase.generation.responseEntryId !== responseEntryId || + phase.generation.usageId !== usageId + ) { + throw new SessionInvariantError("Assistant settlement found another pending request"); + } + if (message.stopReason === "aborted" && current.state.control.status === "running") { + throw new SessionInvariantError("Assistant response is aborted without durable cancellation"); + } + const successful = message.stopReason === "stop" || message.stopReason === "length"; + const error: OperationError = { + code: "assistant_error", + message: message.errorMessage ?? `Unsupported assistant stop reason: ${message.stopReason}`, + }; + const nextPhase: RunState["phase"] = successful + ? { + kind: "checkpoint", + continuation: { kind: "may_finish", includeFinalAssistant: true }, + triggerEntryId: responseEntryId, + } + : { kind: "failure_drain", error, provenance: { kind: "response", entryId: responseEntryId } }; + const result = await mutator.commit({ + writes: [ + { + kind: "entry", + entry: { id: responseEntryId, parentId: restored.leafId, type: "message", message }, + }, + { kind: "register", op: "set", namespace: "lane.leaf", key: lane, value: responseEntryId }, + { + kind: "usage", + row: { id: usageId, usage: message.usage, entryId: responseEntryId, adjustment: false }, + }, + { + kind: "register", + op: "set", + namespace: "op.state", + key: operationId, + value: { ...current.state, latestAssistantEntryId: responseEntryId, phase: nextPhase }, + }, + ], + }); + const entry = (await mutator.getEntries([responseEntryId])).get(responseEntryId); + if (entry === undefined) throw new SessionInvariantError("Committed assistant response is missing"); + return { + entry, + row: { + id: usageId, + seq: result.seqs[2]!, + usage: message.usage, + entryId: responseEntryId, + adjustment: false as const, + }, + }; + }); + return { ...committed, totals: (await this.sessionStorage.getStats()).usage }; + } catch (error) { + if (error instanceof HarnessClosed || error instanceof HarnessFault) throw error; + throw this.fault(error); + } + } + + private async finishRun( + lane: AgentLaneRuntime, + active: ActiveOperation, + error: OperationError | undefined, + options: DriveOptions, + ): Promise { + await lane.breakpoint.hit({ + kind: "run.finish", + description: "Finish the run", + details: { operationId: active.operationId, outcome: error === undefined ? "completed" : "failed" }, + }); + if (deadlineReached(options)) return undefined; + try { + const lastResult = await this.sessionStorage.mutate(lane.name, async (mutator) => { + const restored = await restoreLane(mutator, lane.name); + const current = restored.current; + if ( + current === undefined || + current.operation.operationId !== active.operationId || + current.state.kind !== "run" + ) { + throw new SessionInvariantError("Run finish lost operation ownership"); + } + if ( + current.state.inbox.steer.length !== 0 || + current.state.inbox.followUp.length !== 0 || + current.state.inbox.writes.length !== 0 + ) { + throw new RuntimeSliceNotImplemented("finish(run with queued input)"); + } + if (error === undefined) { + if ( + current.state.phase.kind !== "checkpoint" || + current.state.phase.continuation.kind !== "may_finish" + ) { + throw new SessionInvariantError("Completed run is not at a finish checkpoint"); + } + } else if (current.state.phase.kind !== "failure_drain") { + throw new SessionInvariantError("Failed run is not at failure drain"); + } + if (restored.leafId === null) throw new SessionInvariantError("Run cannot finish at the root"); + const finalAssistantEntryId = current.state.latestAssistantEntryId ?? undefined; + const result: LaneLastResult = + error === undefined + ? { + operationId: active.operationId, + kind: "run", + outcome: "completed", + leafId: restored.leafId, + runCompletion: "assistant", + ...(finalAssistantEntryId === undefined ? {} : { finalAssistantEntryId }), + } + : { + operationId: active.operationId, + kind: "run", + outcome: "failed", + leafId: restored.leafId, + error, + ...(finalAssistantEntryId === undefined ? {} : { finalAssistantEntryId }), + }; + const [toolArgs, preparations] = await Promise.all([ + mutator.listRegisters("op.tool_args", `${active.operationId}:`), + mutator.listRegisters("op.preparation", `${active.operationId}:`), + ]); + const pendingIds = [ + ...current.state.inbox.steer, + ...current.state.inbox.followUp, + ...current.state.inbox.writes, + ...(current.state.control.status === "cancel_requested" + ? [...current.state.control.drainedSteer, ...current.state.control.drainedFollowUp] + : []), + ]; + await mutator.commit({ + writes: [ + { kind: "register", op: "delete", namespace: "op.meta", key: active.operationId }, + { kind: "register", op: "delete", namespace: "op.state", key: active.operationId }, + ...toolArgs.map( + (register) => + ({ kind: "register", op: "delete", namespace: "op.tool_args", key: register.key }) as const, + ), + ...preparations.map( + (register) => + ({ kind: "register", op: "delete", namespace: "op.preparation", key: register.key }) as const, + ), + ...pendingIds.map( + (id) => ({ kind: "register", op: "delete", namespace: "pending.entry", key: id }) as const, + ), + { kind: "register", op: "set", namespace: "lane.lastResult", key: lane.name, value: result }, + { + kind: "register", + op: "set", + namespace: "lane.state", + key: lane.name, + value: { ...restored.laneState, currentOperationId: null }, + }, + ], + }); + return result; + }); + const outcome = await this.sessionStorage.mutate(lane.name, (reader) => + hydrateTerminalOutcome(reader, lastResult), + ); + if (outcome.operation !== "run") + throw new SessionInvariantError("Run terminal result hydrated as another kind"); + this.attachedOperationIds.delete(active.operationId); + this.restoredSuspensions.delete(lane.name); + const finalFields = + outcome.finalEntryId === undefined + ? {} + : { finalEntryId: outcome.finalEntryId, finalMessage: outcome.finalMessage }; + if (outcome.kind === "failed") { + await this.events.emit({ + type: "run_end", + runId: active.operationId, + outcome: "failed", + leafId: outcome.leafId, + error: outcome.error, + ...finalFields, + lane: lane.name, + }); + } else { + await this.events.emit({ + type: "run_end", + runId: active.operationId, + outcome: outcome.kind, + leafId: outcome.leafId, + ...finalFields, + lane: lane.name, + }); + } + return outcome; + } catch (caught) { + if ( + caught instanceof RuntimeSliceNotImplemented || + caught instanceof HarnessClosed || + caught instanceof HarnessFault + ) + throw caught; + throw this.fault(caught); + } } private async removeActiveOperation(lane: string, active: ActiveOperation): Promise { @@ -781,7 +2015,7 @@ class AgentLaneRuntime implements AgentLane constructor(harness: AgentHarnessRuntime, name: string) { this.harness = harness; this.name = name; - this.session = harness.sessionStorage.view(name); + this.session = harness.createPublicSessionView(name); this.breakpoint = new BreakpointBarrier(harness.driveMode); } @@ -795,8 +2029,8 @@ class AgentLaneRuntime implements AgentLane return (await this.harness.sessionStorage.getRegister("lane.lastResult", this.name))?.value; } - accept(_request: OperationRequest): Promise { - return this.unimplementedResult("accept"); + accept(request: OperationRequest): Promise { + return this.harness.acceptLane(this, request); } drive(options: DriveOptions): Promise { @@ -811,16 +2045,24 @@ class AgentLaneRuntime implements AgentLane return this.harness.inspectLane(this); } - prompt(_text: string, _images?: ImageContent[]): Promise; - prompt(_message: AgentMessage | AgentMessage[]): Promise; - prompt(_message: string | AgentMessage | AgentMessage[], _images?: ImageContent[]): Promise { - return this.unimplementedResult("prompt"); + prompt(text: string, images?: ImageContent[]): Promise; + prompt(message: AgentMessage | AgentMessage[]): Promise; + prompt(message: string | AgentMessage | AgentMessage[], images?: ImageContent[]): Promise { + return this.runAccepted( + typeof message === "string" + ? { kind: "prompt", prompt: message, ...(images === undefined ? {} : { images }) } + : { kind: "prompt", prompt: message }, + ); } - skill(_name: string, _additionalInstructions?: string): Promise { - return this.unimplementedResult("skill"); + skill(name: string, additionalInstructions?: string): Promise { + return this.runAccepted({ + kind: "skill", + name, + ...(additionalInstructions === undefined ? {} : { additionalInstructions }), + }); } - promptFromTemplate(_name: string, _args?: string[]): Promise { - return this.unimplementedResult("promptFromTemplate"); + promptFromTemplate(name: string, args?: string[]): Promise { + return this.runAccepted({ kind: "prompt_template", name, ...(args === undefined ? {} : { args }) }); } compact(_options?: { customInstructions?: string }): Promise { return this.unimplementedResult("compact"); @@ -937,6 +2179,53 @@ class AgentLaneRuntime implements AgentLane return this.notImplemented("watch"); } + private async runAccepted( + request: Extract, + ): Promise { + const admission = await this.accept(request); + if (!admission.ok) { + if ( + admission.error._tag === "LaneBusy" || + admission.error._tag === "MissingIdentities" || + admission.error._tag === "InvalidMessage" || + admission.error._tag === "UnknownSkill" || + admission.error._tag === "UnknownTemplate" || + admission.error._tag === "Closed" + ) { + return Result.err(admission.error); + } + throw new SessionInvariantError("Run acceptance returned a structural-operation error"); + } + while (true) { + const driven = await this.drive({ + operationId: admission.value.operationId, + waitForRetry: true, + pollDeferred: true, + }); + if (!driven.ok) { + throw new SessionInvariantError("A convenience drive lost ownership of its accepted operation"); + } + if (driven.value.kind === "yielded") continue; + if (driven.value.kind !== "settled") { + if (driven.value.reason === "missing_identities") { + return Result.ok({ + runId: admission.value.operationId, + kind: "suspended", + reason: "missing_identities", + missing: driven.value.missing, + leafId: await this.getLeafId().then((leaf) => leaf ?? ""), + }); + } + throw new RuntimeSliceNotImplemented(`convenience wait(${driven.value.reason})`); + } + const outcome = driven.value.outcome; + if (outcome.operation !== "run") + throw new SessionInvariantError("Accepted run settled as another operation kind"); + const { operation: _operation, ...run } = outcome; + return Result.ok(run); + } + } + private async getConfiguration(): Promise { this.harness.assertOpen(); try { @@ -993,6 +2282,39 @@ function deadlineReached(options: DriveOptions): boolean { return options.deadline !== undefined && Date.now() >= options.deadline; } +function normalizeRetryPolicy(policy: RetryPolicy): { maxAttempts: number; baseDelayMs: number } { + return { maxAttempts: policy.enabled ? policy.maxRetries + 1 : 1, baseDelayMs: policy.baseDelayMs }; +} + +function isPendingAssistant(message: AgentMessage): boolean { + return message.role === "assistant" && message.stopReason === "pending"; +} + +function createMissingModelStream(model: Model): AssistantMessageEventStream { + const stream = createAssistantMessageEventStream(); + const message: AssistantMessage = { + role: "assistant", + content: [], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "error", + errorMessage: `Model is no longer available: ${model.provider}/${model.id}`, + timestamp: Date.now(), + }; + stream.push({ type: "error", reason: "error", error: message }); + stream.end(message); + return stream; +} + async function hydrateTerminalOutcome( reader: Pick, lastResult: LaneLastResult, @@ -1022,6 +2344,17 @@ async function hydrateTerminalOutcome( throw new SessionInvariantError(`Final assistant ${lastResult.finalAssistantEntryId} is invalid`); } const finalMessage = final?.message as AssistantMessage | undefined; + if (lastResult.outcome === "completed" && lastResult.runCompletion === "assistant") { + if (final === undefined || finalMessage === undefined) { + throw new SessionInvariantError("Completed assistant run has no final assistant"); + } + if (lastResult.finalAssistantEntryId !== lastResult.leafId) { + throw new SessionInvariantError("Completed assistant run final entry is not its leaf"); + } + if (finalMessage.stopReason !== "stop" && finalMessage.stopReason !== "length") { + throw new SessionInvariantError("Completed assistant run has an invalid stop reason"); + } + } const finalFields = final === undefined || finalMessage === undefined ? {} : { finalEntryId: final.id, finalMessage }; if (lastResult.outcome === "failed") { diff --git a/packages/agent/src/harness/hooks.ts b/packages/agent/src/harness/hooks.ts index 3267668b6..28fd52086 100644 --- a/packages/agent/src/harness/hooks.ts +++ b/packages/agent/src/harness/hooks.ts @@ -46,6 +46,10 @@ export class HookRegistry implements Hooks { }; } + has(name: HookName): boolean { + return (this.registrations.get(name)?.length ?? 0) !== 0; + } + /** Invoke one accepted-operation aggregate after synchronously passing its effect gate. */ runWithGate( name: TName, @@ -357,7 +361,7 @@ export class HookRegistry implements Hooks { } } -function applyStreamOptionsPatch( +export function applyStreamOptionsPatch( base: AgentHarnessStreamOptions, patch: AgentHarnessStreamOptionsPatch, ): AgentHarnessStreamOptions { diff --git a/packages/agent/src/harness/restore.ts b/packages/agent/src/harness/restore.ts index 6435ac1e7..7f59ae5bc 100644 --- a/packages/agent/src/harness/restore.ts +++ b/packages/agent/src/harness/restore.ts @@ -6,6 +6,7 @@ import type { LaneState, Operation, OperationState, + RunState, SessionReader, } from "./session/types.ts"; @@ -56,8 +57,11 @@ export async function restoreLane( invariant(lane, `names operation ${operationId}, which is missing op.state`); } - const expectations = new Map(); + const expectations = new Map(); + const forbiddenEntries = new Set(); + const pendingExpectations = new Map(); if (leaf.value !== null) expectations.set(leaf.value, "any"); + for (const id of laneState.value.pendingNextRun) addPendingExpectation(pendingExpectations, id, "message", lane); if (operation !== undefined && operationState !== undefined) { if (operation.value.operationId !== operationId) { invariant(lane, `op.meta/${operationId} has another operation id`); @@ -74,16 +78,73 @@ export async function restoreLane( if (operation.value.sourceLeafId !== null) expectations.set(operation.value.sourceLeafId, "any"); if (operation.value.intent.kind === "run") { for (const id of operation.value.intent.promptEntryIds) expectations.set(id, "message"); + validateRunState( + operationState.value as RunState, + expectations, + forbiddenEntries, + pendingExpectations, + leaf.value, + lane, + ); } else if (operation.value.intent.kind === "navigation" && operation.value.intent.targetId !== null) { expectations.set(operation.value.intent.targetId, "any"); } } - const entries = await reader.getEntries([...expectations.keys()]); + const entryIds = [...new Set([...expectations.keys(), ...forbiddenEntries, ...pendingExpectations.keys()])]; + const [entries, pendingRegisters] = await Promise.all([ + reader.getEntries(entryIds), + Promise.all( + [...pendingExpectations].map(async ([id, expected]) => ({ + id, + expected, + register: await reader.getRegister("pending.entry", id), + })), + ), + ]); for (const [id, expected] of expectations) { const entry = entries.get(id); if (entry === undefined) invariant(lane, `references missing entry ${id}`); - if (expected === "message" && entry.type !== "message") invariant(lane, `entry ${id} is not a message`); + if (expected !== "any" && entry.type !== "message") invariant(lane, `entry ${id} is not a message`); + if (expected === "assistant" && entry.type === "message" && entry.message.role !== "assistant") { + invariant(lane, `entry ${id} is not an assistant message`); + } + } + if (operationState?.value.kind === "run") { + const state = operationState.value; + if ( + state.control.status === "running" && + state.phase.kind === "checkpoint" && + state.phase.continuation.kind === "may_finish" && + state.phase.continuation.includeFinalAssistant && + state.latestAssistantEntryId !== null + ) { + const latest = entries.get(state.latestAssistantEntryId); + if ( + latest?.type !== "message" || + latest.message.role !== "assistant" || + (latest.message.stopReason !== "stop" && latest.message.stopReason !== "length") + ) { + invariant(lane, "finish checkpoint latest assistant has an invalid stop reason"); + } + } + } + for (const id of forbiddenEntries) { + if (entries.has(id)) invariant(lane, `reserved entry ${id} already exists`); + } + for (const pending of pendingRegisters) { + if (pending.register === undefined) invariant(lane, `references missing pending entry ${pending.id}`); + if (entries.has(pending.id)) invariant(lane, `pending entry ${pending.id} is already placed`); + if (pending.expected === "message" && pending.register.value.type !== "message") { + invariant(lane, `pending entry ${pending.id} is not a message`); + } + if ( + pending.register.value.type === "message" && + pending.register.value.payload.role === "assistant" && + pending.register.value.payload.stopReason === "pending" + ) { + invariant(lane, `pending entry ${pending.id} contains a pending assistant`); + } } return { @@ -98,6 +159,109 @@ export async function restoreLane( }; } +function validateRunState( + state: RunState, + expectations: Map, + forbiddenEntries: Set, + pendingExpectations: Map, + leafId: string | null, + lane: string, +): void { + if (state.latestAssistantEntryId !== null) + addEntryExpectation(expectations, state.latestAssistantEntryId, "assistant"); + for (const id of state.inbox.steer) addPendingExpectation(pendingExpectations, id, "message", lane); + for (const id of state.inbox.followUp) addPendingExpectation(pendingExpectations, id, "message", lane); + for (const id of state.inbox.writes) addPendingExpectation(pendingExpectations, id, "any", lane); + if (state.control.status === "cancel_requested") { + for (const id of state.control.drainedSteer) addPendingExpectation(pendingExpectations, id, "message", lane); + for (const id of state.control.drainedFollowUp) addPendingExpectation(pendingExpectations, id, "message", lane); + } + + const phase = state.phase; + if (phase.kind === "checkpoint") { + addEntryExpectation(expectations, phase.triggerEntryId, "any"); + if (phase.continuation.kind === "need_assistant" && leafId !== phase.triggerEntryId) { + invariant(lane, "need-assistant checkpoint trigger is not the lane leaf"); + } + if (phase.continuation.kind === "may_finish" && phase.continuation.includeFinalAssistant) { + if (state.latestAssistantEntryId === null) invariant(lane, "finish checkpoint has no latest assistant"); + if (phase.triggerEntryId !== state.latestAssistantEntryId) { + invariant(lane, "finish checkpoint trigger is not the latest assistant"); + } + } + return; + } + if (phase.kind === "assistant") { + const generation = phase.generation; + addEntryExpectation(expectations, generation.context.triggerEntryId, "any"); + validatePositiveInteger(generation.context.retryPolicy.maxAttempts, lane, "generation maxAttempts"); + validateNonNegativeInteger(generation.context.retryPolicy.baseDelayMs, lane, "generation baseDelayMs"); + if (generation.status === "ready") { + validatePositiveInteger(generation.nextAttempt, lane, "generation nextAttempt"); + if (generation.nextAttempt === 1 && leafId !== generation.context.triggerEntryId) { + invariant(lane, "initial generation trigger is not the lane leaf"); + } + if (generation.nextAttempt > generation.context.retryPolicy.maxAttempts) { + invariant(lane, "generation nextAttempt exceeds maxAttempts"); + } + return; + } + if (generation.status === "retry_wait") { + validatePositiveInteger(generation.nextAttempt, lane, "generation nextAttempt"); + validateNonNegativeInteger(generation.notBefore, lane, "generation notBefore"); + return; + } + validatePositiveInteger(generation.attempt, lane, "generation attempt"); + if (generation.attempt > generation.context.retryPolicy.maxAttempts) { + invariant(lane, "generation attempt exceeds maxAttempts"); + } + if (generation.attempt === 1 && leafId !== generation.context.triggerEntryId) { + invariant(lane, "initial generation trigger is not the lane leaf"); + } + validateNonNegativeInteger(generation.intendedOutputLimit, lane, "generation intendedOutputLimit"); + validateNonNegativeInteger(generation.contextWindow, lane, "generation contextWindow"); + if (generation.responseEntryId === generation.usageId) + invariant(lane, "generation response and usage ids collide"); + forbiddenEntries.add(generation.responseEntryId); + return; + } + if (phase.kind === "failure_drain" && phase.provenance.kind === "response") { + addEntryExpectation(expectations, phase.provenance.entryId, "assistant"); + if (state.latestAssistantEntryId !== phase.provenance.entryId) { + invariant(lane, "response failure provenance is not the latest assistant"); + } + } +} + +function addEntryExpectation( + expectations: Map, + id: string, + expected: "any" | "message" | "assistant", +): void { + const current = expectations.get(id); + if (current === "assistant" || current === expected) return; + if (current === "message" && expected === "any") return; + expectations.set(id, expected); +} + +function addPendingExpectation( + expectations: Map, + id: string, + expected: "any" | "message", + lane: string, +): void { + if (expectations.has(id)) invariant(lane, `queues pending entry ${id} more than once`); + expectations.set(id, expected); +} + +function validatePositiveInteger(value: number, lane: string, field: string): void { + if (!Number.isSafeInteger(value) || value <= 0) invariant(lane, `${field} is invalid`); +} + +function validateNonNegativeInteger(value: number, lane: string, field: string): void { + if (!Number.isSafeInteger(value) || value < 0) invariant(lane, `${field} is invalid`); +} + function invariant(lane: string, message: string): never { throw new SessionInvariantError(`Lane ${JSON.stringify(lane)} ${message}`); } diff --git a/packages/agent/test/harness/agent-harness-r2.test.ts b/packages/agent/test/harness/agent-harness-r2.test.ts new file mode 100644 index 000000000..6ad2e67f2 --- /dev/null +++ b/packages/agent/test/harness/agent-harness-r2.test.ts @@ -0,0 +1,549 @@ +import { + createModels, + type FauxProviderHandle, + fauxAssistantMessage, + fauxProvider, + type MutableModels, +} from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it } from "vitest"; +import { + AgentHarness, + type AgentHarness as AgentHarnessInstance, + type HarnessEvent, + MemorySessionRepo, + type Session, +} from "../../src/index.ts"; +import { waitForTick } from "../utils/wait-for-tick.ts"; + +interface Fixture { + harness: AgentHarnessInstance; + session: Session; + repo: MemorySessionRepo; + faux: FauxProviderHandle; + models: MutableModels; +} + +const fixtures: Fixture[] = []; + +async function createFixture(options: { drive?: "automatic" | "manual" } = {}): Promise { + const repo = new MemorySessionRepo(); + const session = await repo.create({}); + const faux = fauxProvider(); + const models = createModels(); + models.setProvider(faux.provider); + const { harness } = await AgentHarness.create({ + session, + models, + model: faux.getModel(), + activeToolNames: [], + drive: options.drive, + }); + const fixture = { harness, session, repo, faux, models }; + fixtures.push(fixture); + return fixture; +} + +function captureLifecycle(harness: AgentHarnessInstance): HarnessEvent[] { + const events: HarnessEvent[] = []; + for (const type of [ + "run_start", + "run_resume", + "turn_start", + "message_start", + "message_update", + "message_end", + "entry_added", + "usage", + "turn_end", + "run_end", + ] as const) { + harness.events.on(type, (event) => { + events.push(event); + }); + } + return events; +} + +async function waitForAction(harness: AgentHarnessInstance): Promise { + for (let attempt = 0; attempt < 100; attempt++) { + if ((await harness.peekAction()) !== undefined) return; + await waitForTick(); + } + throw new Error("action did not park"); +} + +function deferred(): { promise: Promise; resolve(): void } { + let resolvePromise: (() => void) | undefined; + const promise = new Promise((resolve) => { + resolvePromise = resolve; + }); + return { promise, resolve: () => resolvePromise?.() }; +} + +afterEach(async () => { + for (const fixture of fixtures.splice(0)) { + await fixture.harness.close(); + await fixture.repo.close(); + } +}); + +describe("AgentHarness R2 minimal run", () => { + it("accepts durably without starting the provider, then completes one no-tool generation", async () => { + const { harness, session, faux } = await createFixture(); + faux.setResponses([fauxAssistantMessage("answer", { timestamp: 20 })]); + const events = captureLifecycle(harness); + const operationId = session.idGenerator.next(); + + const admission = await harness.accept({ kind: "prompt", operationId, prompt: "question" }); + expect(admission).toMatchObject({ ok: true, value: { operationId, kind: "run" } }); + expect(faux.state.callCount).toBe(0); + expect((await harness.inspectExecution()).current).toMatchObject({ id: operationId, status: "suspended" }); + expect((await session.getRegister("op.meta", operationId))?.value.operationId).toBe(operationId); + expect(events.map((event) => event.type)).toEqual(["run_start", "message_start", "message_end", "entry_added"]); + + const driven = await harness.drive({ operationId }); + expect(driven).toMatchObject({ + ok: true, + value: { + kind: "settled", + operationId, + outcome: { + operation: "run", + runId: operationId, + kind: "completed", + finalMessage: { role: "assistant", stopReason: "stop" }, + }, + }, + }); + expect(faux.state.callCount).toBe(1); + expect(await session.getRegister("op.meta", operationId)).toBeUndefined(); + expect(await session.getRegister("op.state", operationId)).toBeUndefined(); + expect(await session.getRegister("lane.state", "main")).toMatchObject({ value: { currentOperationId: null } }); + expect(await session.getRegister("lane.lastResult", "main")).toMatchObject({ + value: { operationId, kind: "run", outcome: "completed", runCompletion: "assistant" }, + }); + + const types = events.map((event) => event.type); + expect(types.slice(4, 6)).toEqual(["turn_start", "message_start"]); + expect(types.slice(6, -5).every((type) => type === "message_update")).toBe(true); + expect(types.slice(-5)).toEqual(["message_end", "entry_added", "usage", "turn_end", "run_end"]); + }); + + it("applies context, request-option, and response hooks around the captured request", async () => { + const { harness, faux } = await createFixture(); + await harness.setThinkingLevel("high"); + await harness.setStreamOptions({ headers: { base: "yes" } }); + harness.hooks.on("transform_context", ({ messages }) => ({ + messages: [...messages, { role: "user", content: "context hook", timestamp: 2 }], + })); + harness.hooks.on("before_request", () => ({ streamOptions: { headers: { extra: "yes" } } })); + harness.hooks.on("after_response", ({ message }) => ({ + message: { ...message, content: [{ type: "text", text: "post-hook" }] }, + })); + faux.setResponses([ + (context, options) => { + expect(context.messages.at(-1)).toMatchObject({ role: "user", content: "context hook" }); + expect(options).toMatchObject({ reasoning: "high", headers: { base: "yes", extra: "yes" } }); + return fauxAssistantMessage("provider"); + }, + ]); + const result = await harness.prompt("go"); + expect(result).toMatchObject({ + ok: true, + value: { finalMessage: { content: [{ type: "text", text: "post-hook" }] } }, + }); + }); + + it("implements prompt as acceptance followed by the same drive path", async () => { + const { harness, faux } = await createFixture(); + faux.setResponses([fauxAssistantMessage("done")]); + const result = await harness.prompt("go"); + expect(result).toMatchObject({ + ok: true, + value: { kind: "completed", finalMessage: { role: "assistant", stopReason: "stop" } }, + }); + }); + + it("rejects caller-supplied pending assistants without opening an operation", async () => { + const { harness } = await createFixture(); + const pending = fauxAssistantMessage("draft", { stopReason: "pending" }); + expect(await harness.accept({ kind: "prompt", prompt: pending })).toMatchObject({ + ok: false, + error: { _tag: "InvalidMessage", reason: "pending_assistant" }, + }); + expect((await harness.inspectExecution()).current).toBeNull(); + }); + + it("expands run resources and rejects unknown or empty prompts without opening an operation", async () => { + const { harness } = await createFixture(); + await harness.setResources({ + skills: [{ name: "review", description: "Review", content: "Check it", filePath: "/skills/review/SKILL.md" }], + promptTemplates: [{ name: "hello", content: "Hello $1" }], + }); + const skill = await harness.accept({ kind: "skill", name: "review", additionalInstructions: "carefully" }); + expect(skill.ok).toBe(true); + if (skill.ok) { + const meta = await harness.session.getEntry((await harness.getLeafId())!); + expect(meta).toMatchObject({ type: "message", message: { content: expect.stringContaining("carefully") } }); + await harness.close(); + } + + const second = await createFixture(); + await second.harness.setResources({ promptTemplates: [{ name: "hello", content: "Hello $1" }] }); + const template = await second.harness.accept({ kind: "prompt_template", name: "hello", args: ["Ada"] }); + expect(template.ok).toBe(true); + expect(await second.harness.session.getEntry((await second.harness.getLeafId())!)).toMatchObject({ + message: { content: "Hello Ada" }, + }); + + const third = await createFixture(); + expect(await third.harness.accept({ kind: "skill", name: "missing" })).toMatchObject({ + ok: false, + error: { _tag: "UnknownSkill" }, + }); + expect(await third.harness.accept({ kind: "prompt", prompt: "" })).toMatchObject({ + ok: false, + error: { _tag: "InvalidMessage" }, + }); + expect((await third.harness.inspectExecution()).current).toBeNull(); + }); + + it("persists before_run injections, system override, and per-handler resume data", async () => { + const { harness, session } = await createFixture(); + harness.hooks.on( + "before_run", + () => ({ + messages: [{ role: "user", content: "hook", timestamp: 2 }], + systemPrompt: "override", + resumeData: { token: "saved" }, + }), + { id: "extension" }, + ); + const accepted = await harness.accept({ kind: "prompt", prompt: "caller" }); + if (!accepted.ok) throw accepted.error; + const operation = await session.getRegister("op.meta", accepted.value.operationId); + expect(operation?.value.intent).toMatchObject({ + kind: "run", + systemPromptOverride: "override", + resumeData: { extension: { token: "saved" } }, + }); + if (operation?.value.intent.kind !== "run") throw new Error("missing run operation"); + expect(operation.value.intent.promptEntryIds).toHaveLength(1); + const branch = await harness.session.findEntriesOnBranch({ order: "oldestFirst" }); + expect(branch).toHaveLength(2); + expect(branch[1]).toMatchObject({ message: { content: "hook" } }); + }); + + it("rejects unresolved model and active-tool identities before writing an operation", async () => { + const modelFixture = await createFixture(); + 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: [] }, + }); + expect((await modelFixture.harness.inspectExecution()).current).toBeNull(); + + const toolFixture = await createFixture(); + 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"] }, + }); + expect((await toolFixture.harness.inspectExecution()).current).toBeNull(); + }); + + it("holds a stable admission reservation across before_run and lets a matching drive wait", async () => { + const { harness, session, faux } = await createFixture(); + faux.setResponses([fauxAssistantMessage("ok")]); + const gate = deferred(); + const operationId = session.idGenerator.next(); + harness.hooks.on( + "before_run", + async ({ runId }) => { + expect(runId).toBe(operationId); + await gate.promise; + return { messages: [{ role: "user", content: "injected", timestamp: 2 }], resumeData: { accepted: true } }; + }, + { id: "test" }, + ); + + const acceptance = harness.accept({ kind: "prompt", operationId, prompt: "hello" }); + await waitForTick(); + const competing = await harness.accept({ kind: "prompt", prompt: "other" }); + expect(competing).toMatchObject({ ok: false, error: { _tag: "LaneBusy", operationId } }); + const drive = harness.drive({ operationId }); + expect(faux.state.callCount).toBe(0); + gate.resolve(); + expect(await acceptance).toMatchObject({ ok: true, value: { operationId } }); + expect(await drive).toMatchObject({ ok: true, value: { kind: "settled", operationId } }); + expect((await session.getRegister("lane.lastResult", "main"))?.value.operationId).toBe(operationId); + }); + + it("makes leaf-moving tree writes wait for a pre-acceptance reservation", async () => { + const { harness, session } = await createFixture(); + const gate = deferred(); + harness.hooks.on( + "before_run", + async () => { + await gate.promise; + return undefined; + }, + { id: "wait" }, + ); + const acceptance = harness.accept({ kind: "prompt", prompt: "accepted" }); + await waitForTick(); + const append = harness.session.appendMessage({ role: "user", content: "later", timestamp: 3 }); + await waitForTick(); + expect((await session.getRegister("lane.leaf", "main"))?.value).toBeNull(); + gate.resolve(); + const accepted = await acceptance; + if (!accepted.ok) throw accepted.error; + const appendedId = await append; + expect((await session.getRegister("pending.entry", appendedId))?.value).toMatchObject({ + type: "message", + payload: { content: "later" }, + }); + expect((await session.getRegister("op.state", accepted.value.operationId))?.value).toMatchObject({ + inbox: { writes: [appendedId] }, + }); + }); + + it("captures and places pending next-run payloads before the caller prompt", async () => { + const { harness, session } = await createFixture(); + const pendingId = session.idGenerator.next(); + await session.mutate("main", async (mutator) => { + const laneState = await mutator.getRegister("lane.state", "main"); + if (laneState === undefined) throw new Error("missing lane state"); + await mutator.commit({ + writes: [ + { + kind: "register", + op: "set", + namespace: "pending.entry", + key: pendingId, + value: { type: "message", payload: { role: "user", content: "queued", timestamp: 1 } }, + }, + { + kind: "register", + op: "set", + namespace: "lane.state", + key: "main", + value: { ...laneState.value, pendingNextRun: [pendingId] }, + }, + ], + }); + }); + + const accepted = await harness.accept({ kind: "prompt", prompt: "caller" }); + expect(accepted.ok).toBe(true); + const branch = await harness.session.findEntriesOnBranch({ order: "oldestFirst" }); + expect( + branch.map((entry) => + entry.type === "message" && entry.message.role === "user" ? entry.message.content : "custom", + ), + ).toEqual(["queued", [{ type: "text", text: "caller" }]]); + expect(await session.getRegister("pending.entry", pendingId)).toBeUndefined(); + expect((await session.getRegister("lane.state", "main"))?.value.pendingNextRun).toEqual([]); + }); + + it("publishes stable manual breakpoints for the same durable run", async () => { + const { harness, faux, session } = await createFixture({ drive: "manual" }); + faux.setResponses([fauxAssistantMessage("manual")]); + const accepted = await harness.accept({ kind: "prompt", prompt: "step" }); + if (!accepted.ok) throw accepted.error; + const drive = harness.drive({ operationId: accepted.value.operationId }); + const actions: string[] = []; + while (true) { + await waitForAction(harness); + const parked = await harness.peekAction(); + if (parked?.kind === "run.finish") { + const state = await session.getRegister("op.state", accepted.value.operationId); + const leaf = await session.getRegister("lane.leaf", "main"); + expect(state?.value).toMatchObject({ + latestAssistantEntryId: leaf?.value, + phase: { kind: "checkpoint", triggerEntryId: leaf?.value }, + }); + } + const action = await harness.executeAction(); + if (action !== undefined) actions.push(action.kind); + if (action?.kind === "run.finish") break; + } + expect(await drive).toMatchObject({ ok: true, value: { kind: "settled" } }); + expect(actions).toEqual([ + "runtime.dispatch", + "run.generation_ready", + "assistant.intent", + "assistant_request", + "assistant.settlement", + "run.finish", + ]); + }); + + it("re-resolves the durable model identity when the provider request starts", async () => { + const { harness, faux, models } = await createFixture({ drive: "manual" }); + faux.setResponses([fauxAssistantMessage("must not be requested")]); + const accepted = await harness.accept({ kind: "prompt", prompt: "race" }); + if (!accepted.ok) throw accepted.error; + const drive = harness.drive({ operationId: accepted.value.operationId }); + for (const expected of ["runtime.dispatch", "run.generation_ready", "assistant.intent"] as const) { + await waitForAction(harness); + expect((await harness.executeAction())?.kind).toBe(expected); + } + await waitForAction(harness); + expect(await harness.peekAction()).toMatchObject({ kind: "assistant_request" }); + models.deleteProvider(faux.provider.id); + await harness.executeAction(); + await harness.runToCompletion(); + expect(await drive).toMatchObject({ + ok: true, + value: { kind: "settled", outcome: { kind: "failed", error: { code: "assistant_error" } } }, + }); + expect(faux.state.callCount).toBe(0); + }); + + it("closes at the provider boundary without starting the request or advancing the intent", async () => { + const fixture = await createFixture({ drive: "manual" }); + fixture.faux.setResponses([fauxAssistantMessage("must not start")]); + const accepted = await fixture.harness.accept({ kind: "prompt", prompt: "close" }); + if (!accepted.ok) throw accepted.error; + const drive = fixture.harness.drive({ operationId: accepted.value.operationId }); + for (const expected of ["runtime.dispatch", "run.generation_ready", "assistant.intent"] as const) { + await waitForAction(fixture.harness); + expect((await fixture.harness.executeAction())?.kind).toBe(expected); + } + await waitForAction(fixture.harness); + expect(await fixture.harness.peekAction()).toMatchObject({ kind: "assistant_request" }); + await fixture.harness.close(); + await expect(drive).rejects.toMatchObject({ name: "HarnessClosed" }); + expect(fixture.faux.state.callCount).toBe(0); + fixtures.splice(fixtures.indexOf(fixture), 1); + const reopened = await fixture.repo.open(fixture.session.metadata); + expect((await reopened.getRegister("op.state", accepted.value.operationId))?.value).toMatchObject({ + phase: { kind: "assistant", generation: { status: "effect_pending" } }, + }); + await reopened.close(); + await fixture.repo.close(); + }); + + it.each([ + { target: "assistant.settlement", phase: "assistant", generationStatus: "effect_pending" }, + { target: "run.finish", phase: "checkpoint", generationStatus: undefined }, + ] as const)( + "closes at $target without starting the following commit", + async ({ target, phase, generationStatus }) => { + const fixture = await createFixture({ drive: "manual" }); + fixture.faux.setResponses([fauxAssistantMessage("boundary")]); + const accepted = await fixture.harness.accept({ kind: "prompt", prompt: "boundary" }); + if (!accepted.ok) throw accepted.error; + const drive = fixture.harness.drive({ operationId: accepted.value.operationId }); + while (true) { + await waitForAction(fixture.harness); + if ((await fixture.harness.peekAction())?.kind === target) break; + await fixture.harness.executeAction(); + } + await fixture.harness.close(); + await expect(drive).rejects.toMatchObject({ name: "HarnessClosed" }); + fixtures.splice(fixtures.indexOf(fixture), 1); + const reopened = await fixture.repo.open(fixture.session.metadata); + const state = (await reopened.getRegister("op.state", accepted.value.operationId))?.value; + expect(state).toMatchObject({ + phase: { + kind: phase, + ...(generationStatus === undefined ? {} : { generation: { status: generationStatus } }), + }, + }); + await reopened.close(); + await fixture.repo.close(); + }, + ); + + it("reopens a durable generation-ready state and performs the request", async () => { + const fixture = await createFixture({ drive: "manual" }); + fixture.faux.setResponses([fauxAssistantMessage("ready after reopen")]); + const accepted = await fixture.harness.accept({ kind: "prompt", prompt: "ready" }); + if (!accepted.ok) throw accepted.error; + const firstDrive = fixture.harness.drive({ operationId: accepted.value.operationId }); + for (const expected of ["runtime.dispatch", "run.generation_ready"] as const) { + await waitForAction(fixture.harness); + expect((await fixture.harness.executeAction())?.kind).toBe(expected); + } + await waitForAction(fixture.harness); + expect(await fixture.harness.peekAction()).toMatchObject({ kind: "assistant.intent" }); + await fixture.harness.close(); + await expect(firstDrive).rejects.toMatchObject({ name: "HarnessClosed" }); + fixtures.splice(fixtures.indexOf(fixture), 1); + + const reopenedSession = await fixture.repo.open(fixture.session.metadata); + const models = createModels(); + models.setProvider(fixture.faux.provider); + const created = await AgentHarness.create({ + session: reopenedSession, + models, + model: fixture.faux.getModel(), + activeToolNames: [], + }); + const reopened: Fixture = { + harness: created.harness, + session: reopenedSession, + repo: fixture.repo, + faux: fixture.faux, + models, + }; + fixtures.push(reopened); + expect(await reopened.harness.drive({ operationId: accepted.value.operationId })).toMatchObject({ + ok: true, + value: { kind: "settled", outcome: { kind: "completed" } }, + }); + expect(fixture.faux.state.callCount).toBe(1); + }); + + it("settles adapter-reported errors durably as failed runs", async () => { + const { harness, faux, session } = await createFixture(); + faux.setResponses([fauxAssistantMessage([], { stopReason: "error", errorMessage: "adapter failed" })]); + const result = await harness.prompt("fail"); + expect(result).toMatchObject({ + ok: true, + value: { kind: "failed", error: { code: "assistant_error", message: "adapter failed" } }, + }); + expect(await session.getRegister("lane.lastResult", "main")).toMatchObject({ + value: { outcome: "failed", error: { message: "adapter failed" } }, + }); + }); + + it("reopens an accepted checkpoint without activating it and resumes on matching drive", async () => { + const fixture = await createFixture(); + fixture.faux.setResponses([fauxAssistantMessage("resumed")]); + const accepted = await fixture.harness.accept({ kind: "prompt", prompt: "persist" }); + if (!accepted.ok) throw accepted.error; + const metadata = fixture.session.metadata; + await fixture.harness.close(); + fixtures.splice(fixtures.indexOf(fixture), 1); + + const reopenedSession = await fixture.repo.open(metadata); + const models = createModels(); + models.setProvider(fixture.faux.provider); + const created = await AgentHarness.create({ + session: reopenedSession, + models, + model: fixture.faux.getModel(), + activeToolNames: [], + }); + const reopened: Fixture = { + harness: created.harness, + session: reopenedSession, + repo: fixture.repo, + faux: fixture.faux, + models, + }; + fixtures.push(reopened); + expect(created.suspended).toMatchObject([{ operationId: accepted.value.operationId, reason: "crash" }]); + expect(fixture.faux.state.callCount).toBe(0); + const resumeEvents: HarnessEvent[] = []; + reopened.harness.events.on("run_resume", (event) => { + resumeEvents.push(event); + }); + const driven = await reopened.harness.drive({ operationId: accepted.value.operationId }); + expect(driven).toMatchObject({ ok: true, value: { kind: "settled" } }); + expect(resumeEvents).toMatchObject([{ type: "run_resume", recovery: true }]); + }); +}); diff --git a/packages/agent/test/harness/agent-harness-runtime.test.ts b/packages/agent/test/harness/agent-harness-runtime.test.ts index d848a0714..355027fa0 100644 --- a/packages/agent/test/harness/agent-harness-runtime.test.ts +++ b/packages/agent/test/harness/agent-harness-runtime.test.ts @@ -199,16 +199,15 @@ describe("AgentHarness runtime shell", () => { expect((await harness.inspectExecution()).current?.status).toBe("running"); const second = harness.drive({ operationId: current.operationId, deadline: 0 }); await new Promise((resolve) => setTimeout(resolve, 0)); - await harness.executeAction(); + await harness.close(); const [left, right] = await Promise.allSettled([first, second]); expect(left.status).toBe("rejected"); expect(right.status).toBe("rejected"); if (left.status === "rejected" && right.status === "rejected") { expect(left.reason).toBe(right.reason); - expect(left.reason).toMatchObject({ name: "RuntimeSliceNotImplemented" }); + expect(left.reason).toBeInstanceOf(HarnessClosed); } - expect((await harness.inspectExecution()).current?.status).toBe("suspended"); }); it("faults inspection when process and durable operation ownership disagree", async () => { diff --git a/packages/agent/test/harness/restore.test.ts b/packages/agent/test/harness/restore.test.ts index ea9bb7a71..5c3657f25 100644 --- a/packages/agent/test/harness/restore.test.ts +++ b/packages/agent/test/harness/restore.test.ts @@ -112,6 +112,116 @@ describe("restoreLane base validation", () => { await expect(restoreLane(session, "main")).rejects.toThrow(`missing entry ${missingId}`); }); + it("rejects missing pending-next-run payloads", async () => { + const session = await createConfiguredSession(); + const pendingId = session.idGenerator.next(); + await session.mutate("main", (mutator) => + mutator.commit({ + writes: [ + { + kind: "register", + op: "set", + namespace: "lane.state", + key: "main", + value: { currentOperationId: null, pendingNextRun: [pendingId] }, + }, + ], + }), + ); + await expect(restoreLane(session, "main")).rejects.toThrow(`missing pending entry ${pendingId}`); + }); + + it("rejects invalid R2 checkpoint and generation relationships", async () => { + const session = await createConfiguredSession(); + const { operationId, promptEntryId, state } = await installRun(session); + if (state.kind !== "run") throw new Error("expected run state"); + const writeState = (value: OperationState) => + session.mutate("main", (mutator) => + mutator.commit({ + writes: [{ kind: "register", op: "set", namespace: "op.state", key: operationId, value }], + }), + ); + + await writeState({ + ...state, + latestAssistantEntryId: null, + phase: { + kind: "checkpoint", + continuation: { kind: "may_finish", includeFinalAssistant: true }, + triggerEntryId: promptEntryId, + }, + }); + await expect(restoreLane(session, "main")).rejects.toThrow(/finish checkpoint has no latest assistant/); + + await writeState({ + ...state, + latestAssistantEntryId: promptEntryId, + phase: { + kind: "checkpoint", + continuation: { kind: "may_finish", includeFinalAssistant: true }, + triggerEntryId: promptEntryId, + }, + }); + await expect(restoreLane(session, "main")).rejects.toThrow(`entry ${promptEntryId} is not an assistant message`); + + const responseEntryId = session.idGenerator.next(); + await writeState({ + ...state, + phase: { + kind: "assistant", + generation: { + status: "effect_pending", + context: { + stepId: session.idGenerator.next(), + triggerEntryId: promptEntryId, + configuration: { + model: { provider: "test", modelId: "model" }, + thinkingLevel: "off", + activeToolNames: [], + }, + streamOptions: {}, + retryPolicy: { maxAttempts: 1, baseDelayMs: 0 }, + overflowRecoveryUsed: false, + }, + attempt: 1, + responseEntryId, + usageId: responseEntryId, + intendedOutputLimit: 1, + contextWindow: 1, + }, + }, + }); + await expect(restoreLane(session, "main")).rejects.toThrow(/response and usage ids collide/); + + await writeState({ + ...state, + phase: { + kind: "assistant", + generation: { + status: "effect_pending", + context: { + stepId: session.idGenerator.next(), + triggerEntryId: promptEntryId, + configuration: { + model: { provider: "test", modelId: "model" }, + thinkingLevel: "off", + activeToolNames: [], + }, + streamOptions: {}, + retryPolicy: { maxAttempts: 1, baseDelayMs: 0 }, + overflowRecoveryUsed: false, + }, + attempt: 1, + responseEntryId: promptEntryId, + usageId: session.idGenerator.next(), + intendedOutputLimit: 1, + contextWindow: 1, + }, + }, + }); + await expect(restoreLane(session, "main")).rejects.toThrow(`reserved entry ${promptEntryId} already exists`); + }); + it("rejects missing operation registers and invalid base references", async () => { const session = await createConfiguredSession(); const { operationId, operation, state } = await installRun(session); diff --git a/packages/agent/test/harness/scratch.ts b/packages/agent/test/harness/scratch.ts new file mode 100644 index 000000000..4f7a6668e --- /dev/null +++ b/packages/agent/test/harness/scratch.ts @@ -0,0 +1,48 @@ +// Minimal real-provider AgentHarness example. +// Run from packages/agent: node test/harness/scratch.ts +// Requires ANTHROPIC_API_KEY. + +import { createModels } from "../../../ai/src/models.ts"; +import { anthropicProvider } from "../../../ai/src/providers/anthropic.ts"; +import { AgentHarness, MemorySessionRepo } from "../../src/index.ts"; + +const models = createModels(); +models.setProvider(anthropicProvider()); + +const model = models.getModel("anthropic", "claude-haiku-4-5"); +if (!model) throw new Error("model not found"); + +const auth = await models.getAuth(model.provider); +console.log(`model: ${model.provider}/${model.id}`); +console.log(`auth: ${auth ? `configured via ${auth.source}` : "not configured"}\n`); +if (!auth) process.exit(1); + +const repo = new MemorySessionRepo(); +const session = await repo.create({}); +const { harness } = await AgentHarness.create({ + session, + models, + model, + activeToolNames: [], + systemPrompt: "You are terse.", +}); + +harness.events.on("message_update", ({ event }) => { + if (event.type === "text_delta") process.stdout.write(event.delta); +}); + +try { + process.stdout.write("assistant: "); + const result = await harness.prompt("Spit out one sentence"); + process.stdout.write("\n"); + + if (!result.ok) throw result.error; + console.log(`result: ${result.value.kind}`); + if ("finalMessage" in result.value && result.value.finalMessage) { + console.log(`stop: ${result.value.finalMessage.stopReason}`); + console.log(`cost: $${result.value.finalMessage.usage.cost.total.toFixed(6)}`); + } +} finally { + await harness.close(); + await repo.close(); +} diff --git a/packages/agent/test/utils/wait-for-tick.ts b/packages/agent/test/utils/wait-for-tick.ts new file mode 100644 index 000000000..72ad2876a --- /dev/null +++ b/packages/agent/test/utils/wait-for-tick.ts @@ -0,0 +1,3 @@ +export function waitForTick(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +}