From 51f3090790c152530dbf591fab40fc84c8f196d0 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Sun, 13 Sep 2026 04:00:27 +0200 Subject: [PATCH] feat(agent): align pico foundation with v2 contract --- packages/agent/src/harness/pico/addresses.ts | 242 +++++- packages/agent/src/harness/pico/core.ts | 3 + packages/agent/src/harness/pico/entries.ts | 74 +- packages/agent/src/harness/pico/hooks.ts | 62 ++ packages/agent/src/harness/pico/index.ts | 162 +++- .../agent/src/harness/pico/memory-storage.ts | 221 +++++- packages/agent/src/harness/pico/runtime.ts | 245 +++--- packages/agent/src/harness/pico/session.ts | 334 +++++--- packages/agent/src/harness/pico/storage.ts | 31 +- packages/agent/src/harness/pico/tasks.ts | 181 +++-- .../test/harness/pico/memory-storage.test.ts | 570 +++++++++++++- .../agent/test/harness/pico/types.compile.ts | 744 +++++++++++------- 12 files changed, 2179 insertions(+), 690 deletions(-) create mode 100644 packages/agent/src/harness/pico/hooks.ts diff --git a/packages/agent/src/harness/pico/addresses.ts b/packages/agent/src/harness/pico/addresses.ts index a9bd69afa..dab63da24 100644 --- a/packages/agent/src/harness/pico/addresses.ts +++ b/packages/agent/src/harness/pico/addresses.ts @@ -6,6 +6,10 @@ export type Scope = | { readonly type: "task"; readonly taskId: Id } | { readonly type: "shared"; readonly id: Id }; +export type PublicScope = Exclude; +export type StickyScope = Extract; +export type ConversationScope = Extract; + declare const addressType: unique symbol; export interface Address { @@ -19,6 +23,7 @@ export interface Address { export interface Value extends Address { readonly kind: "value"; + readonly default?: T; } export interface List extends Address { @@ -30,48 +35,239 @@ export interface Element { readonly value: T; } -type StickyOptions = { readonly key?: string; readonly rewind?: false }; -type ConversationOptions = { readonly key?: string; readonly rewind: boolean }; -type StickyScope = Extract; -type ConversationScope = Extract; -type Options = StickyOptions | ConversationOptions; +export interface UnboundValue { + readonly namespace: string; + readonly key?: string; + readonly rewind: R; + readonly default?: T; + readonly [addressType]?: T; + bind(conversationId: Id): Value; +} -export function defineValue( - scope: StickyScope, - namespace: string, - options?: StickyOptions, -): Value; -export function defineValue( - scope: ConversationScope, - namespace: string, - options: ConversationOptions, -): Value; -export function defineValue(scope: Scope, namespace: string, options?: Options): Value { +export interface UnboundList { + readonly namespace: string; + readonly key?: string; + readonly rewind: R; + readonly [addressType]?: T; + bind(conversationId: Id): List; +} + +export type ConversationValueDefinition = UnboundValue; +export type ConversationListDefinition = UnboundList; +export type PayloadOf = D extends Address + ? BoundPayload + : D extends UnboundValue + ? DefinitionPayload + : never; +export type ResolvedValue = D extends { readonly default: infer V } ? V : PayloadOf | undefined; + +type DefaultedUnboundValue = Omit, "bind" | "default"> & { + readonly default: T; + bind(conversationId: Id): Value & { readonly default: T }; +}; + +type StickyOptions = { readonly key?: string; readonly default?: T }; +type DefaultedStickyOptions = { readonly key?: string; readonly default: T }; +type ConversationOptions = { + readonly key?: string; + readonly rewind: boolean; + readonly default?: T; +}; +type DefaultedConversationOptions = { + readonly key?: string; + readonly rewind: boolean; + readonly default: T; +}; +type InternalOptions = StickyOptions & { readonly rewind?: boolean }; + +function assertPublicNamespace(namespace: string): void { + if (namespace.startsWith("pi.")) throw new Error(`Namespace ${namespace} is reserved`); +} + +function assertPublicScope(scope: Scope): asserts scope is PublicScope { + if (scope.type === "shared") throw new Error("Shared scope is internal"); +} + +function value(scope: Scope, namespace: string, options?: InternalOptions): Value { return Object.freeze({ ...options, scope, namespace, - kind: "value", + kind: "value" as const, rewind: options?.rewind ?? false, }); } +function list(scope: Scope, namespace: string, options?: InternalOptions): List { + return Object.freeze({ + ...options, + scope, + namespace, + kind: "list" as const, + rewind: options?.rewind ?? false, + }); +} + +export function defineValue( + scope: StickyScope, + namespace: string, + options: DefaultedStickyOptions, +): Value & { readonly default: T }; +export function defineValue( + scope: StickyScope, + namespace: string, + options?: { readonly key?: string }, +): Value; +export function defineValue( + scope: ConversationScope, + namespace: string, + options: DefaultedConversationOptions, +): Value & { readonly default: T }; +export function defineValue( + scope: ConversationScope, + namespace: string, + options: { readonly key?: string; readonly rewind: boolean }, +): Value; +export function defineValue( + scope: Scope, + namespace: string, + options?: InternalOptions, +): Value { + assertPublicScope(scope); + assertPublicNamespace(namespace); + return value(scope, namespace, options); +} + export function defineList( scope: StickyScope, namespace: string, - options?: StickyOptions, + options?: { readonly key?: string }, ): List; export function defineList( scope: ConversationScope, namespace: string, - options: ConversationOptions, + options: Omit, "default">, ): List; -export function defineList(scope: Scope, namespace: string, options?: Options): List { +export function defineList( + scope: Scope, + namespace: string, + options?: InternalOptions, +): List { + assertPublicScope(scope); + assertPublicNamespace(namespace); + return list(scope, namespace, options); +} + +export function conversationValue( + namespace: string, + options: { readonly key?: string; readonly rewind: true; readonly default: T }, +): DefaultedUnboundValue; +export function conversationValue( + namespace: string, + options: { readonly key?: string; readonly rewind: false; readonly default: T }, +): DefaultedUnboundValue; +export function conversationValue( + namespace: string, + options: { readonly key?: string; readonly rewind: true }, +): UnboundValue; +export function conversationValue( + namespace: string, + options: { readonly key?: string; readonly rewind: false }, +): UnboundValue; +export function conversationValue( + namespace: string, + options: ConversationOptions, +): UnboundValue { + assertPublicNamespace(namespace); + return unboundValue(namespace, options); +} + +export function conversationList( + namespace: string, + options: { readonly key?: string; readonly rewind: true }, +): UnboundList; +export function conversationList( + namespace: string, + options: { readonly key?: string; readonly rewind: false }, +): UnboundList; +export function conversationList( + namespace: string, + options: Omit, "default">, +): UnboundList { + assertPublicNamespace(namespace); + return unboundList(namespace, options); +} + +export function sessionValue( + namespace: string, + options: { readonly key?: string; readonly default: T }, +): Value & { readonly default: T }; +export function sessionValue(namespace: string, options?: { readonly key?: string }): Value; +export function sessionValue(namespace: string, options?: StickyOptions): Value { + return defineValue({ type: "session" }, namespace, options); +} + +/** Internal constructor for fixed `pi.*` definitions and shared task output state. */ +export function defineInternalValue( + scope: Scope, + namespace: string, + options?: InternalOptions, +): Value { + return value(scope, namespace, options); +} + +/** Internal constructor for fixed `pi.*` definitions and shared task output state. */ +export function defineInternalList( + scope: Scope, + namespace: string, + options?: InternalOptions, +): List { + return list(scope, namespace, options); +} + +/** Internal constructor for fixed conversation configuration definitions. */ +export function internalConversationValue( + namespace: string, + options: { readonly key?: string; readonly rewind: R; readonly default: T }, +): DefaultedUnboundValue; +export function internalConversationValue( + namespace: string, + options: { readonly key?: string; readonly rewind: R }, +): UnboundValue; +export function internalConversationValue( + namespace: string, + options: { readonly key?: string; readonly rewind: R; readonly default?: T }, +): UnboundValue { + return unboundValue(namespace, options); +} + +function unboundValue( + namespace: string, + options: { readonly key?: string; readonly rewind: R; readonly default?: T }, +): UnboundValue { return Object.freeze({ ...options, - scope, namespace, - kind: "list", - rewind: options?.rewind ?? false, + bind: (conversationId: Id) => + value({ type: "conversation", conversationId }, namespace, { + key: options.key, + rewind: options.rewind, + default: options.default, + }), + }); +} + +function unboundList( + namespace: string, + options: { readonly key?: string; readonly rewind: R }, +): UnboundList { + return Object.freeze({ + ...options, + namespace, + bind: (conversationId: Id) => + list({ type: "conversation", conversationId }, namespace, { + key: options.key, + rewind: options.rewind, + }), }); } diff --git a/packages/agent/src/harness/pico/core.ts b/packages/agent/src/harness/pico/core.ts index fa71922d9..2467ccd63 100644 --- a/packages/agent/src/harness/pico/core.ts +++ b/packages/agent/src/harness/pico/core.ts @@ -1,5 +1,8 @@ +import type { JsonRepresentation } from "@earendil-works/chord"; + export type JsonPrimitive = null | boolean | number | string; export type JsonValue = JsonPrimitive | JsonObject | readonly JsonValue[]; +export type Stored = JsonRepresentation; export interface JsonObject { readonly [key: string]: JsonValue; diff --git a/packages/agent/src/harness/pico/entries.ts b/packages/agent/src/harness/pico/entries.ts index ead28a2ef..843783fd4 100644 --- a/packages/agent/src/harness/pico/entries.ts +++ b/packages/agent/src/harness/pico/entries.ts @@ -1,17 +1,22 @@ -import type { AssistantMessage, Message, ToolResultMessage, UserMessage } from "@earendil-works/pi-ai"; -import type { Id, JsonValue } from "./core.ts"; +import type { AssistantMessage, Message, ToolResultMessage, Usage, UserMessage } from "@earendil-works/pi-ai"; +import type { Id, JsonValue, Stored } from "./core.ts"; export type ContextEdit = | { readonly target: Id; readonly action: "omit" } - | { readonly target: Id; readonly action: "replace"; readonly messages: readonly Message[] }; + | { readonly target: Id; readonly action: "replace"; readonly messages: readonly Stored[] }; -export interface EntryIdentity { +export interface Entry { readonly id: Id; readonly conversationId: Id; readonly kind: string; readonly byTaskId?: Id; + readonly data?: JsonValue; + readonly model?: readonly Stored[]; + readonly head?: Id; + readonly edits?: readonly ContextEdit[]; } +export type EntryIdentity = Pick; export type EntryBase = EntryIdentity; export interface EntryData { @@ -19,7 +24,7 @@ export interface EntryData { } export interface ModelProjection { - readonly model: readonly M[]; + readonly model: readonly Stored[]; } export interface ContextHead { @@ -30,10 +35,8 @@ export interface ContextEdits { readonly edits: readonly ContextEdit[]; } -export type Entry = EntryBase & Partial; - export type EntryInput = Omit & - (E extends ContextHead ? { readonly head: Id | "self" } : { readonly head?: never }); + (E extends { readonly head: Id } ? { readonly head: Id | "self" } : { readonly head?: never }); export interface EntryKind { readonly kind: string; @@ -55,12 +58,49 @@ export interface Conversation { readonly owner?: Id; } -export type UserEntry = EntryBase & { readonly model: readonly [UserMessage] }; -export type AssistantEntry = EntryBase & { readonly model: readonly [AssistantMessage] }; -export type ToolResultEntry = EntryBase & - EntryData & { readonly model: readonly [ToolResultMessage] }; -export type NoticeEntry = EntryBase & Partial & { readonly model: readonly [UserMessage] }; -export type SummaryEntry = EntryBase & - EntryData<{ readonly through: Id }> & { readonly model: readonly [UserMessage] } & ContextHead; -export type HandoffEntry = EntryBase & { readonly model: readonly [UserMessage] } & ContextHead; -export type ResetEntry = EntryBase & ContextHead; +export type AssistantEntryData = { readonly attempt: number }; +export type UsageEntryData = { readonly attempt: number; readonly usage?: Stored; readonly error: string }; +export type ToolControl = { + readonly terminate?: true; + readonly handoff?: string; + readonly addTools?: readonly string[]; +}; +export type ToolDiagnostic = { + readonly severity: "info" | "warn" | "error"; + readonly message: string; + readonly code?: string; +}; +export type ToolUsage = { readonly [key: string]: number }; +export type ToolResultData = { + readonly details?: JsonValue; + readonly usage?: ToolUsage; + readonly diagnostics?: readonly ToolDiagnostic[]; + readonly control?: ToolControl; + readonly truncated?: { readonly bytes: number; readonly lines: number }; +}; +export type SectionRecord = + | { readonly key: string; readonly action: "set"; readonly value: JsonValue; readonly rendered: string } + | { readonly key: string; readonly action: "remove" }; +export type SystemEntryData = { readonly baseline?: true; readonly sections: readonly SectionRecord[] }; + +export type UserEntry = EntryBase & { + readonly model: readonly [Stored]; + readonly data?: { readonly continuation: true; readonly from: Id }; +}; +export type AssistantEntry = EntryBase & { + readonly model: readonly [Stored]; + readonly data?: AssistantEntryData; +}; +export type ToolResultEntry = EntryBase & { + readonly model: readonly [Stored]; + readonly data: ToolResultData; +}; +export type NoticeEntry = EntryBase & { readonly model: readonly [Stored]; readonly data?: JsonValue }; +export type UsageEntry = EntryBase & { readonly data: UsageEntryData }; +export type SummaryEntry = EntryBase & { + readonly model: readonly [Stored]; + readonly data: { readonly through: Id }; + readonly head: Id; +}; +export type HandoffEntry = EntryBase & { readonly model: readonly [Stored]; readonly head: Id }; +export type ResetEntry = EntryBase & { readonly head: Id }; diff --git a/packages/agent/src/harness/pico/hooks.ts b/packages/agent/src/harness/pico/hooks.ts new file mode 100644 index 000000000..9cb20fafd --- /dev/null +++ b/packages/agent/src/harness/pico/hooks.ts @@ -0,0 +1,62 @@ +import type { Context } from "@earendil-works/chord"; +import type { Id } from "./core.ts"; + +declare const hookIn: unique symbol; +declare const hookOut: unique symbol; + +export type Fold = "collect" | "first" | "chain"; + +export interface HookPoint { + readonly fold: F; + readonly onThrow: "skip" | "abort"; + readonly [hookIn]?: In; + readonly [hookOut]?: Out; +} + +export function defineHookPoint(options: { + readonly fold: "collect"; + readonly onThrow: "skip" | "abort"; +}): HookPoint; +export function defineHookPoint(options: { + readonly fold: "first"; + readonly onThrow: "skip" | "abort"; +}): HookPoint; +export function defineHookPoint>(options: { + readonly fold: "chain"; + readonly onThrow: "skip" | "abort"; +}): HookPoint; +export function defineHookPoint(options: { + readonly fold: Fold; + readonly onThrow: "skip" | "abort"; +}): HookPoint { + return Object.freeze(options); +} + +export type HookPoints = Record>; +export type EmptyHookPoints = Record; +export type HookIn

= P extends HookPoint ? In : never; +export type HookOut

= P extends HookPoint ? Out : never; + +// biome-ignore lint/suspicious/noConfusingVoidType: void allows handlers with no return value. +type HookHandlerResult

= HookOut

| void; +export type HookHandler

= ( + input: HookIn

, + info: HookInfo, + ctx: Context, +) => HookHandlerResult

| Promise>; + +export interface HookInfo { + readonly kind: string; + readonly taskId: Id; + readonly conversationId: Id; +} + +export interface HookRunner { + run

(point: P, input: HookIn

, ctx: Context): Promise>; +} + +export type HookResult

= P extends { fold: "collect" } + ? { readonly outputs: readonly HookOut

[]; readonly threw?: unknown } + : P extends { fold: "first" } + ? { readonly output?: HookOut

; readonly threw?: unknown } + : { readonly output: HookIn

; readonly threw?: unknown }; diff --git a/packages/agent/src/harness/pico/index.ts b/packages/agent/src/harness/pico/index.ts index f6807b27d..1f0f85d01 100644 --- a/packages/agent/src/harness/pico/index.ts +++ b/packages/agent/src/harness/pico/index.ts @@ -1,8 +1,154 @@ -export * from "./addresses.ts"; -export * from "./core.ts"; -export * from "./entries.ts"; -export * from "./memory-storage.ts"; -export * from "./runtime.ts"; -export * from "./session.ts"; -export * from "./storage.ts"; -export * from "./tasks.ts"; +export type { + Address, + ConversationListDefinition, + ConversationScope, + ConversationValueDefinition, + Element, + List, + PayloadOf, + PublicScope, + ResolvedValue, + Scope, + StickyScope, + UnboundList, + UnboundValue, + Value, +} from "./addresses.ts"; +export { + conversationList, + conversationValue, + defineList, + defineValue, + sessionValue, +} from "./addresses.ts"; +export type { Id, JsonObject, JsonPrimitive, JsonValue, Seq, Stored } from "./core.ts"; +export type { + AssistantEntry, + AssistantEntryData, + ContextEdit, + ContextEdits, + ContextHead, + Conversation, + Entry, + EntryBase, + EntryData, + EntryIdentity, + EntryInput, + EntryKind, + HandoffEntry, + ModelProjection, + NoticeEntry, + ResetEntry, + SectionRecord, + SummaryEntry, + SystemEntryData, + ToolControl, + ToolDiagnostic, + ToolResultData, + ToolResultEntry, + ToolUsage, + UsageEntry, + UsageEntryData, + UserEntry, +} from "./entries.ts"; +export { defineEntry } from "./entries.ts"; +export type { + Fold, + HookHandler, + HookIn, + HookInfo, + HookOut, + HookPoint, + HookPoints, + HookResult, + HookRunner, +} from "./hooks.ts"; +export { defineHookPoint } from "./hooks.ts"; +export { MemoryStorage } from "./memory-storage.ts"; +export type { + AbortClosure, + AbortFinalTx, + AbortTaskConversation, + AbortTaskRuntime, + AbortTx, + Acceptance, + BaseTaskTx, + BoundPublicState, + BoundTxState, + ConversationCreateSpec, + ConversationTx, + FinalOutput, + FinalTx, + GenerationOutput, + HostTx, + InputHandle, + InputOutcome, + PublicList, + PublicValue, + QueuedInput, + ReadonlyTaskOutput, + RuntimeOutput, + ScratchReader, + SendInput, + StickyPublicValue, + StoredEntryDraft, + StoredInputResult, + TaskConversation, + TaskOutput, + TaskRuntime, + TerminalClosure, + TerminalInputOutcome, + ToolInput, + ToolOutputState, + TxList, + TxReaders, + TxState, + TxValue, + UserInput, +} from "./runtime.ts"; +export { Closed, Faulted, NestedLineOperation, ReadAfterWrite } from "./session.ts"; +export type { + ConversationScan, + EntryScan, + NewTask, + Page, + PageQuery, + StateWrite, + Storage, + TaskScan, + Write, +} from "./storage.ts"; +export { InvalidHistoryPosition, OutputRetired, ScratchRetired } from "./storage.ts"; +export type { + AbortedOf, + AnyDefinedKind, + CheckpointOf, + Completion, + ConfigBundle, + DistributiveOmit, + ExactJsonInput, + FailureOf, + InputOf, + NoCheckpoint, + NoExtra, + OrdinaryKind, + OutputOf, + OutputState, + PayloadsOf, + ResultOf, + RunningTask, + StoredTaskOutputRef, + Task, + TaskBase, + TaskCheckpoint, + TaskKind, + TaskKindDefinition, + TaskKindFactory, + TaskOf, + TaskOutcome, + TaskOutputDefinition, + TaskOutputField, + TaskOutputKind, + TaskOutputRef, +} from "./tasks.ts"; +export { defineTask, defineTaskOutput } from "./tasks.ts"; diff --git a/packages/agent/src/harness/pico/memory-storage.ts b/packages/agent/src/harness/pico/memory-storage.ts index fede04287..a3b126702 100644 --- a/packages/agent/src/harness/pico/memory-storage.ts +++ b/packages/agent/src/harness/pico/memory-storage.ts @@ -6,9 +6,12 @@ import { type ConversationScan, type EntryScan, InvalidHistoryPosition, + OutputRetired, type Page, type PageQuery, + ScratchRetired, type Storage, + type StoredConversation, type TaskScan, type Write, } from "./storage.ts"; @@ -25,7 +28,12 @@ type ListHistoryWrite = type StoredState = | { readonly kind: "value"; readonly rewind: false; value?: JsonValue } - | { readonly kind: "list"; readonly rewind: false; readonly elements: Map> } + | { + readonly kind: "list"; + readonly rewind: false; + readonly elements: Map>; + readonly known: Set; + } | { readonly kind: "value"; readonly rewind: true; readonly writes: ValueHistoryWrite[] } | { readonly kind: "list"; readonly rewind: true; readonly writes: ListHistoryWrite[] }; @@ -53,6 +61,12 @@ interface StoredEntry { readonly entry: Entry; } +interface ProspectiveWrites { + readonly conversations: Map; + readonly entries: Map; + readonly listAppends: { readonly seq: Seq; readonly address: List; readonly elementId: Id }[]; +} + function rewindable( address: Address, ): address is Address & { readonly scope: Extract; readonly rewind: true } { @@ -75,7 +89,8 @@ function page( ): Page { const items: T[] = []; let more = false; - for (const value of values) { + const ordered = [...values].sort((left, right) => (ascending ? id(left) - id(right) : id(right) - id(left))); + for (const value of ordered) { const valueId = id(value); if ( !matches(value) || @@ -94,8 +109,9 @@ function page( export class MemoryStorage implements Storage { private nextObjectId: Id = 1; + private committedIdHighWater: Id = 0; private nextSeq: Seq = 1; - private readonly conversations = new Map(); + private readonly conversations = new Map(); private readonly entries = new Map(); private readonly entriesByConversation = new Map(); private readonly tasks = new Map(); @@ -106,19 +122,24 @@ export class MemoryStorage implements Storage { private readonly taskState = new Map(); private readonly sharedState = new Map(); + static create(): MemoryStorage { + return new MemoryStorage(); + } + nextId(): Id { return this.nextObjectId++; } async commit(writes: readonly Write[], _ctx: Context): Promise { + this.validateListRemovals(writes); const seqs = writes.map((_, index) => this.nextSeq + index); const affectedOutputs = new Set(); - let nextObjectId = this.nextObjectId; + let batchMaxId = this.committedIdHighWater; for (const write of writes) { - if (write.type === "conversation.create") nextObjectId = Math.max(nextObjectId, write.conversation.id + 1); - else if (write.type === "entry.append") nextObjectId = Math.max(nextObjectId, write.entry.id + 1); - else if (write.type === "task.create") nextObjectId = Math.max(nextObjectId, write.task.id + 1); - else if (write.type === "list.append") nextObjectId = Math.max(nextObjectId, write.element.id + 1); + if (write.type === "conversation.create") batchMaxId = Math.max(batchMaxId, write.conversation.id); + else if (write.type === "entry.append") batchMaxId = Math.max(batchMaxId, write.entry.id); + else if (write.type === "task.create") batchMaxId = Math.max(batchMaxId, write.task.id); + else if (write.type === "list.append") batchMaxId = Math.max(batchMaxId, write.element.id); if (write.type !== "task.create" && write.type !== "task.set") continue; const currentOutput = this.tasks.get(write.task.id)?.output?.id; if (currentOutput !== undefined) affectedOutputs.add(currentOutput); @@ -131,23 +152,28 @@ export class MemoryStorage implements Storage { for (const id of affectedOutputs) { if (!this.outputReferences.has(id)) this.sharedState.delete(id); } - this.nextObjectId = nextObjectId; + this.committedIdHighWater = batchMaxId; + this.nextObjectId = Math.max(this.nextObjectId, batchMaxId + 1); this.nextSeq += writes.length; return seqs; } + async close(_ctx: Context): Promise { + this.nextObjectId = this.committedIdHighWater + 1; + } + async getConversations(ids: readonly Id[], _ctx: Context): Promise> { const result = new Map(); for (const id of ids) { const conversation = this.conversations.get(id); - if (conversation !== undefined) result.set(id, conversation); + if (conversation !== undefined) result.set(id, this.publicConversation(conversation)); } return result; } async scanConversations(query: ConversationScan, _ctx: Context): Promise> { return page( - this.conversations.values(), + [...this.conversations.values()].map((conversation) => this.publicConversation(conversation)), query, (conversation) => conversation.id, (conversation) => @@ -167,18 +193,24 @@ export class MemoryStorage implements Storage { } async scanEntries(query: EntryScan, _ctx: Context): Promise> { - return page( - this.visibleEntries(query.conversationId, query.through), - query, - (entry) => entry.id, - (entry) => query.kind === undefined || entry.kind === query.kind, - false, - ); + const cursor = query.cursor === undefined ? undefined : this.historyPosition(query.conversationId, query.cursor); + const items: Entry[] = []; + let more = false; + for (const stored of this.visibleEntries(query.conversationId, query.through)) { + if (cursor !== undefined && stored.seq >= cursor) continue; + if (query.kind !== undefined && stored.entry.kind !== query.kind) continue; + if (items.length === query.limit) { + more = true; + break; + } + items.push(stored.entry); + } + return { items, ...(more && items.length > 0 ? { next: items.at(-1)!.id } : {}) }; } async newestHead(conversationId: Id, at: Id, _ctx: Context): Promise { - for (const entry of this.visibleEntries(conversationId, at)) { - if (entry.head !== undefined) return entry; + for (const stored of this.visibleEntries(conversationId, at)) { + if (stored.entry.head !== undefined) return stored.entry; } return undefined; } @@ -210,8 +242,10 @@ export class MemoryStorage implements Storage { async getValue(address: Value, at: Id | undefined, _ctx: Context): Promise { if (at !== undefined) { if (!rewindable(address)) throw new InvalidHistoryPosition(at); + this.assertScopeReadable(address); return this.rewindValue(address, this.historyPosition(address.scope.conversationId, at)); } + this.assertScopeReadable(address); if (rewindable(address)) return this.rewindValue(address, undefined); const state = this.storedState(address); return state?.kind === "value" && !state.rewind ? (state.value as T | undefined) : undefined; @@ -224,26 +258,31 @@ export class MemoryStorage implements Storage { ): Promise[]> { if (at !== undefined) { if (!rewindable(address)) throw new InvalidHistoryPosition(at); + this.assertScopeReadable(address); return this.rewindListElements(address, this.historyPosition(address.scope.conversationId, at)); } + this.assertScopeReadable(address); if (rewindable(address)) return this.rewindListElements(address, undefined); const state = this.storedState(address); if (state?.kind !== "list" || state.rewind) return []; - return [...(state.elements.values() as Iterable>)]; + return [...(state.elements.values() as Iterable>)].sort((left, right) => left.id - right.id); } - private historyPosition(conversationId: Id, at: Id): Seq { - const stored = this.entries.get(at); + private historyPosition(conversationId: Id, at: Id, prospective?: ProspectiveWrites): Seq { + const stored = prospective?.entries.get(at) ?? this.entries.get(at); if (stored === undefined) throw new InvalidHistoryPosition(at); let currentId = conversationId; let cap: Seq | undefined; while (true) { if (stored.entry.conversationId === currentId && (cap === undefined || stored.seq <= cap)) return stored.seq; - const conversation = this.conversations.get(currentId); + const conversation = prospective?.conversations.get(currentId) ?? this.conversations.get(currentId); if (conversation?.parent === undefined) throw new InvalidHistoryPosition(at); - const parentAtSeq = this.entries.get(conversation.parent.at)?.seq; - if (parentAtSeq === undefined) throw new InvalidHistoryPosition(at); + const parentAtSeq = this.historyPosition( + conversation.parent.conversationId, + conversation.parent.at, + prospective, + ); cap = Math.min(cap ?? parentAtSeq, parentAtSeq); currentId = conversation.parent.conversationId; } @@ -294,24 +333,134 @@ export class MemoryStorage implements Storage { else elements.clear(); } } - return [...elements.values()]; + return [...elements.values()].sort((left, right) => left.id - right.id); } - private *visibleEntries(conversationId: Id, through: Id | undefined): Iterable { + private *visibleEntries(conversationId: Id, through: Id | undefined): Iterable { let conversation = this.conversations.get(conversationId); - let cap = through; + let cap = through === undefined ? undefined : this.historyPosition(conversationId, through); while (conversation !== undefined) { const entries = this.entriesByConversation.get(conversation.id) ?? []; for (let index = entries.length - 1; index >= 0; index--) { - const entry = entries[index]!.entry; - if (cap === undefined || entry.id <= cap) yield entry; + const stored = entries[index]!; + if (cap === undefined || stored.seq <= cap) yield stored; } if (conversation.parent === undefined) return; - cap = Math.min(cap ?? conversation.parent.at, conversation.parent.at); + const parentAtSeq = this.historyPosition(conversation.parent.conversationId, conversation.parent.at); + cap = Math.min(cap ?? parentAtSeq, parentAtSeq); conversation = this.conversations.get(conversation.parent.conversationId); } } + private validateListRemovals(writes: readonly Write[]): void { + const prospective: ProspectiveWrites = { + conversations: new Map(), + entries: new Map(), + listAppends: [], + }; + for (let index = 0; index < writes.length; index++) { + const write = writes[index]!; + const seq = this.nextSeq + index; + if (write.type === "conversation.create") { + prospective.conversations.set(write.conversation.id, write.conversation); + } else if (write.type === "entry.append") { + prospective.entries.set(write.entry.id, { seq, entry: write.entry }); + } else if (write.type === "list.append") { + prospective.listAppends.push({ seq, address: write.address, elementId: write.element.id }); + } else if ( + write.type === "list.remove" && + !this.listKnowsElement(write.address, write.elementId, prospective) + ) { + throw new Error(`List element ${write.elementId} does not belong to this list lineage`); + } + } + } + + private listKnowsElement(address: List, elementId: Id, prospective?: ProspectiveWrites): boolean { + if (!rewindable(address)) { + const state = this.storedState(address); + return ( + (state?.kind === "list" && !state.rewind && state.known.has(elementId)) || + (prospective?.listAppends.some( + (append) => append.elementId === elementId && this.sameAddress(append.address, address), + ) ?? + false) + ); + } + let conversationId = address.scope.conversationId; + let cap: Seq | undefined; + while (true) { + const state = this.conversationStoredState(conversationId, address); + const appendedProspectively = prospective?.listAppends.some( + (append) => + append.elementId === elementId && + append.address.scope.type === "conversation" && + append.address.scope.conversationId === conversationId && + append.address.rewind === address.rewind && + append.address.namespace === address.namespace && + append.address.key === address.key && + (cap === undefined || append.seq <= cap), + ); + if ( + appendedProspectively || + (state?.kind === "list" && + state.rewind && + state.writes.some( + (write) => + write.type === "append" && + write.element.id === elementId && + (cap === undefined || write.seq <= cap), + )) + ) { + return true; + } + const conversation = prospective?.conversations.get(conversationId) ?? this.conversations.get(conversationId); + if (conversation?.parent === undefined) return false; + const parentAtSeq = this.historyPosition( + conversation.parent.conversationId, + conversation.parent.at, + prospective, + ); + cap = Math.min(cap ?? parentAtSeq, parentAtSeq); + conversationId = conversation.parent.conversationId; + } + } + + private sameAddress(left: Address, right: Address): boolean { + if ( + left.kind !== right.kind || + left.rewind !== right.rewind || + left.namespace !== right.namespace || + left.key !== right.key || + left.scope.type !== right.scope.type + ) { + return false; + } + if (left.scope.type === "session") return true; + if (left.scope.type === "conversation" && right.scope.type === "conversation") { + return left.scope.conversationId === right.scope.conversationId; + } + if (left.scope.type === "task" && right.scope.type === "task") return left.scope.taskId === right.scope.taskId; + return left.scope.type === "shared" && right.scope.type === "shared" && left.scope.id === right.scope.id; + } + + private publicConversation(conversation: StoredConversation): Conversation { + return { + id: conversation.id, + ...(conversation.parent === undefined ? {} : { parent: conversation.parent }), + ...(conversation.owner === undefined ? {} : { owner: conversation.owner }), + }; + } + + private assertScopeReadable(address: Address): void { + if (address.scope.type === "task") { + const task = this.tasks.get(address.scope.taskId); + if (task === undefined || task.status === "terminal") throw new ScratchRetired(address.scope.taskId); + } else if (address.scope.type === "shared" && !this.outputReferences.has(address.scope.id)) { + throw new OutputRetired(address.scope.id); + } + } + private apply(write: Write, seq: Seq): void { if (write.type === "conversation.create") { this.conversations.set(write.conversation.id, write.conversation); @@ -345,8 +494,10 @@ export class MemoryStorage implements Storage { else if (write.type === "list.remove") state.writes.push({ seq, type: "remove", elementId: write.elementId }); else state.writes.push({ seq, type: "clear" }); - } else if (write.type === "list.append") state.elements.set(write.element.id, write.element); - else if (write.type === "list.remove") state.elements.delete(write.elementId); + } else if (write.type === "list.append") { + state.known.add(write.element.id); + state.elements.set(write.element.id, write.element); + } else if (write.type === "list.remove") state.elements.delete(write.elementId); else state.elements.clear(); } } @@ -382,7 +533,7 @@ export class MemoryStorage implements Storage { private listState(address: List): Extract { const initial: StoredState = rewindable(address) ? { kind: "list", rewind: true, writes: [] } - : { kind: "list", rewind: false, elements: new Map() }; + : { kind: "list", rewind: false, elements: new Map(), known: new Set() }; return this.ensureState(address, initial) as Extract; } diff --git a/packages/agent/src/harness/pico/runtime.ts b/packages/agent/src/harness/pico/runtime.ts index a4ae202a3..fe5929864 100644 --- a/packages/agent/src/harness/pico/runtime.ts +++ b/packages/agent/src/harness/pico/runtime.ts @@ -1,16 +1,18 @@ import type { Context } from "@earendil-works/chord"; -import type { ImageContent, Message, TextContent } from "@earendil-works/pi-ai"; -import type { Element, List, Value } from "./addresses.ts"; -import type { Id, JsonValue } from "./core.ts"; -import type { ContextEdit, Conversation, Entry, EntryInput, EntryKind } from "./entries.ts"; +import type { AssistantMessage, ImageContent, Message, TextContent, ToolCall } from "@earendil-works/pi-ai"; +import type { Element, List, PayloadOf, ResolvedValue, UnboundList, UnboundValue, Value } from "./addresses.ts"; +import type { Id, JsonValue, Stored } from "./core.ts"; +import type { AssistantEntry, ContextEdit, Conversation, Entry, EntryInput, EntryKind, UserEntry } from "./entries.ts"; +import type { EmptyHookPoints, HookPoints, HookRunner } from "./hooks.ts"; import type { - AnyDefinedTaskKind, - AnyTaskKind, + AnyDefinedKind, Completion, ExactJsonInput, InputOf, NoExtra, + OrdinaryKind, OutputOf, + OutputState, RunningTask, Task, TaskCheckpoint, @@ -24,59 +26,66 @@ export interface TxReaders { getEntry(kind: EntryKind, id: Id): Promise; getEntries(ids: readonly Id[]): Promise>; getTask(id: Id): Promise; - getTask(kind: K, id: Id): Promise | undefined>; + getTask(kind: K, id: Id): Promise | undefined>; getTasks(ids: readonly Id[]): Promise>; } -export interface TxValue { - get(at?: Id): Promise; +export interface TxValue { + get(at?: Id): Promise; set(value: T): void; delete(): void; } export interface TxList { - read(): Promise[]>; - read(at: Id): Promise[]>; + read(at?: Id): Promise[]>; append(value: T): Id; remove(id: Id): void; clear(): void; } -export interface Acceptance { +export interface TxState { + value>(address: D): TxValue, ResolvedValue>; + list(address: List): TxList; +} + +export interface BoundTxState { + value | UnboundValue>(address: D): TxValue, ResolvedValue>; + list(address: List | UnboundList): TxList; +} + +export type Acceptance = { + readonly kind: "send" | "write"; readonly requestId?: string; readonly conversationId: Id; readonly inputId: Id; -} +}; -interface TaskSpecBase { +export type TaskSpec = { readonly conversationId?: Id; - readonly input: I; + readonly input: InputOf; readonly after?: readonly Id[]; readonly background?: true; -} - -export type TaskSpec = TaskSpecBase & - ([O] extends [never] ? { readonly output?: never } : { readonly output?: TaskOutputRef }); +} & ([OutputOf] extends [never] ? { readonly output?: never } : { readonly output?: TaskOutputRef> }); export interface ConversationCreateSpec { - readonly parent?: { readonly conversationId: Id; readonly at: Id }; + readonly parent?: { readonly conversationId: Id; readonly at: Id | "start" }; } export type UserInput = string | readonly (TextContent | ImageContent)[]; -export interface StoredEntryDraft { +export type StoredEntryDraft = { readonly kind: string; readonly data?: JsonValue; - readonly model?: readonly Message[]; + readonly model?: readonly Stored[]; readonly head?: Id | "self"; readonly edits?: readonly ContextEdit[]; -} +}; export type QueuedInput = - | { readonly mode: "steer" | "followUp"; readonly input: UserInput; readonly requestId?: string } + | { readonly mode: "steer" | "followUp"; readonly input: Stored; readonly requestId?: string } | { readonly mode: "write"; readonly entry: StoredEntryDraft; readonly requestId?: string }; -export type InputResult = +export type StoredInputResult = | { readonly status: "queued"; readonly requestId?: string } | { readonly status: "placed"; readonly requestId?: string; readonly entry: Id } | { readonly status: "done"; readonly requestId?: string; readonly entry: Id; readonly answer?: Id } @@ -88,7 +97,24 @@ export type InputResult = readonly detail?: string; }; -export type TerminalInputResult = Extract; +export type InputOutcome = + | { readonly status: "queued"; readonly requestId?: string } + | { readonly status: "placed"; readonly requestId?: string; readonly input: UserEntry } + | { + readonly status: "done"; + readonly requestId?: string; + readonly input: UserEntry; + readonly answer?: { readonly entry: AssistantEntry; readonly message: AssistantMessage }; + } + | { + readonly status: "unanswered"; + readonly requestId?: string; + readonly input?: UserEntry; + readonly reason: "terminated" | "aborted" | "failed" | "stale"; + readonly detail?: string; + }; + +export type TerminalInputOutcome = Extract; export interface InternalAcceptOptions { readonly input: UserInput; @@ -105,28 +131,36 @@ export interface SendInput { export interface InputHandle { readonly id: Id; readonly conversationId: Id; - result(ctx: Context): Promise; - wait(ctx: Context): Promise; + result(ctx: Context): Promise; + wait(ctx: Context): Promise; abort(ctx: Context): Promise<"aborted" | "already_placed" | "not_found">; } -interface TaskCreator { - task, OutputOf>>( +interface TaskCreator> { + task & Extra>( kind: K, - spec: NoExtra, OutputOf>, S> & { - readonly input: ExactJsonInput, S["input"]>; - }, + spec: NoExtra & Extra, S> & { readonly input: ExactJsonInput, S["input"]> }, ): Id; } -interface TaskTxBase extends TxReaders { - checkpoint(value: C): void; +export interface HostTx extends TxReaders, TxState, TaskCreator { createConversation(spec: ConversationCreateSpec): Id; - value(address: Value): TxValue; - list(address: List): TxList; + write(conversationId: Id, kind: EntryKind, input: EntryInput): Promise; } -export type BaseTaskTx = TaskTxBase & TaskCreator; +export interface ConversationTx extends TxReaders, BoundTxState, TaskCreator { + createConversation(spec: ConversationCreateSpec): Id; + write(kind: EntryKind, input: EntryInput): Promise; +} + +interface TaskTxBase extends TxReaders, BoundTxState { + checkpoint(value: C): void; + createConversation(spec: ConversationCreateSpec): Id; +} + +export interface BaseTaskTx extends TaskTxBase, TaskCreator { + write(kind: EntryKind, input: EntryInput): Promise; +} export interface InternalAdmissionTx { accept(conversationId: Id, options: InternalAcceptOptions): Promise; @@ -139,15 +173,21 @@ export interface InternalAdmissionTx { ): Promise; } -export type CoreTaskTx = TaskTxBase & - TaskCreator & +export type CoreTaskTx = Omit, "task" | "write"> & + TaskCreator & InternalAdmissionTx & { entry(kind: EntryKind, conversationId: Id, input: EntryInput): Id; }; -export interface PublicValue { - get(ctx: Context): Promise; - get(at: Id, ctx: Context): Promise; +export interface PublicValue { + get(ctx: Context): Promise; + get(at: Id, ctx: Context): Promise; + set(value: T, ctx: Context): Promise; + delete(ctx: Context): Promise; +} + +export interface StickyPublicValue { + get(ctx: Context): Promise; set(value: T, ctx: Context): Promise; delete(ctx: Context): Promise; } @@ -160,50 +200,49 @@ export interface PublicList { clear(ctx: Context): Promise; } -export interface TaskOutput { +export interface BoundPublicState { + value | UnboundValue>(address: D): PublicValue, ResolvedValue>; + list(address: List | UnboundList): PublicList; +} + +export interface TaskOutput { readonly ref: TaskOutputRef; read(ctx: Context): Promise; mutate(mutator: (state: O) => undefined, ctx: Context): Promise; replace(value: O, ctx: Context): Promise; } -export interface ReadonlyTaskOutput { +export interface ReadonlyTaskOutput { readonly ref: TaskOutputRef; read(ctx: Context): Promise; } -export type RuntimeOutput = [O] extends [never] ? unknown : { readonly output: TaskOutput }; -export type AbortRuntimeOutput = [O] extends [never] - ? unknown +export type RuntimeOutput = [O] extends [never] + ? Record + : { readonly output: TaskOutput }; +export type AbortRuntimeOutput = [O] extends [never] + ? Record : { readonly output: ReadonlyTaskOutput }; -export type FinalOutput = [O] extends [never] ? unknown : { readonly output: O }; - -export interface ScratchTx { - value(address: Value): TxValue; - list(address: List): TxList; -} +export type FinalOutput = [O] extends [never] ? Record : { readonly output: O }; export interface ScratchReader { - value(address: Value): Pick, "get">; + value>(address: D): Pick, ResolvedValue>, "get">; list(address: List): Pick, "read">; } -export interface TaskConversation { +export interface TaskConversation extends BoundPublicState { readonly id: Id; send(input: SendInput, ctx: Context): Promise; - value(address: Value): PublicValue; - list(address: List): PublicList; } -export interface AbortTaskConversation { +export interface AbortTaskConversation extends BoundPublicState { readonly id: Id; - value(address: Value): PublicValue; - list(address: List): PublicList; } -interface TaskRuntimeBase { +interface TaskRuntimeBase { readonly taskId: Id; - scratch(build: (tx: ScratchTx) => T | Promise, ctx: Context): Promise; + scratch(build: (tx: TxState) => T | Promise, ctx: Context): Promise; + readonly hooks: HookRunner; conversation(id: Id, ctx: Context): Promise; waitForTask(id: Id, ctx: Context): Promise; abortTask(id: Id, ctx: Context): Promise<"marked" | "terminal">; @@ -211,48 +250,43 @@ interface TaskRuntimeBase { sleep(untilMs: number, ctx: Context): Promise; } -interface CommitRuntime { +interface CommitRuntime { commit(build: (tx: Tx, current: RunningTask) => T | Promise, ctx: Context): Promise; } -export type TaskRuntime = TaskRuntimeBase & - CommitRuntime> & - RuntimeOutput; - -export type CoreTaskRuntime = TaskRuntimeBase & - CommitRuntime> & - RuntimeOutput; - -type TerminalClosureFor< - Tx, +export type TaskRuntime< I extends JsonValue, C extends TaskCheckpoint, - R extends JsonValue, - F extends JsonValue, - O extends object, -> = (tx: Tx & FinalOutput, current: RunningTask) => Completion | Promise>; + O extends OutputState = never, + H extends HookPoints = EmptyHookPoints, +> = TaskRuntimeBase & CommitRuntime> & RuntimeOutput; -export type FinalTx = BaseTaskTx & FinalOutput; -export type CoreFinalTx = CoreTaskTx & FinalOutput; +export type CoreTaskRuntime< + I extends JsonValue, + C extends TaskCheckpoint, + O extends OutputState = never, + H extends HookPoints = EmptyHookPoints, +> = TaskRuntimeBase & CommitRuntime> & RuntimeOutput; + +export type FinalTx = BaseTaskTx & FinalOutput; +export type CoreFinalTx = CoreTaskTx & FinalOutput; export type TerminalClosure< I extends JsonValue, C extends TaskCheckpoint, R extends JsonValue, F extends JsonValue, - O extends object, -> = TerminalClosureFor, I, C, R, F, O>; + O extends OutputState, +> = (tx: FinalTx, current: RunningTask) => Completion | Promise>; export type CoreTerminalClosure< I extends JsonValue, C extends TaskCheckpoint, R extends JsonValue, F extends JsonValue, - O extends object, -> = TerminalClosureFor, I, C, R, F, O>; + O extends OutputState, +> = (tx: CoreFinalTx, current: RunningTask) => Completion | Promise>; -export interface AbortTx extends TxReaders { +export interface AbortTx extends TxReaders, BoundTxState { checkpoint(value: C): void; - value(address: Value): TxValue; - list(address: List): TxList; markTask(id: Id): Promise<"marked" | "terminal">; } @@ -266,43 +300,44 @@ export interface CoreAbortTx extends AbortTx { ): Promise; } -type AbortClosureFor = ( - tx: Tx & FinalOutput, +export type AbortFinalTx = AbortTx & FinalOutput; +export type CoreAbortFinalTx = CoreAbortTx & FinalOutput; +export type AbortClosure = ( + tx: AbortFinalTx, current: RunningTask, ) => A | Promise; - -export type AbortFinalTx = AbortTx & FinalOutput; -export type CoreAbortFinalTx = CoreAbortTx & FinalOutput; -export type AbortClosure< - I extends JsonValue, - C extends TaskCheckpoint, - A extends JsonValue, - O extends object, -> = AbortClosureFor, I, C, A, O>; export type CoreAbortClosure< I extends JsonValue, C extends TaskCheckpoint, A extends JsonValue, - O extends object, -> = AbortClosureFor, I, C, A, O>; + O extends OutputState, +> = (tx: CoreAbortFinalTx, current: RunningTask) => A | Promise; -interface AbortRuntimeBase { +interface AbortRuntimeBase { readonly taskId: Id; scratch(read: (scratch: ScratchReader) => T | Promise, ctx: Context): Promise; + readonly hooks: HookRunner; conversation(id: Id, ctx: Context): Promise; waitForTask(id: Id, ctx: Context): Promise; abortTask(id: Id, ctx: Context): Promise<"marked" | "terminal">; now(): number; + sleep(untilMs: number, ctx: Context): Promise; } export type AbortTaskRuntime< I extends JsonValue, C extends TaskCheckpoint, - O extends object = never, -> = AbortRuntimeBase & CommitRuntime> & AbortRuntimeOutput; + O extends OutputState = never, + H extends HookPoints = EmptyHookPoints, +> = AbortRuntimeBase & CommitRuntime> & AbortRuntimeOutput; export type CoreAbortTaskRuntime< I extends JsonValue, C extends TaskCheckpoint, - O extends object = never, -> = AbortRuntimeBase & CommitRuntime> & AbortRuntimeOutput; + O extends OutputState = never, + H extends HookPoints = EmptyHookPoints, +> = AbortRuntimeBase & CommitRuntime> & AbortRuntimeOutput; + +export type ToolInput = { readonly assistant: Id; readonly call: Stored }; +export type GenerationOutput = { message?: Stored }; +export type ToolOutputState = { progress?: string; log: string[]; details?: JsonValue }; diff --git a/packages/agent/src/harness/pico/session.ts b/packages/agent/src/harness/pico/session.ts index bfe8d1243..5f7546efb 100644 --- a/packages/agent/src/harness/pico/session.ts +++ b/packages/agent/src/harness/pico/session.ts @@ -1,10 +1,33 @@ +import { AsyncLocalStorage } from "node:async_hooks"; import type { Context } from "@earendil-works/chord"; -import type { Address, Element, List, Value } from "./addresses.ts"; +import { withoutAbortSignal } from "@earendil-works/chord/context"; +import type { Element, List, PayloadOf, ResolvedValue, Value } from "./addresses.ts"; import type { Id, JsonValue } from "./core.ts"; import type { Conversation, Entry } from "./entries.ts"; import type { ConversationScan, EntryScan, NewTask, Page, Storage, TaskScan, Write } from "./storage.ts"; import type { Task } from "./tasks.ts"; +export class Closed extends Error { + constructor() { + super("Session is closed"); + this.name = "Closed"; + } +} + +export class Faulted extends Error { + constructor() { + super("Session is faulted"); + this.name = "Faulted"; + } +} + +export class NestedLineOperation extends Error { + constructor() { + super("Session line operations cannot be nested"); + this.name = "NestedLineOperation"; + } +} + export class ReadAfterWrite extends Error { constructor() { super("Transaction reads must precede writes"); @@ -12,30 +35,29 @@ export class ReadAfterWrite extends Error { } } -export class ScratchRetired extends Error { - readonly taskId: Id; - - constructor(taskId: Id) { - super(`Scratch for task ${taskId} is retired`); - this.name = "ScratchRetired"; - this.taskId = taskId; - } -} - -export class SharedOutputRetired extends Error { - readonly outputId: Id; - - constructor(outputId: Id) { - super(`Shared task output ${outputId} is retired`); - this.name = "SharedOutputRetired"; - this.outputId = outputId; - } -} - export type TaskCreate = Omit & { readonly output?: { readonly id?: Id; readonly kind: string }; }; +interface LineInvocation { + active: boolean; +} + +const lineInvocation = new AsyncLocalStorage(); +const storageOwners = new WeakMap(); + +function assertContextActive(ctx: Context): void { + ctx.abortSignal?.throwIfAborted(); +} + +function resolveStoredValue>( + address: D, + stored: PayloadOf | undefined, +): ResolvedValue { + const fallback = address.default as PayloadOf | undefined; + return (stored === undefined ? fallback : stored) as ResolvedValue; +} + export interface Transaction { getConversation(id: Id): Promise; getEntry(id: Id): Promise; @@ -46,11 +68,11 @@ export interface Transaction { entry(entry: Omit): Id; task(task: TaskCreate): Id; setTask(task: Task): void; - value( - address: Value, + value>( + address: D, ): { - get(at?: Id): Promise; - set(value: T): void; + get(at?: Id): Promise>; + set(value: PayloadOf): void; delete(): void; }; list( @@ -68,7 +90,9 @@ class TransactionState { private readonly ctx: Context; private active = true; private writing = false; + private entryAppended = false; private pendingReads = 0; + private failure: Error | undefined; private readonly writes: Write[] = []; constructor(storage: Storage, ctx: Context) { @@ -87,14 +111,14 @@ class TransactionState { entry: (entry: Omit) => this.entry(entry), task: (task: TaskCreate) => this.task(task), setTask: (task: Task) => this.setTask(task), - value: (address: Value) => this.value(address), + value: >(address: D) => this.value(address), list: (address: List) => this.list(address), }); } finish(): readonly Write[] { this.assertActive(); - if (this.pendingReads !== 0) throw new Error("Transaction has pending reads"); + if (this.pendingReads !== 0) this.poison(new Error("Transaction has pending reads")); this.active = false; return this.writes; } @@ -132,6 +156,7 @@ class TransactionState { entry(entry: Omit): Id { const id = this.allocate(); this.writes.push({ type: "entry.append", entry: { ...entry, id } }); + this.entryAppended = true; return id; } @@ -151,18 +176,28 @@ class TransactionState { this.write({ type: "task.set", task }); } - value( - address: Value, + value>( + address: D, ): { - get(at?: Id): Promise; - set(value: T): void; + get(at?: Id): Promise>; + set(value: PayloadOf): void; delete(): void; } { this.assertActive(); return Object.freeze({ - get: (at?: Id) => this.read(() => this.storage.getValue(address, at, this.ctx)), - set: (value: T) => this.write({ type: "value.set", address, value }), - delete: () => this.write({ type: "value.delete", address }), + get: (at?: Id) => + this.read(async () => { + const stored = await this.storage.getValue(address as unknown as Value>, at, this.ctx); + return resolveStoredValue(address, stored); + }), + set: (value: PayloadOf) => { + this.assertStateWriteOrder(address); + this.write({ type: "value.set", address, value }); + }, + delete: () => { + this.assertStateWriteOrder(address); + this.write({ type: "value.delete", address }); + }, }); } @@ -178,15 +213,29 @@ class TransactionState { return Object.freeze({ read: (at?: Id) => this.read(() => this.storage.readList(address, at, this.ctx)), append: (value: T) => { + this.assertStateWriteOrder(address); const id = this.allocate(); this.writes.push({ type: "list.append", address, element: { id, value } }); return id; }, - remove: (elementId: Id) => this.write({ type: "list.remove", address, elementId }), - clear: () => this.write({ type: "list.clear", address }), + remove: (elementId: Id) => { + this.assertStateWriteOrder(address); + this.write({ type: "list.remove", address, elementId }); + }, + clear: () => { + this.assertStateWriteOrder(address); + this.write({ type: "list.clear", address }); + }, }); } + private assertStateWriteOrder(address: Value | List): void { + this.assertActive(); + if (address.rewind && this.entryAppended) { + this.poison(new Error("Rewindable state writes must precede entry appends")); + } + } + private allocate(): Id { this.assertWritable(); return this.storage.nextId(); @@ -209,102 +258,183 @@ class TransactionState { private assertReadable(): void { this.assertActive(); - if (this.writing) throw new ReadAfterWrite(); + if (this.writing) this.poison(new ReadAfterWrite()); } private assertWritable(): void { this.assertActive(); - if (this.pendingReads !== 0) throw new Error("Transaction writes must await reads"); + if (this.pendingReads !== 0) this.poison(new Error("Transaction writes must await reads")); this.writing = true; } private assertActive(): void { if (!this.active) throw new Error("Transaction is closed"); + if (this.failure !== undefined) throw this.failure; + } + + private poison(error: Error): never { + this.failure ??= error; + throw this.failure; } } export class Session { private readonly storage: Storage; + private readonly ownership = Object.freeze({}); private tail: Promise = Promise.resolve(); + private fault: Faulted | undefined; + private closed = false; + private closeCompletion: Promise | undefined; + private storageCloseCompletion: Promise | undefined; constructor(storage: Storage) { + if (storageOwners.has(storage)) throw new Error("Storage is already owned by another Session"); + storageOwners.set(storage, this.ownership); this.storage = storage; } - async commit(build: (tx: Transaction) => T | Promise, ctx: Context): Promise { + commit(build: (tx: Transaction) => T | Promise, ctx: Context): Promise { + try { + assertContextActive(ctx); + } catch (error) { + return Promise.reject(error); + } + return this.onLine(async () => { + this.assertUsable(); + assertContextActive(ctx); + const tx = new TransactionState(this.storage, ctx); + try { + const result = await build(tx.facade()); + const writes = tx.finish(); + if (writes.length !== 0) { + try { + await this.storage.commit(writes, withoutAbortSignal(ctx)); + } catch (error) { + this.fault = new Faulted(); + try { + await this.closeStorage(ctx); + } catch {} + throw error; + } + } + return result; + } finally { + tx.seal(); + } + }); + } + + close(ctx: Context): Promise { + if (this.nested()) return Promise.reject(new NestedLineOperation()); + this.closeCompletion ??= this.enqueueLine(async () => { + this.closed = true; + await this.closeStorage(ctx); + }); + return this.closeCompletion; + } + + getConversations(ids: readonly Id[], ctx: Context): Promise> { + return this.onLine(() => { + this.assertUsable(); + return this.storage.getConversations(ids, ctx); + }); + } + + scanConversations(query: ConversationScan, ctx: Context): Promise> { + return this.onLine(() => { + this.assertUsable(); + return this.storage.scanConversations(query, ctx); + }); + } + + getEntries(ids: readonly Id[], ctx: Context): Promise> { + return this.onLine(() => { + this.assertUsable(); + return this.storage.getEntries(ids, ctx); + }); + } + + scanEntries(query: EntryScan, ctx: Context): Promise> { + return this.onLine(() => { + this.assertUsable(); + return this.storage.scanEntries(query, ctx); + }); + } + + newestHead(conversationId: Id, at: Id, ctx: Context): Promise { + return this.onLine(() => { + this.assertUsable(); + return this.storage.newestHead(conversationId, at, ctx); + }); + } + + getTasks(ids: readonly Id[], ctx: Context): Promise> { + return this.onLine(() => { + this.assertUsable(); + return this.storage.getTasks(ids, ctx); + }); + } + + scanTasks(query: TaskScan, ctx: Context): Promise> { + return this.onLine(() => { + this.assertUsable(); + return this.storage.scanTasks(query, ctx); + }); + } + + getValue>(address: D, at: Id | undefined, ctx: Context): Promise> { + return this.onLine(async () => { + this.assertUsable(); + const stored = await this.storage.getValue(address as unknown as Value>, at, ctx); + return resolveStoredValue(address, stored); + }); + } + + readList(address: List, at: Id | undefined, ctx: Context): Promise[]> { + return this.onLine(() => { + this.assertUsable(); + return this.storage.readList(address, at, ctx); + }); + } + + private onLine(operation: () => T | Promise): Promise { + if (this.nested()) return Promise.reject(new NestedLineOperation()); + return this.enqueueLine(operation); + } + + private enqueueLine(operation: () => T | Promise): Promise { const previous = this.tail; let release!: () => void; this.tail = new Promise((resolve) => { release = resolve; }); - await previous; - const tx = new TransactionState(this.storage, ctx); - try { - const result = await build(tx.facade()); - const writes = tx.finish(); - if (writes.length !== 0) await this.storage.commit(writes, ctx); - return result; - } finally { - tx.seal(); - release(); - } + return (async () => { + await previous; + const invocation: LineInvocation = { active: true }; + try { + return await lineInvocation.run(invocation, operation); + } finally { + invocation.active = false; + release(); + } + })(); } - getConversations(ids: readonly Id[], ctx: Context): Promise> { - return this.storage.getConversations(ids, ctx); + private nested(): boolean { + return lineInvocation.getStore()?.active === true; } - scanConversations(query: ConversationScan, ctx: Context): Promise> { - return this.storage.scanConversations(query, ctx); + private closeStorage(ctx: Context): Promise { + this.storageCloseCompletion ??= Promise.resolve() + .then(() => this.storage.close(withoutAbortSignal(ctx))) + .then(() => { + if (storageOwners.get(this.storage) === this.ownership) storageOwners.delete(this.storage); + }); + return this.storageCloseCompletion; } - getEntries(ids: readonly Id[], ctx: Context): Promise> { - return this.storage.getEntries(ids, ctx); - } - - scanEntries(query: EntryScan, ctx: Context): Promise> { - return this.storage.scanEntries(query, ctx); - } - - newestHead(conversationId: Id, at: Id, ctx: Context): Promise { - return this.storage.newestHead(conversationId, at, ctx); - } - - getTasks(ids: readonly Id[], ctx: Context): Promise> { - return this.storage.getTasks(ids, ctx); - } - - scanTasks(query: TaskScan, ctx: Context): Promise> { - return this.storage.scanTasks(query, ctx); - } - - async getValue(address: Value, at: Id | undefined, ctx: Context): Promise { - await this.assertReadable(address, ctx); - return this.storage.getValue(address, at, ctx); - } - - async readList( - address: List, - at: Id | undefined, - ctx: Context, - ): Promise[]> { - await this.assertReadable(address, ctx); - return this.storage.readList(address, at, ctx); - } - - private async outputIsLive(id: Id, ctx: Context): Promise { - return ( - (await this.storage.scanTasks({ outputId: id, statuses: ["pending", "running"], limit: 1 }, ctx)).items - .length > 0 - ); - } - - private async assertReadable(address: Address, ctx: Context): Promise { - if (address.scope.type === "task") { - const task = (await this.storage.getTasks([address.scope.taskId], ctx)).get(address.scope.taskId); - if (task === undefined || task.status === "terminal") throw new ScratchRetired(address.scope.taskId); - } else if (address.scope.type === "shared" && !(await this.outputIsLive(address.scope.id, ctx))) { - throw new SharedOutputRetired(address.scope.id); - } + private assertUsable(): void { + if (this.fault !== undefined) throw this.fault; + if (this.closed) throw new Closed(); } } diff --git a/packages/agent/src/harness/pico/storage.ts b/packages/agent/src/harness/pico/storage.ts index e1786273f..fbf507d84 100644 --- a/packages/agent/src/harness/pico/storage.ts +++ b/packages/agent/src/harness/pico/storage.ts @@ -1,8 +1,8 @@ import type { Context } from "@earendil-works/chord"; import type { Element, List, Value } from "./addresses.ts"; import type { Id, JsonValue, Seq } from "./core.ts"; -import type { Conversation, Entry } from "./entries.ts"; -import type { Task } from "./tasks.ts"; +import type { Conversation, Entry, SectionRecord } from "./entries.ts"; +import type { Task, TaskBase } from "./tasks.ts"; export interface PageQuery { /** Last item returned by the same scan. */ @@ -15,6 +15,26 @@ export interface Page { readonly next?: Id; } +export class ScratchRetired extends Error { + readonly taskId: Id; + + constructor(taskId: Id) { + super(`Scratch for task ${taskId} is retired`); + this.name = "ScratchRetired"; + this.taskId = taskId; + } +} + +export class OutputRetired extends Error { + readonly outputId: Id; + + constructor(outputId: Id) { + super(`Task output ${outputId} is retired`); + this.name = "OutputRetired"; + this.outputId = outputId; + } +} + export class InvalidHistoryPosition extends Error { readonly at: Id; @@ -25,7 +45,7 @@ export class InvalidHistoryPosition extends Error { } } -export type NewTask = Omit & { +export type NewTask = Omit & { readonly status: "pending"; readonly owns: readonly []; readonly abort?: never; @@ -40,9 +60,11 @@ export type StateWrite = | { readonly type: "list.remove"; readonly address: List; readonly elementId: Id } | { readonly type: "list.clear"; readonly address: List }; +export type StoredConversation = Conversation & { readonly sectionSeed?: readonly SectionRecord[] }; + export type Write = | StateWrite - | { readonly type: "conversation.create"; readonly conversation: Conversation } + | { readonly type: "conversation.create"; readonly conversation: StoredConversation } | { readonly type: "entry.append"; readonly entry: Entry } | { readonly type: "task.create"; readonly task: NewTask } | { readonly type: "task.set"; readonly task: Task }; @@ -78,4 +100,5 @@ export interface Storage { scanTasks(query: TaskScan, ctx: Context): Promise>; getValue(address: Value, at: Id | undefined, ctx: Context): Promise; readList(address: List, at: Id | undefined, ctx: Context): Promise[]>; + close(ctx: Context): Promise; } diff --git a/packages/agent/src/harness/pico/tasks.ts b/packages/agent/src/harness/pico/tasks.ts index 9066fabab..355f628e4 100644 --- a/packages/agent/src/harness/pico/tasks.ts +++ b/packages/agent/src/harness/pico/tasks.ts @@ -1,5 +1,7 @@ import type { Context } from "@earendil-works/chord"; +import type { ConversationValueDefinition } from "./addresses.ts"; import type { Id, JsonObject, JsonValue } from "./core.ts"; +import type { EmptyHookPoints, HookPoints } from "./hooks.ts"; import type { AbortClosure, AbortTaskRuntime, @@ -16,45 +18,64 @@ export interface TaskCheckpoint extends JsonObject { } export type NoCheckpoint = never; +export type OutputState = JsonObject | JsonValue[]; +export type ConfigBundle = Record>; +export type ReservedConfigKeys = "set" | "get"; +export type KeysOf = U extends unknown ? keyof U : never; +export type Collides = [ + Extract]> | Reserved>, +] extends [never] + ? false + : true; +export type AnyCollision = true extends { + [N in keyof M]: Collides; +}[keyof M] + ? true + : false; +export type DisjointBundles = AnyCollision extends false + ? true + : false; declare const taskOutputType: unique symbol; +const taskKindBrand: unique symbol = Symbol("pico.taskKind"); -export interface TaskOutputKind { +export interface TaskOutputKind { readonly kind: string; readonly [taskOutputType]?: O; } -export function defineTaskOutput(kind: string): TaskOutputKind { +export function defineTaskOutput(kind: string): TaskOutputKind { return Object.freeze({ kind }); } -export interface TaskOutputSpec { - readonly kind: TaskOutputKind; - initial(input: I): O; -} - export interface StoredTaskOutputRef { readonly id: Id; readonly kind: string; } -export interface TaskOutputRef extends StoredTaskOutputRef { +export interface TaskOutputRef extends StoredTaskOutputRef { readonly [taskOutputType]?: O; } -export type TaskOutputField = [O] extends [never] ? unknown : { readonly output: TaskOutputRef }; +export type TaskOutputField = [O] extends [never] + ? Record + : { readonly output: TaskOutputRef }; -export type TaskOutputDefinition = [O] extends [never] +export type TaskOutputDefinition = [O] extends [never] ? { readonly output?: never } - : { readonly output: TaskOutputSpec }; + : { readonly output: { readonly kind: TaskOutputKind; initial(input: I): O } }; -export type TaskOutcome = +export type TaskOutcome< + R extends JsonValue = JsonValue, + F extends JsonValue = JsonValue, + A extends JsonValue = JsonValue, +> = | { readonly status: "completed"; readonly result: R } | { readonly status: "failed"; readonly failure: F } | { readonly status: "aborted"; readonly result: A } | { readonly status: "orphaned" }; -export interface TaskBase { +export interface TaskBase { readonly id: Id; readonly conversationId: Id; readonly kind: string; @@ -79,7 +100,7 @@ export type Task< | { readonly status: "terminal"; readonly outcome: TaskOutcome } ); -export type RunningTask = Omit< +export type RunningTask = Omit< TaskBase, "output" > & @@ -92,39 +113,31 @@ export type Completion = | { readonly status: "completed"; readonly result: R } | { readonly status: "failed"; readonly failure: F }; -const taskKindBrand: unique symbol = Symbol("pico.taskKind"); - -export interface TaskKindBase { - readonly kind: string; - readonly [taskKindBrand]: "ordinary"; -} - -export interface CoreTaskKindBase { - readonly kind: string; - readonly [taskKindBrand]: "core"; -} - interface TaskKindMethods< I extends JsonValue, C extends TaskCheckpoint, R extends JsonValue, F extends JsonValue, A extends JsonValue, - O extends object, -> extends TaskKindBase { + O extends OutputState, + H extends HookPoints, +> { + readonly kind: string; + readonly hooks?: H; + readonly config?: ConfigBundle; execute( task: RunningTask, - runtime: TaskRuntime, + runtime: TaskRuntime, ctx: Context, ): Promise>; recover( task: RunningTask, - runtime: TaskRuntime, + runtime: TaskRuntime, ctx: Context, ): Promise>; abort( task: RunningTask, - runtime: AbortTaskRuntime, + runtime: AbortTaskRuntime, ctx: Context, ): Promise>; } @@ -135,21 +148,25 @@ interface CoreTaskKindMethods< R extends JsonValue, F extends JsonValue, A extends JsonValue, - O extends object, -> extends CoreTaskKindBase { + O extends OutputState, + H extends HookPoints, +> { + readonly kind: string; + readonly hooks?: H; + readonly config?: ConfigBundle; execute( task: RunningTask, - runtime: CoreTaskRuntime, + runtime: CoreTaskRuntime, ctx: Context, ): Promise>; recover( task: RunningTask, - runtime: CoreTaskRuntime, + runtime: CoreTaskRuntime, ctx: Context, ): Promise>; abort( task: RunningTask, - runtime: CoreAbortTaskRuntime, + runtime: CoreAbortTaskRuntime, ctx: Context, ): Promise>; } @@ -160,8 +177,9 @@ export type TaskKind< R extends JsonValue, F extends JsonValue, A extends JsonValue, - O extends object = never, -> = TaskKindMethods & TaskOutputDefinition; + O extends OutputState = never, + H extends HookPoints = EmptyHookPoints, +> = TaskKindMethods & TaskOutputDefinition & { readonly [taskKindBrand]: "ordinary" }; export type CoreTaskKind< I extends JsonValue, @@ -169,19 +187,37 @@ export type CoreTaskKind< R extends JsonValue, F extends JsonValue, A extends JsonValue, - O extends object = never, -> = CoreTaskKindMethods & TaskOutputDefinition; + O extends OutputState = never, + H extends HookPoints = EmptyHookPoints, +> = CoreTaskKindMethods & TaskOutputDefinition & { readonly [taskKindBrand]: "core" }; export type NoExtra = Actual & Record, never>; export type ExactJsonInput = Expected extends readonly JsonValue[] ? Actual : Expected extends JsonObject - ? Actual & Record, never> + ? NoExtra : Actual; -type KindDefinition = Omit; -type KindFactory = >(definition: NoExtra, D>) => D & K; +export type TaskKindDefinition< + I extends JsonValue, + C extends TaskCheckpoint, + R extends JsonValue, + F extends JsonValue, + A extends JsonValue, + O extends OutputState, + H extends HookPoints, +> = Omit, typeof taskKindBrand>; + +export type CoreTaskKindDefinition< + I extends JsonValue, + C extends TaskCheckpoint, + R extends JsonValue, + F extends JsonValue, + A extends JsonValue, + O extends OutputState, + H extends HookPoints, +> = Omit, typeof taskKindBrand>; export type TaskKindFactory< I extends JsonValue, @@ -189,8 +225,10 @@ export type TaskKindFactory< R extends JsonValue, F extends JsonValue, A extends JsonValue, - O extends object, -> = KindFactory>; + O extends OutputState, +> = >( + definition: NoExtra, D> & { readonly hooks?: H }, +) => D & TaskKind; export type CoreTaskKindFactory< I extends JsonValue, @@ -198,8 +236,10 @@ export type CoreTaskKindFactory< R extends JsonValue, F extends JsonValue, A extends JsonValue, - O extends object, -> = KindFactory>; + O extends OutputState, +> = >( + definition: NoExtra, D> & { readonly hooks?: H }, +) => D & CoreTaskKind; export function defineTask< I extends JsonValue, @@ -207,13 +247,15 @@ export function defineTask< R extends JsonValue, F extends JsonValue, A extends JsonValue, - O extends object = never, + O extends OutputState = never, >(): TaskKindFactory { - const define = (definition: KindDefinition>) => + const define = >( + definition: NoExtra, D> & { readonly hooks?: H }, + ): D & TaskKind => Object.freeze({ ...definition, [taskKindBrand]: "ordinary" as const, - }); + }) as D & TaskKind; return define as TaskKindFactory; } @@ -224,24 +266,36 @@ export function defineCoreTask< R extends JsonValue, F extends JsonValue, A extends JsonValue, - O extends object = never, + O extends OutputState = never, >(): CoreTaskKindFactory { - const define = (definition: KindDefinition>) => + const define = >( + definition: NoExtra, D> & { readonly hooks?: H }, + ): D & CoreTaskKind => Object.freeze({ ...definition, [taskKindBrand]: "core" as const, - }); + }) as D & CoreTaskKind; return define as CoreTaskKindFactory; } -export type AnyTaskKind = TaskKindBase; -export type AnyCoreTaskKind = CoreTaskKindBase; -export type AnyDefinedTaskKind = AnyTaskKind | AnyCoreTaskKind; +export type OrdinaryKind = { + readonly kind: string; + readonly [taskKindBrand]: "ordinary"; + readonly hooks?: HookPoints; + readonly config?: ConfigBundle; +}; +export type CoreKind = { + readonly kind: string; + readonly [taskKindBrand]: "core"; + readonly hooks?: HookPoints; + readonly config?: ConfigBundle; +}; +export type AnyDefinedKind = OrdinaryKind | CoreKind; -export type PayloadsOf = K extends TaskKind - ? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O } - : K extends CoreTaskKind - ? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O } +export type PayloadsOf = K extends TaskKind + ? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O; hooks: H } + : K extends CoreTaskKind + ? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O; hooks: H } : never; export type InputOf = PayloadsOf["input"]; export type CheckpointOf = PayloadsOf["checkpoint"]; @@ -249,9 +303,10 @@ export type ResultOf = PayloadsOf["result"]; export type FailureOf = PayloadsOf["failure"]; export type AbortedOf = PayloadsOf["aborted"]; export type OutputOf = PayloadsOf["output"]; -type WithoutOutput = T extends unknown ? Omit : never; +export type DistributiveOmit = T extends unknown ? Omit : never; -export type TaskOf = WithoutOutput< - Task, CheckpointOf, ResultOf, FailureOf, AbortedOf> +export type TaskOf = DistributiveOmit< + Task, CheckpointOf, ResultOf, FailureOf, AbortedOf>, + "output" > & TaskOutputField>; diff --git a/packages/agent/test/harness/pico/memory-storage.test.ts b/packages/agent/test/harness/pico/memory-storage.test.ts index d65483909..c2512fa84 100644 --- a/packages/agent/test/harness/pico/memory-storage.test.ts +++ b/packages/agent/test/harness/pico/memory-storage.test.ts @@ -1,22 +1,39 @@ -import { TODO_CONTEXT } from "@earendil-works/chord/context"; +import { TODO_CONTEXT, withCancel } from "@earendil-works/chord/context"; import { describe, expect, it, vi } from "vitest"; +import { defineInternalList, defineInternalValue } from "../../../src/harness/pico/addresses.ts"; import { + conversationList, + conversationValue, defineList, defineValue, InvalidHistoryPosition, type JsonValue, MemoryStorage, - ReadAfterWrite, + OutputRetired, ScratchRetired, - Session, - SharedOutputRetired, + sessionValue, type Task, +} from "../../../src/harness/pico/index.ts"; +import { + Closed, + Faulted, + NestedLineOperation, + ReadAfterWrite, + Session, type TaskCreate, type Transaction, -} from "../../../src/harness/pico/index.ts"; +} from "../../../src/harness/pico/session.ts"; const ctx = TODO_CONTEXT; +function deferred(): { readonly promise: Promise; readonly resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((settled) => { + resolve = settled; + }); + return { promise, resolve }; +} + function taskSpec( conversationId: number, extra: Partial> = {}, @@ -44,13 +61,13 @@ async function setTask(session: Session, id: number, update: (task: LiveTask) => describe("Pico Session over MemoryStorage", () => { it("separates IDs from sequences and burns IDs from failed callbacks", async () => { - const storage = new MemoryStorage(); + const storage = MemoryStorage.create(); expect(await storage.commit([{ type: "conversation.create", conversation: { id: 10 } }], ctx)).toEqual([1]); const state = defineValue({ type: "session" }, "test.state"); expect(await storage.commit([{ type: "value.set", address: state, value: "set" }], ctx)).toEqual([2]); expect(await new Session(storage).commit((tx) => tx.createConversation({}), ctx)).toBe(11); - const sessionStorage = new MemoryStorage(); + const sessionStorage = MemoryStorage.create(); const commit = vi.spyOn(sessionStorage, "commit"); const session = new Session(sessionStorage); await expect( @@ -71,14 +88,14 @@ describe("Pico Session over MemoryStorage", () => { }, ctx), ).rejects.toBeInstanceOf(ReadAfterWrite); expect(await session.commit((tx) => tx.createConversation({}), ctx)).toBe(4); - expect( - await session.commit(async (tx) => { + await expect( + session.commit(async (tx) => { const read = tx.getConversation(1); expect(() => tx.createConversation({})).toThrow("Transaction writes must await reads"); await read; - return tx.createConversation({}); }, ctx), - ).toBe(5); + ).rejects.toThrow("Transaction writes must await reads"); + expect(await session.commit((tx) => tx.createConversation({}), ctx)).toBe(5); let leaked!: Transaction; await session.commit((tx) => { @@ -97,24 +114,248 @@ describe("Pico Session over MemoryStorage", () => { expect(commit).toHaveBeenCalledTimes(5); }); - it("leaves state and sequence unchanged when persistence rejects", async () => { - const storage = new MemoryStorage(); - const commit = vi.spyOn(storage, "commit").mockRejectedValueOnce(new Error("persistence failed")); - await expect(new Session(storage).commit((tx) => tx.createConversation({}), ctx)).rejects.toThrow( - "persistence failed", - ); + it("serializes direct reads and commits in invocation order", async () => { + const session = new Session(MemoryStorage.create()); + const entered = deferred(); + const release = deferred(); + const holding = session.commit(async () => { + entered.resolve(); + await release.promise; + }, ctx); + await entered.promise; + + let firstReadSettled = false; + const firstRead = session.getConversations([1], ctx).then((conversations) => { + firstReadSettled = true; + return conversations; + }); + const created = session.commit((tx) => tx.createConversation({}), ctx); + const secondRead = session.getConversations([1], ctx); + await Promise.resolve(); + expect(firstReadSettled).toBe(false); + + release.resolve(); + await holding; + expect(await firstRead).toEqual(new Map()); + expect(await created).toBe(1); + expect((await secondRead).has(1)).toBe(true); + }); + + it("checks commit cancellation before queueing and before invoking its callback", async () => { + const storage = MemoryStorage.create(); + const commit = vi.spyOn(storage, "commit"); + const session = new Session(storage); + const entered = deferred(); + const release = deferred(); + const holding = session.commit(async () => { + entered.resolve(); + await release.promise; + }, ctx); + await entered.promise; + + const beforeEntry = withCancel(ctx); + const beforeEntryError = new Error("cancelled before entry"); + beforeEntry.cancel(beforeEntryError); + let beforeEntryCallbackRan = false; + await expect( + session.commit(() => { + beforeEntryCallbackRan = true; + }, beforeEntry.context), + ).rejects.toBe(beforeEntryError); + expect(beforeEntryCallbackRan).toBe(false); + + const beforeCallback = withCancel(ctx); + const beforeCallbackError = new Error("cancelled before callback"); + let beforeCallbackRan = false; + const queued = session.commit(() => { + beforeCallbackRan = true; + }, beforeCallback.context); + beforeCallback.cancel(beforeCallbackError); + release.resolve(); + await holding; + await expect(queued).rejects.toBe(beforeCallbackError); + expect(beforeCallbackRan).toBe(false); + expect(commit).not.toHaveBeenCalled(); + }); + + it("uses a non-cancellable context for admitted persistence", async () => { + const storage = MemoryStorage.create(); + const commit = vi.spyOn(storage, "commit"); + const read = vi.spyOn(storage, "getConversations"); + const session = new Session(storage); + const caller = withCancel(ctx); + const callerError = new Error("cancelled after callback admission"); + expect( + await session.commit((tx) => { + const id = tx.createConversation({}); + caller.cancel(callerError); + return id; + }, caller.context), + ).toBe(1); expect(commit).toHaveBeenCalledTimes(1); + expect(commit.mock.calls[0]![1].abortSignal).toBeUndefined(); + expect(caller.context.abortSignal?.reason).toBe(callerError); + expect((await session.getConversations([1], ctx)).has(1)).toBe(true); + expect(read.mock.calls[0]![1]).toBe(ctx); + }); + + it("rejects nested Session line entry immediately", async () => { + const session = new Session(MemoryStorage.create()); + const otherSession = new Session(MemoryStorage.create()); + expect( + await session.commit(async (tx) => { + await expect(session.getConversations([], ctx)).rejects.toBeInstanceOf(NestedLineOperation); + await expect(session.commit(() => undefined, ctx)).rejects.toBeInstanceOf(NestedLineOperation); + await expect(session.close(ctx)).rejects.toBeInstanceOf(NestedLineOperation); + await expect(otherSession.getConversations([], ctx)).rejects.toBeInstanceOf(NestedLineOperation); + return tx.createConversation({}); + }, ctx), + ).toBe(1); + expect((await session.getConversations([1], ctx)).has(1)).toBe(true); + await otherSession.close(ctx); + }); + + it("rejects a transaction when a caught ReadAfterWrite poisoned it", async () => { + const storage = MemoryStorage.create(); + const commit = vi.spyOn(storage, "commit"); + const session = new Session(storage); + await expect( + session.commit(async (tx) => { + tx.createConversation({}); + try { + await tx.getConversation(1); + } catch (error) { + expect(error).toBeInstanceOf(ReadAfterWrite); + } + }, ctx), + ).rejects.toBeInstanceOf(ReadAfterWrite); + expect(commit).not.toHaveBeenCalled(); + expect(await session.commit((tx) => tx.createConversation({}), ctx)).toBe(2); + }); + + it("reopens from the committed-ID high-water and reuses only uncommitted IDs", async () => { + const storage = MemoryStorage.create(); + expect(storage.nextId()).toBe(1); + await storage.close(ctx); + await storage.close(ctx); + expect(storage.nextId()).toBe(1); + const session = new Session(storage); + + expect(await session.commit((tx) => tx.createConversation({}), ctx)).toBe(2); + expect(storage.nextId()).toBe(3); + await session.close(ctx); + expect(storage.nextId()).toBe(3); + + const list = defineList({ type: "session" }, "test.high-water"); + expect( + await storage.commit( + [ + { type: "list.append", address: list, element: { id: 20, value: "retired" } }, + { type: "list.clear", address: list }, + ], + ctx, + ), + ).toEqual([2, 3]); + await storage.close(ctx); + expect(storage.nextId()).toBe(21); + }); + + it("faults and closes the Session after an uncertain persistence failure", async () => { + const storage = MemoryStorage.create(); + const commit = vi.spyOn(storage, "commit").mockRejectedValueOnce(new Error("persistence failed")); + const close = vi.spyOn(storage, "close"); + const session = new Session(storage); + const caller = withCancel(ctx); + await expect( + session.commit((tx) => { + const id = tx.createConversation({}); + caller.cancel(new Error("cancelled during persistence")); + return id; + }, caller.context), + ).rejects.toThrow("persistence failed"); + expect(commit).toHaveBeenCalledTimes(1); + expect(close).toHaveBeenCalledTimes(1); + expect(close.mock.calls[0]![0].abortSignal).toBeUndefined(); + await expect(session.getConversations([1], ctx)).rejects.toBeInstanceOf(Faulted); + await expect(session.commit((tx) => tx.createConversation({}), ctx)).rejects.toBeInstanceOf(Faulted); + await session.close(ctx); + await session.close(ctx); + expect(close).toHaveBeenCalledTimes(1); + const recovered = new Session(storage); + await recovered.close(ctx); + expect(close).toHaveBeenCalledTimes(2); expect(await storage.getConversations([1], ctx)).toEqual(new Map()); commit.mockRestore(); const state = defineValue({ type: "session" }, "test.after-failure"); expect(await storage.commit([{ type: "value.set", address: state, value: "stored" }], ctx)).toEqual([1]); expect(await storage.getValue(state, undefined, ctx)).toBe("stored"); - expect(storage.nextId()).toBe(2); + expect(storage.nextId()).toBe(1); + }); + + it("makes operations queued behind an uncertain commit observe Faulted", async () => { + const storage = MemoryStorage.create(); + const enteredCommit = deferred(); + const rejectCommit = deferred(); + vi.spyOn(storage, "commit").mockImplementationOnce(async () => { + enteredCommit.resolve(); + await rejectCommit.promise; + throw new Error("uncertain"); + }); + const session = new Session(storage); + const failing = session.commit((tx) => tx.createConversation({}), ctx); + await enteredCommit.promise; + + let laterCallbackRan = false; + const queuedRead = session.scanConversations({ limit: 1 }, ctx); + const queuedCommit = session.commit(() => { + laterCallbackRan = true; + }, ctx); + rejectCommit.resolve(); + + await expect(failing).rejects.toThrow("uncertain"); + await expect(queuedRead).rejects.toBeInstanceOf(Faulted); + await expect(queuedCommit).rejects.toBeInstanceOf(Faulted); + expect(laterCallbackRan).toBe(false); + }); + + it("enforces exclusive Storage ownership until Session close succeeds", async () => { + const storage = MemoryStorage.create(); + const first = new Session(storage); + expect(() => new Session(storage)).toThrow("already owned"); + await first.close(ctx); + const reopened = new Session(storage); + await reopened.close(ctx); + }); + + it("keeps Storage owned when close fails", async () => { + const storage = MemoryStorage.create(); + const close = vi.spyOn(storage, "close").mockRejectedValueOnce(new Error("close failed")); + const session = new Session(storage); + await expect(session.close(ctx)).rejects.toThrow("close failed"); + await expect(session.close(ctx)).rejects.toThrow("close failed"); + expect(close).toHaveBeenCalledTimes(1); + expect(() => new Session(storage)).toThrow("already owned"); + }); + + it("closes once and rejects later low-level Session operations", async () => { + const storage = MemoryStorage.create(); + const close = vi.spyOn(storage, "close"); + const session = new Session(storage); + const caller = withCancel(ctx); + caller.cancel(new Error("close caller cancelled")); + const first = session.close(caller.context); + const second = session.close(ctx); + expect(second).toBe(first); + await first; + expect(close).toHaveBeenCalledTimes(1); + expect(close.mock.calls[0]![0].abortSignal).toBeUndefined(); + await expect(session.getConversations([], ctx)).rejects.toBeInstanceOf(Closed); + await expect(session.commit(() => undefined, ctx)).rejects.toBeInstanceOf(Closed); }); it("uses write sequences for historical state inside one atomic storage batch", async () => { - const storage = new MemoryStorage(); + const storage = MemoryStorage.create(); const conversation = storage.nextId(); const entry = storage.nextId(); const beforeElement = storage.nextId(); @@ -146,8 +387,143 @@ describe("Pico Session over MemoryStorage", () => { expect((await storage.readList(list, undefined, ctx)).map((item) => item.value)).toEqual(["before", "after"]); }); + it("rejects rewindable state writes after an entry in a transaction", async () => { + const storage = MemoryStorage.create(); + const session = new Session(storage); + const value = defineValue({ type: "conversation", conversationId: 1 }, "test.order", { + rewind: true, + }); + await expect( + session.commit((tx) => { + tx.createConversation({}); + tx.entry({ conversationId: 1, kind: "test.entry" }); + try { + tx.value(value).set("too late"); + } catch (error) { + expect(error).toHaveProperty("message", "Rewindable state writes must precede entry appends"); + } + }, ctx), + ).rejects.toThrow("must precede"); + expect(await storage.getConversations([1], ctx)).toEqual(new Map()); + expect(await session.commit((tx) => tx.createConversation({}), ctx)).toBe(3); + }); + + it("validates list removals against their logical lineage", async () => { + const storage = MemoryStorage.create(); + const rootList = defineList({ type: "conversation", conversationId: 1 }, "test.lineage", { + rewind: true, + }); + await storage.commit( + [ + { + type: "conversation.create", + conversation: { + id: 1, + sectionSeed: [{ key: "identity", action: "set", value: "test", rendered: "test" }], + }, + }, + { type: "list.append", address: rootList, element: { id: 2, value: "inherited" } }, + { type: "entry.append", entry: { id: 3, conversationId: 1, kind: "test.fork-point" } }, + { type: "conversation.create", conversation: { id: 4, parent: { conversationId: 1, at: 3 } } }, + { type: "conversation.create", conversation: { id: 5 } }, + ], + ctx, + ); + expect((await storage.getConversations([1], ctx)).get(1)).toEqual({ id: 1 }); + expect((await storage.scanConversations({ limit: 10 }, ctx)).items[0]).toEqual({ id: 1 }); + const childList = defineList({ type: "conversation", conversationId: 4 }, "test.lineage", { + rewind: true, + }); + const unrelatedList = defineList({ type: "conversation", conversationId: 5 }, "test.lineage", { + rewind: true, + }); + const otherDefinition = defineList({ type: "conversation", conversationId: 4 }, "test.other", { + rewind: true, + }); + const marker = defineValue({ type: "session" }, "test.atomic-remove"); + expect( + await storage.commit( + [{ type: "list.append", address: rootList, element: { id: 6, value: "after fork" } }], + ctx, + ), + ).toEqual([6]); + await expect( + storage.commit( + [ + { type: "value.set", address: marker, value: "must not land" }, + { type: "list.remove", address: childList, elementId: 6 }, + ], + ctx, + ), + ).rejects.toThrow("does not belong"); + expect(await storage.getValue(marker, undefined, ctx)).toBeUndefined(); + + expect(await storage.commit([{ type: "list.remove", address: childList, elementId: 2 }], ctx)).toEqual([7]); + expect(await storage.commit([{ type: "list.remove", address: childList, elementId: 2 }], ctx)).toEqual([8]); + expect((await storage.readList(rootList, undefined, ctx)).map(({ id }) => id)).toEqual([2, 6]); + expect(await storage.readList(childList, undefined, ctx)).toEqual([]); + await expect( + storage.commit([{ type: "list.remove", address: unrelatedList, elementId: 2 }], ctx), + ).rejects.toThrow("does not belong"); + await expect( + storage.commit([{ type: "list.remove", address: otherDefinition, elementId: 2 }], ctx), + ).rejects.toThrow("does not belong"); + expect(await storage.commit([{ type: "list.clear", address: childList }], ctx)).toEqual([9]); + }); + + it("validates inherited list removal against earlier writes in the same batch", async () => { + const storage = MemoryStorage.create(); + const session = new Session(storage); + const parentList = defineList({ type: "conversation", conversationId: 1 }, "test.same-batch-lineage", { + rewind: true, + }); + const seeded = await session.commit((tx) => { + const parent = tx.createConversation({}); + const element = tx.list(parentList).append("persisted"); + const at = tx.entry({ conversationId: parent, kind: "test.fork-point" }); + return { parent, element, at }; + }, ctx); + + const child = await session.commit((tx) => { + const id = tx.createConversation({ parent: { conversationId: seeded.parent, at: seeded.at } }); + const childList = defineList({ type: "conversation", conversationId: id }, "test.same-batch-lineage", { + rewind: true, + }); + tx.list(childList).remove(seeded.element); + return id; + }, ctx); + const childList = defineList({ type: "conversation", conversationId: child }, "test.same-batch-lineage", { + rewind: true, + }); + expect(await storage.readList(childList, undefined, ctx)).toEqual([]); + expect((await storage.readList(parentList, undefined, ctx)).map(({ id }) => id)).toEqual([seeded.element]); + + const prospectiveChildList = defineList( + { type: "conversation", conversationId: 22 }, + "test.same-batch-lineage", + { rewind: true }, + ); + expect( + await storage.commit( + [ + { type: "list.append", address: parentList, element: { id: 20, value: "prospective" } }, + { type: "entry.append", entry: { id: 21, conversationId: seeded.parent, kind: "test.next-fork" } }, + { + type: "conversation.create", + conversation: { id: 22, parent: { conversationId: seeded.parent, at: 21 } }, + }, + { type: "list.remove", address: prospectiveChildList, elementId: 20 }, + ], + ctx, + ), + ).toEqual([6, 7, 8, 9]); + expect((await storage.readList(prospectiveChildList, undefined, ctx)).map(({ id }) => id)).toEqual([ + seeded.element, + ]); + }); + it("pages conversations and fork-visible entries", async () => { - const session = new Session(new MemoryStorage()); + const session = new Session(MemoryStorage.create()); const ids = await session.commit((tx) => { const root = tx.createConversation({}); const first = tx.entry({ conversationId: root, kind: "test.first" }); @@ -159,6 +535,12 @@ describe("Pico Session over MemoryStorage", () => { return { root, first, head, fork, forkEntry, late, ownedFork }; }, ctx); expect(ids).toEqual({ root: 1, first: 2, head: 3, fork: 4, forkEntry: 5, late: 6, ownedFork: 7 }); + const filteredIds = await session.commit((tx) => { + const older = tx.entry({ conversationId: ids.root, kind: "test.match" }); + tx.entry({ conversationId: ids.root, kind: "test.other" }); + const newer = tx.entry({ conversationId: ids.root, kind: "test.match" }); + return { older, newer }; + }, ctx); const conversations = await session.scanConversations({ limit: 1 }, ctx); expect(conversations.items.map(({ id }) => id)).toEqual([1]); @@ -189,13 +571,68 @@ describe("Pico Session over MemoryStorage", () => { ({ id }) => id, ), ).toEqual([ids.head]); + const filtered = await session.scanEntries({ conversationId: ids.root, kind: "test.match", limit: 1 }, ctx); + expect(filtered.items.map(({ id }) => id)).toEqual([filteredIds.newer]); + expect(filtered.next).toBe(filteredIds.newer); + const filteredNext = await session.scanEntries( + { conversationId: ids.root, kind: "test.match", cursor: filtered.next, limit: 1 }, + ctx, + ); + expect(filteredNext.items.map(({ id }) => id)).toEqual([filteredIds.older]); + expect(filteredNext.next).toBeUndefined(); expect((await session.getEntries([ids.first, 999_999], ctx)).has(ids.first)).toBe(true); expect((await session.newestHead(1, 3, ctx))?.id).toBe(3); expect(await session.newestHead(4, 5, ctx)).toBeUndefined(); }); + it("uses entry sequences for scans, cutoffs, cursors, and nested fork caps", async () => { + const storage = MemoryStorage.create(); + await storage.commit( + [ + { type: "conversation.create", conversation: { id: 100 } }, + { type: "entry.append", entry: { id: 900, conversationId: 100, kind: "root.old", head: 900 } }, + { type: "entry.append", entry: { id: 700, conversationId: 100, kind: "root.fork" } }, + { type: "conversation.create", conversation: { id: 200, parent: { conversationId: 100, at: 700 } } }, + { type: "entry.append", entry: { id: 950, conversationId: 100, kind: "root.late" } }, + { type: "entry.append", entry: { id: 10, conversationId: 200, kind: "child.fork" } }, + { type: "entry.append", entry: { id: 5, conversationId: 200, kind: "child.late" } }, + { type: "conversation.create", conversation: { id: 300, parent: { conversationId: 200, at: 10 } } }, + { type: "entry.append", entry: { id: 1, conversationId: 300, kind: "grand.local" } }, + { type: "conversation.create", conversation: { id: 400 } }, + { type: "entry.append", entry: { id: 2, conversationId: 400, kind: "unrelated" } }, + ], + ctx, + ); + + expect((await storage.scanEntries({ conversationId: 300, limit: 10 }, ctx)).items.map(({ id }) => id)).toEqual([ + 1, 10, 700, 900, + ]); + expect( + (await storage.scanEntries({ conversationId: 300, through: 10, limit: 10 }, ctx)).items.map(({ id }) => id), + ).toEqual([10, 700, 900]); + expect( + (await storage.scanEntries({ conversationId: 300, through: 700, limit: 10 }, ctx)).items.map(({ id }) => id), + ).toEqual([700, 900]); + const first = await storage.scanEntries({ conversationId: 300, limit: 2 }, ctx); + expect(first.items.map(({ id }) => id)).toEqual([1, 10]); + expect(first.next).toBe(10); + expect( + (await storage.scanEntries({ conversationId: 300, cursor: first.next, limit: 2 }, ctx)).items.map( + ({ id }) => id, + ), + ).toEqual([700, 900]); + expect((await storage.newestHead(300, 10, ctx))?.id).toBe(900); + await expect(storage.scanEntries({ conversationId: 300, through: 2, limit: 10 }, ctx)).rejects.toBeInstanceOf( + InvalidHistoryPosition, + ); + await expect(storage.scanEntries({ conversationId: 300, cursor: 2, limit: 10 }, ctx)).rejects.toBeInstanceOf( + InvalidHistoryPosition, + ); + await expect(storage.newestHead(300, 2, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); + }); + it("buffers values and list elements into one commit", async () => { - const session = new Session(new MemoryStorage()); + const session = new Session(MemoryStorage.create()); const value = defineValue({ type: "conversation", conversationId: 1 }, "test.value", { rewind: false, }); @@ -221,8 +658,63 @@ describe("Pico Session over MemoryStorage", () => { expect(await session.readList(list, undefined, ctx)).toEqual([]); }); + it("resolves value defaults in Session and transaction reads", async () => { + const storage = MemoryStorage.create(); + const session = new Session(storage); + const sessionName = sessionValue("test.default.session", { default: "default" }); + const conversationDefinition = conversationValue("test.default.conversation", { + rewind: true, + default: 7, + }); + const boundConversationValue = conversationDefinition.bind(1); + const initial = await session.commit((tx) => { + tx.createConversation({}); + return tx.entry({ conversationId: 1, kind: "test.initial" }); + }, ctx); + + expect(await session.getValue(sessionName, undefined, ctx)).toBe("default"); + expect(await session.getValue(boundConversationValue, undefined, ctx)).toBe(7); + expect(await session.getValue(boundConversationValue, initial, ctx)).toBe(7); + expect( + await session.commit( + async (tx) => ({ + session: await tx.value(sessionName).get(), + historical: await tx.value(boundConversationValue).get(initial), + }), + ctx, + ), + ).toEqual({ session: "default", historical: 7 }); + + const setAt = await session.commit((tx) => { + tx.value(sessionName).set("set"); + tx.value(boundConversationValue).set(9); + return tx.entry({ conversationId: 1, kind: "test.set" }); + }, ctx); + const deletedAt = await session.commit((tx) => { + tx.value(sessionName).delete(); + tx.value(boundConversationValue).delete(); + return tx.entry({ conversationId: 1, kind: "test.deleted" }); + }, ctx); + + expect(await session.getValue(sessionName, undefined, ctx)).toBe("default"); + expect(await session.getValue(boundConversationValue, setAt, ctx)).toBe(9); + expect(await session.getValue(boundConversationValue, deletedAt, ctx)).toBe(7); + expect(await storage.getValue(sessionName, undefined, ctx)).toBeUndefined(); + expect(await storage.getValue(boundConversationValue, deletedAt, ctx)).toBeUndefined(); + expect( + await session.commit( + async (tx) => ({ + session: await tx.value(sessionName).get(), + current: await tx.value(boundConversationValue).get(), + historical: await tx.value(boundConversationValue).get(deletedAt), + }), + ctx, + ), + ).toEqual({ session: "default", current: 7, historical: 7 }); + }); + it("rewinds values through fork caps and preserves local tombstones", async () => { - const storage = new MemoryStorage(); + const storage = MemoryStorage.create(); const session = new Session(storage); const root = await session.commit((tx) => tx.createConversation({}), ctx); const rootValue = defineValue({ type: "conversation", conversationId: root }, "test.history", { @@ -314,7 +806,7 @@ describe("Pico Session over MemoryStorage", () => { }); const sessionValue = defineValue({ type: "session" }, "test.history"); const taskList = defineList({ type: "task", taskId: 100 }, "test.history"); - const sharedValue = defineValue({ type: "shared", id: 200 }, "test.history"); + const sharedValue = defineInternalValue({ type: "shared", id: 200 }, "test.history"); await expect(storage.getValue(sticky, first, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); await expect(storage.getValue(sessionValue, first, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); await expect(storage.readList(taskList, first, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); @@ -322,7 +814,7 @@ describe("Pico Session over MemoryStorage", () => { }); it("rewinds complete lists through append, remove, clear, and nested fork caps", async () => { - const session = new Session(new MemoryStorage()); + const session = new Session(MemoryStorage.create()); const root = await session.commit((tx) => tx.createConversation({}), ctx); const rootList = defineList({ type: "conversation", conversationId: root }, "test.history-list", { rewind: true, @@ -433,8 +925,22 @@ describe("Pico Session over MemoryStorage", () => { await expect(session.readList(stickyList, childRemove, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); }); + it("reserves internal namespaces and shared address construction", () => { + const sharedAsSession = { type: "shared", id: 1 } as unknown as { readonly type: "session" }; + expect(() => defineValue(sharedAsSession, "plugin.shared")).toThrow("Shared scope is internal"); + expect(() => defineList(sharedAsSession, "plugin.shared")).toThrow("Shared scope is internal"); + expect(() => defineValue({ type: "session" }, "pi.internal")).toThrow("reserved"); + expect(() => defineList({ type: "task", taskId: 1 }, "pi.internal")).toThrow("reserved"); + expect(() => conversationValue("pi.internal", { rewind: true })).toThrow("reserved"); + expect(() => conversationList("pi.internal", { rewind: false })).toThrow("reserved"); + expect(defineInternalList({ type: "shared", id: 1 }, "pi.output").scope).toEqual({ + type: "shared", + id: 1, + }); + }); + it("keeps scopes, kinds, optional keys, and rewind stores separate", async () => { - const session = new Session(new MemoryStorage()); + const session = new Session(MemoryStorage.create()); const conversation = await session.commit((tx) => tx.createConversation({}), ctx); const scope = { type: "conversation", conversationId: conversation } as const; const stickyValue = defineValue(scope, "test.identity", { key: "same", rewind: false }); @@ -467,19 +973,19 @@ describe("Pico Session over MemoryStorage", () => { }); it("stores task snapshots and retires scratch and shared output", async () => { - const storage = new MemoryStorage(); + const storage = MemoryStorage.create(); const session = new Session(storage); const ids = await session.commit((tx) => { const conversation = tx.createConversation({}); const producer = tx.task(taskSpec(conversation, { output: { kind: "test.output" } })); - const output = defineList({ type: "shared", id: producer }, "pi.output"); + const output = defineInternalList({ type: "shared", id: producer }, "pi.output"); tx.list(output).append([["r", { text: "" }]]); const consumer = tx.task(taskSpec(conversation, { output: { id: producer, kind: "test.output" } })); return { conversation, producer, consumer }; }, ctx); expect(ids).toEqual({ conversation: 1, producer: 2, consumer: 4 }); - const output = defineList({ type: "shared", id: ids.producer }, "pi.output"); + const output = defineInternalList({ type: "shared", id: ids.producer }, "pi.output"); const scratch = defineValue({ type: "task", taskId: ids.producer }, "test.scratch"); await session.commit((tx) => { tx.value(scratch).set({ n: 1 }); @@ -495,7 +1001,7 @@ describe("Pico Session over MemoryStorage", () => { outcome: { status: "completed", result: null }, })); await expect(session.getValue(scratch, undefined, ctx)).rejects.toBeInstanceOf(ScratchRetired); - expect(await storage.getValue(scratch, undefined, ctx)).toBeUndefined(); + await expect(storage.getValue(scratch, undefined, ctx)).rejects.toBeInstanceOf(ScratchRetired); await setTask(session, ids.consumer, (task) => ({ ...task, status: "running" })); const replacement = await session.commit(async (tx) => { @@ -511,12 +1017,12 @@ describe("Pico Session over MemoryStorage", () => { status: "terminal", outcome: { status: "completed", result: null }, })); - await expect(session.readList(output, undefined, ctx)).rejects.toBeInstanceOf(SharedOutputRetired); - expect(await storage.readList(output, undefined, ctx)).toEqual([]); + await expect(session.readList(output, undefined, ctx)).rejects.toBeInstanceOf(OutputRetired); + await expect(storage.readList(output, undefined, ctx)).rejects.toBeInstanceOf(OutputRetired); }); it("filters and pages tasks", async () => { - const session = new Session(new MemoryStorage()); + const session = new Session(MemoryStorage.create()); const ids = await session.commit((tx) => { const conversation = tx.createConversation({}); const otherConversation = tx.createConversation({}); diff --git a/packages/agent/test/harness/pico/types.compile.ts b/packages/agent/test/harness/pico/types.compile.ts index e2131739a..f6f154f62 100644 --- a/packages/agent/test/harness/pico/types.compile.ts +++ b/packages/agent/test/harness/pico/types.compile.ts @@ -1,44 +1,107 @@ import type { Context } from "@earendil-works/chord"; import type { - AbortClosure, - AbortedOf, - AbortTaskRuntime, + AssistantMessage, + DeferredHandle, + Message, + RetryPolicy, + ToolCall, + ToolResultMessage, + Usage, + UserMessage, +} from "@earendil-works/pi-ai"; +import { defineInternalList, internalConversationValue } from "../../../src/harness/pico/addresses.ts"; +import type { + Acceptance, + AssistantEntryData, BaseTaskTx, - CheckpointOf, - CoreAbortClosure, - CoreAbortTaskRuntime, - CoreTaskRuntime, - CoreTaskTx, - CoreTerminalClosure, + BoundTxState, + ConversationTx, Entry, + EntryBase, + EntryData, EntryInput, ExactJsonInput, - FailureOf, - InputOf, + GenerationOutput, + HookHandler, + HookPoint, + HookResult, + HostTx, + InputOutcome, + JsonObject, JsonValue, List, - OutputOf, + OutputState, PayloadsOf, + PublicScope, + QueuedInput, ReadonlyTaskOutput, - ResultOf, + ResolvedValue, + RunningTask, + ScratchReader, + SectionRecord, + Stored, + StoredEntryDraft, + StoredInputResult, SummaryEntry, - TaskKindBase, + SystemEntryData, + TaskCheckpoint, + TaskKind, TaskOf, - TaskOutcome, TaskOutput, TaskOutputRef, TaskRuntime, TerminalClosure, - Value, + ToolControl, + ToolDiagnostic, + ToolInput, + ToolOutputState, + ToolResultData, + TxState, + UsageEntryData, + UserInput, } from "../../../src/harness/pico/index.ts"; +import * as Pico from "../../../src/harness/pico/index.ts"; import { - defineCoreTask, + conversationList, + conversationValue, defineEntry, + defineHookPoint, defineList, defineTask, defineTaskOutput, defineValue, + sessionValue, } from "../../../src/harness/pico/index.ts"; +import type { CoreAbortTaskRuntime, CoreTaskRuntime, CoreTaskTx } from "../../../src/harness/pico/runtime.ts"; +import type { DisjointBundles } from "../../../src/harness/pico/tasks.ts"; +import * as TaskDeclarations from "../../../src/harness/pico/tasks.ts"; +import { defineCoreTask } from "../../../src/harness/pico/tasks.ts"; + +// @ts-expect-error privileged core authoring is not part of the public Pico surface +const publicDefineCoreTask = Pico.defineCoreTask; +// @ts-expect-error the task-kind brand is private even within the tasks module +const publicTaskKindBrand = TaskDeclarations.taskKindBrand; +void publicDefineCoreTask; +void publicTaskKindBrand; + +type RemovedPublicExports = [ + // @ts-expect-error low-level Session is internal + Pico.Session, + // @ts-expect-error low-level transaction is internal + Pico.Transaction, + // @ts-expect-error low-level task creation shape is internal + Pico.TaskCreate, + // @ts-expect-error stored conversation metadata is internal + Pico.StoredConversation, + // @ts-expect-error core kind construction types are internal + Pico.CoreTaskKind, + // @ts-expect-error erased core kind is internal + Pico.CoreKind, + // @ts-expect-error collision implementation helpers are internal + Pico.DisjointBundles, +]; +declare const removedPublicExports: RemovedPublicExports; +void removedPublicExports; type Equal = (() => Type extends Left ? 1 : 2) extends () => Type extends Right ? 1 : 2 ? (() => Type extends Right ? 1 : 2) extends () => Type extends Left ? 1 : 2 @@ -46,39 +109,62 @@ type Equal = (() => Type extends Left ? 1 : 2) extends : false : false; type Assert = Condition; -type HasEntry = "entry" extends keyof Surface ? true : false; -type HasOutput = "output" extends keyof Surface ? true : false; +type ExtendsJson = T; +type ExtendsJsonObject = T; -type TestInput = { - readonly prompt: string; - readonly nested: { readonly count: number }; -}; -type TestCheckpoint = { - readonly phase: "ready"; - readonly cursor: number; -}; -type SameCheckpoint = { - readonly phase: "ready"; - readonly cursor: number; -}; -type IncompatibleCheckpoint = { - readonly phase: "ready"; - readonly token: string; -}; -type TestResult = { readonly answer: string }; -type TestFailure = { readonly code: "failed" }; -type TestAborted = { readonly cleaned: boolean }; -type TestOutput = { text: string; chunks: string[] }; - -type JsonAssertions = [ - Assert, - Assert, +type StoredAssertions = [ + ExtendsJsonObject>, + ExtendsJsonObject>, + ExtendsJsonObject>, + ExtendsJsonObject>, + ExtendsJsonObject>, + ExtendsJsonObject>, + ExtendsJsonObject>, + ExtendsJson>, + ExtendsJson, + ExtendsJson, + ExtendsJson, + ExtendsJson, + ExtendsJson, + ExtendsJson, + ExtendsJson, + ExtendsJson, + ExtendsJson, + ExtendsJson, + ExtendsJson, + Assert, + Assert, Assert, Assert<{ readonly value: undefined } extends JsonValue ? false : true>, ]; -const ordinaryKind = defineTask()({ - kind: "test.ordinary", +type Input = { readonly prompt: string; readonly nested: { readonly count: number } }; +type Checkpoint = { readonly phase: "ready"; readonly cursor: number }; +type Result = { readonly answer: string }; +type Failure = { readonly code: "failed" }; +type Aborted = { readonly cleaned: boolean }; +type Progress = { text: string; chunks: string[] }; + +const pluginEnabled = conversationValue("plugin.enabled", { rewind: false, default: false }); +const collected = defineHookPoint<{ readonly value: string }, { readonly seen: true }>({ + fold: "collect", + onThrow: "skip", +}); +const first = defineHookPoint<{ readonly value: string }, { readonly accepted: true }>({ + fold: "first", + onThrow: "skip", +}); +const beforeRun = defineHookPoint<{ readonly prompt: string }, { readonly prompt?: string }>({ + fold: "chain", + onThrow: "skip", +}); +const noReturnHandler: HookHandler = async () => {}; +void noReturnHandler; + +const ordinaryKind = defineTask()({ + kind: "plugin.ordinary", + config: { enabled: pluginEnabled }, + hooks: { beforeRun }, execute() { return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } })); }, @@ -90,7 +176,7 @@ const ordinaryKind = defineTask()({ +const coreKind = defineCoreTask()({ kind: "pi.test_core", execute() { return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } })); @@ -103,8 +189,10 @@ const coreKind = defineCoreTask()({ - kind: "test.same_checkpoint", +const progressOutput = defineTaskOutput("plugin.progress"); +const outputKind = defineTask()({ + kind: "plugin.output", + output: { kind: progressOutput, initial: (input) => ({ text: input.prompt, chunks: [] }) }, execute() { return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } })); }, @@ -116,288 +204,342 @@ const sameCheckpointKind = defineTask("test.output"); -const outputTaskKind = defineTask()({ - kind: "test.with_output", - output: { - kind: testOutputKind, - initial(input) { - return { text: input.prompt, chunks: [] }; - }, - }, - execute() { - return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } })); - }, - recover() { - return Promise.resolve(() => ({ status: "failed", failure: { code: "failed" } })); - }, - abort() { - return Promise.resolve(() => ({ cleaned: true })); - }, -}); - -const incompatibleCheckpointKind = defineTask< - TestInput, - IncompatibleCheckpoint, - TestResult, - TestFailure, - TestAborted ->()({ - kind: "test.incompatible_checkpoint", - execute() { - return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } })); - }, - recover() { - return Promise.resolve(() => ({ status: "failed", failure: { code: "failed" } })); - }, - abort() { - return Promise.resolve(() => ({ cleaned: true })); - }, -}); - -type OrdinaryRuntimeTx = Parameters["commit"]>[0]>[0]; -type CoreRuntimeTx = Parameters["commit"]>[0]>[0]; -type OrdinaryAbortTx = Parameters["commit"]>[0]>[0]; -type CoreRuntimeAbortTx = Parameters["commit"]>[0]>[0]; -type OrdinaryFinalTx = Parameters>[0]; -type CoreFinalTx = Parameters>[0]; -type OrdinaryAbortFinalTx = Parameters>[0]; -type CoreAbortFinalTx = Parameters>[0]; -type OutputFinalTx = Parameters>[0]; -type OutputAbortFinalTx = Parameters>[0]; - +type HookAssertions = [ + Assert>>, + Assert>>, + Assert>>, + Assert< + Equal< + HookResult, + { readonly outputs: readonly { readonly seen: true }[]; readonly threw?: unknown } + > + >, + Assert, { readonly output?: { readonly accepted: true }; readonly threw?: unknown }>>, +]; +type InferredHooks = PayloadsOf["hooks"]; type KindAssertions = [ - Assert<"turn" extends keyof typeof ordinaryKind ? false : true>, - Assert<"turn" extends keyof typeof coreKind ? false : true>, - Assert<"busy" extends keyof TaskOf ? false : true>, - Assert, TestInput>>, - Assert, TestInput>>, - Assert, TestCheckpoint>>, - Assert, TestResult>>, - Assert, TestFailure>>, - Assert, TestAborted>>, - Assert["aborted"], TestAborted>>, - Assert, never>>, - Assert, TestOutput>>, - Assert[1], TaskRuntime>>, - Assert[1], CoreTaskRuntime>>, - Assert[1]["output"], TaskOutput>>, - Assert["mutate"]>[0]>, undefined>>, - Assert[1], AbortTaskRuntime>>, - Assert[1], CoreAbortTaskRuntime>>, - Assert[1]["output"], ReadonlyTaskOutput>>, - Assert[1]>, false>>, - Assert[1]>, true>>, - Assert>, - Assert>, - Assert, false>>, - Assert, true>>, - Assert, false>>, - Assert, true>>, - Assert, false>>, - Assert, true>>, - Assert, false>>, - Assert, true>>, - Assert<"sleep" extends keyof AbortTaskRuntime ? false : true>, + Assert>, + Assert>, + Assert>, + Assert>, + Assert< + Equal< + Parameters[1], + TaskRuntime + > + >, + Assert[1], CoreTaskRuntime>>, + Assert[1]["output"], TaskOutput>>, + Assert[1]["output"], ReadonlyTaskOutput>>, + Assert< + Equal< + Extract, { status: "terminal" }>["outcome"]["status"], + "completed" | "failed" | "aborted" | "orphaned" + > + >, ]; -type KindOutcome = Extract, { readonly status: "terminal" }>["outcome"]; -type OutcomeAssertions = [ - Assert>>, - Assert["result"], TestResult>>, - Assert["failure"], TestFailure>>, - Assert["result"], TestAborted>>, -]; - -const summaryKind = defineEntry("pi.summary"); -type SummaryInput = EntryInput; -type EntryAssertions = [ - Assert>, - Assert>, - Assert<"id" extends keyof SummaryInput ? false : true>, - Assert<"conversationId" extends keyof SummaryInput ? false : true>, - Assert<"kind" extends keyof SummaryInput ? false : true>, - Assert<"byTaskId" extends keyof SummaryInput ? false : true>, -]; - -function verifyTaskInputs( - tx: BaseTaskTx, - outputRef: TaskOutputRef, - wrongOutputRef: TaskOutputRef<{ count: number }>, -): void { - tx.task(ordinaryKind, { - input: { prompt: "hello", nested: { count: 1 } }, - }); - // @ts-expect-error ordinary task transactions cannot instantiate fixed core task kinds - tx.task(coreKind, { input: { prompt: "core", nested: { count: 1 } } }); - tx.task(outputTaskKind, { - input: { prompt: "own output", nested: { count: 1 } }, - }); - tx.task(outputTaskKind, { - input: { prompt: "shared output", nested: { count: 1 } }, - output: outputRef, - }); - // @ts-expect-error a no-output kind cannot receive an output reference - tx.task(ordinaryKind, { input: { prompt: "hello", nested: { count: 1 } }, output: outputRef }); - tx.task(outputTaskKind, { - input: { prompt: "wrong output", nested: { count: 1 } }, - // @ts-expect-error a shared output reference must have the kind's output type - output: wrongOutputRef, - }); - - const exactVariable = { - conversationId: 1, - input: { prompt: "hello", nested: { count: 1 } }, - background: true, - } as const; - tx.task(ordinaryKind, exactVariable); - - // Top-level exactness intentionally does not recursively reject extra nested fields. - tx.task(ordinaryKind, { - input: { prompt: "hello", nested: { count: 1, extension: true } }, - }); - - // @ts-expect-error task input literals reject extra top-level fields - tx.task(ordinaryKind, { input: { prompt: "hello", nested: { count: 1 }, extension: true } }); - - const extraInputVariable = { prompt: "hello", nested: { count: 1 }, extension: true }; - // @ts-expect-error task input variables reject visible extra top-level fields - tx.task(ordinaryKind, { input: extraInputVariable }); - - const inputBase = { prompt: "hello", nested: { count: 1 } }; - // @ts-expect-error task input spreads reject visible extra top-level fields - tx.task(ordinaryKind, { input: { ...inputBase, extension: true } }); - - const extraSpecVariable = { - input: { prompt: "hello", nested: { count: 1 } }, - priority: 1, - }; - // @ts-expect-error task specs reject visible extra top-level fields - tx.task(ordinaryKind, extraSpecVariable); - - tx.checkpoint({ phase: "ready", cursor: 1 }); - // @ts-expect-error every checkpoint requires its phase - tx.checkpoint({ cursor: 1 }); - // @ts-expect-error checkpoint replacement requires the complete kind-specific shape - tx.checkpoint({ phase: "ready" }); - // @ts-expect-error an incompatible checkpoint shape is rejected - tx.checkpoint({ phase: "ready", token: "cursor" }); -} - -function verifyOrdinaryRegistration(register: (kind: TaskKindBase) => void): void { - register(ordinaryKind); - // @ts-expect-error fixed core kinds cannot enter the ordinary task registry - register(coreKind); -} - -function verifyCoreTaskInputs(tx: CoreTaskTx): void { - tx.task(ordinaryKind, { - input: { prompt: "ordinary child", nested: { count: 1 } }, - }); - tx.task(coreKind, { - input: { prompt: "core child", nested: { count: 1 } }, - }); -} - -function verifyOutputMutation( - output: TaskOutput, - readonlyOutput: ReadonlyTaskOutput, +function verifyNoOutputRuntime( + runtime: TaskRuntime, + abortRuntime: Parameters[1], + finalTx: Parameters>[0], + running: RunningTask, ctx: Context, ): void { + // @ts-expect-error no-output runtimes have no output property + void runtime.output; + // @ts-expect-error no-output abort runtimes have no output property + void abortRuntime.output; + // @ts-expect-error no-output final transactions have no output property + void finalTx.output; + // @ts-expect-error no-output running tasks have no output property + void running.output; + void abortRuntime.sleep(1, ctx); +} + +function verifyOutput(output: TaskOutput, readonlyOutput: ReadonlyTaskOutput, ctx: Context): void { void output.mutate((state) => { - state.chunks.push("chunk"); + state.chunks.push("next"); }, ctx); - // @ts-expect-error task-output mutation callbacks must be synchronous + // @ts-expect-error output mutations must be synchronous void output.mutate(async (state) => { - state.chunks.push("chunk"); + state.text = "invalid"; }, ctx); - // @ts-expect-error fresh abort output is read-only + // @ts-expect-error fresh abort receives read-only output void readonlyOutput.mutate; } -function verifyCheckpointCompatibility( - checkpoint: CheckpointOf, - incompatible: CheckpointOf, -): void { - // Checkpoints are structural: an independently declared identical shape is assignable. - const structurallyCompatible: CheckpointOf = checkpoint; - void structurallyCompatible; - // @ts-expect-error a differently shaped kind checkpoint is not assignable - const structurallyIncompatible: CheckpointOf = incompatible; - void structurallyIncompatible; +type JobOutput = { + stdout: string; + stderr: string; + droppedStdout: number; + droppedStderr: number; + exitCode?: number; + occurrence: number; +}; +type JobAssertions = [Assert]; +function verifyJobMutation(output: TaskOutput, snapshot: Readonly, ctx: Context): void { + void output.mutate((state) => { + state.stdout = snapshot.stdout; + }, ctx); } -function verifyOutcomes(outcome: KindOutcome): void { - const completed: KindOutcome = { status: "completed", result: { answer: "done" } }; - const failed: KindOutcome = { status: "failed", failure: { code: "failed" } }; - const aborted: KindOutcome = { status: "aborted", result: { cleaned: true } }; - const orphaned: KindOutcome = { status: "orphaned" }; - // @ts-expect-error completed outcomes require this kind's result payload - const wrongCompleted: KindOutcome = { status: "completed", result: { answer: 1 } }; - // @ts-expect-error failed outcomes require this kind's failure payload - const wrongFailure: KindOutcome = { status: "failed", failure: { code: "other" } }; - // @ts-expect-error aborted outcomes require this kind's abort payload - const wrongAborted: KindOutcome = { status: "aborted", result: { cleaned: "yes" } }; - void outcome; - void completed; - void failed; - void aborted; - void orphaned; - void wrongCompleted; - void wrongFailure; - void wrongAborted; +type NoteEntry = EntryBase & EntryData<{ readonly text: string }>; +type PinnedEntry = EntryBase & + EntryData<{ readonly text: string }> & { + readonly model: readonly [Stored]; + }; +const noteKind = defineEntry("plugin.note"); +const pinnedKind = defineEntry("plugin.pinned"); +const summaryKind = defineEntry("pi.summary"); + +function verifyEntryInputs(tx: ConversationTx, text: string): void { + void tx.write(noteKind, { data: { text } }); + void tx.write(pinnedKind, { + data: { text }, + model: [{ role: "user", content: text, timestamp: 1 }], + }); + void tx.write(summaryKind, { + data: { through: 1 }, + model: [{ role: "user", content: text, timestamp: 1 }], + head: "self", + }); + // @ts-expect-error kind is supplied by the witness + void tx.write(noteKind, { kind: "plugin.note", data: { text } }); + // @ts-expect-error id is supplied by Pico + void tx.write(noteKind, { id: 1, data: { text } }); +} + +type EntryAssertions = [ + Assert, + Assert, + Assert, + Assert["head"], number | "self">>, +]; + +function verifyHostTaskCreation(tx: HostTx): void { + tx.task(ordinaryKind, { conversationId: 1, input: { prompt: "host", nested: { count: 1 } } }); + // @ts-expect-error host task creation requires a conversation ID + tx.task(ordinaryKind, { input: { prompt: "host", nested: { count: 1 } } }); +} + +function verifyTaskCreation( + tx: BaseTaskTx, + coreTx: CoreTaskTx, + outputRef: TaskOutputRef, +): void { + tx.task(ordinaryKind, { input: { prompt: "literal", nested: { count: 1 } } }); + const exact = { input: { prompt: "variable", nested: { count: 1 } } }; + tx.task(ordinaryKind, exact); + tx.task(ordinaryKind, { ...exact }); + // @ts-expect-error ordinary transactions cannot create core tasks + tx.task(coreKind, { input: { prompt: "core", nested: { count: 1 } } }); + coreTx.task(coreKind, { input: { prompt: "core", nested: { count: 1 } } }); + + // @ts-expect-error input literals reject visible top-level extras + tx.task(ordinaryKind, { input: { prompt: "extra", nested: { count: 1 }, extra: 1 } }); + const extraInput = { prompt: "extra", nested: { count: 1 }, extra: 1 }; + // @ts-expect-error input variables reject visible top-level extras + tx.task(ordinaryKind, { input: extraInput }); + const inputBase = { prompt: "extra", nested: { count: 1 } }; + // @ts-expect-error input spreads reject visible top-level extras + tx.task(ordinaryKind, { input: { ...inputBase, extra: 1 } }); + const extraSpec = { input: { prompt: "extra", nested: { count: 1 } }, extra: 1 }; + // @ts-expect-error specs reject visible extras + tx.task(ordinaryKind, extraSpec); + // @ts-expect-error spread specs reject visible extras + tx.task(ordinaryKind, { ...extraSpec }); + + // Deep exactness is outside the contract. + tx.task(ordinaryKind, { input: { prompt: "nested", nested: { count: 1, extension: true } } }); + tx.task(outputKind, { input: { prompt: "output", nested: { count: 1 } }, output: outputRef }); + // @ts-expect-error no-output kinds reject output refs + tx.task(ordinaryKind, { input: { prompt: "output", nested: { count: 1 } }, output: outputRef }); } function verifyAddresses(): void { - const sessionValue = defineValue({ type: "session" }, "plugin.name"); - const taskList = defineList({ type: "task", taskId: 2 }, "plugin.items", { key: "active" }); - const sharedList = defineList({ type: "shared", id: 4 }, "plugin.shared"); - const conversationValue = defineValue({ type: "conversation", conversationId: 3 }, "plugin.result", { + const rewindable = conversationValue("plugin.plan", { rewind: true }); + const defaulted = conversationValue("plugin.plan", { key: "defaulted", rewind: true, default: false }); + const sticky = conversationValue("plugin.mode", { rewind: false }); + const events = conversationList("plugin.events", { rewind: true }); + const plainSession = sessionValue("plugin.plain-name"); + const defaultedSession = sessionValue("plugin.name", { default: "pico" }); + const directPlainSession = defineValue({ type: "session" }, "plugin.direct-session"); + const directDefaultedSession = defineValue({ type: "session" }, "plugin.direct-session-default", { + default: "session", + }); + const directDefaultedTask = defineValue({ type: "task", taskId: 1 }, "plugin.direct-task-default", { + default: "task", + }); + const directPlainConversation = defineValue( + { type: "conversation", conversationId: 1 }, + "plugin.direct-conversation", + { rewind: true }, + ); + const directDefaultedConversation = defineValue( + { type: "conversation", conversationId: 1 }, + "plugin.direct-conversation-default", + { rewind: false, default: false }, + ); + const bound = rewindable.bind(1); + const defaultedBound = defaulted.bind(1); + const internalDefaulted = internalConversationValue("pi.internal-default", { rewind: true, + default: "internal", }); - const conversationList = defineList({ type: "conversation", conversationId: 3 }, "plugin.events", { - rewind: false, - }); - const typedSessionValue: Value = sessionValue; - const typedTaskList: List = taskList; - void typedSessionValue; - void typedTaskList; - void sharedList; - void conversationValue; - void conversationList; + const internalDefaultedBound = internalDefaulted.bind(1); + const boundList: List = events.bind(1); + void sticky; + void bound; + void defaultedBound; + void internalDefaulted; + void internalDefaultedBound; + void boundList; + void plainSession; + void defaultedSession; + void directPlainSession; + void directDefaultedSession; + void directDefaultedTask; + void directPlainConversation; + void directDefaultedConversation; - // @ts-expect-error conversation addresses explicitly select rewind behavior - defineValue({ type: "conversation", conversationId: 3 }, "plugin.value"); + type AddressAssertions = [ + Assert>, + Assert>, + Assert, string | undefined>>, + Assert, string>>, + Assert, string | undefined>>, + Assert, string>>, + Assert, string>>, + Assert, boolean | undefined>>, + Assert, boolean>>, + Assert>, + Assert>, + Assert>, + Assert, boolean | undefined>>, + Assert, boolean>>, + Assert, boolean | undefined>>, + Assert, boolean>>, + Assert>, + Assert, string>>, + Assert, string>>, + Assert>, + Assert extends never ? true : false>, + ]; + const assertions = null as unknown as AddressAssertions; + void assertions; + + // @ts-expect-error public bound constructors do not accept shared scope + defineValue({ type: "shared", id: 1 }, "plugin.shared"); + // @ts-expect-error public bound constructors do not accept shared scope + defineList({ type: "shared", id: 1 }, "plugin.shared"); + // @ts-expect-error conversation addresses require an explicit rewind policy + defineValue({ type: "conversation", conversationId: 1 }, "plugin.value"); // @ts-expect-error session addresses are sticky defineValue({ type: "session" }, "plugin.value", { rewind: true }); - // @ts-expect-error task addresses are never rewindable - defineList({ type: "task", taskId: 2 }, "plugin.list", { rewind: true }); - // @ts-expect-error shared addresses are never rewindable - defineList({ type: "shared", id: 4 }, "plugin.shared", { rewind: true }); - // @ts-expect-error address payloads are strict JSON - defineValue({ type: "session" }, "plugin.date"); + defineInternalList({ type: "shared", id: 1 }, "pi.output"); } -type ExactInputAssertion = Assert< - Equal["extension"], never> ->; +function verifyResolvedTransactionReads(tx: TxState, boundTx: BoundTxState, scratch: ScratchReader): void { + const plainSession = defineValue({ type: "session" }, "plugin.tx-plain"); + const defaultedSession = defineValue({ type: "session" }, "plugin.tx-default", { default: "x" }); + const defaultedTask = defineValue({ type: "task", taskId: 1 }, "plugin.scratch-default", { + default: 1, + }); + const defaultedConversation = defineValue( + { type: "conversation", conversationId: 1 }, + "plugin.tx-conversation-default", + { rewind: true, default: false }, + ); + const plainDefinition = conversationValue("plugin.bound-plain", { rewind: true }); + const defaultedDefinition = conversationValue("plugin.bound-default", { rewind: false, default: 1 }); + const plainRead = tx.value(plainSession).get(); + const defaultedRead = tx.value(defaultedSession).get(); + const defaultedConversationRead = tx.value(defaultedConversation).get(); + const plainDefinitionRead = boundTx.value(plainDefinition).get(); + const defaultedDefinitionRead = boundTx.value(defaultedDefinition).get(); + const scratchRead = scratch.value(defaultedTask).get(); + type ReadAssertions = [ + Assert, string | undefined>>, + Assert, string>>, + Assert, boolean>>, + Assert, number | undefined>>, + Assert, number>>, + Assert, number>>, + ]; + const assertions = null as unknown as ReadAssertions; + void assertions; +} -declare const jsonAssertions: JsonAssertions; +type FakeDuplicateConfigs = { one: { value: 1 }; two: { value: 2 } }; +type FakeReservedConfig = { one: { set: 1 }; two: Record }; +type FakeDuplicateHooks = { one: { before: 1 }; two: { before: 2 } }; +type RealConfigs = { one: { model: 1 }; two: { retry: 2 } }; +type RealHooks = { one: { before: 1 }; two: { after: 2 } }; +type BundleAssertions = [Assert>, Assert>]; +type BadBundleAssertions = [ + // @ts-expect-error duplicate bundle keys are rejected by the assertion + Assert>, + // @ts-expect-error reserved config keys are rejected by the assertion + Assert>, + // @ts-expect-error duplicate hook keys are rejected by the assertion + Assert>, +]; + +type ExactInputAssertion = Assert< + Equal["extension"], never> +>; +type CheckpointAssertion = Assert< + { readonly phase: "deferred"; readonly handle: Stored } extends TaskCheckpoint ? true : false +>; +type HydrationAssertions = [ + Assert["input"] extends Entry ? true : false>, + Assert< + Equal< + Extract["answer"] extends infer A | undefined + ? NonNullable extends { message: infer M } + ? M + : never + : never, + AssistantMessage + > + >, + ExtendsJson, +]; +type KindShape = TaskKind; +type RunningAssertion = Assert<"output" extends keyof RunningTask ? false : true>; + +declare const storedAssertions: StoredAssertions; +declare const hookAssertions: HookAssertions; declare const kindAssertions: KindAssertions; -declare const outcomeAssertions: OutcomeAssertions; +declare const jobAssertions: JobAssertions; declare const entryAssertions: EntryAssertions; +declare const bundleAssertions: BundleAssertions; +declare const badBundleAssertions: BadBundleAssertions; declare const exactInputAssertion: ExactInputAssertion; -declare const entry: Entry; -void jsonAssertions; +declare const checkpointAssertion: CheckpointAssertion; +declare const hydrationAssertions: HydrationAssertions; +declare const kindShape: KindShape; +declare const runningAssertion: RunningAssertion; +void storedAssertions; +void hookAssertions; void kindAssertions; -void outcomeAssertions; +void jobAssertions; void entryAssertions; +void bundleAssertions; +void badBundleAssertions; void exactInputAssertion; -void entry; -void verifyTaskInputs; -void verifyOrdinaryRegistration; -void verifyCoreTaskInputs; -void verifyOutputMutation; -void verifyCheckpointCompatibility; -void verifyOutcomes; +void checkpointAssertion; +void hydrationAssertions; +void kindShape; +void runningAssertion; +void verifyNoOutputRuntime; +void verifyOutput; +void verifyJobMutation; +void verifyEntryInputs; +void verifyHostTaskCreation; +void verifyTaskCreation; void verifyAddresses; +void verifyResolvedTransactionReads; +void (null as Message | null); +void (null as CoreAbortTaskRuntime | null);