diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index 3f806a0ac..8a56f6abe 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -53,14 +53,6 @@ export class OperationMismatch extends TaggedError("OperationMismatch")<{ lastOperationId?: string; message: string; }> {} -export interface MissingIdentityInfo { - tools: string[]; - model?: string; -} - -export class MissingIdentities extends TaggedError("MissingIdentities")< - { lane: string; message: string } & MissingIdentityInfo -> {} export class NoActiveRun extends TaggedError("NoActiveRun")<{ lane: string; message: string }> {} export class NoActiveOperation extends TaggedError("NoActiveOperation")<{ lane: string; message: string }> {} export class NothingToResume extends TaggedError("NothingToResume")<{ lane: string; message: string }> {} @@ -107,11 +99,10 @@ export type OptionalFinalAssistant = | { finalEntryId: string; finalMessage: AssistantMessage } | { finalEntryId?: never; finalMessage?: never }; -export type MissingIdentitySuspension = { - kind: "suspended"; - reason: "missing_identities"; - missing: MissingIdentityInfo; -}; +export interface ActionRequired { + kind: "action_required"; + action: ActionInfo; +} export type RunOutcome = | ({ kind: "completed"; leafId: string } & OptionalFinalAssistant) @@ -124,13 +115,13 @@ export type RunOutcome = finalEntryId: string; deferred: DeferredHandle; } - | (MissingIdentitySuspension & { leafId: string }); + | ActionRequired; export type CompactionOutcome = | { kind: "completed"; leafId: string; entry: CompactionEntry } | { kind: "declined" | "aborted"; leafId: string } | { kind: "failed"; leafId: string; error: OperationError } - | (MissingIdentitySuspension & { leafId: string }); + | ActionRequired; export type NavigationOutcome = | { @@ -141,26 +132,24 @@ export type NavigationOutcome = } | { kind: "declined" | "aborted"; leafId: string | null } | { kind: "failed"; leafId: string | null; error: OperationError } - | (MissingIdentitySuspension & { leafId: string | null }); + | ActionRequired; -export type ResumeOutcome = - | ({ operation: "run"; runId: string } & RunOutcome) - | ({ operation: "compaction"; runId: string } & CompactionOutcome) - | ({ operation: "navigation"; runId: string } & NavigationOutcome); +export type RunOperationOutcome = { operation: "run"; runId: string } & RunOutcome; +export type CompactionOperationOutcome = { operation: "compaction"; runId: string } & CompactionOutcome; +export type NavigationOperationOutcome = { operation: "navigation"; runId: string } & NavigationOutcome; +export type ResumeOutcome = RunOperationOutcome | CompactionOperationOutcome | NavigationOperationOutcome; +export type SettledResumeOutcome = Exclude; export type RunResult = Result< - { runId: string } & RunOutcome, - LaneBusy | MissingIdentities | InvalidMessage | UnknownSkill | UnknownTemplate | Closed ->; -export type CompactionResult = Result< - { runId: string } & CompactionOutcome, - LaneBusy | MissingIdentities | NothingToCompact | Closed + RunOperationOutcome, + LaneBusy | InvalidMessage | UnknownSkill | UnknownTemplate | Closed >; +export type CompactionResult = Result; export type NavigationResult = Result< - { runId: string } & NavigationOutcome, - LaneBusy | MissingIdentities | InvalidNavigation | UnknownTarget | Closed + NavigationOperationOutcome, + LaneBusy | InvalidNavigation | UnknownTarget | Closed >; -export type ResumeResult = Result; +export type ResumeResult = Result; export type QueueResult = Result<{ entryId: string }, NoActiveRun | InvalidMessage | Closed>; export type NextRunResult = Result<{ entryId: string }, InvalidMessage | Closed>; export type CancelQueuedResult = Result<{ kind: "cancelled" | "already_consumed" | "not_found" }, Closed>; @@ -193,7 +182,6 @@ export interface OperationAdmission { export type OperationAdmissionError = | LaneBusy - | MissingIdentities | InvalidMessage | UnknownSkill | UnknownTemplate @@ -210,37 +198,40 @@ export interface DriveOptions { pollDeferred?: boolean; } +export interface ModelIdentity { + provider: string; + modelId: string; +} + +export type OperationStatus = "running" | "open" | "aborting"; + export interface CurrentOperationInfo { id: string; kind: "run" | "compaction" | "navigation"; - status: "running" | "suspended" | "aborting"; startedAt: number; - suspended?: SuspendedOperation; + status: OperationStatus; + capturedModel?: ModelIdentity; } export interface LaneExecutionInfo { lane: string; leafId: string | null; + configuredModel: ModelIdentity; current: CurrentOperationInfo | null; lastResult?: LaneLastResult; } export type TerminalOperationOutcome = - | ({ operation: "run"; runId: string } & Exclude) - | ({ operation: "compaction"; runId: string } & Exclude) - | ({ operation: "navigation"; runId: string } & Exclude); + | ({ operation: "run"; runId: string } & Exclude) + | ({ operation: "compaction"; runId: string } & Exclude) + | ({ operation: "navigation"; runId: string } & Exclude); export type DriveOutcome = | { kind: "settled"; operationId: string; outcome: TerminalOperationOutcome } | { kind: "waiting"; operationId: string; reason: "retry"; notBefore: number } | { kind: "waiting"; operationId: string; reason: "deferred"; deferred: DeferredHandle } - | { - kind: "waiting"; - operationId: string; - reason: "missing_identities"; - missing: MissingIdentityInfo; - } - | { kind: "yielded"; operationId: string }; + | { kind: "yielded"; operationId: string } + | { kind: "action_required"; operationId: string; action: ActionInfo }; export type DriveResult = Result; export type AbortRequestResult = Result< @@ -268,33 +259,16 @@ export interface WatchHandle { export interface LaneInfo { name: string; leafId: string | null; - operation: null | { - id: string; - kind: "run" | "compaction" | "navigation"; - status: "running" | "suspended" | "aborting"; - }; + operation: CurrentOperationInfo | null; } -export type SuspendedOperation = { +export interface OpenOperation { lane: string; operationId: string; kind: "run" | "compaction" | "navigation"; startedAt: number; - prompt?: AgentMessage[]; - aborting?: { steer: AgentMessage[]; followUp: AgentMessage[] }; -} & ( - | { reason: "deferred"; deferred: DeferredHandle; missing?: never } - | { - reason: "missing_identities"; - missing: MissingIdentityInfo; - deferred?: never; - } - | { - reason: "crash"; - deferred?: DeferredHandle; - missing?: MissingIdentityInfo; - } -); + aborting?: true; +} export interface QueuedItem { entryId: string; @@ -305,12 +279,16 @@ export interface LaneSnapshot { lane: string; transcript: Entry[]; leafId: string | null; + lastResult?: LaneLastResult; operation: null | { id: string; kind: "run" | "compaction" | "navigation"; - status: "running" | "suspended" | "aborting"; startedAt: number; - suspended?: SuspendedOperation; + status: OperationStatus; + action?: ActionInfo; + retry?: { attempt: number; maxAttempts: number; nextAttemptAt: number }; + deferred?: { handle: DeferredHandle; poll: number }; + drained?: { steer: QueuedItem[]; followUp: QueuedItem[] }; streamingMessage?: AssistantMessage; runningTools: { toolCallId: string; @@ -318,7 +296,6 @@ export interface LaneSnapshot { args: unknown; partialResult?: AgentToolResult; }[]; - retry?: { attempt: number; maxAttempts: number; nextAttemptAt: number }; }; queues: { steer: QueuedItem[]; followUp: QueuedItem[]; nextRun: QueuedItem[] }; pendingWrites: { @@ -332,7 +309,7 @@ export interface LaneSnapshot { } export interface SessionSnapshot { - lanes: (LaneInfo & { suspended?: SuspendedOperation })[]; + lanes: LaneInfo[]; faulted: boolean; } @@ -340,24 +317,6 @@ export type HarnessEventPayload = | { type: "run_start"; runId: string } | { type: "run_resume"; runId: string } | { type: "run_suspend"; runId: string; reason: "deferred"; deferred: DeferredHandle } - | { - type: "run_suspend"; - runId: string; - reason: "missing_identities"; - missing: MissingIdentityInfo; - } - | { - type: "compaction_suspend"; - runId: string; - reason: "missing_identities"; - missing: MissingIdentityInfo; - } - | { - type: "navigation_suspend"; - runId: string; - reason: "missing_identities"; - missing: MissingIdentityInfo; - } | { type: "run_abort"; runId: string; steer: AgentMessage[]; followUp: AgentMessage[] } | ({ type: "run_end"; runId: string; leafId: string | null } & ( | ({ outcome: "completed" | "aborted" } & OptionalFinalAssistant) @@ -660,8 +619,8 @@ export interface AgentLane { waitForIdle(context: Context): Promise; runWhenIdle(callback: (context: Context) => void | Promise, context: Context): Promise; peekAction(context: Context): Promise; - executeAction(context: Context): Promise; - runToCompletion(context: Context): Promise; + executeAction(context: Context): Promise; + runToCompletion(context: Context): Promise>; getModel(context: Context): Promise | undefined>; setModel(model: Model, context: Context): Promise; getThinkingLevel(context: Context): Promise; @@ -700,7 +659,7 @@ export interface AgentHarnessConstructor { create( options: AgentHarnessOptions, context: Context, - ): Promise<{ harness: AgentHarness; suspended: SuspendedOperation[] }>; + ): Promise<{ harness: AgentHarness; open: OpenOperation[] }>; } /** Runtime constructor for attaching the durable harness to one open session. */ diff --git a/packages/agent/src/harness/config.ts b/packages/agent/src/harness/config.ts index fe0e19ec1..4c8b6aeec 100644 --- a/packages/agent/src/harness/config.ts +++ b/packages/agent/src/harness/config.ts @@ -1,7 +1,5 @@ -import type { Models, RetryPolicy } from "@earendil-works/pi-ai"; -import type { MissingIdentityInfo } from "./agent-harness.ts"; +import type { RetryPolicy } from "@earendil-works/pi-ai"; import type { CompactionSettings } from "./compaction/compaction.ts"; -import type { LaneConfiguration } from "./session/types.ts"; export const DEFAULT_RETRY_POLICY: RetryPolicy = { enabled: true, maxRetries: 3, baseDelayMs: 1_000 }; @@ -25,27 +23,6 @@ export function validateRetryPolicy(policy: RetryPolicy): void { } } -export function hasMissingIdentities(info: MissingIdentityInfo): boolean { - return info.tools.length !== 0 || info.model !== undefined; -} - -export function missingIdentities( - models: Models, - configuration: LaneConfiguration, - tools: readonly { name: string }[], -): MissingIdentityInfo { - const model = models.getModel(configuration.model.provider, configuration.model.modelId); - return { - tools: missingToolIdentities(configuration, tools), - ...(model === undefined ? { model: `${configuration.model.provider}/${configuration.model.modelId}` } : {}), - }; -} - -export function missingToolIdentities(configuration: LaneConfiguration, tools: readonly { name: string }[]): string[] { - const availableTools = new Set(tools.map((tool) => tool.name)); - return configuration.activeToolNames.filter((name) => !availableTools.has(name)); -} - export function validateCompactionSettings(settings: CompactionSettings): void { if ( !Number.isSafeInteger(settings.reserveTokens) || diff --git a/packages/agent/src/harness/events.ts b/packages/agent/src/harness/events.ts index 45a151bb8..f95af5331 100644 --- a/packages/agent/src/harness/events.ts +++ b/packages/agent/src/harness/events.ts @@ -3,10 +3,15 @@ import type { Context } from "./context.ts"; type UntypedEventListener = (event: HarnessEvent, context: Context) => void | Promise; +export interface HarnessEventDelivery { + start(): Promise; +} + /** Passive harness event bus with isolated handler failures. */ export class HarnessEventBus implements Events { private readonly listeners = new Map>(); private readonly watchListeners = new Set(); + private readonly pendingStarts = new Set<() => void>(); private deliveryTail: Promise = Promise.resolve(); private closedError: Error | undefined; @@ -27,10 +32,43 @@ export class HarnessEventBus implements Events { } emit(event: HarnessEvent, context: Context): Promise { - if (this.closedError !== undefined) return Promise.resolve(); - const delivery = this.deliveryTail.then(() => this.deliver(event, true, context)); + return this.enqueue([event], context).start(); + } + + /** + * Bind and reserve one contiguous event batch without permitting listener execution until `start()`. + * Commit owners call this before releasing their serialization line, then call `start()` only after release. + * Prefer `Lane.command`; a dropped or early-started delivery breaks snapshot/event boundary semantics. + */ + enqueue(events: readonly HarnessEvent[], context: Context): HarnessEventDelivery { + if (this.closedError !== undefined || events.length === 0) return { start: () => Promise.resolve() }; + const bound = events.map((event) => { + const payload = structuredClone(event); + return { payload, recipients: this.snapshotRecipients(payload) }; + }); + let release: (() => void) | undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + const delivery = this.deliveryTail.then(async () => { + await gate; + for (const { payload, recipients } of bound) await this.deliver(payload, recipients, true, context); + }); this.deliveryTail = delivery.catch(() => {}); - return delivery; + let started = false; + const start = (): void => { + if (started) return; + started = true; + this.pendingStarts.delete(start); + release?.(); + }; + this.pendingStarts.add(start); + return { + start: () => { + start(); + return delivery; + }, + }; } watch(snapshot: T, filter: (event: HarnessEvent) => boolean, _context: Context): WatchHandle { @@ -56,6 +94,7 @@ export class HarnessEventBus implements Events { close(error: Error): void { this.closedError ??= error; + for (const start of [...this.pendingStarts]) start(); void this.deliveryTail.finally(() => { this.listeners.clear(); this.watchListeners.clear(); @@ -90,9 +129,17 @@ export class HarnessEventBus implements Events { return watcher; } - private async deliver(event: HarnessEvent, reportErrors: boolean, context: Context): Promise { - const listeners = [...(this.listeners.get(event.type) ?? []), ...this.watchListeners]; - for (const listener of listeners) { + private snapshotRecipients(event: HarnessEvent): UntypedEventListener[] { + return [...(this.listeners.get(event.type) ?? []), ...this.watchListeners]; + } + + private async deliver( + event: HarnessEvent, + recipients: readonly UntypedEventListener[], + reportErrors: boolean, + context: Context, + ): Promise { + for (const listener of recipients) { try { await listener(structuredClone(event), context); } catch (error) { @@ -107,7 +154,7 @@ export class HarnessEventBus implements Events { ...(normalized.stack === undefined ? {} : { stack: normalized.stack }), ...(lane === undefined ? {} : { lane }), }; - await this.deliver(handlerError, false, context); + await this.deliver(handlerError, this.snapshotRecipients(handlerError), false, context); } } } diff --git a/packages/agent/src/harness/runtime2/harness.ts b/packages/agent/src/harness/runtime2/harness.ts index 4eb4bd24d..c95b0bc0d 100644 --- a/packages/agent/src/harness/runtime2/harness.ts +++ b/packages/agent/src/harness/runtime2/harness.ts @@ -1,4 +1,4 @@ -import type { Models, RetryPolicy } from "@earendil-works/pi-ai"; +import type { RetryPolicy } from "@earendil-works/pi-ai"; import type { QueueMode } from "../../types.ts"; import type { AgentHarness, @@ -7,26 +7,19 @@ import type { CreateLaneResult, GlobalConfigEventPayload, LaneInfo, + OpenOperation, Resources, - SuspendedOperation, } from "../agent-harness.ts"; import { Closed, HarnessClosed, HarnessFault, InvalidLane, LaneExists, UnknownTarget } from "../agent-harness.ts"; import { type CompactionSettings, DEFAULT_COMPACTION_SETTINGS } from "../compaction/compaction.ts"; -import { - DEFAULT_RETRY_POLICY, - hasMissingIdentities, - missingIdentities, - missingToolIdentities, - validateCompactionSettings, - validateRetryPolicy, - validateToolNames, -} from "../config.ts"; +import { DEFAULT_RETRY_POLICY, validateCompactionSettings, validateRetryPolicy, validateToolNames } from "../config.ts"; import type { Context } from "../context.ts"; -import { HarnessEventBus } from "../events.ts"; +import { HarnessEventBus, type HarnessEventDelivery } from "../events.ts"; import { HookRegistry } from "../hooks.ts"; import { convertToLlm } from "../messages.ts"; import { Result } from "../result.ts"; import { + createLaneWithMutator, SessionInvalidLaneError, SessionInvariantError, SessionLaneExistsError, @@ -47,43 +40,42 @@ export class Harness extends Lane implement readonly events: HarnessEventBus; readonly seed: LaneConfiguration; readonly lanesByName = new Map(); - private config: Config; + private readonly configStore: { value: Config }; closePromise: Promise | undefined; faultError: HarnessFault | undefined; - constructor( - options: AgentHarnessOptions, - seed: LaneConfiguration, - restored: Map, - suspensionsByLane: Map, - ) { + constructor(options: AgentHarnessOptions, seed: LaneConfiguration, restored: Map) { const main = restored.get("main"); if (main === undefined) throw new SessionInvariantError("Session is missing main lane"); + const configStore: { value: Config } = { + value: { + tools: options.tools ?? [], + resources: options.resources ?? {}, + streamOptions: options.streamOptions ?? {}, + retryPolicy: options.retry ?? DEFAULT_RETRY_POLICY, + compaction: options.compaction ?? DEFAULT_COMPACTION_SETTINGS, + steeringMode: options.steeringMode ?? "all", + followUpMode: options.followUpMode ?? "all", + toolExecution: options.toolExecution ?? "parallel", + drive: options.drive ?? "automatic", + toolContext: options.toolContext, + systemPrompt: options.systemPrompt, + toProviderMessages: options.toProviderMessages ?? ((messages) => convertToLlm(messages)), + entryProjectors: options.entryProjectors ?? {}, + }, + }; super( "main", options.session, options.models, main, (cause, context) => this.fault(cause, context), - (event, context) => this.events.emit(event, context), - suspensionsByLane.get("main"), + (events, context) => this.events.enqueue(events, context), + (snapshot, filter, context) => this.events.watch(snapshot, filter, context), + () => configStore.value, ); this.seed = seed; - this.config = { - tools: options.tools ?? [], - resources: options.resources ?? {}, - streamOptions: options.streamOptions ?? {}, - retryPolicy: options.retry ?? DEFAULT_RETRY_POLICY, - compaction: options.compaction ?? DEFAULT_COMPACTION_SETTINGS, - steeringMode: options.steeringMode ?? "all", - followUpMode: options.followUpMode ?? "all", - toolExecution: options.toolExecution ?? "parallel", - drive: options.drive ?? "automatic", - toolContext: options.toolContext, - systemPrompt: options.systemPrompt, - toProviderMessages: options.toProviderMessages ?? ((messages) => convertToLlm(messages)), - entryProjectors: options.entryProjectors ?? {}, - }; + this.configStore = configStore; this.events = new HarnessEventBus(); this.hooks = new HookRegistry((error, hook, lane, context) => this.events.emit( @@ -100,7 +92,7 @@ export class Harness extends Lane implement ); this.lanesByName.set("main", this); for (const [name, state] of restored) { - if (name !== "main") this.lanesByName.set(name, this.buildLane(name, state, suspensionsByLane.get(name))); + if (name !== "main") this.lanesByName.set(name, this.buildLane(name, state)); } } @@ -117,14 +109,7 @@ export class Harness extends Lane implement return executions.map((execution) => ({ name: execution.lane, leafId: execution.leafId, - operation: - execution.current === null - ? null - : { - id: execution.current.id, - kind: execution.current.kind, - status: execution.current.status, - }, + operation: execution.current, })); } @@ -140,11 +125,21 @@ export class Harness extends Lane implement operation: null, }; try { - await this.session.createLane(name, at, this.seed, context); - const lane = this.buildLane(name, state); - if (this.closedError !== undefined) lane.seal(this.closedError); - this.lanesByName.set(name, lane); - await this.events.emit({ type: "lane_created", lane: name, at }, context); + let lane: Lane | undefined; + let delivery: HarnessEventDelivery | undefined; + await this.session.mutate( + name, + async (mutator) => { + await createLaneWithMutator(mutator, name, at, this.seed, context); + lane = this.buildLane(name, state); + if (this.closedError !== undefined) lane.seal(this.closedError); + this.lanesByName.set(name, lane); + delivery = this.events.enqueue([{ type: "lane_created", lane: name, at }], context); + }, + context, + ); + await delivery?.start(); + if (lane === undefined) throw new SessionInvariantError(`Lane ${JSON.stringify(name)} was not published`); return Result.ok(lane); } catch (error) { // Session has no Result layer, so its expected validation failures are thrown. Translate only those failures @@ -225,15 +220,16 @@ export class Harness extends Lane implement throw new SliceNotImplemented("watchSession"); } - private buildLane(name: string, state: LaneState, suspension?: SuspendedOperation): Lane { + private buildLane(name: string, state: LaneState): Lane { return new Lane( name, this.session, this.models, state, (cause, context) => this.fault(cause, context), - (event, context) => this.events.emit(event, context), - suspension, + (events, context) => this.events.enqueue(events, context), + (snapshot, filter, context) => this.events.watch(snapshot, filter, context), + () => this.configStore.value, ); } @@ -242,7 +238,7 @@ export class Harness extends Lane implement _context: Context, ): Promise[TKey]> { this.assertOpen(); - return this.config[key]; + return this.configStore.value[key]; } private async setConfig>( @@ -252,7 +248,7 @@ export class Harness extends Lane implement context: Context, ): Promise { this.assertOpen(); - this.config = { ...this.config, [key]: value }; + this.configStore.value = { ...this.configStore.value, [key]: value }; await this.events.emit({ type: "config_update", property }, context); } @@ -303,7 +299,7 @@ export class Harness extends Lane implement export async function createAgentHarness( options: AgentHarnessOptions, context: Context, -): Promise<{ harness: AgentHarness; suspended: SuspendedOperation[] }> { +): Promise<{ harness: AgentHarness; open: OpenOperation[] }> { const tools = options.tools ?? []; validateToolNames(tools); validateRetryPolicy(options.retry ?? DEFAULT_RETRY_POLICY); @@ -316,76 +312,26 @@ export async function createAgentHarness [descriptor.lane, descriptor])); - return { harness: new Harness(options, seed, restored, suspensionsByLane), suspended }; + const open = [...restored].flatMap(([lane, state]): OpenOperation[] => { + const operation = state.operation; + return operation === null + ? [] + : [ + { + lane, + operationId: operation.meta.operationId, + kind: operation.meta.intent.kind, + startedAt: operation.meta.startedAt, + ...(operation.state.control.status === "cancel_requested" ? { aborting: true as const } : {}), + }, + ]; + }); + return { harness: new Harness(options, seed, restored), open }; } catch (error) { throw new HarnessFault("AgentHarness storage or invariant fault", error); } } -async function describeSuspension( - session: Session, - models: Models, - tools: readonly { name: string }[], - lane: string, - state: LaneState, - context: Context, -): Promise { - const operation = state.operation; - if (operation === null) throw new SessionInvariantError(`Lane ${JSON.stringify(lane)} is not suspended`); - const promptIds = operation.meta.intent.kind === "run" ? operation.meta.intent.promptEntryIds : []; - const deferredSourceId = - operation.state.kind === "run" && operation.state.phase.kind === "deferred" - ? operation.state.phase.deferred.sourceEntryId - : undefined; - const entries = await session.getEntries( - [...new Set([...promptIds, ...(deferredSourceId === undefined ? [] : [deferredSourceId])])], - context, - ); - const prompt = promptIds.map((id) => { - const entry = entries.get(id); - if (entry?.type !== "message") throw new SessionInvariantError(`Prompt entry ${id} is missing`); - return entry.message; - }); - const base = { - lane, - operationId: operation.meta.operationId, - kind: operation.meta.intent.kind, - startedAt: operation.meta.startedAt, - ...(operation.meta.intent.kind === "run" ? { prompt } : {}), - }; - if (deferredSourceId !== undefined) { - const entry = entries.get(deferredSourceId); - if (entry?.type !== "message" || entry.message.role !== "assistant" || entry.message.deferred === undefined) { - throw new SessionInvariantError("Deferred suspension source is invalid"); - } - return { ...base, reason: "deferred", deferred: entry.message.deferred }; - } - if ( - operation.state.kind === "run" && - operation.state.phase.kind === "assistant" && - operation.state.phase.generation.status === "ready" - ) { - const missing = missingIdentities(models, operation.state.phase.generation.context.configuration, tools); - if (hasMissingIdentities(missing)) { - return { ...base, reason: "crash", missing }; - } - } - if (operation.state.kind === "run" && operation.state.phase.kind === "tools") { - const missing = missingToolIdentities(operation.state.phase.batch.configuration, tools); - if (missing.length !== 0) { - return { ...base, reason: "crash", missing: { tools: missing } }; - } - } - return { ...base, reason: "crash" }; -} - async function seedMain(session: Session, seed: LaneConfiguration, context: Context): Promise { await session.mutate( "main", diff --git a/packages/agent/src/harness/runtime2/lane.ts b/packages/agent/src/harness/runtime2/lane.ts index 4084caa36..9d2095db7 100644 --- a/packages/agent/src/harness/runtime2/lane.ts +++ b/packages/agent/src/harness/runtime2/lane.ts @@ -1,36 +1,68 @@ -import type { Api, ImageContent, Model, Models } from "@earendil-works/pi-ai"; +import { + type Api, + type ImageContent, + type Model, + type Models, + reduceAssistantMessageFrames, +} from "@earendil-works/pi-ai"; import type { AgentMessage, ThinkingLevel } from "../../types.ts"; -import type { - AgentLane, - HarnessEvent, - LaneConfigEventPayload, - LaneExecutionInfo, - SuspendedOperation, +import { + type AgentLane, + Closed, + HarnessClosed, + type HarnessEvent, + InvalidMessage, + LaneBusy, + type LaneConfigEventPayload, + type LaneExecutionInfo, + type LaneSnapshot, + type ModelIdentity, + type OperationAdmissionResult, + type OperationRequest, + type QueuedItem, + UnknownSkill, + UnknownTemplate, + type WatchHandle, } from "../agent-harness.ts"; import type { Context } from "../context.ts"; +import type { HarnessEventDelivery } from "../events.ts"; +import { formatPromptTemplateInvocation } from "../prompt-templates.ts"; +import { Result } from "../result.ts"; import { insertEntry } from "../session/commit.ts"; -import { SessionPendingAssistantMessageError } from "../session/session.ts"; +import { SessionInvariantError, SessionPendingAssistantMessageError } from "../session/session.ts"; import type { BranchScan, CommitResult, Entry, LaneLastResult, + Operation, PendingEntry, Session, SessionReader, SessionTree, + StructuralDecision, Write, } from "../session/types.ts"; import { + deleteValue, + entryLabel, laneConfig, laneLeaf, + laneState as laneStateValue, + operationMeta as operationMetaValue, operationState as operationStateValue, + operationToolArgs, + pendingAssistantFrames, pendingEntry, + pendingToolOutput, + sessionName, setValue, } from "../session/values.ts"; -import { type LaneState, SliceNotImplemented } from "./types.ts"; +import { formatSkillInvocation } from "../skills.ts"; +import { type AcceptanceConfig, type LaneState, SliceNotImplemented } from "./types.ts"; -type EventHandler = (event: HarnessEvent, context: Context) => Promise; +type EventHandler = (events: readonly HarnessEvent[], context: Context) => HarnessEventDelivery; +type WatchHandler = (snapshot: T, filter: (event: HarnessEvent) => boolean, context: Context) => WatchHandle; type FaultHandler = (cause: unknown, context: Context) => Error; type Synchronous = TResult extends PromiseLike ? never : TResult; @@ -40,17 +72,46 @@ type LaneCommand = writes: Write[]; next: LaneState; materialize(commit: CommitResult): Synchronous; + events?(commit: CommitResult): HarnessEvent[]; } | { kind: "return"; result: TResult } | { kind: "reject"; error: Error }; -type LaneCommandOutcome = { kind: "return"; result: TResult } | { kind: "reject"; error: Error }; +type LaneCommandOutcome = + | { kind: "return"; result: TResult; delivery?: HarnessEventDelivery } + | { kind: "reject"; error: Error }; function isPromiseLike(value: unknown): value is PromiseLike { if (value === null || (typeof value !== "object" && typeof value !== "function")) return false; return "then" in value && typeof value.then === "function"; } +function structuralModel(structural: StructuralDecision): ModelIdentity | undefined { + return structural.status === "generating" ? structural.generation.context.configuration.model : undefined; +} + +function capturedModel(operation: Operation): ModelIdentity | undefined { + const { state } = operation; + if (state.kind === "compaction") return structuralModel(state.structural); + if (state.kind === "navigation") { + return state.summarize ? structuralModel(state.phase.structural) : undefined; + } + switch (state.phase.kind) { + case "assistant": + return state.phase.generation.context.configuration.model; + case "tools": + return state.phase.batch.configuration.model; + case "compaction": + return structuralModel(state.phase.structural); + case "deferred": + return state.phase.deferred.configuration.model; + case "starting": + case "checkpoint": + case "failure_drain": + return undefined; + } +} + /** Runtime2 implementation of one configured lane. */ export class Lane implements AgentLane { readonly name: string; @@ -60,7 +121,8 @@ export class Lane implements AgentLane { protected readonly models: Models; private readonly onFault: FaultHandler; private readonly onEvent: EventHandler; - private suspension: SuspendedOperation | undefined; + private readonly installWatch: WatchHandler; + private readonly getAcceptanceConfig: () => AcceptanceConfig; state: LaneState; closedError: Error | undefined; @@ -71,7 +133,8 @@ export class Lane implements AgentLane { state: LaneState, onFault: FaultHandler, onEvent: EventHandler, - suspension?: SuspendedOperation, + installWatch: WatchHandler, + getAcceptanceConfig: () => AcceptanceConfig, ) { this.session = session; this.models = models; @@ -91,7 +154,8 @@ export class Lane implements AgentLane { this.state = state; this.onFault = onFault; this.onEvent = onEvent; - this.suspension = suspension; + this.installWatch = installWatch; + this.getAcceptanceConfig = getAcceptanceConfig; } async getLeafId(_context: Context): Promise { @@ -146,14 +210,13 @@ export class Lane implements AgentLane { case "commit": { const commit = await mutator.commit(decision.writes, context); this.state = decision.next; - if (this.suspension?.operationId !== decision.next.operation?.meta.operationId) { - this.suspension = undefined; - } const result = decision.materialize(commit); if (isPromiseLike(result)) { throw new TypeError("Lane command materialize() must be synchronous"); } - return { kind: "return", result }; + const events = decision.events?.(commit) ?? []; + const delivery = events.length === 0 ? undefined : this.onEvent(events, context); + return { kind: "return", result, ...(delivery === undefined ? {} : { delivery }) }; } } } catch (error) { @@ -168,11 +231,213 @@ export class Lane implements AgentLane { throw error; } if (outcome.kind === "reject") throw outcome.error; + await outcome.delivery?.start(); return outcome.result; } - async accept(..._args: Parameters): Promise { - throw new SliceNotImplemented("accept"); + async accept(request: OperationRequest, context: Context): Promise { + if (this.closedError instanceof HarnessClosed) { + return Result.err(new Closed({ message: this.closedError.message })); + } + this.assertOpen(); + if (request.kind === "compaction" || request.kind === "navigation") { + throw new SliceNotImplemented(`accept(${request.kind})`); + } + + const startedAt = Date.now(); + const operationId = request.operationId ?? this.session.idGenerator.next(startedAt); + const acceptanceConfig = this.getAcceptanceConfig(); + let messages: AgentMessage[]; + switch (request.kind) { + case "prompt": + if (typeof request.prompt !== "string") { + messages = Array.isArray(request.prompt) ? request.prompt : [request.prompt]; + } else { + const images = request.images ?? []; + messages = + request.prompt.length === 0 && images.length === 0 + ? [] + : [ + { + role: "user", + content: [ + ...(request.prompt.length === 0 + ? [] + : [{ type: "text" as const, text: request.prompt }]), + ...images, + ], + timestamp: startedAt, + }, + ]; + } + break; + case "skill": { + const skill = acceptanceConfig.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: [{ type: "text", text: formatSkillInvocation(skill, request.additionalInstructions) }], + timestamp: startedAt, + }, + ]; + break; + } + case "prompt_template": { + const template = acceptanceConfig.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}` }), + ); + } + const content = formatPromptTemplateInvocation(template, request.args); + messages = + content.length === 0 + ? [] + : [{ role: "user", content: [{ type: "text", text: content }], timestamp: startedAt }]; + break; + } + } + + for (const message of messages) { + if (message.role === "assistant" && message.stopReason === "pending") { + return Result.err( + new InvalidMessage({ + lane: this.name, + reason: "pending_assistant", + message: "Cannot accept a pending assistant message", + }), + ); + } + } + const prompt = messages.map((message) => ({ id: this.session.idGenerator.next(startedAt), message })); + + return this.command(async (state, reader) => { + if (state.operation !== null) { + return { + kind: "return", + result: Result.err( + new LaneBusy({ + lane: this.name, + operationId: state.operation.meta.operationId, + operationKind: state.operation.meta.intent.kind, + message: `Lane ${JSON.stringify(this.name)} already has an active operation`, + }), + ), + }; + } + const capturedIds = [...state.pendingNextRun]; + const captured = await Promise.all( + capturedIds.map(async (id) => { + const stored = await reader.getValue(pendingEntry(id), context); + if (stored?.value.type !== "message") { + throw new SessionInvariantError(`Pending next-run entry ${id} is missing a message payload`); + } + if (stored.value.payload.role === "assistant" && stored.value.payload.stopReason === "pending") { + throw new SessionInvariantError(`Pending next-run entry ${id} contains a pending assistant message`); + } + return { id, message: stored.value.payload }; + }), + ); + const placed = [...captured, ...prompt]; + if (placed.length === 0) { + return { + kind: "return", + result: Result.err( + new InvalidMessage({ + lane: this.name, + reason: "empty", + message: "Acceptance must append at least one message", + }), + ), + }; + } + + let parentId = state.leafId; + const entryWrites = placed.map(({ id, message }) => { + const write = insertEntry({ id, parentId, type: "message", message }); + parentId = id; + return write; + }); + const meta = { + operationId, + lane: this.name, + sourceLeafId: state.leafId, + startedAt, + intent: { kind: "run" as const, promptEntryIds: prompt.map(({ id }) => id) }, + }; + const operationState = { + kind: "run" as const, + control: { status: "running" as const }, + settings: { + compaction: acceptanceConfig.compaction, + steeringMode: acceptanceConfig.steeringMode, + followUpMode: acceptanceConfig.followUpMode, + toolExecution: acceptanceConfig.toolExecution, + }, + phase: { kind: "starting" as const }, + inbox: { steer: [], followUp: [], writes: [] }, + latestAssistantEntryId: null, + }; + const next: LaneState = { + ...state, + leafId: parentId, + pendingNextRun: [], + operation: { meta, state: operationState }, + }; + return { + kind: "commit", + writes: [ + ...entryWrites, + ...capturedIds.map((id) => deleteValue(pendingEntry(id))), + setValue(laneLeaf(this.name), parentId), + setValue(operationMetaValue(operationId), meta), + setValue(operationStateValue(operationId), operationState), + setValue(laneStateValue(this.name), { currentOperationId: operationId, pendingNextRun: [] }), + ], + next, + materialize: () => Result.ok({ operationId, kind: "run", startedAt }), + events: (commit) => { + const events: HarnessEvent[] = [{ type: "run_start", runId: operationId, lane: this.name }]; + for (const [index, item] of placed.entries()) { + const parent = index === 0 ? state.leafId : placed[index - 1]!.id; + const entry: Entry = { + id: item.id, + parentId: parent, + type: "message", + message: item.message, + seq: commit.seqs[index]!, + timestamp: commit.timestamp, + }; + events.push( + { type: "message_start", runId: operationId, message: item.message, lane: this.name }, + { + type: "message_end", + runId: operationId, + message: item.message, + entryId: item.id, + lane: this.name, + }, + { type: "entry_added", entry, lane: this.name }, + ); + } + if (captured.length !== 0) { + events.push({ + type: "queue_update", + steer: [], + followUp: [], + nextRun: [], + lane: this.name, + }); + } + return events; + }, + }; + }, context); } async drive(..._args: Parameters): Promise { @@ -183,34 +448,34 @@ export class Lane implements AgentLane { throw new SliceNotImplemented("requestAbort"); } - async inspectExecution(_context: Context): Promise { - this.assertOpen(); - const operation = this.state.operation; - if (operation === null) { + inspectExecution(context: Context): Promise { + return this.command((state) => { + const operation = state.operation; + const captured = operation === null ? undefined : capturedModel(operation); + const current = + operation === null + ? null + : { + id: operation.meta.operationId, + kind: operation.meta.intent.kind, + status: + operation.state.control.status === "cancel_requested" + ? ("aborting" as const) + : ("open" as const), + startedAt: operation.meta.startedAt, + ...(captured === undefined ? {} : { capturedModel: captured }), + }; return { - lane: this.name, - leafId: this.state.leafId, - current: null, - ...(this.state.lastResult === undefined ? {} : { lastResult: this.state.lastResult }), + kind: "return", + result: { + lane: this.name, + leafId: state.leafId, + configuredModel: state.configuration.model, + current, + ...(state.lastResult === undefined ? {} : { lastResult: state.lastResult }), + }, }; - } - const status = operation.state.control.status === "cancel_requested" ? "aborting" : "suspended"; - const suspended = - status === "suspended" && this.suspension?.operationId === operation.meta.operationId - ? this.suspension - : undefined; - return { - lane: this.name, - leafId: this.state.leafId, - current: { - id: operation.meta.operationId, - kind: operation.meta.intent.kind, - status, - startedAt: operation.meta.startedAt, - ...(suspended === undefined ? {} : { suspended }), - }, - ...(this.state.lastResult === undefined ? {} : { lastResult: this.state.lastResult }), - }; + }, context); } async prompt( @@ -342,8 +607,222 @@ export class Lane implements AgentLane { ); } - async watch(..._args: Parameters): Promise { - throw new SliceNotImplemented("watch"); + watch(context: Context): Promise> { + return this.command(async (state, reader) => { + const watcher = this.installWatch( + {} as LaneSnapshot, + (event) => event.type === "usage" || !("lane" in event) || event.lane === this.name, + context, + ); + const captured = structuredClone(state); + try { + const transcript = + captured.leafId === null + ? [] + : ( + await reader.scanBranch( + { start: captured.leafId, stopAtType: "compaction", order: "newestFirst" }, + context, + ) + ).reverse(); + const queuedItems = async (ids: readonly string[], description: string): Promise => + Promise.all( + ids.map(async (entryId) => { + const stored = await reader.getValue(pendingEntry(entryId), context); + if (stored?.value.type !== "message") { + throw new SessionInvariantError(`${description} ${entryId} is missing a message payload`); + } + return { entryId, message: stored.value.payload }; + }), + ); + const nextRun = await queuedItems(captured.pendingNextRun, "Pending next-run entry"); + const operation = captured.operation; + const steer = + operation?.state.kind === "run" ? await queuedItems(operation.state.inbox.steer, "Steer entry") : []; + const followUp = + operation?.state.kind === "run" + ? await queuedItems(operation.state.inbox.followUp, "Follow-up entry") + : []; + const pendingWrites = + operation?.state.kind === "run" + ? await Promise.all( + operation.state.inbox.writes.map(async (entryId) => { + const stored = await reader.getValue(pendingEntry(entryId), context); + if (stored === undefined) { + throw new SessionInvariantError(`Pending write ${entryId} is missing its payload`); + } + return stored.value.type === "message" + ? { entryId, type: "message" as const, message: stored.value.payload } + : { + entryId, + type: "custom" as const, + customType: stored.value.customType, + ...(stored.value.payload === undefined ? {} : { data: stored.value.payload }), + }; + }), + ) + : []; + + let operationSnapshot: NonNullable | null = null; + if (operation !== null) { + const runningTools: NonNullable["runningTools"] = []; + let streamingMessage: NonNullable["streamingMessage"]; + let retry: NonNullable["retry"]; + let deferred: NonNullable["deferred"]; + const readStreamingMessage = async (responseEntryId: string) => { + const frames = []; + let cursor: { seq: number } | undefined; + while (frames.length < 10_000) { + const limit = Math.min(1_000, 10_000 - frames.length); + const page = await reader.readList( + pendingAssistantFrames(operation.meta.operationId, responseEntryId), + { order: "asc", limit, ...(cursor === undefined ? {} : { cursor }) }, + context, + ); + frames.push(...page.map(({ value }) => value)); + if (page.length < limit) break; + cursor = { seq: page[page.length - 1]!.seq }; + } + return reduceAssistantMessageFrames(frames); + }; + const readRetry = (generation: { + status: string; + nextAttempt?: number; + notBefore?: number; + context: { retryPolicy: { maxAttempts: number } }; + }) => { + if (generation.status !== "retry_wait") return undefined; + if (generation.nextAttempt === undefined || generation.notBefore === undefined) { + throw new SessionInvariantError("Retry wait is missing retry metadata"); + } + return { + attempt: generation.nextAttempt, + maxAttempts: generation.context.retryPolicy.maxAttempts, + nextAttemptAt: generation.notBefore, + }; + }; + const readStructuralRetry = (structural: StructuralDecision) => { + if (structural.status !== "generating") return undefined; + return readRetry(structural.generation); + }; + + if (operation.state.kind === "run") { + switch (operation.state.phase.kind) { + case "assistant": + retry = readRetry(operation.state.phase.generation); + if (operation.state.phase.generation.status === "effect_pending") { + streamingMessage = await readStreamingMessage( + operation.state.phase.generation.responseEntryId, + ); + } + break; + case "deferred": { + const durable = operation.state.phase.deferred; + const source = (await reader.getEntries([durable.sourceEntryId], context)).get( + durable.sourceEntryId, + ); + if ( + source?.type !== "message" || + source.message.role !== "assistant" || + source.message.deferred === undefined + ) { + throw new SessionInvariantError("Deferred source is missing its assistant handle"); + } + deferred = { handle: source.message.deferred, poll: durable.poll }; + if (durable.status === "effect_pending") { + streamingMessage = await readStreamingMessage(durable.responseEntryId); + } + break; + } + case "tools": { + const { batch } = operation.state.phase; + const assistant = (await reader.getEntries([batch.assistantEntryId], context)).get( + batch.assistantEntryId, + ); + if (assistant?.type !== "message" || assistant.message.role !== "assistant") { + throw new SessionInvariantError("Tool batch assistant entry is invalid"); + } + for (const call of batch.calls) { + if (call.status !== "effect_pending") continue; + const block = assistant.message.content[call.sourceIndex]; + if (block?.type !== "toolCall") { + throw new SessionInvariantError( + `Tool call source index ${call.sourceIndex} does not name a tool-call block`, + ); + } + const args = await reader.getValue( + operationToolArgs(operation.meta.operationId, batch.turnId, call.sourceIndex), + context, + ); + if (args === undefined) { + throw new SessionInvariantError(`Tool call ${block.id} is missing persisted arguments`); + } + const checkpoint = await reader.getValue( + pendingToolOutput(operation.meta.operationId, call.resultEntryId), + context, + ); + runningTools.push({ + toolCallId: block.id, + toolName: block.name, + args: args.value, + ...(checkpoint === undefined ? {} : { partialResult: checkpoint.value }), + }); + } + break; + } + case "compaction": + retry = readStructuralRetry(operation.state.phase.structural); + break; + case "starting": + case "checkpoint": + case "failure_drain": + break; + } + } else if (operation.state.kind === "compaction") { + retry = readStructuralRetry(operation.state.structural); + } else if (operation.state.summarize) { + retry = readStructuralRetry(operation.state.phase.structural); + } + + const drained = + operation.state.control.status === "cancel_requested" + ? { + steer: await queuedItems(operation.state.control.drainedSteer, "Drained steer entry"), + followUp: await queuedItems( + operation.state.control.drainedFollowUp, + "Drained follow-up entry", + ), + } + : undefined; + operationSnapshot = { + id: operation.meta.operationId, + kind: operation.meta.intent.kind, + startedAt: operation.meta.startedAt, + status: operation.state.control.status === "cancel_requested" ? "aborting" : "open", + ...(retry === undefined ? {} : { retry }), + ...(deferred === undefined ? {} : { deferred }), + ...(drained === undefined ? {} : { drained }), + ...(streamingMessage === undefined ? {} : { streamingMessage }), + runningTools, + }; + } + + watcher.snapshot = structuredClone({ + lane: this.name, + transcript, + leafId: captured.leafId, + ...(captured.lastResult === undefined ? {} : { lastResult: captured.lastResult }), + operation: operationSnapshot, + queues: { steer, followUp, nextRun }, + pendingWrites, + faulted: false, + }); + return { kind: "return", result: watcher }; + } catch (error) { + watcher.unsubscribe(); + throw error; + } + }, context); } private async setConfiguration( @@ -351,26 +830,42 @@ export class Lane implements AgentLane { event: (previous: LaneState["configuration"], value: LaneState["configuration"]) => LaneConfigEventPayload, context: Context, ): Promise { - const payload = await this.command((state) => { + await this.command((state) => { const configuration = update(state.configuration); return { kind: "commit", writes: [setValue(laneConfig(this.name), configuration)], next: { ...state, configuration }, - materialize: () => event(state.configuration, configuration), + materialize: () => undefined, + events: () => [{ ...event(state.configuration, configuration), lane: this.name }], }; }, context); - await this.onEvent({ ...payload, lane: this.name }, context); } private async setName(name: string | undefined, context: Context): Promise { - await this.sessionView.setName(name, context); - await this.onEvent({ type: "value_update", value: "session_name", name }, context); + await this.command( + (state) => ({ + kind: "commit", + writes: [name === undefined ? deleteValue(sessionName) : setValue(sessionName, name)], + next: state, + materialize: () => undefined, + events: () => [{ type: "value_update", value: "session_name", name }], + }), + context, + ); } private async setLabel(targetId: string, label: string | undefined, context: Context): Promise { - await this.sessionView.setLabel(targetId, label, context); - await this.onEvent({ type: "value_update", value: "entry_label", targetId, label }, context); + await this.command( + (state) => ({ + kind: "commit", + writes: [label === undefined ? deleteValue(entryLabel(targetId)) : setValue(entryLabel(targetId), label)], + next: state, + materialize: () => undefined, + events: () => [{ type: "value_update", value: "entry_label", targetId, label }], + }), + context, + ); } private async findEntriesOnBranch(query: BranchScan | undefined, context: Context): Promise { @@ -414,32 +909,68 @@ export class Lane implements AgentLane { ], next: { ...state, leafId: id }, materialize: () => id, + events: (commit) => { + const entry: Entry = + pending.type === "message" + ? { + id, + parentId: state.leafId, + type: "message", + message: pending.payload, + seq: commit.seqs[0]!, + timestamp: commit.timestamp, + } + : { + id, + parentId: state.leafId, + type: "custom", + customType: pending.customType, + ...(pending.payload === undefined ? {} : { data: pending.payload }), + seq: commit.seqs[0]!, + timestamp: commit.timestamp, + }; + return pending.type === "message" + ? [ + { type: "message_start", message: pending.payload, lane: this.name }, + { type: "message_end", message: pending.payload, entryId: id, lane: this.name }, + { type: "entry_added", entry, lane: this.name }, + ] + : [{ type: "entry_added", entry, lane: this.name }]; + }, }; } - if (state.operation.state.kind !== "run") { + const operation = state.operation; + if (operation.state.kind !== "run") { return { kind: "reject", - error: new Error( - `Cannot append while structural operation ${state.operation.meta.operationId} is active`, - ), + error: new Error(`Cannot append while structural operation ${operation.meta.operationId} is active`), }; } const operationState = { - ...state.operation.state, + ...operation.state, inbox: { - ...state.operation.state.inbox, - writes: [...state.operation.state.inbox.writes, id], + ...operation.state.inbox, + writes: [...operation.state.inbox.writes, id], }, }; return { kind: "commit", writes: [ setValue(pendingEntry(id), pending), - setValue(operationStateValue(state.operation.meta.operationId), operationState), + setValue(operationStateValue(operation.meta.operationId), operationState), ], - next: { ...state, operation: { meta: state.operation.meta, state: operationState } }, + next: { ...state, operation: { meta: operation.meta, state: operationState } }, materialize: () => id, + events: () => [ + { + type: "write_pending", + runId: operation.meta.operationId, + entryId: id, + entryType: pending.type, + lane: this.name, + }, + ], }; }, context); } diff --git a/packages/agent/src/harness/runtime2/restore.ts b/packages/agent/src/harness/runtime2/restore.ts index d223d5ee5..ef5e6ce89 100644 --- a/packages/agent/src/harness/runtime2/restore.ts +++ b/packages/agent/src/harness/runtime2/restore.ts @@ -50,6 +50,21 @@ async function restoreLaneState(reader: SessionReader, lane: string, context: Co ]); if (meta === undefined) throw new SessionInvariantError(`Operation ${operationId} is missing op.meta`); if (state === undefined) throw new SessionInvariantError(`Operation ${operationId} is missing op.state`); + if (meta.value.operationId !== operationId) { + throw new SessionInvariantError( + `Operation ${operationId} metadata names operation ${JSON.stringify(meta.value.operationId)}`, + ); + } + if (meta.value.lane !== lane) { + throw new SessionInvariantError( + `Operation ${operationId} belongs to lane ${JSON.stringify(meta.value.lane)}, not ${JSON.stringify(lane)}`, + ); + } + if (meta.value.intent.kind !== state.value.kind) { + throw new SessionInvariantError( + `Operation ${operationId} intent ${meta.value.intent.kind} does not match state ${state.value.kind}`, + ); + } operation = { meta: meta.value, state: state.value }; } diff --git a/packages/agent/src/harness/runtime2/types.ts b/packages/agent/src/harness/runtime2/types.ts index 8adaef9c8..52cf51ad5 100644 --- a/packages/agent/src/harness/runtime2/types.ts +++ b/packages/agent/src/harness/runtime2/types.ts @@ -29,6 +29,15 @@ export interface Config { readonly entryProjectors: Readonly["entryProjectors"]>>; } +/** Harness-global settings captured by an accepted run. */ +export interface AcceptanceConfig { + readonly resources: Resources; + readonly compaction: CompactionSettings; + readonly steeringMode: QueueMode; + readonly followUpMode: QueueMode; + readonly toolExecution: "sequential" | "parallel"; +} + /** The current durable state owned by one lane. */ export interface LaneState { readonly leafId: string | null; diff --git a/packages/agent/src/harness/session/memory.ts b/packages/agent/src/harness/session/memory.ts index 933f52a7a..45b02f1a8 100644 --- a/packages/agent/src/harness/session/memory.ts +++ b/packages/agent/src/harness/session/memory.ts @@ -213,6 +213,10 @@ class MemorySessionFacade implements Session { return this.admit(() => this.session.readList(address, options, context)); } + scanBranch(query: StorageBranchScan, context: Context): Promise { + return this.admit(() => this.session.scanBranch(query, context)); + } + view(lane: string): SessionTree { const view = this.session.view(lane); return { diff --git a/packages/agent/src/harness/session/remote.ts b/packages/agent/src/harness/session/remote.ts index 9b9db362b..24499dfc6 100644 --- a/packages/agent/src/harness/session/remote.ts +++ b/packages/agent/src/harness/session/remote.ts @@ -21,6 +21,7 @@ import type { SessionMutator, SessionStats, SessionTree, + StorageBranchScan, Write, } from "./types.ts"; import type { ListElement, ListReadOptions, StoredValue, Value, ValueList } from "./values.ts"; @@ -240,6 +241,10 @@ export class RemoteSession ) as ListElement[]; } + async scanBranch(query: StorageBranchScan, context: Context): Promise { + return expectArray(await this.invoke("session.scanBranch", { query }, context), "branch entries") as Entry[]; + } + async mutate( lane: string, mutation: (mutator: SessionMutator, context: Context) => T | Promise, @@ -396,6 +401,18 @@ class RemoteSessionMutator implements SessionMutator { ) as ListElement[]; } + async scanBranch(query: StorageBranchScan, context: Context): Promise { + this.assertActive(); + return expectArray( + await this.rpc.invoke( + "session.mutation.scanBranch", + { handle: this.handle, mutationId: this.mutationId, query }, + context, + ), + "branch entries", + ) as Entry[]; + } + settle(): Promise { return ( this.commitPromise?.then( diff --git a/packages/agent/src/harness/session/session.ts b/packages/agent/src/harness/session/session.ts index 68d67e895..12cb131ce 100644 --- a/packages/agent/src/harness/session/session.ts +++ b/packages/agent/src/harness/session/session.ts @@ -20,6 +20,7 @@ import type { SessionStats, SessionTree, Storage, + StorageBranchScan, Write, } from "./types.ts"; import { @@ -99,6 +100,65 @@ export class SessionUnknownTargetError extends Error { } } +/** + * Package-internal lane creation procedure for callers that own the lane mutation callback. + * + * TODO: Replace this one-off procedure if lane creation gains a shared command abstraction. The Harness must own the + * new lane's mutation so the durable commit, in-memory lane publication, and recipient-bound `lane_created` enqueue + * all happen before line release. Calling event-free `Session.createLane()` and emitting afterward would permit both + * snapshot-after plus a stale event and delivery to listeners registered after the commit. + */ +export async function createLaneWithMutator( + mutator: SessionMutator, + name: string, + at: string | null, + configuration: LaneConfiguration, + context: Context, +): Promise { + if (mutator.lane !== name) { + throw new SessionInvariantError( + `Lane mutation for ${JSON.stringify(mutator.lane)} cannot create ${JSON.stringify(name)}`, + ); + } + if (name.length === 0) throw new SessionInvalidLaneError(name, "lane name must not be empty"); + + // R1 owns complete idle-lane and current-state validation. Slice 2 only + // distinguishes valid existing lane shapes from partial durable lane state. + const [leaf, storedConfiguration, storedLaneState, lastResult] = await Promise.all([ + mutator.getValue(laneLeaf(name), context), + mutator.getValue(laneConfig(name), context), + mutator.getValue(laneState(name), context), + mutator.getValue(laneLastResult(name), context), + ]); + const presentCount = [leaf, storedConfiguration, storedLaneState, lastResult].filter( + (stored) => stored !== undefined, + ).length; + if ( + leaf !== undefined && + storedLaneState !== undefined && + (storedConfiguration !== undefined || (name === "main" && lastResult === undefined)) + ) { + throw new SessionLaneExistsError(name); + } + if (presentCount !== 0) { + throw new SessionInvariantError(`Lane ${JSON.stringify(name)} has incomplete durable state`); + } + if (at !== null && !(await mutator.getEntries([at], context)).has(at)) { + throw new SessionUnknownTargetError(at); + } + + // R6 adds the harness-wide admission barrier. Until then, close may reject + // this lane job before Storage.commit admits it; admitted commits still drain. + await mutator.commit( + [ + setValueWrite(laneConfig(name), configuration), + setValueWrite(laneLeaf(name), at), + setValueWrite(laneState(name), { currentOperationId: null, pendingNextRun: [] }), + ], + context, + ); +} + class StorageBackedSessionMutator implements SessionMutator { readonly lane: string; private readonly storage: Storage; @@ -155,6 +215,11 @@ class StorageBackedSessionMutator implements SessionMutator { return this.storage.readList(address, options, context); } + scanBranch(query: StorageBranchScan, context: Context): Promise { + this.assertActive(); + return this.storage.scanBranch(query, context); + } + settle(): Promise { return ( this.commitResult?.then( @@ -235,6 +300,11 @@ export class StorageBackedSession { + this.assertOpen(); + return this.storage.scanBranch(query, context); + } + view(lane: string): SessionTree { return { getLeafId: (context) => this.getLeafIdForLane(lane, context), @@ -268,49 +338,8 @@ export class StorageBackedSession { this.assertOpen(); - if (name.length === 0) throw new SessionInvalidLaneError(name, "lane name must not be empty"); - return this.mutate( - name, - async (mutator) => { - // R1 owns complete idle-lane and current-state validation. Slice 2 only - // distinguishes valid existing lane shapes from partial durable lane state. - const [leaf, storedConfiguration, storedLaneState, lastResult] = await Promise.all([ - mutator.getValue(laneLeaf(name), context), - mutator.getValue(laneConfig(name), context), - mutator.getValue(laneState(name), context), - mutator.getValue(laneLastResult(name), context), - ]); - const presentCount = [leaf, storedConfiguration, storedLaneState, lastResult].filter( - (stored) => stored !== undefined, - ).length; - if ( - leaf !== undefined && - storedLaneState !== undefined && - (storedConfiguration !== undefined || (name === "main" && lastResult === undefined)) - ) { - throw new SessionLaneExistsError(name); - } - if (presentCount !== 0) { - throw new SessionInvariantError(`Lane ${JSON.stringify(name)} has incomplete durable state`); - } - if (at !== null && !(await mutator.getEntries([at], context)).has(at)) { - throw new SessionUnknownTargetError(at); - } - - // R6 adds the harness-wide admission barrier. Until then, close may reject - // this lane job before Storage.commit admits it; admitted commits still drain. - await mutator.commit( - [ - setValueWrite(laneConfig(name), configuration), - setValueWrite(laneLeaf(name), at), - setValueWrite(laneState(name), { currentOperationId: null, pendingNextRun: [] }), - ], - context, - ); - return this.view(name); - }, - context, - ); + await this.mutate(name, (mutator) => createLaneWithMutator(mutator, name, at, configuration, context), context); + return this.view(name); } getLeafId(context: Context): Promise { diff --git a/packages/agent/src/harness/session/types.ts b/packages/agent/src/harness/session/types.ts index fc5066bc9..b38b3d718 100644 --- a/packages/agent/src/harness/session/types.ts +++ b/packages/agent/src/harness/session/types.ts @@ -154,20 +154,19 @@ export type Generation = errorMessage: string; }; -export type ToolCall = - | { status: "planned"; sourceIndex: number; resultEntryId: string } - | { - status: "effect_pending"; - sourceIndex: number; - resultEntryId: string; - replay: "never" | "safe"; - } - | { - status: "completed"; - sourceIndex: number; - resultEntryId: string; - terminate: boolean; - }; +interface ToolCallSource { + /** Zero-based index in the assistant message's complete content array, not a filtered tool-call ordinal. */ + sourceIndex: number; + resultEntryId: string; +} + +export type ToolCall = ToolCallSource & + ( + | { status: "planned" } + | { status: "effect_pending"; replay: "never" | "safe" } + | { status: "outcome_ready"; terminate: boolean } + | { status: "completed"; terminate: boolean } + ); export interface ToolBatch { assistantEntryId: string; @@ -243,7 +242,10 @@ export type RunPhase = | { kind: "failure_drain"; error: OperationError; - provenance: { kind: "response"; entryId: string } | { kind: "structural"; taskId: string }; + provenance: + | { kind: "response"; entryId: string } + | { kind: "structural"; taskId: string } + | { kind: "configuration" }; }; export interface RunState { @@ -487,6 +489,8 @@ export interface SessionReader { options: ListReadOptions | undefined, context: Context, ): Promise[]>; + /** Scan a branch from an explicit entry while this reader capability remains valid. */ + scanBranch(query: StorageBranchScan, context: Context): Promise; } /** Callback-scoped write capability bound to one lane. */ diff --git a/packages/agent/test/harness/execution-primitives.test.ts b/packages/agent/test/harness/execution-primitives.test.ts index c10d5481f..f1fb469ca 100644 --- a/packages/agent/test/harness/execution-primitives.test.ts +++ b/packages/agent/test/harness/execution-primitives.test.ts @@ -491,6 +491,54 @@ describe("HarnessEventBus", () => { expect(seen).toEqual(["one:start", "one:end", "two:start", "two:end"]); }); + it("binds listeners and watchers when an event is enqueued", async () => { + const bus = new HarnessEventBus(); + const started = deferred(); + const release = deferred(); + bus.on("run_start", async (event) => { + if (event.runId !== "blocking") return; + started.resolve(); + await release.promise; + }); + const blocking = bus.emit({ type: "run_start", runId: "blocking", lane: "main" }, BACKGROUND_CONTEXT); + await started.promise; + + const queued = bus.emit({ type: "run_start", runId: "queued", lane: "main" }, BACKGROUND_CONTEXT); + const lateListenerEvents: string[] = []; + bus.on("run_start", (event) => { + lateListenerEvents.push(event.runId); + }); + const lateWatcherEvents: string[] = []; + const lateWatcher = bus.watch({}, (event) => event.type === "run_start", BACKGROUND_CONTEXT); + lateWatcher.start((event) => { + if (event.type === "run_start") lateWatcherEvents.push(event.runId); + }); + + release.resolve(); + await Promise.all([blocking, queued]); + expect(lateListenerEvents).toEqual([]); + expect(lateWatcherEvents).toEqual([]); + + await bus.emit({ type: "run_start", runId: "later", lane: "main" }, BACKGROUND_CONTEXT); + expect(lateListenerEvents).toEqual(["later"]); + expect(lateWatcherEvents).toEqual(["later"]); + }); + + it("releases reserved delivery gates during close", async () => { + const bus = new HarnessEventBus(); + const seen: string[] = []; + bus.on("run_start", (event) => { + seen.push(event.runId); + }); + const delivery = bus.enqueue([{ type: "run_start", runId: "reserved", lane: "main" }], BACKGROUND_CONTEXT); + + bus.close(new Error("closed")); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(seen).toEqual(["reserved"]); + await delivery.start(); + }); + it("continues watcher delivery after listener failure and reports it", async () => { const bus = new HarnessEventBus(); const failures: string[] = []; diff --git a/packages/agent/test/harness/memory-session-repo.test.ts b/packages/agent/test/harness/memory-session-repo.test.ts index bb38cccb8..a8e78cba1 100644 --- a/packages/agent/test/harness/memory-session-repo.test.ts +++ b/packages/agent/test/harness/memory-session-repo.test.ts @@ -28,6 +28,7 @@ describe("MemorySessionRepo metadata", () => { await expect(repo.open(first.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("already open"); await Promise.all([admittedWrite, first.close(BACKGROUND_CONTEXT)]); await expect(first.getName(BACKGROUND_CONTEXT)).rejects.toThrow("Session is closed"); + await expect(first.scanBranch({ start: "entry" }, BACKGROUND_CONTEXT)).rejects.toThrow("Session is closed"); await expect(firstView.getName(BACKGROUND_CONTEXT)).rejects.toThrow("Session is closed"); const second = await repo.open(first.metadata, BACKGROUND_CONTEXT); @@ -61,6 +62,9 @@ describe("MemorySessionRepo metadata", () => { options.position = "at"; await commit; + await expect( + source.scanBranch({ start: childId, order: "oldestFirst" }, BACKGROUND_CONTEXT), + ).resolves.toMatchObject([{ id: rootId }, { id: childId }]); const forked = await fork; expect(await forked.getLeafId(BACKGROUND_CONTEXT)).toBe(rootId); await Promise.all([ diff --git a/packages/agent/test/harness/runtime2/accept.test.ts b/packages/agent/test/harness/runtime2/accept.test.ts new file mode 100644 index 000000000..09d193299 --- /dev/null +++ b/packages/agent/test/harness/runtime2/accept.test.ts @@ -0,0 +1,382 @@ +import { createModels, fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { type HarnessEvent, HarnessFault, type OperationRequest } from "../../../src/harness/agent-harness.ts"; +import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; +import { BACKGROUND_CONTEXT, createContextKey, withContextValue } from "../../../src/harness/context.ts"; +import type { Result } from "../../../src/harness/result.ts"; +import { createAgentHarness, Harness } from "../../../src/harness/runtime2/harness.ts"; +import { MemoryStorage } from "../../../src/harness/session/memory.ts"; +import { StorageBackedSession } from "../../../src/harness/session/session.ts"; +import { InstrumentedStorage } from "../../../src/harness/session/testing/index.ts"; +import * as storedValues from "../../../src/harness/session/values.ts"; +import type { AgentMessage, Session } from "../../../src/index.ts"; +import { ControlledMemoryStorage, deferred, FailingMemoryStorage } from "./test-utils.ts"; + +const sessions: Session[] = []; + +async function createSession(storage = new InstrumentedStorage(new MemoryStorage())): Promise<{ + session: Session; + storage: InstrumentedStorage; +}> { + const session = new StorageBackedSession( + { id: `accept-${sessions.length}`, createdAt: 1, storageVersion: 1 }, + storage, + ); + sessions.push(session); + await session.mutate( + "main", + (mutator) => + mutator.commit( + [ + storedValues.setValue(storedValues.laneLeaf("main"), null), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: null, + pendingNextRun: [], + }), + ], + BACKGROUND_CONTEXT, + ), + BACKGROUND_CONTEXT, + ); + return { session, storage }; +} + +function options(session: Session) { + const provider = fauxProvider(); + const models = createModels(); + models.setProvider(provider.provider); + return { session, models, model: provider.getModel() }; +} + +async function createHarness( + beforeCreate?: (session: Session) => Promise, + resources: Parameters[0]["resources"] = {}, +): Promise<{ + harness: Harness; + session: Session; + storage: InstrumentedStorage; + models: ReturnType; +}> { + const created = await createSession(); + await beforeCreate?.(created.session); + const harnessOptions = { ...options(created.session), resources }; + const { harness } = await createAgentHarness(harnessOptions, BACKGROUND_CONTEXT); + if (!(harness instanceof Harness)) throw new Error("Expected runtime2 Harness"); + created.storage.clearCommitAttempts(); + return { harness, ...created, models: harnessOptions.models }; +} + +function unwrap(result: Result): T { + if (!result.ok) throw result.error; + return result.value; +} + +afterEach(async () => { + for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); +}); + +describe("runtime2 atomic run acceptance", () => { + it.each([ + ["text", { kind: "prompt", prompt: "hello" } satisfies OperationRequest, [{ type: "text", text: "hello" }]], + [ + "images", + { + kind: "prompt", + prompt: "", + images: [{ type: "image", data: "aW1hZ2U=", mimeType: "image/png" }], + } satisfies OperationRequest, + [{ type: "image", data: "aW1hZ2U=", mimeType: "image/png" }], + ], + [ + "text and images", + { + kind: "prompt", + prompt: "hello", + images: [{ type: "image", data: "aW1hZ2U=", mimeType: "image/png" }], + } satisfies OperationRequest, + [ + { type: "text", text: "hello" }, + { type: "image", data: "aW1hZ2U=", mimeType: "image/png" }, + ], + ], + ] as const)("accepts normalized %s prompts into starting", async (_name, request, expectedContent) => { + const { harness, session } = await createHarness(); + + const admission = unwrap(await harness.accept(request, BACKGROUND_CONTEXT)); + const operation = harness.state.operation; + if (operation?.state.kind !== "run") throw new Error("Expected accepted run"); + const entryId = operation.meta.intent.kind === "run" ? operation.meta.intent.promptEntryIds[0] : undefined; + if (entryId === undefined) throw new Error("Expected prompt entry"); + const entry = await session.getEntry(entryId, BACKGROUND_CONTEXT); + + expect(admission).toMatchObject({ operationId: operation.meta.operationId, kind: "run" }); + expect(entry).toMatchObject({ type: "message", message: { role: "user", content: expectedContent } }); + expect(operation.state.phase).toEqual({ kind: "starting" }); + expect(operation.state.settings).toEqual({ + compaction: DEFAULT_COMPACTION_SETTINGS, + steeringMode: "all", + followUpMode: "all", + toolExecution: "parallel", + }); + }); + + it("preserves supplied message arrays and commits the exact acceptance write families once", async () => { + const { harness, session, storage, models } = await createHarness(); + const messages: AgentMessage[] = [ + { role: "user", content: "one", timestamp: 10 }, + { role: "user", content: "two", timestamp: 11 }, + ]; + const getModel = vi.spyOn(models, "getModel"); + const contextKey = createContextKey("accept.context"); + const context = withContextValue(contextKey, "source", BACKGROUND_CONTEXT); + const seen: Array<{ event: HarnessEvent; sameContext: boolean }> = []; + for (const type of ["run_start", "message_start", "message_end", "entry_added"] as const) { + harness.events.on(type, (event, eventContext) => { + seen.push({ event, sameContext: eventContext === context }); + }); + } + + const admission = unwrap( + await harness.accept({ kind: "prompt", operationId: "operation", prompt: messages }, context), + ); + + expect(admission).toEqual({ operationId: "operation", kind: "run", startedAt: expect.any(Number) }); + expect(storage.getCommitAttempts()).toHaveLength(1); + expect( + storage + .getCommitAttempts()[0] + ?.map((write) => (write.kind === "value" ? `${write.kind}:${write.op}` : write.kind)), + ).toEqual(["entry", "entry", "value:set", "value:set", "value:set", "value:set"]); + const operation = harness.state.operation; + if (operation?.meta.intent.kind !== "run") throw new Error("Expected run metadata"); + expect(operation.meta).toMatchObject({ + operationId: "operation", + lane: "main", + sourceLeafId: null, + intent: { promptEntryIds: expect.any(Array) }, + }); + expect(operation.meta.intent.promptEntryIds).toHaveLength(2); + expect(await session.getLeafId(BACKGROUND_CONTEXT)).toBe(operation.meta.intent.promptEntryIds[1]); + expect(seen.map(({ event }) => event.type)).toEqual([ + "run_start", + "message_start", + "message_end", + "entry_added", + "message_start", + "message_end", + "entry_added", + ]); + expect(seen.every(({ sameContext }) => sameContext)).toBe(true); + expect(getModel).not.toHaveBeenCalled(); + }); + + it("captures next-run messages before request messages and accepts an otherwise empty request", async () => { + const first = "pending-first"; + const second = "pending-second"; + const firstMessage = { role: "user" as const, content: "first", timestamp: 1 }; + const secondMessage = { role: "user" as const, content: "second", timestamp: 2 }; + const { harness, session, storage } = await createHarness(async (source) => { + await source.mutate( + "main", + (mutator) => + mutator.commit( + [ + storedValues.setValue(storedValues.pendingEntry(first), { + type: "message", + payload: firstMessage, + }), + storedValues.setValue(storedValues.pendingEntry(second), { + type: "message", + payload: secondMessage, + }), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: null, + pendingNextRun: [first, second], + }), + ], + BACKGROUND_CONTEXT, + ), + BACKGROUND_CONTEXT, + ); + }); + const events: string[] = []; + harness.events.on("queue_update", (event) => { + events.push(event.type); + }); + + unwrap(await harness.accept({ kind: "prompt", prompt: "" }, BACKGROUND_CONTEXT)); + + const operation = harness.state.operation; + if (operation?.meta.intent.kind !== "run" || operation.state.kind !== "run") { + throw new Error("Expected accepted run"); + } + expect(operation.meta.intent.promptEntryIds).toEqual([]); + expect(operation.meta.sourceLeafId).toBeNull(); + expect(operation.state.inbox).toEqual({ steer: [], followUp: [], writes: [] }); + expect( + (await session.scanBranch({ start: second, order: "oldestFirst" }, BACKGROUND_CONTEXT)).map(({ id }) => id), + ).toEqual([first, second]); + expect(await session.getValue(storedValues.pendingEntry(first), BACKGROUND_CONTEXT)).toBeUndefined(); + expect(await session.getValue(storedValues.pendingEntry(second), BACKGROUND_CONTEXT)).toBeUndefined(); + expect( + storage + .getCommitAttempts()[0] + ?.map((write) => (write.kind === "value" ? `${write.kind}:${write.op}` : write.kind)), + ).toEqual(["entry", "entry", "value:delete", "value:delete", "value:set", "value:set", "value:set", "value:set"]); + expect(events).toEqual(["queue_update"]); + }); + + it("formats skills and templates before acceptance", async () => { + const resources = { + skills: [ + { name: "review", description: "Review", content: "Inspect it", filePath: "/skills/review/SKILL.md" }, + ], + promptTemplates: [{ name: "fix", content: "Fix $1 then $@" }], + }; + const skill = await createHarness(undefined, resources); + unwrap( + await skill.harness.accept( + { kind: "skill", name: "review", additionalInstructions: "Be strict" }, + BACKGROUND_CONTEXT, + ), + ); + const skillEntry = await skill.session.getEntry(skill.harness.state.leafId!, BACKGROUND_CONTEXT); + expect(skillEntry).toMatchObject({ + type: "message", + message: { content: [{ type: "text", text: expect.stringContaining(' { + const { harness, storage } = await createHarness(); + const pending = fauxAssistantMessage([], { stopReason: "pending" }); + + expect(await harness.accept({ kind: "prompt", prompt: "" }, BACKGROUND_CONTEXT)).toMatchObject({ + ok: false, + error: { _tag: "InvalidMessage", reason: "empty" }, + }); + expect(await harness.accept({ kind: "prompt", prompt: pending }, BACKGROUND_CONTEXT)).toMatchObject({ + ok: false, + error: { _tag: "InvalidMessage", reason: "pending_assistant" }, + }); + expect(await harness.accept({ kind: "skill", name: "missing" }, BACKGROUND_CONTEXT)).toMatchObject({ + ok: false, + error: { _tag: "UnknownSkill" }, + }); + expect(await harness.accept({ kind: "prompt_template", name: "missing" }, BACKGROUND_CONTEXT)).toMatchObject({ + ok: false, + error: { _tag: "UnknownTemplate" }, + }); + expect(storage.getCommitAttempts()).toEqual([]); + }); + + it("serializes concurrent accepts so exactly one wins", async () => { + const { harness, storage } = await createHarness(); + + const results = await Promise.all([ + harness.accept({ kind: "prompt", prompt: "first" }, BACKGROUND_CONTEXT), + harness.accept({ kind: "prompt", prompt: "second" }, BACKGROUND_CONTEXT), + ]); + + expect(results.filter((result) => result.ok)).toHaveLength(1); + expect(results.filter((result) => !result.ok)).toMatchObject([{ error: { _tag: "LaneBusy" } }]); + expect(storage.getCommitAttempts()).toHaveLength(1); + }); + + it("faults on commit failure without publishing acceptance", async () => { + const storage = new FailingMemoryStorage(); + const session = new StorageBackedSession({ id: "failing", createdAt: 1, storageVersion: 1 }, storage); + sessions.push(session); + await session.mutate( + "main", + (mutator) => + mutator.commit( + [ + storedValues.setValue(storedValues.laneLeaf("main"), null), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: null, + pendingNextRun: [], + }), + ], + BACKGROUND_CONTEXT, + ), + BACKGROUND_CONTEXT, + ); + const { harness } = await createAgentHarness(options(session), BACKGROUND_CONTEXT); + if (!(harness instanceof Harness)) throw new Error("Expected runtime2 Harness"); + storage.failure = new Error("accept failed"); + + await expect(harness.accept({ kind: "prompt", prompt: "hello" }, BACKGROUND_CONTEXT)).rejects.toBeInstanceOf( + HarnessFault, + ); + expect(harness.state.operation).toBeNull(); + }); + + it("delivers acceptance listeners outside the lane line after publishing state", async () => { + const { harness } = await createHarness(); + let inspected = false; + harness.events.on("run_start", async () => { + inspected = (await harness.inspectExecution(BACKGROUND_CONTEXT)).current?.status === "open"; + }); + + unwrap(await harness.accept({ kind: "prompt", prompt: "hello" }, BACKGROUND_CONTEXT)); + + expect(inspected).toBe(true); + }); + + it("returns Closed when acceptance starts after close", async () => { + const { harness, storage } = await createHarness(); + await harness.close(BACKGROUND_CONTEXT); + + expect(await harness.accept({ kind: "prompt", prompt: "late" }, BACKGROUND_CONTEXT)).toMatchObject({ + ok: false, + error: { _tag: "Closed" }, + }); + expect(storage.getCommitAttempts()).toEqual([]); + }); + + it("publishes an acceptance admitted before close", async () => { + const storage = new ControlledMemoryStorage(); + const session = new StorageBackedSession({ id: "closing", createdAt: 1, storageVersion: 1 }, storage); + sessions.push(session); + await session.mutate( + "main", + (mutator) => + mutator.commit( + [ + storedValues.setValue(storedValues.laneLeaf("main"), null), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: null, + pendingNextRun: [], + }), + ], + BACKGROUND_CONTEXT, + ), + BACKGROUND_CONTEXT, + ); + const { harness } = await createAgentHarness(options(session), BACKGROUND_CONTEXT); + const started = deferred(); + const release = deferred(); + storage.beforeNextCommit = async () => { + started.resolve(); + await release.promise; + }; + const acceptance = harness.accept({ kind: "prompt", prompt: "hello" }, BACKGROUND_CONTEXT); + await started.promise; + const closing = harness.close(BACKGROUND_CONTEXT); + release.resolve(); + + expect((await acceptance).ok).toBe(true); + await closing; + }); +}); diff --git a/packages/agent/test/harness/runtime2/harness.test.ts b/packages/agent/test/harness/runtime2/harness.test.ts index cd99864ef..8a22897a8 100644 --- a/packages/agent/test/harness/runtime2/harness.test.ts +++ b/packages/agent/test/harness/runtime2/harness.test.ts @@ -1,4 +1,4 @@ -import { createModels, fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai"; +import { createModels, fauxProvider } from "@earendil-works/pi-ai"; import { Type } from "typebox"; import { afterEach, describe, expect, it, vi } from "vitest"; import { AgentHarness, HarnessClosed, HarnessFault } from "../../../src/harness/agent-harness.ts"; @@ -263,6 +263,35 @@ describe("runtime2 AgentHarness", () => { expect(thinkingAtEvent).toBe("high"); }); + it("serializes inspection behind earlier lane commits", async () => { + const storage = new ControlledMemoryStorage(); + const session = await createStorageSession(storage); + const { harness } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); + const commitStarted = deferred(); + const releaseCommit = deferred(); + storage.beforeNextCommit = async () => { + commitStarted.resolve(); + await releaseCommit.promise; + }; + const setting = harness.setThinkingLevel("high", BACKGROUND_CONTEXT); + await commitStarted.promise; + let inspected = false; + const inspection = harness.inspectExecution(BACKGROUND_CONTEXT).then((value) => { + inspected = true; + return value; + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(inspected).toBe(false); + + releaseCommit.resolve(); + await setting; + expect((await inspection).configuredModel).toEqual({ + provider: modelOptions(session).model.provider, + modelId: modelOptions(session).model.id, + }); + expect(await harness.getThinkingLevel(BACKGROUND_CONTEXT)).toBe("high"); + }); + it("rejects invalid config without publication or fault", async () => { const session = await createSession(); const initialTools = [tool("initial")]; @@ -327,7 +356,7 @@ describe("runtime2 AgentHarness", () => { const session = await createSession(); const options = modelOptions(session); - const { harness, suspended } = await createAgentHarness( + const { harness, open } = await createAgentHarness( { ...options, thinkingLevel: "high", @@ -339,7 +368,7 @@ describe("runtime2 AgentHarness", () => { expect(harness).toBeInstanceOf(Harness); expect(await harness.lane("main", BACKGROUND_CONTEXT)).toBe(harness); expect(await harness.getLeafId(BACKGROUND_CONTEXT)).toBeNull(); - expect(suspended).toEqual([]); + expect(open).toEqual([]); expect((await session.getValue(storedValues.laneConfig("main"), BACKGROUND_CONTEXT))?.value).toEqual({ model: { provider: options.model.provider, modelId: options.model.id }, thinkingLevel: "high", @@ -350,6 +379,13 @@ describe("runtime2 AgentHarness", () => { it("appends idle entries through owned lane state", async () => { const session = await createSession(); const { harness } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); + const events: string[] = []; + for (const type of ["message_start", "message_end", "entry_added"] as const) { + harness.events.on(type, async (event) => { + events.push(event.type); + await harness.inspectExecution(BACKGROUND_CONTEXT); + }); + } const entryId = await harness.sessionTree.appendMessage( { role: "user", content: "hello", timestamp: 1 }, @@ -361,6 +397,7 @@ describe("runtime2 AgentHarness", () => { expect(await harness.sessionTree.findEntriesOnBranch({ order: "oldestFirst" }, BACKGROUND_CONTEXT)).toMatchObject( [{ id: entryId, parentId: null, message: { role: "user", content: "hello" } }], ); + expect(events).toEqual(["message_start", "message_end", "entry_added"]); }); it("restores every configured lane without replacing its configuration", async () => { @@ -414,10 +451,13 @@ describe("runtime2 AgentHarness", () => { await harness.setThinkingLevel("low", BACKGROUND_CONTEXT); let durableConfigurationAtEvent: LaneConfiguration | undefined; let laneVisibleAtEvent = false; + let laneInspectionAtEvent: string | undefined; harness.events.on("lane_created", async () => { durableConfigurationAtEvent = (await session.getValue(storedValues.laneConfig("worker"), BACKGROUND_CONTEXT)) ?.value; - laneVisibleAtEvent = (await harness.lane("worker", BACKGROUND_CONTEXT)) !== undefined; + const published = await harness.lane("worker", BACKGROUND_CONTEXT); + laneVisibleAtEvent = published !== undefined; + laneInspectionAtEvent = (await published?.inspectExecution(BACKGROUND_CONTEXT))?.lane; }); const worker = unwrap(await harness.createLane("worker", anchor, BACKGROUND_CONTEXT)); @@ -432,12 +472,58 @@ describe("runtime2 AgentHarness", () => { activeToolNames: ["read"], }); expect(laneVisibleAtEvent).toBe(true); + expect(laneInspectionAtEvent).toBe("worker"); expect((await session.getValue(storedValues.laneState("worker"), BACKGROUND_CONTEXT))?.value).toEqual({ currentOperationId: null, pendingNextRun: [], }); }); + it("binds metadata and lane-creation recipients before releasing their mutation lines", async () => { + const session = await createSession(); + const { harness } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); + const blockStarted = deferred(); + const releaseBlock = deferred(); + let shouldBlock = true; + harness.events.on("config_update", async (event) => { + if (event.property !== "resources" || !shouldBlock) return; + shouldBlock = false; + blockStarted.resolve(); + await releaseBlock.promise; + }); + const blocking = harness.setResources({ skills: [] }, BACKGROUND_CONTEXT); + await blockStarted.promise; + + const naming = harness.sessionTree.setName("bound", BACKGROUND_CONTEXT); + while ((await session.getName(BACKGROUND_CONTEXT)) !== "bound") await Promise.resolve(); + const lateValues = vi.fn(); + harness.events.on("value_update", lateValues); + releaseBlock.resolve(); + await Promise.all([blocking, naming]); + expect(lateValues).not.toHaveBeenCalled(); + + const secondStarted = deferred(); + const secondRelease = deferred(); + let secondBlock = true; + harness.events.on("config_update", async (event) => { + if (event.property !== "streamOptions" || !secondBlock) return; + secondBlock = false; + secondStarted.resolve(); + await secondRelease.promise; + }); + const secondBlocking = harness.setStreamOptions({ timeoutMs: 1 }, BACKGROUND_CONTEXT); + await secondStarted.promise; + const creating = harness.createLane("worker", null, BACKGROUND_CONTEXT); + while ((await session.getValue(storedValues.laneLeaf("worker"), BACKGROUND_CONTEXT)) === undefined) { + await Promise.resolve(); + } + const lateLanes = vi.fn(); + harness.events.on("lane_created", lateLanes); + secondRelease.resolve(); + await Promise.all([secondBlocking, creating]); + expect(lateLanes).not.toHaveBeenCalled(); + }); + it("maps expected lane creation failures without faulting", async () => { const session = await createSession(); const { harness } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); @@ -518,150 +604,14 @@ describe("runtime2 AgentHarness", () => { await expect(worker.getLeafId(BACKGROUND_CONTEXT)).rejects.toBeInstanceOf(HarnessClosed); }); - it("classifies assistant-ready suspensions from captured identities", async () => { + it("attaches from the minimal projection without resolving identities or payload references", async () => { const session = await createSession(); const options = modelOptions(session); await configureMain(session); - const prompt = { role: "user" as const, content: "hello", timestamp: 1 }; - const promptId = await session.appendMessage(prompt, BACKGROUND_CONTEXT); - const meta = operationMeta(session, { kind: "run", promptEntryIds: [promptId] }, 2); - const { operationId } = meta; - const captured: LaneConfiguration = { - model: { provider: "missing-provider", modelId: "missing-model" }, - thinkingLevel: "off", - activeToolNames: ["missing-tool"], - }; - await installOperation(session, meta, { - ...runState(promptId), - phase: { - kind: "assistant", - generation: { - status: "ready", - context: { - stepId: session.idGenerator.next(), - triggerEntryId: promptId, - configuration: captured, - streamOptions: {}, - retryPolicy: { maxAttempts: 1, baseDelayMs: 0 }, - overflowRecoveryUsed: false, - }, - nextAttempt: 1, - }, - }, - }); - - const { harness, suspended } = await createAgentHarness( - { ...options, tools: [tool("available")] }, - BACKGROUND_CONTEXT, - ); - const expected = { - lane: "main", - operationId, - kind: "run" as const, - startedAt: 2, - prompt: [prompt], - reason: "crash" as const, - missing: { tools: ["missing-tool"], model: "missing-provider/missing-model" }, - }; - - expect(suspended).toEqual([expected]); - expect((await harness.inspectExecution(BACKGROUND_CONTEXT)).current?.suspended).toBe(suspended[0]); - const getEntries = vi.spyOn(session, "getEntries"); - await harness.inspectExecution(BACKGROUND_CONTEXT); - await harness.lanes(BACKGROUND_CONTEXT); - expect(getEntries).not.toHaveBeenCalled(); - }); - - it("classifies tool-phase suspensions from captured tools", async () => { - const session = await createSession(); - const options = modelOptions(session); - await configureMain(session); - const meta = operationMeta(session, { kind: "run", promptEntryIds: [] }); + const meta = operationMeta(session, { kind: "run", promptEntryIds: ["missing-prompt"] }, 2); const { operationId } = meta; await installOperation(session, meta, { - ...runState(session.idGenerator.next()), - phase: { - kind: "tools", - batch: { - assistantEntryId: session.idGenerator.next(), - configuration: { ...configuredMain, activeToolNames: ["missing-tool"] }, - turnId: session.idGenerator.next(), - calls: [], - }, - }, - }); - - const { harness, suspended } = await createAgentHarness(options, BACKGROUND_CONTEXT); - - expect(suspended).toEqual([ - { - lane: "main", - operationId, - kind: "run", - startedAt: 1, - prompt: [], - reason: "crash", - missing: { tools: ["missing-tool"] }, - }, - ]); - expect((await harness.inspectExecution(BACKGROUND_CONTEXT)).current?.suspended).toBe(suspended[0]); - }); - - it("restores deferred suspension handles from their source entries", async () => { - const session = await createSession(); - const options = modelOptions(session); - await configureMain(session); - const prompt = { role: "user" as const, content: "defer", timestamp: 1 }; - const promptId = await session.appendMessage(prompt, BACKGROUND_CONTEXT); - const handle = { - provider: options.model.provider, - modelId: options.model.id, - api: options.model.api, - id: "deferred-1", - }; - const sourceId = await session.appendMessage( - fauxAssistantMessage([], { stopReason: "deferred", deferred: handle }), - BACKGROUND_CONTEXT, - ); - const meta = operationMeta(session, { kind: "run", promptEntryIds: [promptId] }, 2); - const { operationId } = meta; - await installOperation(session, meta, { - ...runState(promptId), - phase: { - kind: "deferred", - deferred: { - status: "suspended", - stepId: session.idGenerator.next(), - sourceEntryId: sourceId, - poll: 0, - configuration: configuredMain, - streamOptions: {}, - }, - }, - }); - - const { harness, suspended } = await createAgentHarness(options, BACKGROUND_CONTEXT); - - expect(suspended).toEqual([ - { - lane: "main", - operationId, - kind: "run", - startedAt: 2, - prompt: [prompt], - reason: "deferred", - deferred: handle, - }, - ]); - expect((await harness.inspectExecution(BACKGROUND_CONTEXT)).current?.suspended).toBe(suspended[0]); - }); - - it("faults harness creation when restored suspension payloads are invalid", async () => { - const session = await createSession(); - await configureMain(session); - const meta = operationMeta(session, { kind: "run", promptEntryIds: [] }); - await installOperation(session, meta, { - ...runState(session.idGenerator.next()), + ...runState("missing-trigger"), phase: { kind: "deferred", deferred: { @@ -669,31 +619,61 @@ describe("runtime2 AgentHarness", () => { stepId: session.idGenerator.next(), sourceEntryId: "missing-source", poll: 0, - configuration: configuredMain, + configuration: { + model: { provider: "missing-provider", modelId: "missing-model" }, + thinkingLevel: "off", + activeToolNames: ["missing-tool"], + }, streamOptions: {}, }, }, }); + const getEntries = vi.spyOn(session, "getEntries"); + const getModel = vi.spyOn(options.models, "getModel"); - await expect(createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT)).rejects.toMatchObject({ - name: "HarnessFault", - cause: { name: "SessionInvariantError", message: "Deferred suspension source is invalid" }, - }); - - const missingPromptSession = await createSession(); - await configureMain(missingPromptSession); - await installOperation( - missingPromptSession, - operationMeta(missingPromptSession, { kind: "run", promptEntryIds: ["missing-prompt"] }), - runState(missingPromptSession.idGenerator.next()), + const { harness, open } = await createAgentHarness( + { ...options, tools: [tool("available")] }, + BACKGROUND_CONTEXT, ); - await expect(createAgentHarness(modelOptions(missingPromptSession), BACKGROUND_CONTEXT)).rejects.toMatchObject({ - name: "HarnessFault", - cause: { name: "SessionInvariantError", message: "Prompt entry missing-prompt is missing" }, + + expect(open).toEqual([{ lane: "main", operationId, kind: "run", startedAt: 2 }]); + expect(getEntries).not.toHaveBeenCalled(); + expect(getModel).not.toHaveBeenCalled(); + expect(await harness.inspectExecution(BACKGROUND_CONTEXT)).toMatchObject({ + configuredModel: configuredMain.model, + current: { + id: operationId, + status: "open", + capturedModel: { provider: "missing-provider", modelId: "missing-model" }, + }, }); }); - it("reports restored cancellation as aborting without a suspension descriptor", async () => { + it("does not classify unavailable captured request tools during attachment", async () => { + const session = await createSession(); + const options = modelOptions(session); + await configureMain(session); + const meta = operationMeta(session, { kind: "run", promptEntryIds: [] }); + const { operationId } = meta; + await installOperation(session, meta, { + ...runState("missing-trigger"), + phase: { + kind: "tools", + batch: { + assistantEntryId: "missing-assistant", + configuration: { ...configuredMain, activeToolNames: ["missing-tool"] }, + turnId: session.idGenerator.next(), + calls: [], + }, + }, + }); + + const { open } = await createAgentHarness(options, BACKGROUND_CONTEXT); + + expect(open).toEqual([{ lane: "main", operationId, kind: "run", startedAt: 1 }]); + }); + + it("marks restored cancellation as aborting in the open inventory", async () => { const session = await createSession(); await configureMain(session); const meta = operationMeta(session, { kind: "run", promptEntryIds: [] }); @@ -703,16 +683,21 @@ describe("runtime2 AgentHarness", () => { control: { status: "cancel_requested", requestedAt: 2, drainedSteer: [], drainedFollowUp: [] }, }); - const { harness, suspended } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); + const { harness, open } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); - expect(suspended).toHaveLength(1); + expect(open).toEqual([{ lane: "main", operationId, kind: "run", startedAt: 1, aborting: true }]); expect(await harness.inspectExecution(BACKGROUND_CONTEXT)).toEqual({ lane: "main", leafId: null, + configuredModel: configuredMain.model, current: { id: operationId, kind: "run", status: "aborting", startedAt: 1 }, }); expect(await harness.lanes(BACKGROUND_CONTEXT)).toEqual([ - { name: "main", leafId: null, operation: { id: operationId, kind: "run", status: "aborting" } }, + { + name: "main", + leafId: null, + operation: { id: operationId, kind: "run", status: "aborting", startedAt: 1 }, + }, ]); }); @@ -743,6 +728,7 @@ describe("runtime2 AgentHarness", () => { expect(await harness.inspectExecution(BACKGROUND_CONTEXT)).toEqual({ lane: "main", leafId: null, + configuredModel: configuredMain.model, current: null, lastResult, }); @@ -787,18 +773,21 @@ describe("runtime2 AgentHarness", () => { const state = runState(session.idGenerator.next()); await installOperation(session, meta, state); - const { harness, suspended } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); + const { harness, open } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); - expect(suspended).toEqual([ - { lane: "main", operationId, kind: "run", startedAt: 1, prompt: [], reason: "crash" }, - ]); + expect(open).toEqual([{ lane: "main", operationId, kind: "run", startedAt: 1 }]); expect(await harness.inspectExecution(BACKGROUND_CONTEXT)).toEqual({ lane: "main", leafId: null, - current: { id: operationId, kind: "run", status: "suspended", startedAt: 1, suspended: suspended[0] }, + configuredModel: configuredMain.model, + current: { id: operationId, kind: "run", status: "open", startedAt: 1 }, }); expect(await harness.lanes(BACKGROUND_CONTEXT)).toEqual([ - { name: "main", leafId: null, operation: { id: operationId, kind: "run", status: "suspended" } }, + { + name: "main", + leafId: null, + operation: { id: operationId, kind: "run", status: "open", startedAt: 1 }, + }, ]); const pendingId = await harness.sessionTree.appendCustomEntry("note", { text: "queued" }, BACKGROUND_CONTEXT); if (!(harness instanceof Harness)) throw new Error("missing runtime2 harness"); @@ -854,7 +843,7 @@ describe("runtime2 AgentHarness", () => { expect(await harness.getLeafId(BACKGROUND_CONTEXT)).toBeNull(); expect(await harness.getLastResult(BACKGROUND_CONTEXT)).toBeUndefined(); expect(forbidden).not.toHaveBeenCalled(); - expect(mutate).not.toHaveBeenCalled(); + expect(mutate).toHaveBeenCalledOnce(); expect(getValue).not.toHaveBeenCalled(); expect(scanValues).not.toHaveBeenCalled(); expect(readList).not.toHaveBeenCalled(); @@ -884,9 +873,9 @@ describe("runtime2 AgentHarness", () => { phase: { kind: "ready_to_commit" }, }; await installOperation(session, meta, state); - const { harness, suspended } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); + const { harness, open } = await createAgentHarness(modelOptions(session), BACKGROUND_CONTEXT); - expect(suspended).toEqual([{ lane: "main", operationId, kind: "navigation", startedAt: 1, reason: "crash" }]); + expect(open).toEqual([{ lane: "main", operationId, kind: "navigation", startedAt: 1 }]); await expect( harness.sessionTree.appendMessage({ role: "user", content: "later", timestamp: 2 }, BACKGROUND_CONTEXT), ).rejects.toThrow(`Cannot append while structural operation ${operationId} is active`); diff --git a/packages/agent/test/harness/runtime2/lane.test.ts b/packages/agent/test/harness/runtime2/lane.test.ts index dc1686ad3..4f7c3f4ad 100644 --- a/packages/agent/test/harness/runtime2/lane.test.ts +++ b/packages/agent/test/harness/runtime2/lane.test.ts @@ -1,6 +1,7 @@ import { type Api, createModels, fauxProvider, type Model } from "@earendil-works/pi-ai"; import { afterEach, describe, expect, it } from "vitest"; import { HarnessClosed } from "../../../src/harness/agent-harness.ts"; +import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; import { BACKGROUND_CONTEXT } from "../../../src/harness/context.ts"; import { Lane } from "../../../src/harness/runtime2/lane.ts"; import { restoreLane } from "../../../src/harness/runtime2/restore.ts"; @@ -51,7 +52,15 @@ async function createLane(): Promise<{ models, await restoreLane(session, "main", BACKGROUND_CONTEXT), (cause) => (cause instanceof Error ? cause : new Error(String(cause))), - () => Promise.resolve(), + () => ({ start: () => Promise.resolve() }), + (snapshot) => ({ snapshot, start: () => {}, unsubscribe: () => {} }), + () => ({ + resources: {}, + compaction: DEFAULT_COMPACTION_SETTINGS, + steeringMode: "all", + followUpMode: "all", + toolExecution: "parallel", + }), ), model: faux.getModel(), session, diff --git a/packages/agent/test/harness/runtime2/restore.test.ts b/packages/agent/test/harness/runtime2/restore.test.ts index cdea7e278..ab3a496c2 100644 --- a/packages/agent/test/harness/runtime2/restore.test.ts +++ b/packages/agent/test/harness/runtime2/restore.test.ts @@ -112,6 +112,44 @@ describe("runtime2 lane restore", () => { expect(restored.operation).toEqual({ meta, state }); }); + it("validates current-operation identity, lane ownership, and intent/state compatibility", async () => { + for (const corruption of ["identity", "lane", "kind"] as const) { + const session = await createSession(); + const operationId = session.idGenerator.next(); + const base: OperationMeta = { + operationId, + lane: "main", + sourceLeafId: null, + startedAt: 1, + intent: { kind: "run", promptEntryIds: [] }, + }; + const meta: OperationMeta = + corruption === "identity" + ? { ...base, operationId: "different-operation" } + : corruption === "lane" + ? { ...base, lane: "worker" } + : { ...base, intent: { kind: "navigation", targetId: null, summarize: false } }; + await session.mutate( + "main", + (mutator) => + mutator.commit( + [ + storedValues.setValue(storedValues.operationMeta(operationId), meta), + storedValues.setValue(storedValues.operationState(operationId), runState("trigger")), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: operationId, + pendingNextRun: [], + }), + ], + BACKGROUND_CONTEXT, + ), + BACKGROUND_CONTEXT, + ); + + await expect(restoreLane(session, "main", BACKGROUND_CONTEXT)).rejects.toBeInstanceOf(Error); + } + }); + it.each([ [storedValues.laneLeaf("main").namespace, storedValues.deleteValue(storedValues.laneLeaf("main"))], [storedValues.laneConfig("main").namespace, storedValues.deleteValue(storedValues.laneConfig("main"))], diff --git a/packages/agent/test/harness/runtime2/watch.test.ts b/packages/agent/test/harness/runtime2/watch.test.ts new file mode 100644 index 000000000..fbffc3f65 --- /dev/null +++ b/packages/agent/test/harness/runtime2/watch.test.ts @@ -0,0 +1,466 @@ +import { type AssistantMessageFrame, createModels, fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { type HarnessEvent, HarnessFault, type WatchHandle } from "../../../src/harness/agent-harness.ts"; +import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; +import { BACKGROUND_CONTEXT, type Context, createContextKey, withContextValue } from "../../../src/harness/context.ts"; +import { createAgentHarness, Harness } from "../../../src/harness/runtime2/harness.ts"; +import * as sessionWrites from "../../../src/harness/session/commit.ts"; +import { MemoryStorage } from "../../../src/harness/session/memory.ts"; +import { StorageBackedSession } from "../../../src/harness/session/session.ts"; +import type { + OperationMeta, + RunPhase, + RunState, + Session, + StorageBranchScan, + Write, +} from "../../../src/harness/session/types.ts"; +import * as storedValues from "../../../src/harness/session/values.ts"; +import { deferred } from "./test-utils.ts"; + +const sessions: Session[] = []; +const configuration = { + model: { provider: "configured", modelId: "model" }, + thinkingLevel: "off" as const, + activeToolNames: [], +}; + +async function createSession(storage: MemoryStorage = new MemoryStorage()): Promise { + const session = new StorageBackedSession( + { id: `watch-${sessions.length}`, createdAt: 1, storageVersion: 1 }, + storage, + ); + sessions.push(session); + await session.mutate( + "main", + (mutator) => + mutator.commit( + [ + storedValues.setValue(storedValues.laneLeaf("main"), null), + storedValues.setValue(storedValues.laneConfig("main"), configuration), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: null, + pendingNextRun: [], + }), + ], + BACKGROUND_CONTEXT, + ), + BACKGROUND_CONTEXT, + ); + return session; +} + +async function commit(session: Session, writes: Write[]): Promise { + await session.mutate("main", (mutator) => mutator.commit(writes, BACKGROUND_CONTEXT), BACKGROUND_CONTEXT); +} + +function runState(phase: RunPhase, control: RunState["control"] = { status: "running" }): RunState { + return { + kind: "run", + control, + settings: { + compaction: DEFAULT_COMPACTION_SETTINGS, + steeringMode: "all", + followUpMode: "all", + toolExecution: "parallel", + }, + phase, + inbox: { steer: [], followUp: [], writes: [] }, + latestAssistantEntryId: null, + }; +} + +async function attach(session: Session): Promise> { + const provider = fauxProvider(); + const models = createModels(); + models.setProvider(provider.provider); + const { harness } = await createAgentHarness({ session, models, model: provider.getModel() }, BACKGROUND_CONTEXT); + if (!(harness instanceof Harness)) throw new Error("Expected runtime2 Harness"); + return harness; +} + +class BlockingScanStorage extends MemoryStorage { + block = false; + readonly started = deferred(); + readonly release = deferred(); + + override async scanBranch(query: StorageBranchScan, context: Context) { + if (this.block) { + this.started.resolve(); + await this.release.promise; + } + return super.scanBranch(query, context); + } +} + +afterEach(async () => { + for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); +}); + +describe("runtime2 lane watch", () => { + it("captures a compaction-bounded transcript and isolates the returned snapshot", async () => { + const session = await createSession(); + await commit(session, [ + sessionWrites.insertEntry({ id: "root", parentId: null, type: "custom", customType: "root" }), + sessionWrites.insertEntry({ + id: "compact", + parentId: "root", + type: "compaction", + summary: "summary", + retainedTail: [], + tokensBefore: 10, + fromHook: false, + }), + sessionWrites.insertEntry({ + id: "after", + parentId: "compact", + type: "message", + message: { role: "user", content: "after", timestamp: 2 }, + }), + storedValues.setValue(storedValues.laneLeaf("main"), "after"), + ]); + const harness = await attach(session); + + const first = await harness.watch(BACKGROUND_CONTEXT); + expect(first.snapshot.transcript.map(({ id }) => id)).toEqual(["compact", "after"]); + expect(first.snapshot).toMatchObject({ + lane: "main", + leafId: "after", + operation: null, + queues: { steer: [], followUp: [], nextRun: [] }, + pendingWrites: [], + faulted: false, + }); + first.snapshot.transcript.length = 0; + first.snapshot.leafId = null; + const second = await harness.watch(BACKGROUND_CONTEXT); + expect(second.snapshot.transcript.map(({ id }) => id)).toEqual(["compact", "after"]); + expect(second.snapshot.leafId).toBe("after"); + first.unsubscribe(); + second.unsubscribe(); + }); + + it("dereferences queues, pending writes, deferred handles, and abort drains", async () => { + const session = await createSession(); + const handle = { provider: "provider", modelId: "model", api: "test", id: "deferred" }; + const sourceMessage = fauxAssistantMessage([], { stopReason: "deferred", deferred: handle }); + const ids = { + next: "next", + steer: "steer", + follow: "follow", + write: "write", + drainedSteer: "drained-steer", + drainedFollow: "drained-follow", + }; + const payload = (content: string) => ({ + type: "message" as const, + payload: { role: "user" as const, content, timestamp: 1 }, + }); + const operationId = session.idGenerator.next(); + const state = runState( + { + kind: "deferred", + deferred: { + status: "suspended", + stepId: "step", + sourceEntryId: "source", + poll: 3, + configuration, + streamOptions: {}, + }, + }, + { + status: "cancel_requested", + requestedAt: 2, + drainedSteer: [ids.drainedSteer], + drainedFollowUp: [ids.drainedFollow], + }, + ); + state.inbox = { steer: [ids.steer], followUp: [ids.follow], writes: [ids.write] }; + const meta: OperationMeta = { + operationId, + lane: "main", + sourceLeafId: null, + startedAt: 1, + intent: { kind: "run", promptEntryIds: [] }, + }; + await commit(session, [ + sessionWrites.insertEntry({ id: "source", parentId: null, type: "message", message: sourceMessage }), + ...Object.values(ids).map((id) => + storedValues.setValue( + storedValues.pendingEntry(id), + id === ids.write ? { type: "custom", customType: "note", payload: { id } } : payload(id), + ), + ), + storedValues.setValue(storedValues.operationMeta(operationId), meta), + storedValues.setValue(storedValues.operationState(operationId), state), + storedValues.setValue(storedValues.laneLeaf("main"), "source"), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: operationId, + pendingNextRun: [ids.next], + }), + ]); + const harness = await attach(session); + + const watch = await harness.watch(BACKGROUND_CONTEXT); + + expect(watch.snapshot.queues.steer).toMatchObject([{ entryId: ids.steer, message: { content: ids.steer } }]); + expect(watch.snapshot.queues.followUp).toMatchObject([{ entryId: ids.follow, message: { content: ids.follow } }]); + expect(watch.snapshot.queues.nextRun).toMatchObject([{ entryId: ids.next, message: { content: ids.next } }]); + expect(watch.snapshot.pendingWrites).toEqual([ + { entryId: ids.write, type: "custom", customType: "note", data: { id: ids.write } }, + ]); + expect(watch.snapshot.operation).toMatchObject({ + id: operationId, + status: "aborting", + deferred: { handle, poll: 3 }, + drained: { + steer: [{ entryId: ids.drainedSteer }], + followUp: [{ entryId: ids.drainedFollow }], + }, + runningTools: [], + }); + watch.unsubscribe(); + }); + + it("omits streaming presentation when an effect-pending response has no frames", async () => { + const session = await createSession(); + const operationId = session.idGenerator.next(); + await commit(session, [ + storedValues.setValue(storedValues.operationMeta(operationId), { + operationId, + lane: "main", + sourceLeafId: null, + startedAt: 1, + intent: { kind: "run", promptEntryIds: [] }, + }), + storedValues.setValue( + storedValues.operationState(operationId), + runState({ + kind: "assistant", + generation: { + status: "effect_pending", + context: { + stepId: "step", + triggerEntryId: "trigger", + configuration, + streamOptions: {}, + retryPolicy: { maxAttempts: 2, baseDelayMs: 0 }, + overflowRecoveryUsed: false, + }, + attempt: 1, + responseEntryId: "response-without-frames", + usageId: "usage", + intendedOutputLimit: 100, + contextWindow: 1000, + }, + }), + ), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: operationId, + pendingNextRun: [], + }), + ]); + const harness = await attach(session); + + const watch = await harness.watch(BACKGROUND_CONTEXT); + + expect(watch.snapshot.operation).not.toHaveProperty("streamingMessage"); + watch.unsubscribe(); + }); + + it("reduces assistant frames and renders only effect-pending tools with full-content indexes", async () => { + const frameSession = await createSession(); + const partial = fauxAssistantMessage([], { stopReason: "pending" }); + const frames: AssistantMessageFrame[] = [ + { type: "start", partial }, + { type: "text_start", contentIndex: 0, content: { type: "text", text: "" } }, + { type: "text_delta", contentIndex: 0, delta: "partial" }, + ]; + const frameOperationId = frameSession.idGenerator.next(); + await commit(frameSession, [ + storedValues.setValue(storedValues.operationMeta(frameOperationId), { + operationId: frameOperationId, + lane: "main", + sourceLeafId: null, + startedAt: 1, + intent: { kind: "run", promptEntryIds: [] }, + }), + storedValues.setValue( + storedValues.operationState(frameOperationId), + runState({ + kind: "assistant", + generation: { + status: "effect_pending", + context: { + stepId: "step", + triggerEntryId: "trigger", + configuration, + streamOptions: {}, + retryPolicy: { maxAttempts: 2, baseDelayMs: 0 }, + overflowRecoveryUsed: false, + }, + attempt: 1, + responseEntryId: "response", + usageId: "usage", + intendedOutputLimit: 100, + contextWindow: 1000, + }, + }), + ), + ...frames.map((frame) => + storedValues.appendList(storedValues.pendingAssistantFrames(frameOperationId, "response"), frame), + ), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: frameOperationId, + pendingNextRun: [], + }), + ]); + const frameHarness = await attach(frameSession); + const frameWatch = await frameHarness.watch(BACKGROUND_CONTEXT); + expect(frameWatch.snapshot.operation?.streamingMessage?.content).toEqual([{ type: "text", text: "partial" }]); + frameWatch.unsubscribe(); + + const toolSession = await createSession(); + const assistant = fauxAssistantMessage([ + { type: "text", text: "before" }, + { type: "toolCall", id: "call", name: "read", arguments: { live: false } }, + { type: "toolCall", id: "call-without-checkpoint", name: "write", arguments: { live: false } }, + ]); + const toolOperationId = toolSession.idGenerator.next(); + await commit(toolSession, [ + sessionWrites.insertEntry({ id: "assistant", parentId: null, type: "message", message: assistant }), + storedValues.setValue(storedValues.operationMeta(toolOperationId), { + operationId: toolOperationId, + lane: "main", + sourceLeafId: null, + startedAt: 1, + intent: { kind: "run", promptEntryIds: [] }, + }), + storedValues.setValue( + storedValues.operationState(toolOperationId), + runState({ + kind: "tools", + batch: { + assistantEntryId: "assistant", + configuration, + turnId: "turn", + calls: [ + { status: "planned", sourceIndex: 1, resultEntryId: "planned" }, + { status: "effect_pending", sourceIndex: 1, resultEntryId: "result", replay: "safe" }, + { + status: "effect_pending", + sourceIndex: 2, + resultEntryId: "without-checkpoint", + replay: "never", + }, + { status: "outcome_ready", sourceIndex: 1, resultEntryId: "ready", terminate: false }, + ], + }, + }), + ), + storedValues.setValue(storedValues.operationToolArgs(toolOperationId, "turn", 1), { path: "file" }), + storedValues.setValue(storedValues.operationToolArgs(toolOperationId, "turn", 2), { path: "output" }), + storedValues.setValue(storedValues.pendingToolOutput(toolOperationId, "result"), { + content: [{ type: "text", text: "partial" }], + details: { bytes: 1 }, + }), + storedValues.setValue(storedValues.laneLeaf("main"), "assistant"), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: toolOperationId, + pendingNextRun: [], + }), + ]); + const toolHarness = await attach(toolSession); + const toolWatch = await toolHarness.watch(BACKGROUND_CONTEXT); + expect(toolWatch.snapshot.operation?.runningTools).toEqual([ + { + toolCallId: "call", + toolName: "read", + args: { path: "file" }, + partialResult: { content: [{ type: "text", text: "partial" }], details: { bytes: 1 } }, + }, + { + toolCallId: "call-without-checkpoint", + toolName: "write", + args: { path: "output" }, + }, + ]); + expect(toolWatch.snapshot.operation?.runningTools[1]).not.toHaveProperty("partialResult"); + toolWatch.unsubscribe(); + }); + + it("faults required payload corruption and unsubscribes the incomplete watcher", async () => { + const session = await createSession(); + await commit(session, [ + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: null, + pendingNextRun: ["missing"], + }), + ]); + const harness = await attach(session); + let unsubscribed = false; + const originalWatch = harness.events.watch.bind(harness.events); + vi.spyOn(harness.events, "watch").mockImplementation( + (snapshot: T, filter: (event: HarnessEvent) => boolean, context: Context): WatchHandle => { + const handle = originalWatch(snapshot, filter, context); + return { + ...handle, + unsubscribe: () => { + unsubscribed = true; + handle.unsubscribe(); + }, + }; + }, + ); + + await expect(harness.watch(BACKGROUND_CONTEXT)).rejects.toBeInstanceOf(HarnessFault); + expect(unsubscribed).toBe(true); + }); + + it("returns snapshot-before plus buffered events when watch wins the lane line", async () => { + const storage = new BlockingScanStorage(); + const session = await createSession(storage); + await commit(session, [ + sessionWrites.insertEntry({ id: "root", parentId: null, type: "custom", customType: "root" }), + storedValues.setValue(storedValues.laneLeaf("main"), "root"), + ]); + const harness = await attach(session); + storage.block = true; + const watchPromise = harness.watch(BACKGROUND_CONTEXT); + await storage.started.promise; + const sourceKey = createContextKey("watch.event.source"); + const sourceContext = withContextValue(sourceKey, "append", BACKGROUND_CONTEXT); + const append = harness.sessionTree.appendMessage({ role: "user", content: "later", timestamp: 2 }, sourceContext); + storage.release.resolve(); + const watch = await watchPromise; + expect(watch.snapshot.transcript.map(({ id }) => id)).toEqual(["root"]); + const seen: Array<{ type: string; sameContext: boolean }> = []; + watch.start((event, eventContext) => { + seen.push({ type: event.type, sameContext: eventContext === sourceContext }); + }); + await append; + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(seen.map(({ type }) => type)).toEqual(["message_start", "message_end", "entry_added"]); + expect(seen.every(({ sameContext }) => sameContext)).toBe(true); + watch.unsubscribe(); + }); + + it("returns snapshot-after without replay when publication wins", async () => { + const session = await createSession(); + const harness = await attach(session); + const entryId = await harness.sessionTree.appendMessage( + { role: "user", content: "existing", timestamp: 1 }, + BACKGROUND_CONTEXT, + ); + + const watch = await harness.watch(BACKGROUND_CONTEXT); + const seen = vi.fn(); + watch.start(seen); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(watch.snapshot.transcript.map(({ id }) => id)).toEqual([entryId]); + expect(seen).not.toHaveBeenCalled(); + watch.unsubscribe(); + }); +}); diff --git a/packages/agent/test/harness/session-create-lane.test.ts b/packages/agent/test/harness/session-create-lane.test.ts index 9d3bd8250..3dcc5f3a0 100644 --- a/packages/agent/test/harness/session-create-lane.test.ts +++ b/packages/agent/test/harness/session-create-lane.test.ts @@ -5,7 +5,11 @@ import { BACKGROUND_CONTEXT } from "../../src/harness/context.ts"; import { describe, expect, expectTypeOf, it } from "vitest"; import * as sessionWrites from "../../src/harness/session/commit.ts"; import { MemoryStorage } from "../../src/harness/session/memory.ts"; -import { SessionInvariantError, StorageBackedSession } from "../../src/harness/session/session.ts"; +import { + createLaneWithMutator, + SessionInvariantError, + StorageBackedSession, +} from "../../src/harness/session/session.ts"; import { InstrumentedStorage, StorageDecorator } from "../../src/harness/session/testing/index.ts"; import type { Entry, @@ -101,6 +105,32 @@ class BlockingCommitStorage extends StorageDecorator { describe("StorageBackedSession.createLane", () => { it("completes the package-internal Session contract", () => { expectTypeOf().toMatchTypeOf(); + expectTypeOf(createLaneWithMutator).returns.toEqualTypeOf>(); + }); + + it("exposes the same procedure to callers that own the lane mutation", async () => { + const storage = new InstrumentedStorage(new MemoryStorage({ now: () => NOW })); + const session = new StorageBackedSession(metadata, storage); + await commitSession(session, rootTransaction()); + storage.clearCommitAttempts(); + + await session.mutate( + "shared", + (mutator) => createLaneWithMutator(mutator, "shared", ROOT_ID, configuration, BACKGROUND_CONTEXT), + BACKGROUND_CONTEXT, + ); + + expect(storage.getCommitAttempts()).toEqual([expectedLaneWrites("shared", ROOT_ID, configuration)]); + await expect(session.view("shared").getLeafId(BACKGROUND_CONTEXT)).resolves.toBe(ROOT_ID); + await expect( + session.mutate( + "different-line", + (mutator) => createLaneWithMutator(mutator, "other", null, configuration, BACKGROUND_CONTEXT), + BACKGROUND_CONTEXT, + ), + ).rejects.toBeInstanceOf(SessionInvariantError); + expect(storage.getCommitAttempts()).toHaveLength(1); + await session.close(BACKGROUND_CONTEXT); }); it("atomically creates configured lane views at an entry or at the root", async () => { diff --git a/packages/agent/test/harness/storage-backed-session.test.ts b/packages/agent/test/harness/storage-backed-session.test.ts index 27d821786..e2337c6ea 100644 --- a/packages/agent/test/harness/storage-backed-session.test.ts +++ b/packages/agent/test/harness/storage-backed-session.test.ts @@ -93,6 +93,37 @@ describe("StorageBackedSession", () => { await session.close(BACKGROUND_CONTEXT); }); + it("exposes explicit branch scans through the Session and callback-scoped mutator", async () => { + const storage = new MemoryStorage({ now: () => NOW }); + const session = new StorageBackedSession(metadata, storage); + const childId = "00000000-0000-7000-8000-000000000002"; + await commitSession(session, [ + sessionWrites.insertEntry({ id: ENTRY_ID, parentId: null, type: "custom", customType: "root" }), + sessionWrites.insertEntry({ id: childId, parentId: ENTRY_ID, type: "custom", customType: "child" }), + ]); + + await expect( + session.scanBranch({ start: childId, order: "oldestFirst" }, BACKGROUND_CONTEXT), + ).resolves.toMatchObject([{ id: ENTRY_ID }, { id: childId }]); + let captured: SessionMutator | undefined; + await session.mutate( + "main", + async (mutator) => { + captured = mutator; + await expect(mutator.scanBranch({ start: childId, limit: 1 }, BACKGROUND_CONTEXT)).resolves.toMatchObject([ + { id: childId }, + ]); + }, + BACKGROUND_CONTEXT, + ); + const invalidated = captured; + if (invalidated === undefined) throw new Error("Expected captured mutator"); + expect(() => invalidated.scanBranch({ start: childId }, BACKGROUND_CONTEXT)).toThrow( + "outside its mutation callback", + ); + await session.close(BACKGROUND_CONTEXT); + }); + it("rejects pending assistant entries at the durable session write boundary", async () => { const storage = new InstrumentedStorage(new MemoryStorage({ now: () => NOW })); const session = new StorageBackedSession(metadata, storage); @@ -226,5 +257,6 @@ describe("StorageBackedSession", () => { await expect(session.scanValues(storedValues.sessionName, BACKGROUND_CONTEXT)).rejects.toThrow( "Session is closed", ); + await expect(session.scanBranch({ start: ENTRY_ID }, BACKGROUND_CONTEXT)).rejects.toThrow("Session is closed"); }); }); diff --git a/packages/agent/test/harness/types.test.ts b/packages/agent/test/harness/types.test.ts index c2a0c02ce..ea832a410 100644 --- a/packages/agent/test/harness/types.test.ts +++ b/packages/agent/test/harness/types.test.ts @@ -18,6 +18,7 @@ import type { CustomEntry, Deferred, DriveResult, + Entry, EntryProjector, EntryWrite, Generation, @@ -45,6 +46,7 @@ import type { SessionCreateOptions, SessionMetadata, SessionMutator, + SessionReader, SessionRepo, SessionSearchHit, SessionSearchService, @@ -52,6 +54,7 @@ import type { SessionTree, SettledAssistantMessage, Storage, + StorageBranchScan, StructuralDecision, SummaryContext, SummaryGeneration, @@ -110,7 +113,8 @@ const generations = [ const toolCalls = [ { status: "planned", sourceIndex: 0, resultEntryId: "result-0" }, { status: "effect_pending", sourceIndex: 1, resultEntryId: "result-1", replay: "safe" }, - { status: "completed", sourceIndex: 2, resultEntryId: "result-2", terminate: false }, + { status: "outcome_ready", sourceIndex: 2, resultEntryId: "result-2", terminate: true }, + { status: "completed", sourceIndex: 3, resultEntryId: "result-3", terminate: false }, ] satisfies ToolCall[]; const deferredStates = [ { @@ -148,7 +152,7 @@ const structuralDecisions = [ { status: "generating", taskId: "task", generation: summaryGenerations[0] }, ] satisfies StructuralDecision[]; -const runPhases = [ +const runPhases: RunPhase[] = [ { kind: "starting" }, checkpoint, { kind: "assistant", generation: generations[0] }, @@ -168,7 +172,25 @@ const runPhases = [ error: { code: "provider", message: "failed" }, provenance: { kind: "response", entryId: "response" }, }, -] satisfies RunPhase[]; + { + kind: "failure_drain", + error: { + code: "model_unavailable", + message: "Model is unavailable", + details: { provider: "provider", modelId: "model" }, + }, + provenance: { kind: "configuration" }, + }, + { + kind: "failure_drain", + error: { + code: "configured_tools_unavailable", + message: "Configured tools are unavailable", + details: { tools: ["read"] }, + }, + provenance: { kind: "configuration" }, + }, +]; const runState = { kind: "run", @@ -179,7 +201,7 @@ const runState = { followUpMode: "one-at-a-time", toolExecution: "parallel", }, - phase: runPhases[0], + phase: runPhases[0]!, inbox: { steer: [], followUp: [], writes: [] }, latestAssistantEntryId: null, } satisfies RunState; @@ -312,7 +334,7 @@ it("covers the complete durable storage and Part 3 discriminants", () => { expectTypeOf().toEqualTypeOf<"run" | "compaction" | "navigation">(); expectTypeOf().toEqualTypeOf<"running" | "cancel_requested">(); expectTypeOf().toEqualTypeOf<"ready" | "effect_pending" | "retry_wait">(); - expectTypeOf().toEqualTypeOf<"planned" | "effect_pending" | "completed">(); + expectTypeOf().toEqualTypeOf<"planned" | "effect_pending" | "outcome_ready" | "completed">(); expectTypeOf().toEqualTypeOf<"suspended" | "effect_pending">(); expectTypeOf().toEqualTypeOf<"ready" | "effect_pending" | "retry_wait">(); expectTypeOf().toEqualTypeOf<"deciding" | "generating">(); @@ -363,6 +385,9 @@ it("covers storage, session, repository, search, and identity signatures", () => context: Context, ) => Promise<{ firstSeq: number; seqs: number[]; timestamp: number }> >(); + expectTypeOf().toEqualTypeOf< + (query: StorageBranchScan, context: Context) => Promise + >(); expectTypeOf().toEqualTypeOf< (name: string, at: string | null, laneConfiguration: LaneConfiguration, context: Context) => Promise >(); @@ -379,7 +404,7 @@ it("covers Part 5 results, events, hooks, snapshots, tools, and stream options", type RunErrorTag = Extract["error"]["_tag"]; type CancelKind = Extract["value"]["kind"]; expectTypeOf().toEqualTypeOf< - "LaneBusy" | "MissingIdentities" | "InvalidMessage" | "UnknownSkill" | "UnknownTemplate" | "Closed" + "LaneBusy" | "InvalidMessage" | "UnknownSkill" | "UnknownTemplate" | "Closed" >(); expectTypeOf().toEqualTypeOf<"cancelled" | "already_consumed" | "not_found">(); expectTypeOf().toEqualTypeOf< @@ -408,9 +433,7 @@ it("covers Part 5 results, events, hooks, snapshots, tools, and stream options", | "config_update" | "compaction_start" | "compaction_end" - | "compaction_suspend" | "navigation_start" - | "navigation_suspend" | "navigation_end" | "lane_created" | "usage" diff --git a/packages/coding-agent/src/experimental/harness-wire-adapter.ts b/packages/coding-agent/src/experimental/harness-wire-adapter.ts index c2648bb42..cb84a6ea9 100644 --- a/packages/coding-agent/src/experimental/harness-wire-adapter.ts +++ b/packages/coding-agent/src/experimental/harness-wire-adapter.ts @@ -289,24 +289,24 @@ function toWireRunValue(value: Extract["value"]) }; } case "suspended": - if (value.reason === "deferred") { - return { - kind: "suspended", - reason: "deferred", - runId: value.runId, - leafId: value.leafId, - finalEntryId: value.finalEntryId, - deferred: toWireDeferred(value.deferred), - }; - } return { kind: "suspended", - reason: "missing_identities", + reason: "deferred", runId: value.runId, leafId: value.leafId, - missing: { - tools: [...value.missing.tools], - models: value.missing.model === undefined ? [] : [value.missing.model], + finalEntryId: value.finalEntryId, + deferred: toWireDeferred(value.deferred), + }; + case "action_required": + return { + kind: "action_required", + runId: value.runId, + action: { + kind: value.action.kind, + description: value.action.description, + ...(value.action.details === undefined + ? {} + : { details: toWireJsonValue(value.action.details, "action details") }), }, }; default: @@ -324,14 +324,6 @@ function toWireRunError(error: Extract["error"] operationKind: error.operationKind, message: error.message, }; - case "MissingIdentities": - return { - _tag: "MissingIdentities", - lane: error.lane, - tools: [...error.tools], - models: error.model === undefined ? [] : [error.model], - message: error.message, - }; case "InvalidMessage": return { _tag: "InvalidMessage", lane: error.lane, reason: error.reason, message: error.message }; case "UnknownSkill": @@ -647,42 +639,6 @@ function toWireFrame(event: Extract): } } -function toWireMissing(missing: { tools: readonly string[]; model?: string }): { tools: string[]; models: string[] } { - return { tools: [...missing.tools], models: missing.model === undefined ? [] : [missing.model] }; -} - -function toWireSuspendedOperation( - suspended: NonNullable["suspended"]>, -): NonNullable["suspended"]> { - const base = { - lane: suspended.lane, - operationId: suspended.operationId, - kind: suspended.kind, - startedAt: suspended.startedAt, - ...(suspended.prompt === undefined ? {} : { prompt: suspended.prompt.map(toWireMessage) }), - ...(suspended.aborting === undefined - ? {} - : { - aborting: { - steer: suspended.aborting.steer.map(toWireMessage), - followUp: suspended.aborting.followUp.map(toWireMessage), - }, - }), - }; - if (suspended.reason === "deferred") { - return { ...base, reason: "deferred", deferred: toWireDeferred(suspended.deferred) }; - } - if (suspended.reason === "missing_identities") { - return { ...base, reason: "missing_identities", missing: toWireMissing(suspended.missing) }; - } - return { - ...base, - reason: "crash", - ...(suspended.deferred === undefined ? {} : { deferred: toWireDeferred(suspended.deferred) }), - ...(suspended.missing === undefined ? {} : { missing: toWireMissing(suspended.missing) }), - }; -} - export function toWireLaneSnapshot(snapshot: HarnessLaneSnapshot): LaneSnapshot { return { lane: snapshot.lane, @@ -696,9 +652,44 @@ export function toWireLaneSnapshot(snapshot: HarnessLaneSnapshot): LaneSnapshot kind: snapshot.operation.kind, status: snapshot.operation.status, startedAt: snapshot.operation.startedAt, - ...(snapshot.operation.suspended === undefined + ...(snapshot.operation.action === undefined ? {} - : { suspended: toWireSuspendedOperation(snapshot.operation.suspended) }), + : { + action: { + kind: snapshot.operation.action.kind, + description: snapshot.operation.action.description, + ...(snapshot.operation.action.details === undefined + ? {} + : { + details: toWireJsonValue( + snapshot.operation.action.details, + "snapshot action details", + ), + }), + }, + }), + ...(snapshot.operation.deferred === undefined + ? {} + : { + deferred: { + handle: toWireDeferred(snapshot.operation.deferred.handle), + poll: snapshot.operation.deferred.poll, + }, + }), + ...(snapshot.operation.drained === undefined + ? {} + : { + drained: { + steer: snapshot.operation.drained.steer.map((item) => ({ + entryId: item.entryId, + message: toWireMessage(item.message), + })), + followUp: snapshot.operation.drained.followUp.map((item) => ({ + entryId: item.entryId, + message: toWireMessage(item.message), + })), + }, + }), ...(snapshot.operation.streamingMessage === undefined ? {} : { streamingMessage: toWireAssistantMessage(snapshot.operation.streamingMessage) }), @@ -742,23 +733,14 @@ export function toWireLaneEvent(event: HarnessEvent): LaneEvent | undefined { case "run_resume": return { type: event.type, runId: event.runId, ...base, lane: event.lane }; case "run_suspend": - return event.reason === "deferred" - ? { - type: "run_suspend", - runId: event.runId, - reason: "deferred", - deferred: event.deferred, - ...base, - lane: event.lane, - } - : { - type: "run_suspend", - runId: event.runId, - reason: "missing_identities", - missing: toWireMissing(event.missing), - ...base, - lane: event.lane, - }; + return { + type: "run_suspend", + runId: event.runId, + reason: "deferred", + deferred: toWireDeferred(event.deferred), + ...base, + lane: event.lane, + }; case "run_abort": return { type: "run_abort", @@ -818,9 +800,7 @@ export function toWireLaneEvent(event: HarnessEvent): LaneEvent | undefined { case "config_update": case "compaction_start": case "compaction_end": - case "compaction_suspend": case "navigation_start": - case "navigation_suspend": case "navigation_end": case "lane_created": case "usage": diff --git a/packages/coding-agent/src/experimental/session-worker.ts b/packages/coding-agent/src/experimental/session-worker.ts index c9dd73c0c..13d0ec825 100644 --- a/packages/coding-agent/src/experimental/session-worker.ts +++ b/packages/coding-agent/src/experimental/session-worker.ts @@ -576,18 +576,12 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo harness.events.on("compaction_start", (event) => lifecycle?.operationStarted("compaction", event.lane, event.runId), ), - harness.events.on("compaction_suspend", (event) => - lifecycle?.operationStopped("compaction", event.lane, event.runId), - ), harness.events.on("compaction_end", (event) => lifecycle?.operationStopped("compaction", event.lane, event.runId), ), harness.events.on("navigation_start", (event) => lifecycle?.operationStarted("navigation", event.lane, event.runId), ), - harness.events.on("navigation_suspend", (event) => - lifecycle?.operationStopped("navigation", event.lane, event.runId), - ), harness.events.on("navigation_end", (event) => lifecycle?.operationStopped("navigation", event.lane, event.runId), ), diff --git a/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts b/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts index c88f259a4..a2be44672 100644 --- a/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts +++ b/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts @@ -5,7 +5,6 @@ import { type RunResult as HarnessRunResult, InvalidMessage, LaneBusy, - MissingIdentities, UnknownSkill, UnknownTemplate, } from "@earendil-works/pi-agent-core"; @@ -121,11 +120,12 @@ describe("Harness wire adapter", () => { test.each([ { ok: true, - value: { kind: "completed", runId: "run-1", leafId: "leaf-1" }, + value: { operation: "run", kind: "completed", runId: "run-1", leafId: "leaf-1" }, }, { ok: true, value: { + operation: "run", kind: "completed", runId: "run-1", leafId: "leaf-1", @@ -135,11 +135,12 @@ describe("Harness wire adapter", () => { }, { ok: true, - value: { kind: "aborted", runId: "run-1", leafId: "leaf-1" }, + value: { operation: "run", kind: "aborted", runId: "run-1", leafId: "leaf-1" }, }, { ok: true, value: { + operation: "run", kind: "aborted", runId: "run-1", leafId: "leaf-1", @@ -150,6 +151,7 @@ describe("Harness wire adapter", () => { { ok: true, value: { + operation: "run", kind: "failed", runId: "run-1", leafId: "leaf-1", @@ -159,6 +161,7 @@ describe("Harness wire adapter", () => { { ok: true, value: { + operation: "run", kind: "failed", runId: "run-1", leafId: "leaf-1", @@ -170,6 +173,7 @@ describe("Harness wire adapter", () => { { ok: true, value: { + operation: "run", kind: "suspended", reason: "deferred", runId: "run-1", @@ -178,33 +182,21 @@ describe("Harness wire adapter", () => { deferred: { provider: "test", modelId: "model", api: "test", id: "deferred-1", data: { row: 1 } }, }, }, + { + ok: true, + value: { + operation: "run", + kind: "action_required", + runId: "run-1", + action: { kind: "confirm", description: "Confirm", details: { safe: true } }, + }, + }, ] satisfies HarnessRunResult[])("projects every Run outcome %#", (result) => { const wire = toWireRunResult(result); expect(Check(RunResultSchema, wire)).toBe(true); - expect(wire).toEqual(result); - }); - - test("projects singular Harness model identities to the wire list", () => { - const wire = toWireRunResult({ - ok: true, - value: { - kind: "suspended", - reason: "missing_identities", - runId: "run-1", - leafId: "leaf-1", - missing: { tools: ["read"], model: "missing-model" }, - }, - }); - expect(wire).toEqual({ - ok: true, - value: { - kind: "suspended", - reason: "missing_identities", - runId: "run-1", - leafId: "leaf-1", - missing: { tools: ["read"], models: ["missing-model"] }, - }, - }); + if (!result.ok || !wire.ok) throw new Error("Expected successful run result"); + const { operation: _operation, ...expected } = result.value; + expect(wire.value).toEqual(expected); }); test.each([ @@ -214,7 +206,6 @@ describe("Harness wire adapter", () => { operationKind: "run", message: "busy", }), - new MissingIdentities({ lane: "main", tools: ["read"], model: "missing-model", message: "missing" }), new InvalidMessage({ lane: "main", reason: "invalid", message: "invalid" }), new UnknownSkill({ name: "skill", message: "unknown" }), new UnknownTemplate({ name: "template", message: "unknown" }), @@ -281,6 +272,7 @@ describe("Harness wire adapter", () => { const result: HarnessRunResult = { ok: true, value: { + operation: "run", kind: "completed", runId: "run-1", leafId: "leaf-1", diff --git a/packages/protocol/src/harness.ts b/packages/protocol/src/harness.ts index 533116de7..aeb8caa13 100644 --- a/packages/protocol/src/harness.ts +++ b/packages/protocol/src/harness.ts @@ -168,9 +168,10 @@ const OperationErrorSchema = StrictObject({ message: Type.String(), details: Type.Optional(JsonValueSchema), }); -const MissingIdentityListSchema = StrictObject({ - tools: Type.Array(Type.String()), - models: Type.Array(Type.String()), +const ActionInfoSchema = StrictObject({ + kind: Type.String(), + description: Type.String(), + details: Type.Optional(JsonValueSchema), }); const RunValueSchema = Type.Union([ StrictObject({ kind: Type.Literal("completed"), runId: IdSchema, leafId: IdSchema }), @@ -212,11 +213,9 @@ const RunValueSchema = Type.Union([ deferred: DeferredHandleSchema, }), StrictObject({ - kind: Type.Literal("suspended"), - reason: Type.Literal("missing_identities"), + kind: Type.Literal("action_required"), runId: IdSchema, - leafId: IdSchema, - missing: MissingIdentityListSchema, + action: ActionInfoSchema, }), ]); const RunErrorSchema = Type.Union([ @@ -227,13 +226,6 @@ const RunErrorSchema = Type.Union([ operationKind: Type.Union([Type.Literal("run"), Type.Literal("compaction"), Type.Literal("navigation")]), message: Type.String(), }), - StrictObject({ - _tag: Type.Literal("MissingIdentities"), - lane: Type.String(), - tools: Type.Array(Type.String()), - models: Type.Array(Type.String()), - message: Type.String(), - }), StrictObject({ _tag: Type.Literal("InvalidMessage"), lane: Type.String(), @@ -363,34 +355,6 @@ export const LaneEntrySchema = Type.Union([ ]); export type LaneEntry = Static; -const MissingIdentityListSchemaForSnapshot = StrictObject({ - tools: Type.Array(Type.String()), - models: Type.Array(Type.String()), -}); -const SuspendedOperationBase = { - lane: Type.String(), - operationId: IdSchema, - kind: Type.Union([Type.Literal("run"), Type.Literal("compaction"), Type.Literal("navigation")]), - startedAt: TimestampSchema, - prompt: Type.Optional(Type.Array(PromptMessageSchema)), - aborting: Type.Optional( - StrictObject({ steer: Type.Array(PromptMessageSchema), followUp: Type.Array(PromptMessageSchema) }), - ), -}; -const SuspendedOperationSchema = Type.Union([ - StrictObject({ ...SuspendedOperationBase, reason: Type.Literal("deferred"), deferred: DeferredHandleSchema }), - StrictObject({ - ...SuspendedOperationBase, - reason: Type.Literal("missing_identities"), - missing: MissingIdentityListSchemaForSnapshot, - }), - StrictObject({ - ...SuspendedOperationBase, - reason: Type.Literal("crash"), - deferred: Type.Optional(DeferredHandleSchema), - missing: Type.Optional(MissingIdentityListSchemaForSnapshot), - }), -]); const QueuedItemSchema = StrictObject({ entryId: IdSchema, message: PromptMessageSchema }); const RunningToolSchema = StrictObject({ toolCallId: Type.String(), @@ -401,9 +365,13 @@ const RunningToolSchema = StrictObject({ const LaneOperationSchema = StrictObject({ id: IdSchema, kind: Type.Union([Type.Literal("run"), Type.Literal("compaction"), Type.Literal("navigation")]), - status: Type.Union([Type.Literal("running"), Type.Literal("suspended"), Type.Literal("aborting")]), + status: Type.Union([Type.Literal("running"), Type.Literal("open"), Type.Literal("aborting")]), startedAt: TimestampSchema, - suspended: Type.Optional(SuspendedOperationSchema), + action: Type.Optional(ActionInfoSchema), + deferred: Type.Optional(StrictObject({ handle: DeferredHandleSchema, poll: Type.Integer({ minimum: 0 }) })), + drained: Type.Optional( + StrictObject({ steer: Type.Array(QueuedItemSchema), followUp: Type.Array(QueuedItemSchema) }), + ), streamingMessage: Type.Optional(AssistantMessageSchema), runningTools: Type.Array(RunningToolSchema), retry: Type.Optional( @@ -462,13 +430,6 @@ export const LaneEventSchema = Type.Union([ deferred: DeferredHandleSchema, ...LaneEventBase, }), - StrictObject({ - type: Type.Literal("run_suspend"), - runId: IdSchema, - reason: Type.Literal("missing_identities"), - missing: MissingIdentityListSchemaForSnapshot, - ...LaneEventBase, - }), StrictObject({ type: Type.Literal("run_abort"), runId: IdSchema, diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts index a74f1ad83..9f6f2404d 100644 --- a/packages/protocol/test/protocol.test.ts +++ b/packages/protocol/test/protocol.test.ts @@ -402,11 +402,9 @@ describe("protocol validation", () => { { ok: true, value: { - kind: "suspended", - reason: "missing_identities", + kind: "action_required", runId: "run-1", - leafId: "leaf-1", - missing: { tools: ["tool"], models: [] }, + action: { kind: "confirm", description: "Confirm", details: { safe: true } }, }, }, { diff --git a/packages/server/src/remote-session-manager.ts b/packages/server/src/remote-session-manager.ts index 7305d9a08..8fc07cee7 100644 --- a/packages/server/src/remote-session-manager.ts +++ b/packages/server/src/remote-session-manager.ts @@ -4,12 +4,14 @@ import type { BranchScan, Context, EntryQuery, + EntryType, JsonValue, LaneConfiguration, ListReadOptions, Session, SessionMetadata, SessionMutator, + StorageBranchScan, Value, ValueList, Write, @@ -101,6 +103,8 @@ export class RemoteSessionManager { await expect(capturedMutator?.getValue(applicationValue, BACKGROUND_CONTEXT)).rejects.toThrow( "outside its mutation callback", ); + await expect(capturedMutator?.scanBranch({ start: "entry" }, BACKGROUND_CONTEXT)).rejects.toThrow( + "outside its mutation callback", + ); + + const rootId = "00000000-0000-7000-8000-000000000001"; + const childId = "00000000-0000-7000-8000-000000000002"; + await remote.mutate( + "main", + (mutator, context) => + mutator.commit( + [ + insertEntry({ id: rootId, parentId: null, type: "custom", customType: "root" }), + insertEntry({ id: childId, parentId: rootId, type: "custom", customType: "child" }), + ], + context, + ), + BACKGROUND_CONTEXT, + ); + await expect( + remote.scanBranch({ start: childId, order: "oldestFirst" }, BACKGROUND_CONTEXT), + ).resolves.toMatchObject([{ id: rootId }, { id: childId }]); + const child = (await remote.getEntries([childId], BACKGROUND_CONTEXT)).get(childId); + if (child === undefined) throw new Error("Expected child entry"); + await expect( + remote.scanBranch( + { + start: childId, + stopAtId: rootId, + type: "custom", + customType: "root", + order: "newestFirst", + limit: 2, + cursor: { seq: child.seq }, + }, + BACKGROUND_CONTEXT, + ), + ).resolves.toMatchObject([{ id: rootId }]); + await expect( + remote.scanBranch({ start: childId, stopAtType: "custom" }, BACKGROUND_CONTEXT), + ).resolves.toMatchObject([{ id: childId }]); + await remote.mutate( + "main", + async (mutator, context) => { + await expect(mutator.scanBranch({ start: childId, limit: 1 }, context)).resolves.toMatchObject([ + { id: childId }, + ]); + }, + BACKGROUND_CONTEXT, + ); await remote.setValue(applicationValue, "tree-write", BACKGROUND_CONTEXT); await expect(remote.getValue(applicationValue, BACKGROUND_CONTEXT)).resolves.toMatchObject({ @@ -105,6 +156,32 @@ describe("RemoteSession", () => { await remote.close(BACKGROUND_CONTEXT); }); + test("validates callback-scoped branch scan RPC arguments", async () => { + const { remote } = await createRemoteSession("remote-scan-validation"); + const invalidQueries: unknown[] = [ + null, + {}, + { start: "" }, + { start: "entry", type: "invalid" }, + { start: "entry", order: "sideways" }, + { start: "entry", limit: Number.NaN }, + { start: "entry", cursor: { seq: "one" } }, + ]; + for (const query of invalidQueries) { + await expect(remote.scanBranch(query as StorageBranchScan, BACKGROUND_CONTEXT)).rejects.toThrow(); + } + await remote.mutate( + "main", + async (mutator, context) => { + await expect( + mutator.scanBranch({ start: "entry", stopAtType: "invalid" } as unknown as StorageBranchScan, context), + ).rejects.toThrow("query.stopAtType"); + }, + BACKGROUND_CONTEXT, + ); + await remote.close(BACKGROUND_CONTEXT); + }); + test("expires a wedged mutation and releases its lane", async () => { const { remote } = await createRemoteSession("remote-expiry", 10); const applicationValue = value("app.test", "expiry"); diff --git a/packages/session-backends/sqlite-node/src/sqlite/session.ts b/packages/session-backends/sqlite-node/src/sqlite/session.ts index 286fcf989..28b57f699 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/session.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/session.ts @@ -12,6 +12,7 @@ import type { SessionMutator, SessionStats, SessionTree, + StorageBranchScan, StoredValue, Value, ValueList, @@ -86,6 +87,10 @@ export class SqliteOpenSession implements Session { return this.admit(() => this.session.readList(address, options, context)); } + scanBranch(query: StorageBranchScan, context: Context): Promise { + return this.admit(() => this.session.scanBranch(query, context)); + } + view(lane: string): SessionTree { const view = this.session.view(lane); return { diff --git a/packages/session-backends/sqlite-node/test/repo.test.ts b/packages/session-backends/sqlite-node/test/repo.test.ts index 24c657060..f5e17a48c 100644 --- a/packages/session-backends/sqlite-node/test/repo.test.ts +++ b/packages/session-backends/sqlite-node/test/repo.test.ts @@ -78,6 +78,36 @@ describe("SqliteSessionRepo", () => { }); }); + it("exposes explicit branch scans through the open-session facade", async () => { + await withTempDir(async (directory) => { + const repo = new SqliteSessionRepo({ + directory, + databaseFactory: createNodeSqliteFactory(), + now: () => 1_700_000_000_000, + }); + const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT); + await session.mutate( + "main", + (mutator) => + mutator.commit( + [ + sessionWrites.insertEntry({ id: "root", parentId: null, type: "custom", customType: "root" }), + sessionWrites.insertEntry({ id: "child", parentId: "root", type: "custom", customType: "child" }), + ], + BACKGROUND_CONTEXT, + ), + BACKGROUND_CONTEXT, + ); + + expect(await session.scanBranch({ start: "child", order: "oldestFirst" }, BACKGROUND_CONTEXT)).toMatchObject([ + { id: "root" }, + { id: "child" }, + ]); + await session.close(BACKGROUND_CONTEXT); + await expect(session.scanBranch({ start: "child" }, BACKGROUND_CONTEXT)).rejects.toThrow("Session is closed"); + }); + }); + it("rejects duplicate create without deleting the existing database", async () => { await withTempDir(async (directory) => { const repo = new SqliteSessionRepo({