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

feat(agent): align pico foundation with v2 contract

Browse source
This commit is contained in:
Mario Zechner 2026-09-13 04:00:27 +02:00
parent 99a3948c4c
commit 51f3090790
12 changed files with 2179 additions and 690 deletions
Download patch file Download diff file Expand all files Collapse all files

242
packages/agent/src/harness/pico/addresses.ts
View file

@ -6,6 +6,10 @@ export type Scope =
| { readonly type: "task"; readonly taskId: Id }
| { readonly type: "shared"; readonly id: Id };
export type PublicScope = Exclude<Scope, { type: "shared" }>;
export type StickyScope = Extract<PublicScope, { type: "session" | "task" }>;
export type ConversationScope = Extract<PublicScope, { type: "conversation" }>;
declare const addressType: unique symbol;
export interface Address<T extends JsonValue = JsonValue> {
@ -19,6 +23,7 @@ export interface Address<T extends JsonValue = JsonValue> {
export interface Value<T extends JsonValue> extends Address<T> {
readonly kind: "value";
readonly default?: T;
}
export interface List<T extends JsonValue> extends Address<T> {
@ -30,48 +35,239 @@ export interface Element<T extends JsonValue> {
readonly value: T;
}
type StickyOptions = { readonly key?: string; readonly rewind?: false };
type ConversationOptions = { readonly key?: string; readonly rewind: boolean };
type StickyScope = Extract<Scope, { type: "session" | "task" | "shared" }>;
type ConversationScope = Extract<Scope, { type: "conversation" }>;
type Options = StickyOptions | ConversationOptions;
export interface UnboundValue<T extends JsonValue, R extends boolean = boolean> {
readonly namespace: string;
readonly key?: string;
readonly rewind: R;
readonly default?: T;
readonly [addressType]?: T;
bind(conversationId: Id): Value<T>;
}
export function defineValue<T extends JsonValue>(
scope: StickyScope,
namespace: string,
options?: StickyOptions,
): Value<T>;
export function defineValue<T extends JsonValue>(
scope: ConversationScope,
namespace: string,
options: ConversationOptions,
): Value<T>;
export function defineValue<T extends JsonValue>(scope: Scope, namespace: string, options?: Options): Value<T> {
export interface UnboundList<T extends JsonValue, R extends boolean = boolean> {
readonly namespace: string;
readonly key?: string;
readonly rewind: R;
readonly [addressType]?: T;
bind(conversationId: Id): List<T>;
}
export type ConversationValueDefinition<T extends JsonValue, R extends boolean = boolean> = UnboundValue<T, R>;
export type ConversationListDefinition<T extends JsonValue, R extends boolean = boolean> = UnboundList<T, R>;
export type PayloadOf<D> = D extends Address<infer BoundPayload>
? BoundPayload
: D extends UnboundValue<infer DefinitionPayload, boolean>
? DefinitionPayload
: never;
export type ResolvedValue<D> = D extends { readonly default: infer V } ? V : PayloadOf<D> | undefined;
type DefaultedUnboundValue<T extends JsonValue, R extends boolean> = Omit<UnboundValue<T, R>, "bind" | "default"> & {
readonly default: T;
bind(conversationId: Id): Value<T> & { readonly default: T };
};
type StickyOptions<T extends JsonValue> = { readonly key?: string; readonly default?: T };
type DefaultedStickyOptions<T extends JsonValue> = { readonly key?: string; readonly default: T };
type ConversationOptions<T extends JsonValue> = {
readonly key?: string;
readonly rewind: boolean;
readonly default?: T;
};
type DefaultedConversationOptions<T extends JsonValue> = {
readonly key?: string;
readonly rewind: boolean;
readonly default: T;
};
type InternalOptions<T extends JsonValue> = StickyOptions<T> & { 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<T extends JsonValue>(scope: Scope, namespace: string, options?: InternalOptions<T>): Value<T> {
return Object.freeze({
...options,
scope,
namespace,
kind: "value",
kind: "value" as const,
rewind: options?.rewind ?? false,
});
}
function list<T extends JsonValue>(scope: Scope, namespace: string, options?: InternalOptions<T>): List<T> {
return Object.freeze({
...options,
scope,
namespace,
kind: "list" as const,
rewind: options?.rewind ?? false,
});
}
export function defineValue<T extends JsonValue>(
scope: StickyScope,
namespace: string,
options: DefaultedStickyOptions<T>,
): Value<T> & { readonly default: T };
export function defineValue<T extends JsonValue>(
scope: StickyScope,
namespace: string,
options?: { readonly key?: string },
): Value<T>;
export function defineValue<T extends JsonValue>(
scope: ConversationScope,
namespace: string,
options: DefaultedConversationOptions<T>,
): Value<T> & { readonly default: T };
export function defineValue<T extends JsonValue>(
scope: ConversationScope,
namespace: string,
options: { readonly key?: string; readonly rewind: boolean },
): Value<T>;
export function defineValue<T extends JsonValue>(
scope: Scope,
namespace: string,
options?: InternalOptions<T>,
): Value<T> {
assertPublicScope(scope);
assertPublicNamespace(namespace);
return value(scope, namespace, options);
}
export function defineList<T extends JsonValue>(
scope: StickyScope,
namespace: string,
options?: StickyOptions,
options?: { readonly key?: string },
): List<T>;
export function defineList<T extends JsonValue>(
scope: ConversationScope,
namespace: string,
options: ConversationOptions,
options: Omit<ConversationOptions<T>, "default">,
): List<T>;
export function defineList<T extends JsonValue>(scope: Scope, namespace: string, options?: Options): List<T> {
export function defineList<T extends JsonValue>(
scope: Scope,
namespace: string,
options?: InternalOptions<T>,
): List<T> {
assertPublicScope(scope);
assertPublicNamespace(namespace);
return list(scope, namespace, options);
}
export function conversationValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: true; readonly default: T },
): DefaultedUnboundValue<T, true>;
export function conversationValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: false; readonly default: T },
): DefaultedUnboundValue<T, false>;
export function conversationValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: true },
): UnboundValue<T, true>;
export function conversationValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: false },
): UnboundValue<T, false>;
export function conversationValue<T extends JsonValue>(
namespace: string,
options: ConversationOptions<T>,
): UnboundValue<T> {
assertPublicNamespace(namespace);
return unboundValue(namespace, options);
}
export function conversationList<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: true },
): UnboundList<T, true>;
export function conversationList<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: false },
): UnboundList<T, false>;
export function conversationList<T extends JsonValue>(
namespace: string,
options: Omit<ConversationOptions<T>, "default">,
): UnboundList<T> {
assertPublicNamespace(namespace);
return unboundList(namespace, options);
}
export function sessionValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly default: T },
): Value<T> & { readonly default: T };
export function sessionValue<T extends JsonValue>(namespace: string, options?: { readonly key?: string }): Value<T>;
export function sessionValue<T extends JsonValue>(namespace: string, options?: StickyOptions<T>): Value<T> {
return defineValue({ type: "session" }, namespace, options);
}
/** Internal constructor for fixed `pi.*` definitions and shared task output state. */
export function defineInternalValue<T extends JsonValue>(
scope: Scope,
namespace: string,
options?: InternalOptions<T>,
): Value<T> {
return value(scope, namespace, options);
}
/** Internal constructor for fixed `pi.*` definitions and shared task output state. */
export function defineInternalList<T extends JsonValue>(
scope: Scope,
namespace: string,
options?: InternalOptions<T>,
): List<T> {
return list(scope, namespace, options);
}
/** Internal constructor for fixed conversation configuration definitions. */
export function internalConversationValue<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R; readonly default: T },
): DefaultedUnboundValue<T, R>;
export function internalConversationValue<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R },
): UnboundValue<T, R>;
export function internalConversationValue<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R; readonly default?: T },
): UnboundValue<T, R> {
return unboundValue(namespace, options);
}
function unboundValue<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R; readonly default?: T },
): UnboundValue<T, R> {
return Object.freeze({
...options,
scope,
namespace,
kind: "list",
rewind: options?.rewind ?? false,
bind: (conversationId: Id) =>
value<T>({ type: "conversation", conversationId }, namespace, {
key: options.key,
rewind: options.rewind,
default: options.default,
}),
});
}
function unboundList<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R },
): UnboundList<T, R> {
return Object.freeze({
...options,
namespace,
bind: (conversationId: Id) =>
list<T>({ type: "conversation", conversationId }, namespace, {
key: options.key,
rewind: options.rewind,
}),
});
}

3
packages/agent/src/harness/pico/core.ts
View file

@ -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<T> = JsonRepresentation<T>;
export interface JsonObject {
readonly [key: string]: JsonValue;

74
packages/agent/src/harness/pico/entries.ts
View file

@ -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<Message>[] };
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<Message>[];
readonly head?: Id;
readonly edits?: readonly ContextEdit[];
}
export type EntryIdentity = Pick<Entry, "id" | "conversationId" | "kind" | "byTaskId">;
export type EntryBase = EntryIdentity;
export interface EntryData<D extends JsonValue = JsonValue> {
@ -19,7 +24,7 @@ export interface EntryData<D extends JsonValue = JsonValue> {
}
export interface ModelProjection<M extends Message = Message> {
readonly model: readonly M[];
readonly model: readonly Stored<M>[];
}
export interface ContextHead {
@ -30,10 +35,8 @@ export interface ContextEdits {
readonly edits: readonly ContextEdit[];
}
export type Entry = EntryBase & Partial<EntryData & ModelProjection & ContextHead & ContextEdits>;
export type EntryInput<E extends Entry> = Omit<E, keyof EntryIdentity | "head"> &
(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<E extends Entry = Entry> {
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<JsonValue> & { readonly model: readonly [ToolResultMessage<JsonValue>] };
export type NoticeEntry = EntryBase & Partial<EntryData> & { 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<Usage>; 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<UserMessage>];
readonly data?: { readonly continuation: true; readonly from: Id };
};
export type AssistantEntry = EntryBase & {
readonly model: readonly [Stored<AssistantMessage>];
readonly data?: AssistantEntryData;
};
export type ToolResultEntry = EntryBase & {
readonly model: readonly [Stored<ToolResultMessage>];
readonly data: ToolResultData;
};
export type NoticeEntry = EntryBase & { readonly model: readonly [Stored<UserMessage>]; readonly data?: JsonValue };
export type UsageEntry = EntryBase & { readonly data: UsageEntryData };
export type SummaryEntry = EntryBase & {
readonly model: readonly [Stored<UserMessage>];
readonly data: { readonly through: Id };
readonly head: Id;
};
export type HandoffEntry = EntryBase & { readonly model: readonly [Stored<UserMessage>]; readonly head: Id };
export type ResetEntry = EntryBase & { readonly head: Id };

62
packages/agent/src/harness/pico/hooks.ts Normal file
View file

@ -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<In, Out, F extends Fold = Fold> {
readonly fold: F;
readonly onThrow: "skip" | "abort";
readonly [hookIn]?: In;
readonly [hookOut]?: Out;
}
export function defineHookPoint<In, Out = void>(options: {
readonly fold: "collect";
readonly onThrow: "skip" | "abort";
}): HookPoint<In, Out, "collect">;
export function defineHookPoint<In, Out = void>(options: {
readonly fold: "first";
readonly onThrow: "skip" | "abort";
}): HookPoint<In, Out, "first">;
export function defineHookPoint<In extends object, Out extends Partial<In>>(options: {
readonly fold: "chain";
readonly onThrow: "skip" | "abort";
}): HookPoint<In, Out, "chain">;
export function defineHookPoint(options: {
readonly fold: Fold;
readonly onThrow: "skip" | "abort";
}): HookPoint<unknown, unknown> {
return Object.freeze(options);
}
export type HookPoints = Record<string, HookPoint<unknown, unknown>>;
export type EmptyHookPoints = Record<never, never>;
export type HookIn<P> = P extends HookPoint<infer In, unknown> ? In : never;
export type HookOut<P> = P extends HookPoint<unknown, infer Out> ? Out : never;
// biome-ignore lint/suspicious/noConfusingVoidType: void allows handlers with no return value.
type HookHandlerResult<P> = HookOut<P> | void;
export type HookHandler<P> = (
input: HookIn<P>,
info: HookInfo,
ctx: Context,
) => HookHandlerResult<P> | Promise<HookHandlerResult<P>>;
export interface HookInfo {
readonly kind: string;
readonly taskId: Id;
readonly conversationId: Id;
}
export interface HookRunner<H extends HookPoints> {
run<P extends H[keyof H]>(point: P, input: HookIn<P>, ctx: Context): Promise<HookResult<P>>;
}
export type HookResult<P> = P extends { fold: "collect" }
? { readonly outputs: readonly HookOut<P>[]; readonly threw?: unknown }
: P extends { fold: "first" }
? { readonly output?: HookOut<P>; readonly threw?: unknown }
: { readonly output: HookIn<P>; readonly threw?: unknown };

162
packages/agent/src/harness/pico/index.ts
View file

@ -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";

221
packages/agent/src/harness/pico/memory-storage.ts
View file

@ -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<Id, Element<JsonValue>> }
| {
readonly kind: "list";
readonly rewind: false;
readonly elements: Map<Id, Element<JsonValue>>;
readonly known: Set<Id>;
}
| { 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<Id, StoredConversation>;
readonly entries: Map<Id, StoredEntry>;
readonly listAppends: { readonly seq: Seq; readonly address: List<JsonValue>; readonly elementId: Id }[];
}
function rewindable(
address: Address,
): address is Address & { readonly scope: Extract<Address["scope"], { type: "conversation" }>; readonly rewind: true } {
@ -75,7 +89,8 @@ function page<T>(
): Page<T> {
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<T>(
export class MemoryStorage implements Storage {
private nextObjectId: Id = 1;
private committedIdHighWater: Id = 0;
private nextSeq: Seq = 1;
private readonly conversations = new Map<Id, Conversation>();
private readonly conversations = new Map<Id, StoredConversation>();
private readonly entries = new Map<Id, StoredEntry>();
private readonly entriesByConversation = new Map<Id, StoredEntry[]>();
private readonly tasks = new Map<Id, Task>();
@ -106,19 +122,24 @@ export class MemoryStorage implements Storage {
private readonly taskState = new Map<Id, ScopeState>();
private readonly sharedState = new Map<Id, ScopeState>();
static create(): MemoryStorage {
return new MemoryStorage();
}
nextId(): Id {
return this.nextObjectId++;
}
async commit(writes: readonly Write[], _ctx: Context): Promise<readonly Seq[]> {
this.validateListRemovals(writes);
const seqs = writes.map((_, index) => this.nextSeq + index);
const affectedOutputs = new Set<Id>();
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<void> {
this.nextObjectId = this.committedIdHighWater + 1;
}
async getConversations(ids: readonly Id[], _ctx: Context): Promise<ReadonlyMap<Id, Conversation>> {
const result = new Map<Id, Conversation>();
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<Page<Conversation>> {
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<Page<Entry>> {
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<Entry | undefined> {
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<T extends JsonValue>(address: Value<T>, at: Id | undefined, _ctx: Context): Promise<T | undefined> {
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<readonly Element<T>[]> {
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<Element<T>>)];
return [...(state.elements.values() as Iterable<Element<T>>)].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<Entry> {
private *visibleEntries(conversationId: Id, through: Id | undefined): Iterable<StoredEntry> {
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<JsonValue>, 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<JsonValue>): Extract<StoredState, { kind: "list" }> {
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<StoredState, { kind: "list" }>;
}

245
packages/agent/src/harness/pico/runtime.ts
View file

@ -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<E extends Entry>(kind: EntryKind<E>, id: Id): Promise<E | undefined>;
getEntries(ids: readonly Id[]): Promise<ReadonlyMap<Id, Entry>>;
getTask(id: Id): Promise<Task | undefined>;
getTask<K extends AnyDefinedTaskKind>(kind: K, id: Id): Promise<TaskOf<K> | undefined>;
getTask<K extends AnyDefinedKind>(kind: K, id: Id): Promise<TaskOf<K> | undefined>;
getTasks(ids: readonly Id[]): Promise<ReadonlyMap<Id, Task>>;
}
export interface TxValue<T extends JsonValue> {
get(at?: Id): Promise<T | undefined>;
export interface TxValue<T extends JsonValue, Read = T | undefined> {
get(at?: Id): Promise<Read>;
set(value: T): void;
delete(): void;
}
export interface TxList<T extends JsonValue> {
read(): Promise<readonly Element<T>[]>;
read(at: Id): Promise<readonly Element<T>[]>;
read(at?: Id): Promise<readonly Element<T>[]>;
append(value: T): Id;
remove(id: Id): void;
clear(): void;
}
export interface Acceptance {
export interface TxState {
value<D extends Value<JsonValue>>(address: D): TxValue<PayloadOf<D>, ResolvedValue<D>>;
list<T extends JsonValue>(address: List<T>): TxList<T>;
}
export interface BoundTxState {
value<D extends Value<JsonValue> | UnboundValue<JsonValue>>(address: D): TxValue<PayloadOf<D>, ResolvedValue<D>>;
list<T extends JsonValue>(address: List<T> | UnboundList<T>): TxList<T>;
}
export type Acceptance = {
readonly kind: "send" | "write";
readonly requestId?: string;
readonly conversationId: Id;
readonly inputId: Id;
}
};
interface TaskSpecBase<I extends JsonValue> {
export type TaskSpec<K extends AnyDefinedKind> = {
readonly conversationId?: Id;
readonly input: I;
readonly input: InputOf<K>;
readonly after?: readonly Id[];
readonly background?: true;
}
export type TaskSpec<I extends JsonValue, O extends object = never> = TaskSpecBase<I> &
([O] extends [never] ? { readonly output?: never } : { readonly output?: TaskOutputRef<O> });
} & ([OutputOf<K>] extends [never] ? { readonly output?: never } : { readonly output?: TaskOutputRef<OutputOf<K>> });
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<Message>[];
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<UserInput>; 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<InputResult, { status: "done" | "unanswered" }>;
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<InputOutcome, { status: "done" | "unanswered" }>;
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<InputResult | undefined>;
wait(ctx: Context): Promise<TerminalInputResult>;
result(ctx: Context): Promise<InputOutcome | undefined>;
wait(ctx: Context): Promise<TerminalInputOutcome>;
abort(ctx: Context): Promise<"aborted" | "already_placed" | "not_found">;
}
interface TaskCreator<Kinds extends AnyDefinedTaskKind> {
task<K extends Kinds, S extends TaskSpec<InputOf<K>, OutputOf<K>>>(
interface TaskCreator<Kinds extends AnyDefinedKind, Extra = Record<never, never>> {
task<K extends Kinds, S extends TaskSpec<K> & Extra>(
kind: K,
spec: NoExtra<TaskSpec<InputOf<K>, OutputOf<K>>, S> & {
readonly input: ExactJsonInput<InputOf<K>, S["input"]>;
},
spec: NoExtra<TaskSpec<K> & Extra, S> & { readonly input: ExactJsonInput<InputOf<K>, S["input"]> },
): Id;
}
interface TaskTxBase<C extends TaskCheckpoint> extends TxReaders {
checkpoint(value: C): void;
export interface HostTx extends TxReaders, TxState, TaskCreator<OrdinaryKind, { readonly conversationId: Id }> {
createConversation(spec: ConversationCreateSpec): Id;
value<T extends JsonValue>(address: Value<T>): TxValue<T>;
list<T extends JsonValue>(address: List<T>): TxList<T>;
write<E extends Entry>(conversationId: Id, kind: EntryKind<E>, input: EntryInput<E>): Promise<Id>;
}
export type BaseTaskTx<C extends TaskCheckpoint> = TaskTxBase<C> & TaskCreator<AnyTaskKind>;
export interface ConversationTx extends TxReaders, BoundTxState, TaskCreator<OrdinaryKind> {
createConversation(spec: ConversationCreateSpec): Id;
write<E extends Entry>(kind: EntryKind<E>, input: EntryInput<E>): Promise<Id>;
}
interface TaskTxBase<C extends TaskCheckpoint> extends TxReaders, BoundTxState {
checkpoint(value: C): void;
createConversation(spec: ConversationCreateSpec): Id;
}
export interface BaseTaskTx<C extends TaskCheckpoint> extends TaskTxBase<C>, TaskCreator<OrdinaryKind> {
write<E extends Entry>(kind: EntryKind<E>, input: EntryInput<E>): Promise<Id>;
}
export interface InternalAdmissionTx {
accept(conversationId: Id, options: InternalAcceptOptions): Promise<Acceptance>;
@ -139,15 +173,21 @@ export interface InternalAdmissionTx {
): Promise<Acceptance>;
}
export type CoreTaskTx<C extends TaskCheckpoint> = TaskTxBase<C> &
TaskCreator<AnyDefinedTaskKind> &
export type CoreTaskTx<C extends TaskCheckpoint> = Omit<BaseTaskTx<C>, "task" | "write"> &
TaskCreator<AnyDefinedKind> &
InternalAdmissionTx & {
entry<E extends Entry>(kind: EntryKind<E>, conversationId: Id, input: EntryInput<E>): Id;
};
export interface PublicValue<T extends JsonValue> {
get(ctx: Context): Promise<T | undefined>;
get(at: Id, ctx: Context): Promise<T | undefined>;
export interface PublicValue<T extends JsonValue, Read = T | undefined> {
get(ctx: Context): Promise<Read>;
get(at: Id, ctx: Context): Promise<Read>;
set(value: T, ctx: Context): Promise<void>;
delete(ctx: Context): Promise<void>;
}
export interface StickyPublicValue<T extends JsonValue, Read = T | undefined> {
get(ctx: Context): Promise<Read>;
set(value: T, ctx: Context): Promise<void>;
delete(ctx: Context): Promise<void>;
}
@ -160,50 +200,49 @@ export interface PublicList<T extends JsonValue> {
clear(ctx: Context): Promise<void>;
}
export interface TaskOutput<O extends object> {
export interface BoundPublicState {
value<D extends Value<JsonValue> | UnboundValue<JsonValue>>(address: D): PublicValue<PayloadOf<D>, ResolvedValue<D>>;
list<T extends JsonValue>(address: List<T> | UnboundList<T>): PublicList<T>;
}
export interface TaskOutput<O extends OutputState> {
readonly ref: TaskOutputRef<O>;
read(ctx: Context): Promise<O>;
mutate(mutator: (state: O) => undefined, ctx: Context): Promise<void>;
replace(value: O, ctx: Context): Promise<void>;
}
export interface ReadonlyTaskOutput<O extends object> {
export interface ReadonlyTaskOutput<O extends OutputState> {
readonly ref: TaskOutputRef<O>;
read(ctx: Context): Promise<O>;
}
export type RuntimeOutput<O extends object> = [O] extends [never] ? unknown : { readonly output: TaskOutput<O> };
export type AbortRuntimeOutput<O extends object> = [O] extends [never]
? unknown
export type RuntimeOutput<O extends OutputState> = [O] extends [never]
? Record<never, never>
: { readonly output: TaskOutput<O> };
export type AbortRuntimeOutput<O extends OutputState> = [O] extends [never]
? Record<never, never>
: { readonly output: ReadonlyTaskOutput<O> };
export type FinalOutput<O extends object> = [O] extends [never] ? unknown : { readonly output: O };
export interface ScratchTx {
value<T extends JsonValue>(address: Value<T>): TxValue<T>;
list<T extends JsonValue>(address: List<T>): TxList<T>;
}
export type FinalOutput<O extends OutputState> = [O] extends [never] ? Record<never, never> : { readonly output: O };
export interface ScratchReader {
value<T extends JsonValue>(address: Value<T>): Pick<TxValue<T>, "get">;
value<D extends Value<JsonValue>>(address: D): Pick<TxValue<PayloadOf<D>, ResolvedValue<D>>, "get">;
list<T extends JsonValue>(address: List<T>): Pick<TxList<T>, "read">;
}
export interface TaskConversation {
export interface TaskConversation extends BoundPublicState {
readonly id: Id;
send(input: SendInput, ctx: Context): Promise<InputHandle>;
value<T extends JsonValue>(address: Value<T>): PublicValue<T>;
list<T extends JsonValue>(address: List<T>): PublicList<T>;
}
export interface AbortTaskConversation {
export interface AbortTaskConversation extends BoundPublicState {
readonly id: Id;
value<T extends JsonValue>(address: Value<T>): PublicValue<T>;
list<T extends JsonValue>(address: List<T>): PublicList<T>;
}
interface TaskRuntimeBase {
interface TaskRuntimeBase<H extends HookPoints> {
readonly taskId: Id;
scratch<T>(build: (tx: ScratchTx) => T | Promise<T>, ctx: Context): Promise<T>;
scratch<T>(build: (tx: TxState) => T | Promise<T>, ctx: Context): Promise<T>;
readonly hooks: HookRunner<H>;
conversation(id: Id, ctx: Context): Promise<TaskConversation | undefined>;
waitForTask(id: Id, ctx: Context): Promise<Task>;
abortTask(id: Id, ctx: Context): Promise<"marked" | "terminal">;
@ -211,48 +250,43 @@ interface TaskRuntimeBase {
sleep(untilMs: number, ctx: Context): Promise<void>;
}
interface CommitRuntime<I extends JsonValue, C extends TaskCheckpoint, O extends object, Tx> {
interface CommitRuntime<I extends JsonValue, C extends TaskCheckpoint, O extends OutputState, Tx> {
commit<T>(build: (tx: Tx, current: RunningTask<I, C, O>) => T | Promise<T>, ctx: Context): Promise<T>;
}
export type TaskRuntime<I extends JsonValue, C extends TaskCheckpoint, O extends object = never> = TaskRuntimeBase &
CommitRuntime<I, C, O, BaseTaskTx<C>> &
RuntimeOutput<O>;
export type CoreTaskRuntime<I extends JsonValue, C extends TaskCheckpoint, O extends object = never> = TaskRuntimeBase &
CommitRuntime<I, C, O, CoreTaskTx<C>> &
RuntimeOutput<O>;
type TerminalClosureFor<
Tx,
export type TaskRuntime<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
O extends object,
> = (tx: Tx & FinalOutput<O>, current: RunningTask<I, C, O>) => Completion<R, F> | Promise<Completion<R, F>>;
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = TaskRuntimeBase<H> & CommitRuntime<I, C, O, BaseTaskTx<C>> & RuntimeOutput<O>;
export type FinalTx<C extends TaskCheckpoint, O extends object> = BaseTaskTx<C> & FinalOutput<O>;
export type CoreFinalTx<C extends TaskCheckpoint, O extends object> = CoreTaskTx<C> & FinalOutput<O>;
export type CoreTaskRuntime<
I extends JsonValue,
C extends TaskCheckpoint,
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = TaskRuntimeBase<H> & CommitRuntime<I, C, O, CoreTaskTx<C>> & RuntimeOutput<O>;
export type FinalTx<C extends TaskCheckpoint, O extends OutputState> = BaseTaskTx<C> & FinalOutput<O>;
export type CoreFinalTx<C extends TaskCheckpoint, O extends OutputState> = CoreTaskTx<C> & FinalOutput<O>;
export type TerminalClosure<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
O extends object,
> = TerminalClosureFor<BaseTaskTx<C>, I, C, R, F, O>;
O extends OutputState,
> = (tx: FinalTx<C, O>, current: RunningTask<I, C, O>) => Completion<R, F> | Promise<Completion<R, F>>;
export type CoreTerminalClosure<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
O extends object,
> = TerminalClosureFor<CoreTaskTx<C>, I, C, R, F, O>;
O extends OutputState,
> = (tx: CoreFinalTx<C, O>, current: RunningTask<I, C, O>) => Completion<R, F> | Promise<Completion<R, F>>;
export interface AbortTx<C extends TaskCheckpoint> extends TxReaders {
export interface AbortTx<C extends TaskCheckpoint> extends TxReaders, BoundTxState {
checkpoint(value: C): void;
value<T extends JsonValue>(address: Value<T>): TxValue<T>;
list<T extends JsonValue>(address: List<T>): TxList<T>;
markTask(id: Id): Promise<"marked" | "terminal">;
}
@ -266,43 +300,44 @@ export interface CoreAbortTx<C extends TaskCheckpoint> extends AbortTx<C> {
): Promise<Acceptance>;
}
type AbortClosureFor<Tx, I extends JsonValue, C extends TaskCheckpoint, A extends JsonValue, O extends object> = (
tx: Tx & FinalOutput<O>,
export type AbortFinalTx<C extends TaskCheckpoint, O extends OutputState> = AbortTx<C> & FinalOutput<O>;
export type CoreAbortFinalTx<C extends TaskCheckpoint, O extends OutputState> = CoreAbortTx<C> & FinalOutput<O>;
export type AbortClosure<I extends JsonValue, C extends TaskCheckpoint, A extends JsonValue, O extends OutputState> = (
tx: AbortFinalTx<C, O>,
current: RunningTask<I, C, O>,
) => A | Promise<A>;
export type AbortFinalTx<C extends TaskCheckpoint, O extends object> = AbortTx<C> & FinalOutput<O>;
export type CoreAbortFinalTx<C extends TaskCheckpoint, O extends object> = CoreAbortTx<C> & FinalOutput<O>;
export type AbortClosure<
I extends JsonValue,
C extends TaskCheckpoint,
A extends JsonValue,
O extends object,
> = AbortClosureFor<AbortTx<C>, I, C, A, O>;
export type CoreAbortClosure<
I extends JsonValue,
C extends TaskCheckpoint,
A extends JsonValue,
O extends object,
> = AbortClosureFor<CoreAbortTx<C>, I, C, A, O>;
O extends OutputState,
> = (tx: CoreAbortFinalTx<C, O>, current: RunningTask<I, C, O>) => A | Promise<A>;
interface AbortRuntimeBase {
interface AbortRuntimeBase<H extends HookPoints> {
readonly taskId: Id;
scratch<T>(read: (scratch: ScratchReader) => T | Promise<T>, ctx: Context): Promise<T>;
readonly hooks: HookRunner<H>;
conversation(id: Id, ctx: Context): Promise<AbortTaskConversation | undefined>;
waitForTask(id: Id, ctx: Context): Promise<Task>;
abortTask(id: Id, ctx: Context): Promise<"marked" | "terminal">;
now(): number;
sleep(untilMs: number, ctx: Context): Promise<void>;
}
export type AbortTaskRuntime<
I extends JsonValue,
C extends TaskCheckpoint,
O extends object = never,
> = AbortRuntimeBase & CommitRuntime<I, C, O, AbortTx<C>> & AbortRuntimeOutput<O>;
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = AbortRuntimeBase<H> & CommitRuntime<I, C, O, AbortTx<C>> & AbortRuntimeOutput<O>;
export type CoreAbortTaskRuntime<
I extends JsonValue,
C extends TaskCheckpoint,
O extends object = never,
> = AbortRuntimeBase & CommitRuntime<I, C, O, CoreAbortTx<C>> & AbortRuntimeOutput<O>;
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = AbortRuntimeBase<H> & CommitRuntime<I, C, O, CoreAbortTx<C>> & AbortRuntimeOutput<O>;
export type ToolInput = { readonly assistant: Id; readonly call: Stored<ToolCall> };
export type GenerationOutput = { message?: Stored<AssistantMessage> };
export type ToolOutputState = { progress?: string; log: string[]; details?: JsonValue };

334
packages/agent/src/harness/pico/session.ts
View file

@ -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<NewTask, "id" | "output"> & {
readonly output?: { readonly id?: Id; readonly kind: string };
};
interface LineInvocation {
active: boolean;
}
const lineInvocation = new AsyncLocalStorage<LineInvocation>();
const storageOwners = new WeakMap<Storage, object>();
function assertContextActive(ctx: Context): void {
ctx.abortSignal?.throwIfAborted();
}
function resolveStoredValue<D extends Value<JsonValue>>(
address: D,
stored: PayloadOf<D> | undefined,
): ResolvedValue<D> {
const fallback = address.default as PayloadOf<D> | undefined;
return (stored === undefined ? fallback : stored) as ResolvedValue<D>;
}
export interface Transaction {
getConversation(id: Id): Promise<Conversation | undefined>;
getEntry(id: Id): Promise<Entry | undefined>;
@ -46,11 +68,11 @@ export interface Transaction {
entry(entry: Omit<Entry, "id">): Id;
task(task: TaskCreate): Id;
setTask(task: Task): void;
value<T extends JsonValue>(
address: Value<T>,
value<D extends Value<JsonValue>>(
address: D,
): {
get(at?: Id): Promise<T | undefined>;
set(value: T): void;
get(at?: Id): Promise<ResolvedValue<D>>;
set(value: PayloadOf<D>): void;
delete(): void;
};
list<T extends JsonValue>(
@ -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<Entry, "id">) => this.entry(entry),
task: (task: TaskCreate) => this.task(task),
setTask: (task: Task) => this.setTask(task),
value: <T extends JsonValue>(address: Value<T>) => this.value(address),
value: <D extends Value<JsonValue>>(address: D) => this.value(address),
list: <T extends JsonValue>(address: List<T>) => 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<Entry, "id">): 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<T extends JsonValue>(
address: Value<T>,
value<D extends Value<JsonValue>>(
address: D,
): {
get(at?: Id): Promise<T | undefined>;
set(value: T): void;
get(at?: Id): Promise<ResolvedValue<D>>;
set(value: PayloadOf<D>): 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<PayloadOf<D>>, at, this.ctx);
return resolveStoredValue(address, stored);
}),
set: (value: PayloadOf<D>) => {
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<JsonValue> | List<JsonValue>): 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<void> = Promise.resolve();
private fault: Faulted | undefined;
private closed = false;
private closeCompletion: Promise<void> | undefined;
private storageCloseCompletion: Promise<void> | 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<T>(build: (tx: Transaction) => T | Promise<T>, ctx: Context): Promise<T> {
commit<T>(build: (tx: Transaction) => T | Promise<T>, ctx: Context): Promise<T> {
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<void> {
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<ReadonlyMap<Id, Conversation>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.getConversations(ids, ctx);
});
}
scanConversations(query: ConversationScan, ctx: Context): Promise<Page<Conversation>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.scanConversations(query, ctx);
});
}
getEntries(ids: readonly Id[], ctx: Context): Promise<ReadonlyMap<Id, Entry>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.getEntries(ids, ctx);
});
}
scanEntries(query: EntryScan, ctx: Context): Promise<Page<Entry>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.scanEntries(query, ctx);
});
}
newestHead(conversationId: Id, at: Id, ctx: Context): Promise<Entry | undefined> {
return this.onLine(() => {
this.assertUsable();
return this.storage.newestHead(conversationId, at, ctx);
});
}
getTasks(ids: readonly Id[], ctx: Context): Promise<ReadonlyMap<Id, Task>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.getTasks(ids, ctx);
});
}
scanTasks(query: TaskScan, ctx: Context): Promise<Page<Task>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.scanTasks(query, ctx);
});
}
getValue<D extends Value<JsonValue>>(address: D, at: Id | undefined, ctx: Context): Promise<ResolvedValue<D>> {
return this.onLine(async () => {
this.assertUsable();
const stored = await this.storage.getValue(address as unknown as Value<PayloadOf<D>>, at, ctx);
return resolveStoredValue(address, stored);
});
}
readList<T extends JsonValue>(address: List<T>, at: Id | undefined, ctx: Context): Promise<readonly Element<T>[]> {
return this.onLine(() => {
this.assertUsable();
return this.storage.readList(address, at, ctx);
});
}
private onLine<T>(operation: () => T | Promise<T>): Promise<T> {
if (this.nested()) return Promise.reject(new NestedLineOperation());
return this.enqueueLine(operation);
}
private enqueueLine<T>(operation: () => T | Promise<T>): Promise<T> {
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<ReadonlyMap<Id, Conversation>> {
return this.storage.getConversations(ids, ctx);
private nested(): boolean {
return lineInvocation.getStore()?.active === true;
}
scanConversations(query: ConversationScan, ctx: Context): Promise<Page<Conversation>> {
return this.storage.scanConversations(query, ctx);
private closeStorage(ctx: Context): Promise<void> {
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<ReadonlyMap<Id, Entry>> {
return this.storage.getEntries(ids, ctx);
}
scanEntries(query: EntryScan, ctx: Context): Promise<Page<Entry>> {
return this.storage.scanEntries(query, ctx);
}
newestHead(conversationId: Id, at: Id, ctx: Context): Promise<Entry | undefined> {
return this.storage.newestHead(conversationId, at, ctx);
}
getTasks(ids: readonly Id[], ctx: Context): Promise<ReadonlyMap<Id, Task>> {
return this.storage.getTasks(ids, ctx);
}
scanTasks(query: TaskScan, ctx: Context): Promise<Page<Task>> {
return this.storage.scanTasks(query, ctx);
}
async getValue<T extends JsonValue>(address: Value<T>, at: Id | undefined, ctx: Context): Promise<T | undefined> {
await this.assertReadable(address, ctx);
return this.storage.getValue(address, at, ctx);
}
async readList<T extends JsonValue>(
address: List<T>,
at: Id | undefined,
ctx: Context,
): Promise<readonly Element<T>[]> {
await this.assertReadable(address, ctx);
return this.storage.readList(address, at, ctx);
}
private async outputIsLive(id: Id, ctx: Context): Promise<boolean> {
return (
(await this.storage.scanTasks({ outputId: id, statuses: ["pending", "running"], limit: 1 }, ctx)).items
.length > 0
);
}
private async assertReadable(address: Address, ctx: Context): Promise<void> {
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();
}
}

31
packages/agent/src/harness/pico/storage.ts
View file

@ -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<T> {
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<Task, "abort" | "checkpoint" | "outcome" | "owns" | "status"> & {
export type NewTask = Omit<TaskBase, "abort" | "checkpoint" | "owns"> & {
readonly status: "pending";
readonly owns: readonly [];
readonly abort?: never;
@ -40,9 +60,11 @@ export type StateWrite =
| { readonly type: "list.remove"; readonly address: List<JsonValue>; readonly elementId: Id }
| { readonly type: "list.clear"; readonly address: List<JsonValue> };
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<Page<Task>>;
getValue<T extends JsonValue>(address: Value<T>, at: Id | undefined, ctx: Context): Promise<T | undefined>;
readList<T extends JsonValue>(address: List<T>, at: Id | undefined, ctx: Context): Promise<readonly Element<T>[]>;
close(ctx: Context): Promise<void>;
}

181
packages/agent/src/harness/pico/tasks.ts
View file

@ -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<string, ConversationValueDefinition<JsonValue>>;
export type ReservedConfigKeys = "set" | "get";
export type KeysOf<U> = U extends unknown ? keyof U : never;
export type Collides<M, N extends keyof M, Reserved extends PropertyKey> = [
Extract<keyof M[N], KeysOf<M[Exclude<keyof M, N>]> | Reserved>,
] extends [never]
? false
: true;
export type AnyCollision<M, Reserved extends PropertyKey> = true extends {
[N in keyof M]: Collides<M, N, Reserved>;
}[keyof M]
? true
: false;
export type DisjointBundles<M, Reserved extends PropertyKey = never> = AnyCollision<M, Reserved> extends false
? true
: false;
declare const taskOutputType: unique symbol;
const taskKindBrand: unique symbol = Symbol("pico.taskKind");
export interface TaskOutputKind<O extends object> {
export interface TaskOutputKind<O extends OutputState> {
readonly kind: string;
readonly [taskOutputType]?: O;
}
export function defineTaskOutput<O extends object>(kind: string): TaskOutputKind<O> {
export function defineTaskOutput<O extends OutputState>(kind: string): TaskOutputKind<O> {
return Object.freeze({ kind });
}
export interface TaskOutputSpec<I extends JsonValue, O extends object> {
readonly kind: TaskOutputKind<O>;
initial(input: I): O;
}
export interface StoredTaskOutputRef {
readonly id: Id;
readonly kind: string;
}
export interface TaskOutputRef<O extends object> extends StoredTaskOutputRef {
export interface TaskOutputRef<O extends OutputState> extends StoredTaskOutputRef {
readonly [taskOutputType]?: O;
}
export type TaskOutputField<O extends object> = [O] extends [never] ? unknown : { readonly output: TaskOutputRef<O> };
export type TaskOutputField<O extends OutputState> = [O] extends [never]
? Record<never, never>
: { readonly output: TaskOutputRef<O> };
export type TaskOutputDefinition<I extends JsonValue, O extends object> = [O] extends [never]
export type TaskOutputDefinition<I extends JsonValue, O extends OutputState> = [O] extends [never]
? { readonly output?: never }
: { readonly output: TaskOutputSpec<I, O> };
: { readonly output: { readonly kind: TaskOutputKind<O>; initial(input: I): O } };
export type TaskOutcome<R extends JsonValue, F extends JsonValue, A extends JsonValue> =
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<I extends JsonValue, C extends TaskCheckpoint> {
export interface TaskBase<I extends JsonValue = JsonValue, C extends TaskCheckpoint = TaskCheckpoint> {
readonly id: Id;
readonly conversationId: Id;
readonly kind: string;
@ -79,7 +100,7 @@ export type Task<
| { readonly status: "terminal"; readonly outcome: TaskOutcome<R, F, A> }
);
export type RunningTask<I extends JsonValue, C extends TaskCheckpoint, O extends object = never> = Omit<
export type RunningTask<I extends JsonValue, C extends TaskCheckpoint, O extends OutputState = never> = Omit<
TaskBase<I, C>,
"output"
> &
@ -92,39 +113,31 @@ export type Completion<R extends JsonValue, F extends JsonValue> =
| { 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<I, C, O>,
runtime: TaskRuntime<I, C, O>,
runtime: TaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<TerminalClosure<I, C, R, F, O>>;
recover(
task: RunningTask<I, C, O>,
runtime: TaskRuntime<I, C, O>,
runtime: TaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<TerminalClosure<I, C, R, F, O>>;
abort(
task: RunningTask<I, C, O>,
runtime: AbortTaskRuntime<I, C, O>,
runtime: AbortTaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<AbortClosure<I, C, A, O>>;
}
@ -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<I, C, O>,
runtime: CoreTaskRuntime<I, C, O>,
runtime: CoreTaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<CoreTerminalClosure<I, C, R, F, O>>;
recover(
task: RunningTask<I, C, O>,
runtime: CoreTaskRuntime<I, C, O>,
runtime: CoreTaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<CoreTerminalClosure<I, C, R, F, O>>;
abort(
task: RunningTask<I, C, O>,
runtime: CoreAbortTaskRuntime<I, C, O>,
runtime: CoreAbortTaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<CoreAbortClosure<I, C, A, O>>;
}
@ -160,8 +177,9 @@ export type TaskKind<
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends object = never,
> = TaskKindMethods<I, C, R, F, A, O> & TaskOutputDefinition<I, O>;
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = TaskKindMethods<I, C, R, F, A, O, H> & TaskOutputDefinition<I, O> & { 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<I, C, R, F, A, O> & TaskOutputDefinition<I, O>;
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = CoreTaskKindMethods<I, C, R, F, A, O, H> & TaskOutputDefinition<I, O> & { readonly [taskKindBrand]: "core" };
export type NoExtra<Expected, Actual extends Expected> = Actual & Record<Exclude<keyof Actual, keyof Expected>, never>;
export type ExactJsonInput<Expected extends JsonValue, Actual extends Expected> = Expected extends readonly JsonValue[]
? Actual
: Expected extends JsonObject
? Actual & Record<Exclude<keyof Actual, keyof Expected>, never>
? NoExtra<Expected, Actual>
: Actual;
type KindDefinition<K> = Omit<K, typeof taskKindBrand>;
type KindFactory<K> = <D extends KindDefinition<K>>(definition: NoExtra<KindDefinition<K>, 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<TaskKind<I, C, R, F, A, O, H>, 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<CoreTaskKind<I, C, R, F, A, O, H>, 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<TaskKind<I, C, R, F, A, O>>;
O extends OutputState,
> = <H extends HookPoints, D extends TaskKindDefinition<I, C, R, F, A, O, H>>(
definition: NoExtra<TaskKindDefinition<I, C, R, F, A, O, H>, D> & { readonly hooks?: H },
) => D & TaskKind<I, C, R, F, A, O, H>;
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<CoreTaskKind<I, C, R, F, A, O>>;
O extends OutputState,
> = <H extends HookPoints, D extends CoreTaskKindDefinition<I, C, R, F, A, O, H>>(
definition: NoExtra<CoreTaskKindDefinition<I, C, R, F, A, O, H>, D> & { readonly hooks?: H },
) => D & CoreTaskKind<I, C, R, F, A, O, H>;
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<I, C, R, F, A, O> {
const define = (definition: KindDefinition<TaskKind<I, C, R, F, A, O>>) =>
const define = <H extends HookPoints, D extends TaskKindDefinition<I, C, R, F, A, O, H>>(
definition: NoExtra<TaskKindDefinition<I, C, R, F, A, O, H>, D> & { readonly hooks?: H },
): D & TaskKind<I, C, R, F, A, O, H> =>
Object.freeze({
...definition,
[taskKindBrand]: "ordinary" as const,
});
}) as D & TaskKind<I, C, R, F, A, O, H>;
return define as TaskKindFactory<I, C, R, F, A, O>;
}
@ -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<I, C, R, F, A, O> {
const define = (definition: KindDefinition<CoreTaskKind<I, C, R, F, A, O>>) =>
const define = <H extends HookPoints, D extends CoreTaskKindDefinition<I, C, R, F, A, O, H>>(
definition: NoExtra<CoreTaskKindDefinition<I, C, R, F, A, O, H>, D> & { readonly hooks?: H },
): D & CoreTaskKind<I, C, R, F, A, O, H> =>
Object.freeze({
...definition,
[taskKindBrand]: "core" as const,
});
}) as D & CoreTaskKind<I, C, R, F, A, O, H>;
return define as CoreTaskKindFactory<I, C, R, F, A, O>;
}
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> = K extends TaskKind<infer I, infer C, infer R, infer F, infer A, infer O>
? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O }
: K extends CoreTaskKind<infer I, infer C, infer R, infer F, infer A, infer O>
? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O }
export type PayloadsOf<K> = K extends TaskKind<infer I, infer C, infer R, infer F, infer A, infer O, infer H>
? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O; hooks: H }
: K extends CoreTaskKind<infer I, infer C, infer R, infer F, infer A, infer O, infer H>
? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O; hooks: H }
: never;
export type InputOf<K> = PayloadsOf<K>["input"];
export type CheckpointOf<K> = PayloadsOf<K>["checkpoint"];
@ -249,9 +303,10 @@ export type ResultOf<K> = PayloadsOf<K>["result"];
export type FailureOf<K> = PayloadsOf<K>["failure"];
export type AbortedOf<K> = PayloadsOf<K>["aborted"];
export type OutputOf<K> = PayloadsOf<K>["output"];
type WithoutOutput<T> = T extends unknown ? Omit<T, "output"> : never;
export type DistributiveOmit<T, K extends PropertyKey> = T extends unknown ? Omit<T, K> : never;
export type TaskOf<K extends AnyDefinedTaskKind> = WithoutOutput<
Task<InputOf<K>, CheckpointOf<K>, ResultOf<K>, FailureOf<K>, AbortedOf<K>>
export type TaskOf<K> = DistributiveOmit<
Task<InputOf<K>, CheckpointOf<K>, ResultOf<K>, FailureOf<K>, AbortedOf<K>>,
"output"
> &
TaskOutputField<OutputOf<K>>;

570
packages/agent/test/harness/pico/memory-storage.test.ts
View file

@ -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<void>; readonly resolve: () => void } {
let resolve!: () => void;
const promise = new Promise<void>((settled) => {
resolve = settled;
});
return { promise, resolve };
}
function taskSpec(
conversationId: number,
extra: Partial<Pick<TaskCreate, "kind" | "output" | "background">> = {},
@ -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<string>({ 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<string>({ 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<string>({ 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<string>({ 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<string>({ 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<string>({ type: "conversation", conversationId: 4 }, "test.lineage", {
rewind: true,
});
const unrelatedList = defineList<string>({ type: "conversation", conversationId: 5 }, "test.lineage", {
rewind: true,
});
const otherDefinition = defineList<string>({ type: "conversation", conversationId: 4 }, "test.other", {
rewind: true,
});
const marker = defineValue<string>({ 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<string>({ 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<string>({ type: "conversation", conversationId: id }, "test.same-batch-lineage", {
rewind: true,
});
tx.list(childList).remove(seeded.element);
return id;
}, ctx);
const childList = defineList<string>({ 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<string>(
{ 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<JsonValue>({ 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<string>("test.default.session", { default: "default" });
const conversationDefinition = conversationValue<number>("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<string>({ type: "conversation", conversationId: root }, "test.history", {
@ -314,7 +806,7 @@ describe("Pico Session over MemoryStorage", () => {
});
const sessionValue = defineValue<string>({ type: "session" }, "test.history");
const taskList = defineList<string>({ type: "task", taskId: 100 }, "test.history");
const sharedValue = defineValue<string>({ type: "shared", id: 200 }, "test.history");
const sharedValue = defineInternalValue<string>({ 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<string>({ 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<string>(sharedAsSession, "plugin.shared")).toThrow("Shared scope is internal");
expect(() => defineList<string>(sharedAsSession, "plugin.shared")).toThrow("Shared scope is internal");
expect(() => defineValue<string>({ type: "session" }, "pi.internal")).toThrow("reserved");
expect(() => defineList<string>({ type: "task", taskId: 1 }, "pi.internal")).toThrow("reserved");
expect(() => conversationValue<string>("pi.internal", { rewind: true })).toThrow("reserved");
expect(() => conversationList<string>("pi.internal", { rewind: false })).toThrow("reserved");
expect(defineInternalList<JsonValue>({ 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<string>(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<JsonValue>({ type: "shared", id: producer }, "pi.output");
const output = defineInternalList<JsonValue>({ 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<JsonValue>({ type: "shared", id: ids.producer }, "pi.output");
const output = defineInternalList<JsonValue>({ type: "shared", id: ids.producer }, "pi.output");
const scratch = defineValue<JsonValue>({ 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({});

744
packages/agent/test/harness/pico/types.compile.ts
View file

@ -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<Left, Right> = (<Type>() => Type extends Left ? 1 : 2) extends <Type>() => Type extends Right ? 1 : 2
? (<Type>() => Type extends Right ? 1 : 2) extends <Type>() => Type extends Left ? 1 : 2
@ -46,39 +109,62 @@ type Equal<Left, Right> = (<Type>() => Type extends Left ? 1 : 2) extends <Type>
: false
: false;
type Assert<Condition extends true> = Condition;
type HasEntry<Surface> = "entry" extends keyof Surface ? true : false;
type HasOutput<Surface> = "output" extends keyof Surface ? true : false;
type ExtendsJson<T extends JsonValue> = T;
type ExtendsJsonObject<T extends JsonObject> = 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<TestInput extends JsonValue ? true : false>,
Assert<readonly [string, { readonly value: number }] extends JsonValue ? true : false>,
type StoredAssertions = [
ExtendsJsonObject<Stored<UserMessage>>,
ExtendsJsonObject<Stored<AssistantMessage>>,
ExtendsJsonObject<Stored<ToolResultMessage>>,
ExtendsJsonObject<Stored<ToolCall>>,
ExtendsJsonObject<Stored<Usage>>,
ExtendsJsonObject<Stored<DeferredHandle>>,
ExtendsJsonObject<Stored<RetryPolicy>>,
ExtendsJson<Stored<UserInput>>,
ExtendsJson<ToolInput>,
ExtendsJson<UsageEntryData>,
ExtendsJson<QueuedInput>,
ExtendsJson<StoredEntryDraft>,
ExtendsJson<SystemEntryData>,
ExtendsJson<AssistantEntryData>,
ExtendsJson<ToolResultData>,
ExtendsJson<Acceptance>,
ExtendsJson<SectionRecord>,
ExtendsJson<ToolControl>,
ExtendsJson<ToolDiagnostic>,
Assert<GenerationOutput extends OutputState ? true : false>,
Assert<ToolOutputState extends OutputState ? true : false>,
Assert<Date extends JsonValue ? false : true>,
Assert<{ readonly value: undefined } extends JsonValue ? false : true>,
];
const ordinaryKind = defineTask<TestInput, TestCheckpoint, TestResult, TestFailure, TestAborted>()({
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<boolean>("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<typeof beforeRun> = async () => {};
void noReturnHandler;
const ordinaryKind = defineTask<Input, Checkpoint, Result, Failure, Aborted>()({
kind: "plugin.ordinary",
config: { enabled: pluginEnabled },
hooks: { beforeRun },
execute() {
return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } }));
},
@ -90,7 +176,7 @@ const ordinaryKind = defineTask<TestInput, TestCheckpoint, TestResult, TestFailu
},
});
const coreKind = defineCoreTask<TestInput, TestCheckpoint, TestResult, TestFailure, TestAborted>()({
const coreKind = defineCoreTask<Input, Checkpoint, Result, Failure, Aborted>()({
kind: "pi.test_core",
execute() {
return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } }));
@ -103,8 +189,10 @@ const coreKind = defineCoreTask<TestInput, TestCheckpoint, TestResult, TestFailu
},
});
const sameCheckpointKind = defineTask<TestInput, SameCheckpoint, TestResult, TestFailure, TestAborted>()({
kind: "test.same_checkpoint",
const progressOutput = defineTaskOutput<Progress>("plugin.progress");
const outputKind = defineTask<Input, Checkpoint, Result, Failure, Aborted, Progress>()({
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<TestInput, SameCheckpoint, TestResult, Tes
},
});
const testOutputKind = defineTaskOutput<TestOutput>("test.output");
const outputTaskKind = defineTask<TestInput, TestCheckpoint, TestResult, TestFailure, TestAborted, TestOutput>()({
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<Parameters<TaskRuntime<TestInput, TestCheckpoint>["commit"]>[0]>[0];
type CoreRuntimeTx = Parameters<Parameters<CoreTaskRuntime<TestInput, TestCheckpoint>["commit"]>[0]>[0];
type OrdinaryAbortTx = Parameters<Parameters<AbortTaskRuntime<TestInput, TestCheckpoint>["commit"]>[0]>[0];
type CoreRuntimeAbortTx = Parameters<Parameters<CoreAbortTaskRuntime<TestInput, TestCheckpoint>["commit"]>[0]>[0];
type OrdinaryFinalTx = Parameters<TerminalClosure<TestInput, TestCheckpoint, TestResult, TestFailure, never>>[0];
type CoreFinalTx = Parameters<CoreTerminalClosure<TestInput, TestCheckpoint, TestResult, TestFailure, never>>[0];
type OrdinaryAbortFinalTx = Parameters<AbortClosure<TestInput, TestCheckpoint, TestAborted, never>>[0];
type CoreAbortFinalTx = Parameters<CoreAbortClosure<TestInput, TestCheckpoint, TestAborted, never>>[0];
type OutputFinalTx = Parameters<TerminalClosure<TestInput, TestCheckpoint, TestResult, TestFailure, TestOutput>>[0];
type OutputAbortFinalTx = Parameters<AbortClosure<TestInput, TestCheckpoint, TestAborted, TestOutput>>[0];
type HookAssertions = [
Assert<Equal<typeof collected, HookPoint<{ readonly value: string }, { readonly seen: true }, "collect">>>,
Assert<Equal<typeof first, HookPoint<{ readonly value: string }, { readonly accepted: true }, "first">>>,
Assert<Equal<typeof beforeRun, HookPoint<{ readonly prompt: string }, { readonly prompt?: string }, "chain">>>,
Assert<
Equal<
HookResult<typeof collected>,
{ readonly outputs: readonly { readonly seen: true }[]; readonly threw?: unknown }
>
>,
Assert<Equal<HookResult<typeof first>, { readonly output?: { readonly accepted: true }; readonly threw?: unknown }>>,
];
type InferredHooks = PayloadsOf<typeof ordinaryKind>["hooks"];
type KindAssertions = [
Assert<"turn" extends keyof typeof ordinaryKind ? false : true>,
Assert<"turn" extends keyof typeof coreKind ? false : true>,
Assert<"busy" extends keyof TaskOf<typeof ordinaryKind> ? false : true>,
Assert<Equal<InputOf<typeof ordinaryKind>, TestInput>>,
Assert<Equal<InputOf<typeof coreKind>, TestInput>>,
Assert<Equal<CheckpointOf<typeof ordinaryKind>, TestCheckpoint>>,
Assert<Equal<ResultOf<typeof ordinaryKind>, TestResult>>,
Assert<Equal<FailureOf<typeof ordinaryKind>, TestFailure>>,
Assert<Equal<AbortedOf<typeof ordinaryKind>, TestAborted>>,
Assert<Equal<PayloadsOf<typeof ordinaryKind>["aborted"], TestAborted>>,
Assert<Equal<OutputOf<typeof ordinaryKind>, never>>,
Assert<Equal<OutputOf<typeof outputTaskKind>, TestOutput>>,
Assert<Equal<Parameters<typeof ordinaryKind.execute>[1], TaskRuntime<TestInput, TestCheckpoint>>>,
Assert<Equal<Parameters<typeof coreKind.execute>[1], CoreTaskRuntime<TestInput, TestCheckpoint>>>,
Assert<Equal<Parameters<typeof outputTaskKind.execute>[1]["output"], TaskOutput<TestOutput>>>,
Assert<Equal<ReturnType<Parameters<TaskOutput<TestOutput>["mutate"]>[0]>, undefined>>,
Assert<Equal<Parameters<typeof ordinaryKind.abort>[1], AbortTaskRuntime<TestInput, TestCheckpoint>>>,
Assert<Equal<Parameters<typeof coreKind.abort>[1], CoreAbortTaskRuntime<TestInput, TestCheckpoint>>>,
Assert<Equal<Parameters<typeof outputTaskKind.abort>[1]["output"], ReadonlyTaskOutput<TestOutput>>>,
Assert<Equal<HasOutput<Parameters<typeof ordinaryKind.execute>[1]>, false>>,
Assert<Equal<HasOutput<Parameters<typeof outputTaskKind.execute>[1]>, true>>,
Assert<Equal<OutputFinalTx["output"], TestOutput>>,
Assert<Equal<OutputAbortFinalTx["output"], TestOutput>>,
Assert<Equal<HasEntry<OrdinaryRuntimeTx>, false>>,
Assert<Equal<HasEntry<CoreRuntimeTx>, true>>,
Assert<Equal<HasEntry<OrdinaryFinalTx>, false>>,
Assert<Equal<HasEntry<CoreFinalTx>, true>>,
Assert<Equal<HasEntry<OrdinaryAbortTx>, false>>,
Assert<Equal<HasEntry<CoreRuntimeAbortTx>, true>>,
Assert<Equal<HasEntry<OrdinaryAbortFinalTx>, false>>,
Assert<Equal<HasEntry<CoreAbortFinalTx>, true>>,
Assert<"sleep" extends keyof AbortTaskRuntime<TestInput, TestCheckpoint> ? false : true>,
Assert<Equal<typeof ordinaryKind.config.enabled, typeof pluginEnabled>>,
Assert<Equal<typeof ordinaryKind.hooks.beforeRun, typeof beforeRun>>,
Assert<Equal<keyof InferredHooks, "beforeRun">>,
Assert<Equal<InferredHooks["beforeRun"], typeof beforeRun>>,
Assert<
Equal<
Parameters<typeof ordinaryKind.execute>[1],
TaskRuntime<Input, Checkpoint, never, typeof ordinaryKind.hooks>
>
>,
Assert<Equal<Parameters<typeof coreKind.execute>[1], CoreTaskRuntime<Input, Checkpoint>>>,
Assert<Equal<Parameters<typeof outputKind.execute>[1]["output"], TaskOutput<Progress>>>,
Assert<Equal<Parameters<typeof outputKind.abort>[1]["output"], ReadonlyTaskOutput<Progress>>>,
Assert<
Equal<
Extract<TaskOf<typeof ordinaryKind>, { status: "terminal" }>["outcome"]["status"],
"completed" | "failed" | "aborted" | "orphaned"
>
>,
];
type KindOutcome = Extract<TaskOf<typeof ordinaryKind>, { readonly status: "terminal" }>["outcome"];
type OutcomeAssertions = [
Assert<Equal<KindOutcome, TaskOutcome<TestResult, TestFailure, TestAborted>>>,
Assert<Equal<Extract<KindOutcome, { readonly status: "completed" }>["result"], TestResult>>,
Assert<Equal<Extract<KindOutcome, { readonly status: "failed" }>["failure"], TestFailure>>,
Assert<Equal<Extract<KindOutcome, { readonly status: "aborted" }>["result"], TestAborted>>,
];
const summaryKind = defineEntry<SummaryEntry>("pi.summary");
type SummaryInput = EntryInput<SummaryEntry>;
type EntryAssertions = [
Assert<Equal<typeof summaryKind.kind, string>>,
Assert<Equal<SummaryInput["head"], number | "self">>,
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<TestCheckpoint>,
outputRef: TaskOutputRef<TestOutput>,
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<TestCheckpoint>): 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<TestOutput>,
readonlyOutput: ReadonlyTaskOutput<TestOutput>,
function verifyNoOutputRuntime(
runtime: TaskRuntime<Input, Checkpoint>,
abortRuntime: Parameters<typeof ordinaryKind.abort>[1],
finalTx: Parameters<TerminalClosure<Input, Checkpoint, Result, Failure, never>>[0],
running: RunningTask<Input, Checkpoint>,
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<Progress>, readonlyOutput: ReadonlyTaskOutput<Progress>, 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<typeof sameCheckpointKind>,
incompatible: CheckpointOf<typeof incompatibleCheckpointKind>,
): void {
// Checkpoints are structural: an independently declared identical shape is assignable.
const structurallyCompatible: CheckpointOf<typeof ordinaryKind> = checkpoint;
void structurallyCompatible;
// @ts-expect-error a differently shaped kind checkpoint is not assignable
const structurallyIncompatible: CheckpointOf<typeof ordinaryKind> = incompatible;
void structurallyIncompatible;
type JobOutput = {
stdout: string;
stderr: string;
droppedStdout: number;
droppedStderr: number;
exitCode?: number;
occurrence: number;
};
type JobAssertions = [Assert<JobOutput extends OutputState ? true : false>];
function verifyJobMutation(output: TaskOutput<JobOutput>, snapshot: Readonly<JobOutput>, 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<UserMessage>];
};
const noteKind = defineEntry<NoteEntry>("plugin.note");
const pinnedKind = defineEntry<PinnedEntry>("plugin.pinned");
const summaryKind = defineEntry<SummaryEntry>("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<NoteEntry extends Entry ? true : false>,
Assert<PinnedEntry extends Entry ? true : false>,
Assert<SummaryEntry extends Entry ? true : false>,
Assert<Equal<EntryInput<SummaryEntry>["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<Checkpoint>,
coreTx: CoreTaskTx<Checkpoint>,
outputRef: TaskOutputRef<Progress>,
): 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<string>({ type: "session" }, "plugin.name");
const taskList = defineList<number>({ type: "task", taskId: 2 }, "plugin.items", { key: "active" });
const sharedList = defineList<number>({ type: "shared", id: 4 }, "plugin.shared");
const conversationValue = defineValue<TestResult>({ type: "conversation", conversationId: 3 }, "plugin.result", {
const rewindable = conversationValue<boolean>("plugin.plan", { rewind: true });
const defaulted = conversationValue<boolean>("plugin.plan", { key: "defaulted", rewind: true, default: false });
const sticky = conversationValue<string>("plugin.mode", { rewind: false });
const events = conversationList<string>("plugin.events", { rewind: true });
const plainSession = sessionValue<string>("plugin.plain-name");
const defaultedSession = sessionValue<string>("plugin.name", { default: "pico" });
const directPlainSession = defineValue<string>({ type: "session" }, "plugin.direct-session");
const directDefaultedSession = defineValue<string>({ type: "session" }, "plugin.direct-session-default", {
default: "session",
});
const directDefaultedTask = defineValue<string>({ type: "task", taskId: 1 }, "plugin.direct-task-default", {
default: "task",
});
const directPlainConversation = defineValue<boolean>(
{ type: "conversation", conversationId: 1 },
"plugin.direct-conversation",
{ rewind: true },
);
const directDefaultedConversation = defineValue<boolean>(
{ 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<string, true>("pi.internal-default", {
rewind: true,
default: "internal",
});
const conversationList = defineList<string>({ type: "conversation", conversationId: 3 }, "plugin.events", {
rewind: false,
});
const typedSessionValue: Value<string> = sessionValue;
const typedTaskList: List<number> = taskList;
void typedSessionValue;
void typedTaskList;
void sharedList;
void conversationValue;
void conversationList;
const internalDefaultedBound = internalDefaulted.bind(1);
const boundList: List<string> = 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<string>({ type: "conversation", conversationId: 3 }, "plugin.value");
type AddressAssertions = [
Assert<Equal<(typeof rewindable)["rewind"], true>>,
Assert<Equal<(typeof sticky)["rewind"], false>>,
Assert<Equal<ResolvedValue<typeof plainSession>, string | undefined>>,
Assert<Equal<ResolvedValue<typeof defaultedSession>, string>>,
Assert<Equal<ResolvedValue<typeof directPlainSession>, string | undefined>>,
Assert<Equal<ResolvedValue<typeof directDefaultedSession>, string>>,
Assert<Equal<ResolvedValue<typeof directDefaultedTask>, string>>,
Assert<Equal<ResolvedValue<typeof directPlainConversation>, boolean | undefined>>,
Assert<Equal<ResolvedValue<typeof directDefaultedConversation>, boolean>>,
Assert<Equal<(typeof directDefaultedSession)["default"], string>>,
Assert<Equal<(typeof directDefaultedTask)["default"], string>>,
Assert<Equal<(typeof directDefaultedConversation)["default"], boolean>>,
Assert<Equal<ResolvedValue<typeof rewindable>, boolean | undefined>>,
Assert<Equal<ResolvedValue<typeof defaulted>, boolean>>,
Assert<Equal<ResolvedValue<typeof bound>, boolean | undefined>>,
Assert<Equal<ResolvedValue<typeof defaultedBound>, boolean>>,
Assert<Equal<(typeof defaultedBound)["default"], boolean>>,
Assert<Equal<ResolvedValue<typeof internalDefaulted>, string>>,
Assert<Equal<ResolvedValue<typeof internalDefaultedBound>, string>>,
Assert<Equal<(typeof defaultedSession)["default"], string>>,
Assert<Extract<PublicScope, { type: "shared" }> 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<string>({ type: "shared", id: 1 }, "plugin.shared");
// @ts-expect-error public bound constructors do not accept shared scope
defineList<string>({ type: "shared", id: 1 }, "plugin.shared");
// @ts-expect-error conversation addresses require an explicit rewind policy
defineValue<string>({ type: "conversation", conversationId: 1 }, "plugin.value");
// @ts-expect-error session addresses are sticky
defineValue<string>({ type: "session" }, "plugin.value", { rewind: true });
// @ts-expect-error task addresses are never rewindable
defineList<string>({ type: "task", taskId: 2 }, "plugin.list", { rewind: true });
// @ts-expect-error shared addresses are never rewindable
defineList<string>({ type: "shared", id: 4 }, "plugin.shared", { rewind: true });
// @ts-expect-error address payloads are strict JSON
defineValue<Date>({ type: "session" }, "plugin.date");
defineInternalList<JsonValue>({ type: "shared", id: 1 }, "pi.output");
}
type ExactInputAssertion = Assert<
Equal<ExactJsonInput<TestInput, TestInput & { readonly extension: boolean }>["extension"], never>
>;
function verifyResolvedTransactionReads(tx: TxState, boundTx: BoundTxState, scratch: ScratchReader): void {
const plainSession = defineValue<string>({ type: "session" }, "plugin.tx-plain");
const defaultedSession = defineValue<string>({ type: "session" }, "plugin.tx-default", { default: "x" });
const defaultedTask = defineValue<number>({ type: "task", taskId: 1 }, "plugin.scratch-default", {
default: 1,
});
const defaultedConversation = defineValue<boolean>(
{ type: "conversation", conversationId: 1 },
"plugin.tx-conversation-default",
{ rewind: true, default: false },
);
const plainDefinition = conversationValue<number>("plugin.bound-plain", { rewind: true });
const defaultedDefinition = conversationValue<number>("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<Equal<Awaited<typeof plainRead>, string | undefined>>,
Assert<Equal<Awaited<typeof defaultedRead>, string>>,
Assert<Equal<Awaited<typeof defaultedConversationRead>, boolean>>,
Assert<Equal<Awaited<typeof plainDefinitionRead>, number | undefined>>,
Assert<Equal<Awaited<typeof defaultedDefinitionRead>, number>>,
Assert<Equal<Awaited<typeof scratchRead>, 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<never, never> };
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<DisjointBundles<RealConfigs, "set" | "get">>, Assert<DisjointBundles<RealHooks>>];
type BadBundleAssertions = [
// @ts-expect-error duplicate bundle keys are rejected by the assertion
Assert<DisjointBundles<FakeDuplicateConfigs, "set" | "get">>,
// @ts-expect-error reserved config keys are rejected by the assertion
Assert<DisjointBundles<FakeReservedConfig, "set" | "get">>,
// @ts-expect-error duplicate hook keys are rejected by the assertion
Assert<DisjointBundles<FakeDuplicateHooks>>,
];
type ExactInputAssertion = Assert<
Equal<ExactJsonInput<Input, Input & { readonly extension: boolean }>["extension"], never>
>;
type CheckpointAssertion = Assert<
{ readonly phase: "deferred"; readonly handle: Stored<DeferredHandle> } extends TaskCheckpoint ? true : false
>;
type HydrationAssertions = [
Assert<Extract<InputOutcome, { status: "placed" }>["input"] extends Entry ? true : false>,
Assert<
Equal<
Extract<InputOutcome, { status: "done" }>["answer"] extends infer A | undefined
? NonNullable<A> extends { message: infer M }
? M
: never
: never,
AssistantMessage
>
>,
ExtendsJson<StoredInputResult>,
];
type KindShape = TaskKind<Input, Checkpoint, Result, Failure, Aborted>;
type RunningAssertion = Assert<"output" extends keyof RunningTask<Input, Checkpoint> ? 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<Input, Checkpoint> | null);