diff --git a/packages/agent/src/harness/pico/index.ts b/packages/agent/src/harness/pico/index.ts index a5c9e4381..f6807b27d 100644 --- a/packages/agent/src/harness/pico/index.ts +++ b/packages/agent/src/harness/pico/index.ts @@ -1,5 +1,8 @@ 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"; diff --git a/packages/agent/src/harness/pico/memory-storage.ts b/packages/agent/src/harness/pico/memory-storage.ts new file mode 100644 index 000000000..fede04287 --- /dev/null +++ b/packages/agent/src/harness/pico/memory-storage.ts @@ -0,0 +1,427 @@ +import type { Context } from "@earendil-works/chord"; +import type { Address, Element, List, Value } from "./addresses.ts"; +import type { Id, JsonValue, Seq } from "./core.ts"; +import type { Conversation, Entry } from "./entries.ts"; +import { + type ConversationScan, + type EntryScan, + InvalidHistoryPosition, + type Page, + type PageQuery, + type Storage, + type TaskScan, + type Write, +} from "./storage.ts"; +import type { Task } from "./tasks.ts"; + +type ValueHistoryWrite = + | { readonly seq: Seq; readonly type: "set"; readonly value: JsonValue } + | { readonly seq: Seq; readonly type: "delete" }; + +type ListHistoryWrite = + | { readonly seq: Seq; readonly type: "append"; readonly element: Element } + | { readonly seq: Seq; readonly type: "remove"; readonly elementId: Id } + | { readonly seq: Seq; readonly type: "clear" }; + +type StoredState = + | { readonly kind: "value"; readonly rewind: false; value?: JsonValue } + | { readonly kind: "list"; readonly rewind: false; readonly elements: Map> } + | { readonly kind: "value"; readonly rewind: true; readonly writes: ValueHistoryWrite[] } + | { readonly kind: "list"; readonly rewind: true; readonly writes: ListHistoryWrite[] }; + +type RewindListState = Extract; + +interface AddressState { + value?: Extract; + list?: Extract; +} + +interface NamespaceState { + unkeyed?: AddressState; + readonly keyed: Map; +} + +type ScopeState = Map; + +interface CappedListState { + readonly state: RewindListState | undefined; + readonly cap: Seq | undefined; +} + +interface StoredEntry { + readonly seq: Seq; + readonly entry: Entry; +} + +function rewindable( + address: Address, +): address is Address & { readonly scope: Extract; readonly rewind: true } { + return address.scope.type === "conversation" && address.rewind; +} + +function scopeId(address: Address): Id | undefined { + if (address.scope.type === "conversation") return address.scope.conversationId; + if (address.scope.type === "task") return address.scope.taskId; + if (address.scope.type === "shared") return address.scope.id; + return undefined; +} + +function page( + values: Iterable, + query: PageQuery, + id: (value: T) => Id, + matches: (value: T) => boolean, + ascending: boolean, +): Page { + const items: T[] = []; + let more = false; + for (const value of values) { + const valueId = id(value); + if ( + !matches(value) || + (query.cursor !== undefined && (ascending ? valueId <= query.cursor : valueId >= query.cursor)) + ) { + continue; + } + if (items.length === query.limit) { + more = true; + break; + } + items.push(value); + } + return { items, ...(more && items.length > 0 ? { next: id(items.at(-1)!) } : {}) }; +} + +export class MemoryStorage implements Storage { + private nextObjectId: Id = 1; + private nextSeq: Seq = 1; + private readonly conversations = new Map(); + private readonly entries = new Map(); + private readonly entriesByConversation = new Map(); + private readonly tasks = new Map(); + private readonly outputReferences = new Map(); + private readonly sessionState: ScopeState = new Map(); + private readonly conversationStickyState = new Map(); + private readonly conversationRewindableState = new Map(); + private readonly taskState = new Map(); + private readonly sharedState = new Map(); + + nextId(): Id { + return this.nextObjectId++; + } + + async commit(writes: readonly Write[], _ctx: Context): Promise { + const seqs = writes.map((_, index) => this.nextSeq + index); + const affectedOutputs = new Set(); + let nextObjectId = this.nextObjectId; + 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 !== "task.create" && write.type !== "task.set") continue; + const currentOutput = this.tasks.get(write.task.id)?.output?.id; + if (currentOutput !== undefined) affectedOutputs.add(currentOutput); + if (write.task.output !== undefined) affectedOutputs.add(write.task.output.id); + } + for (let index = 0; index < writes.length; index++) this.apply(writes[index]!, seqs[index]!); + for (const write of writes) { + if (write.type === "task.set" && write.task.status === "terminal") this.taskState.delete(write.task.id); + } + for (const id of affectedOutputs) { + if (!this.outputReferences.has(id)) this.sharedState.delete(id); + } + this.nextObjectId = nextObjectId; + this.nextSeq += writes.length; + return seqs; + } + + async getConversations(ids: readonly Id[], _ctx: Context): Promise> { + const result = new Map(); + for (const id of ids) { + const conversation = this.conversations.get(id); + if (conversation !== undefined) result.set(id, conversation); + } + return result; + } + + async scanConversations(query: ConversationScan, _ctx: Context): Promise> { + return page( + this.conversations.values(), + query, + (conversation) => conversation.id, + (conversation) => + (query.parent === undefined || conversation.parent?.conversationId === query.parent) && + (query.owner === undefined || conversation.owner === query.owner), + true, + ); + } + + async getEntries(ids: readonly Id[], _ctx: Context): Promise> { + const result = new Map(); + for (const id of ids) { + const stored = this.entries.get(id); + if (stored !== undefined) result.set(id, stored.entry); + } + return result; + } + + async scanEntries(query: EntryScan, _ctx: Context): Promise> { + return page( + this.visibleEntries(query.conversationId, query.through), + query, + (entry) => entry.id, + (entry) => query.kind === undefined || entry.kind === query.kind, + false, + ); + } + + async newestHead(conversationId: Id, at: Id, _ctx: Context): Promise { + for (const entry of this.visibleEntries(conversationId, at)) { + if (entry.head !== undefined) return entry; + } + return undefined; + } + + async getTasks(ids: readonly Id[], _ctx: Context): Promise> { + const result = new Map(); + for (const id of ids) { + const task = this.tasks.get(id); + if (task !== undefined) result.set(id, task); + } + return result; + } + + async scanTasks(query: TaskScan, _ctx: Context): Promise> { + return page( + this.tasks.values(), + query, + (task) => task.id, + (task) => + (query.conversationIds === undefined || query.conversationIds.includes(task.conversationId)) && + (query.statuses === undefined || query.statuses.includes(task.status)) && + (query.kind === undefined || task.kind === query.kind) && + (query.abort === undefined || (task.abort === true) === query.abort) && + (query.outputId === undefined || task.output?.id === query.outputId), + true, + ); + } + + async getValue(address: Value, at: Id | undefined, _ctx: Context): Promise { + if (at !== undefined) { + if (!rewindable(address)) throw new InvalidHistoryPosition(at); + return this.rewindValue(address, this.historyPosition(address.scope.conversationId, at)); + } + 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; + } + + async readList( + address: List, + at: Id | undefined, + _ctx: Context, + ): Promise[]> { + if (at !== undefined) { + if (!rewindable(address)) throw new InvalidHistoryPosition(at); + return this.rewindListElements(address, this.historyPosition(address.scope.conversationId, at)); + } + 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>)]; + } + + private historyPosition(conversationId: Id, at: Id): Seq { + const stored = 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); + if (conversation?.parent === undefined) throw new InvalidHistoryPosition(at); + const parentAtSeq = this.entries.get(conversation.parent.at)?.seq; + if (parentAtSeq === undefined) throw new InvalidHistoryPosition(at); + cap = Math.min(cap ?? parentAtSeq, parentAtSeq); + currentId = conversation.parent.conversationId; + } + } + + private rewindValue(address: Value, cap: Seq | undefined): T | undefined { + let conversationId = address.scope.type === "conversation" ? address.scope.conversationId : undefined; + while (conversationId !== undefined) { + const state = this.conversationStoredState(conversationId, address); + if (state?.kind === "value" && state.rewind) { + for (let index = state.writes.length - 1; index >= 0; index--) { + const write = state.writes[index]!; + if (cap !== undefined && write.seq > cap) continue; + return write.type === "set" ? (write.value as T) : undefined; + } + } + const conversation = this.conversations.get(conversationId); + if (conversation?.parent === undefined) return undefined; + const parentAtSeq = this.historyPosition(conversation.parent.conversationId, conversation.parent.at); + cap = Math.min(cap ?? parentAtSeq, parentAtSeq); + conversationId = conversation.parent.conversationId; + } + return undefined; + } + + private rewindListElements( + address: List & { readonly scope: Extract }, + cap: Seq | undefined, + ): readonly Element[] { + const ancestry: CappedListState[] = []; + let conversationId = address.scope.conversationId; + while (true) { + const stored = this.conversationStoredState(conversationId, address); + ancestry.push({ state: stored?.kind === "list" && stored.rewind ? stored : undefined, cap }); + const conversation = this.conversations.get(conversationId); + if (conversation?.parent === undefined) break; + const parentAtSeq = this.historyPosition(conversation.parent.conversationId, conversation.parent.at); + cap = Math.min(cap ?? parentAtSeq, parentAtSeq); + conversationId = conversation.parent.conversationId; + } + + const elements = new Map>(); + for (const segment of ancestry.reverse()) { + for (const write of segment.state?.writes ?? []) { + if (segment.cap !== undefined && write.seq > segment.cap) continue; + if (write.type === "append") elements.set(write.element.id, write.element as Element); + else if (write.type === "remove") elements.delete(write.elementId); + else elements.clear(); + } + } + return [...elements.values()]; + } + + private *visibleEntries(conversationId: Id, through: Id | undefined): Iterable { + let conversation = this.conversations.get(conversationId); + let cap = 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; + } + if (conversation.parent === undefined) return; + cap = Math.min(cap ?? conversation.parent.at, conversation.parent.at); + conversation = this.conversations.get(conversation.parent.conversationId); + } + } + + private apply(write: Write, seq: Seq): void { + if (write.type === "conversation.create") { + this.conversations.set(write.conversation.id, write.conversation); + } else if (write.type === "entry.append") { + const entry = write.entry; + const stored = { seq, entry }; + this.entries.set(entry.id, stored); + const entries = this.entriesByConversation.get(entry.conversationId); + if (entries === undefined) this.entriesByConversation.set(entry.conversationId, [stored]); + else entries.push(stored); + } else if (write.type === "task.create" || write.type === "task.set") { + const current = this.tasks.get(write.task.id); + if (current?.status !== "terminal" && current?.output !== undefined) + this.removeOutputReference(current.output.id); + this.tasks.set(write.task.id, write.task); + if (write.task.status !== "terminal" && write.task.output !== undefined) { + this.outputReferences.set(write.task.output.id, (this.outputReferences.get(write.task.output.id) ?? 0) + 1); + } + } else if (write.type === "value.set") { + const state = this.valueState(write.address); + if (state.rewind) state.writes.push({ seq, type: "set", value: write.value }); + else state.value = write.value; + } else if (write.type === "value.delete") { + const state = this.valueState(write.address); + if (state.rewind) state.writes.push({ seq, type: "delete" }); + else state.value = undefined; + } else { + const state = this.listState(write.address); + if (state.rewind) { + if (write.type === "list.append") state.writes.push({ seq, type: "append", element: write.element }); + 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 state.elements.clear(); + } + } + + private removeOutputReference(id: Id): void { + const count = this.outputReferences.get(id)!; + if (count === 1) this.outputReferences.delete(id); + else this.outputReferences.set(id, count - 1); + } + + private storedState(address: Address): StoredState | undefined { + return this.stateInScope(this.scopeState(address, false), address); + } + + private conversationStoredState(conversationId: Id, address: Address): StoredState | undefined { + const scopes = address.rewind ? this.conversationRewindableState : this.conversationStickyState; + return this.stateInScope(scopes.get(conversationId), address); + } + + private stateInScope(scope: ScopeState | undefined, address: Address): StoredState | undefined { + const namespace = scope?.get(address.namespace); + const state = address.key === undefined ? namespace?.unkeyed : namespace?.keyed.get(address.key); + return address.kind === "value" ? state?.value : state?.list; + } + + private valueState(address: Value): Extract { + const initial: StoredState = rewindable(address) + ? { kind: "value", rewind: true, writes: [] } + : { kind: "value", rewind: false }; + return this.ensureState(address, initial) as Extract; + } + + private listState(address: List): Extract { + const initial: StoredState = rewindable(address) + ? { kind: "list", rewind: true, writes: [] } + : { kind: "list", rewind: false, elements: new Map() }; + return this.ensureState(address, initial) as Extract; + } + + private ensureState(address: Address, initial: StoredState): StoredState { + const scope = this.scopeState(address, true)!; + let namespace = scope.get(address.namespace); + if (namespace === undefined) { + namespace = { keyed: new Map() }; + scope.set(address.namespace, namespace); + } + let state = address.key === undefined ? namespace.unkeyed : namespace.keyed.get(address.key); + if (state === undefined) { + state = {}; + if (address.key === undefined) namespace.unkeyed = state; + else namespace.keyed.set(address.key, state); + } + const stored = address.kind === "value" ? state.value : state.list; + if (stored !== undefined) return stored; + if (initial.kind === "value") state.value = initial; + else state.list = initial; + return initial; + } + + private scopeState(address: Address, create: boolean): ScopeState | undefined { + if (address.scope.type === "session") return this.sessionState; + const id = scopeId(address)!; + const scopes = + address.scope.type === "conversation" + ? address.rewind + ? this.conversationRewindableState + : this.conversationStickyState + : address.scope.type === "task" + ? this.taskState + : this.sharedState; + let state = scopes.get(id); + if (state === undefined && create) { + state = new Map(); + scopes.set(id, state); + } + return state; + } +} diff --git a/packages/agent/src/harness/pico/session.ts b/packages/agent/src/harness/pico/session.ts new file mode 100644 index 000000000..bfe8d1243 --- /dev/null +++ b/packages/agent/src/harness/pico/session.ts @@ -0,0 +1,310 @@ +import type { Context } from "@earendil-works/chord"; +import type { Address, Element, List, 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 ReadAfterWrite extends Error { + constructor() { + super("Transaction reads must precede writes"); + this.name = "ReadAfterWrite"; + } +} + +export class ScratchRetired extends Error { + readonly taskId: Id; + + constructor(taskId: Id) { + super(`Scratch for task ${taskId} is retired`); + this.name = "ScratchRetired"; + this.taskId = taskId; + } +} + +export class SharedOutputRetired extends Error { + readonly outputId: Id; + + constructor(outputId: Id) { + super(`Shared task output ${outputId} is retired`); + this.name = "SharedOutputRetired"; + this.outputId = outputId; + } +} + +export type TaskCreate = Omit & { + readonly output?: { readonly id?: Id; readonly kind: string }; +}; + +export interface Transaction { + getConversation(id: Id): Promise; + getEntry(id: Id): Promise; + getEntries(ids: readonly Id[]): Promise>; + getTask(id: Id): Promise; + getTasks(ids: readonly Id[]): Promise>; + createConversation(conversation: Omit): Id; + entry(entry: Omit): Id; + task(task: TaskCreate): Id; + setTask(task: Task): void; + value( + address: Value, + ): { + get(at?: Id): Promise; + set(value: T): void; + delete(): void; + }; + list( + address: List, + ): { + read(at?: Id): Promise[]>; + append(value: T): Id; + remove(id: Id): void; + clear(): void; + }; +} + +class TransactionState { + private readonly storage: Storage; + private readonly ctx: Context; + private active = true; + private writing = false; + private pendingReads = 0; + private readonly writes: Write[] = []; + + constructor(storage: Storage, ctx: Context) { + this.storage = storage; + this.ctx = ctx; + } + + facade(): Transaction { + return Object.freeze({ + getConversation: (id: Id) => this.getConversation(id), + getEntry: (id: Id) => this.getEntry(id), + getEntries: (ids: readonly Id[]) => this.getEntries(ids), + getTask: (id: Id) => this.getTask(id), + getTasks: (ids: readonly Id[]) => this.getTasks(ids), + createConversation: (conversation: Omit) => this.createConversation(conversation), + entry: (entry: Omit) => this.entry(entry), + task: (task: TaskCreate) => this.task(task), + setTask: (task: Task) => this.setTask(task), + value: (address: Value) => this.value(address), + list: (address: List) => this.list(address), + }); + } + + finish(): readonly Write[] { + this.assertActive(); + if (this.pendingReads !== 0) throw new Error("Transaction has pending reads"); + this.active = false; + return this.writes; + } + + seal(): void { + this.active = false; + } + + async getConversation(id: Id): Promise { + return (await this.read(() => this.storage.getConversations([id], this.ctx))).get(id); + } + + async getEntry(id: Id): Promise { + return (await this.getEntries([id])).get(id); + } + + getEntries(ids: readonly Id[]): Promise> { + return this.read(() => this.storage.getEntries(ids, this.ctx)); + } + + async getTask(id: Id): Promise { + return (await this.getTasks([id])).get(id); + } + + getTasks(ids: readonly Id[]): Promise> { + return this.read(() => this.storage.getTasks(ids, this.ctx)); + } + + createConversation(conversation: Omit): Id { + const id = this.allocate(); + this.writes.push({ type: "conversation.create", conversation: { ...conversation, id } }); + return id; + } + + entry(entry: Omit): Id { + const id = this.allocate(); + this.writes.push({ type: "entry.append", entry: { ...entry, id } }); + return id; + } + + task(task: TaskCreate): Id { + const id = this.allocate(); + const { output: requestedOutput, ...base } = task; + const output = + requestedOutput === undefined ? undefined : { id: requestedOutput.id ?? id, kind: requestedOutput.kind }; + this.writes.push({ + type: "task.create", + task: { ...base, id, ...(output === undefined ? {} : { output }) }, + }); + return id; + } + + setTask(task: Task): void { + this.write({ type: "task.set", task }); + } + + value( + address: Value, + ): { + get(at?: Id): Promise; + set(value: T): 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 }), + }); + } + + list( + address: List, + ): { + read(at?: Id): Promise[]>; + append(value: T): Id; + remove(id: Id): void; + clear(): void; + } { + this.assertActive(); + return Object.freeze({ + read: (at?: Id) => this.read(() => this.storage.readList(address, at, this.ctx)), + append: (value: T) => { + 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 }), + }); + } + + private allocate(): Id { + this.assertWritable(); + return this.storage.nextId(); + } + + private write(write: Write): void { + this.assertWritable(); + this.writes.push(write); + } + + private async read(read: () => Promise): Promise { + this.assertReadable(); + this.pendingReads++; + try { + return await read(); + } finally { + this.pendingReads--; + } + } + + private assertReadable(): void { + this.assertActive(); + if (this.writing) throw new ReadAfterWrite(); + } + + private assertWritable(): void { + this.assertActive(); + if (this.pendingReads !== 0) throw new Error("Transaction writes must await reads"); + this.writing = true; + } + + private assertActive(): void { + if (!this.active) throw new Error("Transaction is closed"); + } +} + +export class Session { + private readonly storage: Storage; + private tail: Promise = Promise.resolve(); + + constructor(storage: Storage) { + this.storage = storage; + } + + async commit(build: (tx: Transaction) => T | Promise, ctx: Context): Promise { + const previous = this.tail; + let release!: () => void; + this.tail = new Promise((resolve) => { + release = resolve; + }); + await previous; + const tx = new TransactionState(this.storage, ctx); + try { + const result = await build(tx.facade()); + const writes = tx.finish(); + if (writes.length !== 0) await this.storage.commit(writes, ctx); + return result; + } finally { + tx.seal(); + release(); + } + } + + getConversations(ids: readonly Id[], ctx: Context): Promise> { + return this.storage.getConversations(ids, ctx); + } + + scanConversations(query: ConversationScan, ctx: Context): Promise> { + return this.storage.scanConversations(query, ctx); + } + + getEntries(ids: readonly Id[], ctx: Context): Promise> { + return this.storage.getEntries(ids, ctx); + } + + scanEntries(query: EntryScan, ctx: Context): Promise> { + return this.storage.scanEntries(query, ctx); + } + + newestHead(conversationId: Id, at: Id, ctx: Context): Promise { + return this.storage.newestHead(conversationId, at, ctx); + } + + getTasks(ids: readonly Id[], ctx: Context): Promise> { + return this.storage.getTasks(ids, ctx); + } + + scanTasks(query: TaskScan, ctx: Context): Promise> { + return this.storage.scanTasks(query, ctx); + } + + async getValue(address: Value, at: Id | undefined, ctx: Context): Promise { + await this.assertReadable(address, ctx); + return this.storage.getValue(address, at, ctx); + } + + async readList( + address: List, + at: Id | undefined, + ctx: Context, + ): Promise[]> { + await this.assertReadable(address, ctx); + return this.storage.readList(address, at, ctx); + } + + private async outputIsLive(id: Id, ctx: Context): Promise { + return ( + (await this.storage.scanTasks({ outputId: id, statuses: ["pending", "running"], limit: 1 }, ctx)).items + .length > 0 + ); + } + + private async assertReadable(address: Address, ctx: Context): Promise { + if (address.scope.type === "task") { + const task = (await this.storage.getTasks([address.scope.taskId], ctx)).get(address.scope.taskId); + if (task === undefined || task.status === "terminal") throw new ScratchRetired(address.scope.taskId); + } else if (address.scope.type === "shared" && !(await this.outputIsLive(address.scope.id, ctx))) { + throw new SharedOutputRetired(address.scope.id); + } + } +} diff --git a/packages/agent/src/harness/pico/storage.ts b/packages/agent/src/harness/pico/storage.ts new file mode 100644 index 000000000..e1786273f --- /dev/null +++ b/packages/agent/src/harness/pico/storage.ts @@ -0,0 +1,81 @@ +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"; + +export interface PageQuery { + /** Last item returned by the same scan. */ + readonly cursor?: Id; + readonly limit: number; +} + +export interface Page { + readonly items: readonly T[]; + readonly next?: Id; +} + +export class InvalidHistoryPosition extends Error { + readonly at: Id; + + constructor(at: Id) { + super(`Entry ${at} is not a valid history position for this address`); + this.name = "InvalidHistoryPosition"; + this.at = at; + } +} + +export type NewTask = Omit & { + readonly status: "pending"; + readonly owns: readonly []; + readonly abort?: never; + readonly checkpoint?: never; + readonly outcome?: never; +}; + +export type StateWrite = + | { readonly type: "value.set"; readonly address: Value; readonly value: JsonValue } + | { readonly type: "value.delete"; readonly address: Value } + | { readonly type: "list.append"; readonly address: List; readonly element: Element } + | { readonly type: "list.remove"; readonly address: List; readonly elementId: Id } + | { readonly type: "list.clear"; readonly address: List }; + +export type Write = + | StateWrite + | { readonly type: "conversation.create"; readonly conversation: Conversation } + | { readonly type: "entry.append"; readonly entry: Entry } + | { readonly type: "task.create"; readonly task: NewTask } + | { readonly type: "task.set"; readonly task: Task }; + +export type ConversationScan = PageQuery & { + readonly parent?: Id; + readonly owner?: Id; +}; + +export type EntryScan = PageQuery & { + readonly conversationId: Id; + readonly kind?: string; + readonly through?: Id; +}; + +export type TaskScan = PageQuery & { + readonly conversationIds?: readonly Id[]; + readonly statuses?: readonly Task["status"][]; + readonly kind?: string; + readonly abort?: boolean; + readonly outputId?: Id; +}; + +export interface Storage { + nextId(): Id; + commit(writes: readonly Write[], ctx: Context): Promise; + getConversations(ids: readonly Id[], ctx: Context): Promise>; + scanConversations(query: ConversationScan, ctx: Context): Promise>; + getEntries(ids: readonly Id[], ctx: Context): Promise>; + scanEntries(query: EntryScan, ctx: Context): Promise>; + newestHead(conversationId: Id, at: Id, ctx: Context): Promise; + getTasks(ids: readonly Id[], ctx: Context): Promise>; + scanTasks(query: TaskScan, ctx: Context): Promise>; + getValue(address: Value, at: Id | undefined, ctx: Context): Promise; + readList(address: List, at: Id | undefined, ctx: Context): Promise[]>; +} diff --git a/packages/agent/test/harness/pico/memory-storage.test.ts b/packages/agent/test/harness/pico/memory-storage.test.ts new file mode 100644 index 000000000..d65483909 --- /dev/null +++ b/packages/agent/test/harness/pico/memory-storage.test.ts @@ -0,0 +1,571 @@ +import { TODO_CONTEXT } from "@earendil-works/chord/context"; +import { describe, expect, it, vi } from "vitest"; +import { + defineList, + defineValue, + InvalidHistoryPosition, + type JsonValue, + MemoryStorage, + ReadAfterWrite, + ScratchRetired, + Session, + SharedOutputRetired, + type Task, + type TaskCreate, + type Transaction, +} from "../../../src/harness/pico/index.ts"; + +const ctx = TODO_CONTEXT; + +function taskSpec( + conversationId: number, + extra: Partial> = {}, +): TaskCreate { + return { + conversationId, + kind: "test.task", + input: { value: conversationId }, + after: [], + owns: [], + status: "pending", + ...extra, + }; +} + +type LiveTask = Task & { readonly status: "pending" | "running"; readonly outcome?: never }; + +async function setTask(session: Session, id: number, update: (task: LiveTask) => Task): Promise { + await session.commit(async (tx) => { + const task = await tx.getTask(id); + if (task === undefined || task.status === "terminal") throw new Error(`Task ${id} is not live`); + tx.setTask(update(task)); + }, ctx); +} + +describe("Pico Session over MemoryStorage", () => { + it("separates IDs from sequences and burns IDs from failed callbacks", async () => { + const storage = new MemoryStorage(); + expect(await storage.commit([{ type: "conversation.create", conversation: { id: 10 } }], ctx)).toEqual([1]); + const state = defineValue({ type: "session" }, "test.state"); + expect(await storage.commit([{ type: "value.set", address: state, value: "set" }], ctx)).toEqual([2]); + expect(await new Session(storage).commit((tx) => tx.createConversation({}), ctx)).toBe(11); + + const sessionStorage = new MemoryStorage(); + const commit = vi.spyOn(sessionStorage, "commit"); + const session = new Session(sessionStorage); + await expect( + session.commit((tx) => { + tx.createConversation({}); + throw new Error("reject"); + }, ctx), + ).rejects.toThrow("reject"); + expect(commit).not.toHaveBeenCalled(); + expect(await sessionStorage.getConversations([1], ctx)).toEqual(new Map()); + expect(await session.commit((tx) => tx.createConversation({}), ctx)).toBe(2); + expect(commit).toHaveBeenCalledTimes(1); + await expect(commit.mock.results[0]!.value).resolves.toEqual([1]); + await expect( + session.commit(async (tx) => { + tx.createConversation({}); + await tx.getConversation(1); + }, ctx), + ).rejects.toBeInstanceOf(ReadAfterWrite); + expect(await session.commit((tx) => tx.createConversation({}), ctx)).toBe(4); + expect( + await 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); + + let leaked!: Transaction; + await session.commit((tx) => { + leaked = tx; + }, ctx); + expect(() => leaked.createConversation({})).toThrow("Transaction is closed"); + + const concurrent = await Promise.all([ + session.commit(async (tx) => { + await tx.getConversation(1); + return tx.createConversation({}); + }, ctx), + session.commit((tx) => tx.createConversation({}), ctx), + ]); + expect(concurrent).toEqual([6, 7]); + 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", + ); + expect(commit).toHaveBeenCalledTimes(1); + expect(await storage.getConversations([1], ctx)).toEqual(new Map()); + commit.mockRestore(); + + const state = defineValue({ type: "session" }, "test.after-failure"); + expect(await storage.commit([{ type: "value.set", address: state, value: "stored" }], ctx)).toEqual([1]); + expect(await storage.getValue(state, undefined, ctx)).toBe("stored"); + expect(storage.nextId()).toBe(2); + }); + + it("uses write sequences for historical state inside one atomic storage batch", async () => { + const storage = new MemoryStorage(); + const conversation = storage.nextId(); + const entry = storage.nextId(); + const beforeElement = storage.nextId(); + const afterElement = storage.nextId(); + const value = defineValue({ type: "conversation", conversationId: conversation }, "test.value", { + rewind: true, + }); + const list = defineList({ type: "conversation", conversationId: conversation }, "test.list", { + rewind: true, + }); + + expect( + await storage.commit( + [ + { type: "conversation.create", conversation: { id: conversation } }, + { type: "value.set", address: value, value: "before" }, + { type: "list.append", address: list, element: { id: beforeElement, value: "before" } }, + { type: "entry.append", entry: { id: entry, conversationId: conversation, kind: "test.entry" } }, + { type: "value.set", address: value, value: "after" }, + { type: "list.append", address: list, element: { id: afterElement, value: "after" } }, + ], + ctx, + ), + ).toEqual([1, 2, 3, 4, 5, 6]); + expect(entry).toBe(2); + expect(await storage.getValue(value, entry, ctx)).toBe("before"); + expect(await storage.getValue(value, undefined, ctx)).toBe("after"); + expect((await storage.readList(list, entry, ctx)).map((item) => item.value)).toEqual(["before"]); + expect((await storage.readList(list, undefined, ctx)).map((item) => item.value)).toEqual(["before", "after"]); + }); + + it("pages conversations and fork-visible entries", async () => { + const session = new Session(new MemoryStorage()); + const ids = await session.commit((tx) => { + const root = tx.createConversation({}); + const first = tx.entry({ conversationId: root, kind: "test.first" }); + const head = tx.entry({ conversationId: root, kind: "test.head", head: first }); + const fork = tx.createConversation({ parent: { conversationId: root, at: first } }); + const forkEntry = tx.entry({ conversationId: fork, kind: "fork.entry" }); + const late = tx.entry({ conversationId: root, kind: "root.late" }); + const ownedFork = tx.createConversation({ parent: { conversationId: root, at: first }, owner: 99 }); + 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 conversations = await session.scanConversations({ limit: 1 }, ctx); + expect(conversations.items.map(({ id }) => id)).toEqual([1]); + expect( + (await session.scanConversations({ cursor: conversations.next, limit: 1 }, ctx)).items.map(({ id }) => id), + ).toEqual([4]); + expect((await session.getConversations([ids.ownedFork, 999_999], ctx)).has(ids.ownedFork)).toBe(true); + expect((await session.scanConversations({ parent: ids.root, limit: 10 }, ctx)).items.map(({ id }) => id)).toEqual( + [ids.fork, ids.ownedFork], + ); + expect((await session.scanConversations({ owner: 99, limit: 10 }, ctx)).items.map(({ id }) => id)).toEqual([ + ids.ownedFork, + ]); + expect( + (await session.scanConversations({ parent: ids.root, owner: 99, limit: 10 }, ctx)).items.map(({ id }) => id), + ).toEqual([ids.ownedFork]); + + const fork = await session.scanEntries({ conversationId: 4, limit: 1 }, ctx); + expect(fork.items.map(({ id }) => id)).toEqual([5]); + expect( + (await session.scanEntries({ conversationId: 4, cursor: fork.next, limit: 1 }, ctx)).items.map(({ id }) => id), + ).toEqual([2]); + expect( + (await session.scanEntries({ conversationId: 1, through: 3, limit: 10 }, ctx)).items.map(({ id }) => id), + ).toEqual([3, 2]); + expect( + (await session.scanEntries({ conversationId: ids.root, kind: "test.head", limit: 10 }, ctx)).items.map( + ({ id }) => id, + ), + ).toEqual([ids.head]); + 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("buffers values and list elements into one commit", async () => { + const session = new Session(new MemoryStorage()); + const value = defineValue({ type: "conversation", conversationId: 1 }, "test.value", { + rewind: false, + }); + const list = defineList({ type: "conversation", conversationId: 1 }, "test.list", { rewind: false }); + const ids = await session.commit((tx) => { + const conversation = tx.createConversation({}); + tx.value(value).set({ version: 1 }); + const first = tx.list(list).append("a"); + const second = tx.list(list).append("b"); + return { conversation, first, second }; + }, ctx); + expect(ids).toEqual({ conversation: 1, first: 2, second: 3 }); + expect(await session.getValue(value, undefined, ctx)).toEqual({ version: 1 }); + + expect((await session.readList(list, undefined, ctx)).map(({ value }) => value)).toEqual(["a", "b"]); + + await session.commit((tx) => { + tx.value(value).delete(); + tx.list(list).remove(ids.first); + tx.list(list).clear(); + }, ctx); + expect(await session.getValue(value, undefined, ctx)).toBeUndefined(); + expect(await session.readList(list, undefined, ctx)).toEqual([]); + }); + + it("rewinds values through fork caps and preserves local tombstones", async () => { + const storage = new MemoryStorage(); + const session = new Session(storage); + const root = await session.commit((tx) => tx.createConversation({}), ctx); + const rootValue = defineValue({ type: "conversation", conversationId: root }, "test.history", { + rewind: true, + }); + const first = await session.commit((tx) => { + tx.value(rootValue).set("root-first"); + return tx.entry({ conversationId: root, kind: "root.first" }); + }, ctx); + const deleted = await session.commit((tx) => { + tx.value(rootValue).delete(); + return tx.entry({ conversationId: root, kind: "root.deleted" }); + }, ctx); + const late = await session.commit((tx) => { + tx.value(rootValue).set("root-late"); + return tx.entry({ conversationId: root, kind: "root.late" }); + }, ctx); + + expect(await session.getValue(rootValue, undefined, ctx)).toBe("root-late"); + expect(await session.getValue(rootValue, first, ctx)).toBe("root-first"); + expect(await session.getValue(rootValue, deleted, ctx)).toBeUndefined(); + expect(await session.getValue(rootValue, late, ctx)).toBe("root-late"); + + const child = await session.commit( + (tx) => tx.createConversation({ parent: { conversationId: root, at: first } }), + ctx, + ); + const childValue = defineValue({ type: "conversation", conversationId: child }, "test.history", { + rewind: true, + }); + expect(await session.getValue(childValue, undefined, ctx)).toBe("root-first"); + const childDeleted = await session.commit((tx) => { + tx.value(childValue).delete(); + return tx.entry({ conversationId: child, kind: "child.deleted" }); + }, ctx); + const grandchild = await session.commit( + (tx) => tx.createConversation({ parent: { conversationId: child, at: childDeleted } }), + ctx, + ); + const grandchildValue = defineValue( + { type: "conversation", conversationId: grandchild }, + "test.history", + { rewind: true }, + ); + await session.commit((tx) => { + tx.value(childValue).set("child-late"); + tx.entry({ conversationId: child, kind: "child.late" }); + }, ctx); + + expect(await session.getValue(childValue, undefined, ctx)).toBe("child-late"); + expect(await session.getValue(grandchildValue, undefined, ctx)).toBeUndefined(); + expect(await session.getValue(grandchildValue, first, ctx)).toBe("root-first"); + expect(await session.getValue(grandchildValue, childDeleted, ctx)).toBeUndefined(); + + const sibling = await session.commit( + (tx) => tx.createConversation({ parent: { conversationId: root, at: first } }), + ctx, + ); + const siblingValue = defineValue({ type: "conversation", conversationId: sibling }, "test.history", { + rewind: true, + }); + expect(await session.getValue(siblingValue, undefined, ctx)).toBe("root-first"); + await session.commit((tx) => tx.value(siblingValue).set("sibling"), ctx); + expect(await session.getValue(siblingValue, undefined, ctx)).toBe("sibling"); + expect(await session.getValue(childValue, undefined, ctx)).toBe("child-late"); + expect(await session.getValue(grandchildValue, undefined, ctx)).toBeUndefined(); + + const deepAncestorFork = await session.commit( + (tx) => tx.createConversation({ parent: { conversationId: grandchild, at: first } }), + ctx, + ); + const deepAncestorValue = defineValue( + { type: "conversation", conversationId: deepAncestorFork }, + "test.history", + { rewind: true }, + ); + expect(await session.getValue(deepAncestorValue, undefined, ctx)).toBe("root-first"); + await expect(session.getValue(childValue, deleted, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); + await expect(session.getValue(childValue, 999_999, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); + + const unrelated = await session.commit((tx) => { + const conversation = tx.createConversation({}); + return tx.entry({ conversationId: conversation, kind: "unrelated" }); + }, ctx); + await expect(session.getValue(childValue, unrelated, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); + + const sticky = defineValue({ type: "conversation", conversationId: root }, "test.sticky", { + rewind: false, + }); + const sessionValue = defineValue({ type: "session" }, "test.history"); + const taskList = defineList({ type: "task", taskId: 100 }, "test.history"); + const sharedValue = defineValue({ type: "shared", id: 200 }, "test.history"); + 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); + await expect(storage.getValue(sharedValue, first, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); + }); + + it("rewinds complete lists through append, remove, clear, and nested fork caps", async () => { + const session = new Session(new MemoryStorage()); + const root = await session.commit((tx) => tx.createConversation({}), ctx); + const rootList = defineList({ type: "conversation", conversationId: root }, "test.history-list", { + rewind: true, + }); + const first = await session.commit((tx) => { + const a = tx.list(rootList).append("a"); + const b = tx.list(rootList).append("b"); + const entry = tx.entry({ conversationId: root, kind: "root.first" }); + return { a, b, entry }; + }, ctx); + const second = await session.commit((tx) => { + tx.list(rootList).remove(first.a); + const c = tx.list(rootList).append("c"); + const entry = tx.entry({ conversationId: root, kind: "root.second" }); + return { c, entry }; + }, ctx); + const third = await session.commit((tx) => { + tx.list(rootList).clear(); + const d = tx.list(rootList).append("d"); + const entry = tx.entry({ conversationId: root, kind: "root.third" }); + return { d, entry }; + }, ctx); + await session.commit((tx) => tx.list(rootList).append("e"), ctx); + + expect((await session.readList(rootList, first.entry, ctx)).map((item) => item.value)).toEqual(["a", "b"]); + expect((await session.readList(rootList, second.entry, ctx)).map((item) => item.value)).toEqual(["b", "c"]); + expect((await session.readList(rootList, third.entry, ctx)).map((item) => item.value)).toEqual(["d"]); + expect((await session.readList(rootList, undefined, ctx)).map((item) => item.value)).toEqual(["d", "e"]); + + const child = await session.commit( + (tx) => tx.createConversation({ parent: { conversationId: root, at: second.entry } }), + ctx, + ); + const childList = defineList({ type: "conversation", conversationId: child }, "test.history-list", { + rewind: true, + }); + const childAppend = await session.commit((tx) => { + tx.list(childList).append("x"); + return tx.entry({ conversationId: child, kind: "child.append" }); + }, ctx); + const childRemove = await session.commit((tx) => { + tx.list(childList).remove(first.b); + return tx.entry({ conversationId: child, kind: "child.remove" }); + }, ctx); + const grandchild = await session.commit( + (tx) => tx.createConversation({ parent: { conversationId: child, at: childRemove } }), + ctx, + ); + const grandchildList = defineList( + { type: "conversation", conversationId: grandchild }, + "test.history-list", + { rewind: true }, + ); + const childClear = await session.commit((tx) => { + tx.list(childList).clear(); + tx.list(childList).append("y"); + return tx.entry({ conversationId: child, kind: "child.clear" }); + }, ctx); + await session.commit((tx) => { + tx.list(grandchildList).append("z"); + tx.entry({ conversationId: grandchild, kind: "grandchild.append" }); + }, ctx); + + expect((await session.readList(childList, childAppend, ctx)).map((item) => item.value)).toEqual(["b", "c", "x"]); + expect((await session.readList(childList, childRemove, ctx)).map((item) => item.value)).toEqual(["c", "x"]); + expect((await session.readList(childList, childClear, ctx)).map((item) => item.value)).toEqual(["y"]); + expect((await session.readList(childList, undefined, ctx)).map((item) => item.value)).toEqual(["y"]); + expect((await session.readList(grandchildList, undefined, ctx)).map((item) => item.value)).toEqual([ + "c", + "x", + "z", + ]); + expect((await session.readList(grandchildList, childRemove, ctx)).map((item) => item.value)).toEqual(["c", "x"]); + + const sibling = await session.commit( + (tx) => tx.createConversation({ parent: { conversationId: root, at: second.entry } }), + ctx, + ); + const siblingList = defineList({ type: "conversation", conversationId: sibling }, "test.history-list", { + rewind: true, + }); + expect((await session.readList(siblingList, undefined, ctx)).map((item) => item.value)).toEqual(["b", "c"]); + await session.commit((tx) => tx.list(siblingList).append("s"), ctx); + expect((await session.readList(siblingList, undefined, ctx)).map((item) => item.value)).toEqual(["b", "c", "s"]); + expect((await session.readList(childList, undefined, ctx)).map((item) => item.value)).toEqual(["y"]); + + const deepAncestorFork = await session.commit( + (tx) => tx.createConversation({ parent: { conversationId: grandchild, at: second.entry } }), + ctx, + ); + const deepAncestorList = defineList( + { type: "conversation", conversationId: deepAncestorFork }, + "test.history-list", + { rewind: true }, + ); + expect((await session.readList(deepAncestorList, undefined, ctx)).map((item) => item.value)).toEqual(["b", "c"]); + + const unrelated = await session.commit((tx) => { + const conversation = tx.createConversation({}); + return tx.entry({ conversationId: conversation, kind: "unrelated" }); + }, ctx); + const stickyList = defineList({ type: "conversation", conversationId: child }, "test.sticky", { + rewind: false, + }); + await expect(session.readList(childList, third.entry, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); + await expect(session.readList(childList, unrelated, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); + await expect(session.readList(childList, 999_999, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); + await expect(session.readList(stickyList, childRemove, ctx)).rejects.toBeInstanceOf(InvalidHistoryPosition); + }); + + it("keeps scopes, kinds, optional keys, and rewind stores separate", async () => { + const session = new Session(new MemoryStorage()); + const conversation = await session.commit((tx) => tx.createConversation({}), ctx); + const scope = { type: "conversation", conversationId: conversation } as const; + const stickyValue = defineValue(scope, "test.identity", { key: "same", rewind: false }); + const rewindValue = defineValue(scope, "test.identity", { key: "same", rewind: true }); + const stickyList = defineList(scope, "test.identity", { key: "same", rewind: false }); + const rewindList = defineList(scope, "test.identity", { key: "same", rewind: true }); + const absentKey = defineValue(scope, "test.optional-key", { rewind: false }); + const undefinedKey = defineValue(scope, "test.optional-key", { key: undefined, rewind: false }); + const emptyKey = defineValue(scope, "test.optional-key", { key: "", rewind: false }); + const sessionValue = defineValue({ type: "session" }, "test.identity", { key: "same" }); + + await session.commit((tx) => { + tx.value(stickyValue).set("sticky-value"); + tx.value(rewindValue).set("rewind-value"); + tx.list(stickyList).append("sticky-list"); + tx.list(rewindList).append("rewind-list"); + tx.value(absentKey).set("absent"); + tx.value(emptyKey).set("empty"); + tx.value(sessionValue).set("session"); + }, ctx); + + expect(await session.getValue(stickyValue, undefined, ctx)).toBe("sticky-value"); + expect(await session.getValue(rewindValue, undefined, ctx)).toBe("rewind-value"); + expect((await session.readList(stickyList, undefined, ctx)).map((item) => item.value)).toEqual(["sticky-list"]); + expect((await session.readList(rewindList, undefined, ctx)).map((item) => item.value)).toEqual(["rewind-list"]); + expect(await session.getValue(absentKey, undefined, ctx)).toBe("absent"); + expect(await session.getValue(undefinedKey, undefined, ctx)).toBe("absent"); + expect(await session.getValue(emptyKey, undefined, ctx)).toBe("empty"); + expect(await session.getValue(sessionValue, undefined, ctx)).toBe("session"); + }); + + it("stores task snapshots and retires scratch and shared output", async () => { + const storage = new MemoryStorage(); + const session = new Session(storage); + const ids = await session.commit((tx) => { + const conversation = tx.createConversation({}); + const producer = tx.task(taskSpec(conversation, { output: { kind: "test.output" } })); + const output = defineList({ type: "shared", id: producer }, "pi.output"); + tx.list(output).append([["r", { text: "" }]]); + const consumer = tx.task(taskSpec(conversation, { output: { id: producer, kind: "test.output" } })); + return { conversation, producer, consumer }; + }, ctx); + expect(ids).toEqual({ conversation: 1, producer: 2, consumer: 4 }); + + const output = defineList({ type: "shared", id: ids.producer }, "pi.output"); + const scratch = defineValue({ type: "task", taskId: ids.producer }, "test.scratch"); + await session.commit((tx) => { + tx.value(scratch).set({ n: 1 }); + tx.list(output).append([["s", ["text"], "done"]]); + tx.list(output).append([["s", ["status"], "complete"]]); + }, ctx); + + await setTask(session, ids.producer, (task) => ({ ...task, status: "running" })); + await setTask(session, ids.producer, (task) => ({ ...task, checkpoint: { phase: "working" } })); + await setTask(session, ids.producer, (task) => ({ + ...task, + status: "terminal", + outcome: { status: "completed", result: null }, + })); + await expect(session.getValue(scratch, undefined, ctx)).rejects.toBeInstanceOf(ScratchRetired); + expect(await storage.getValue(scratch, undefined, ctx)).toBeUndefined(); + + await setTask(session, ids.consumer, (task) => ({ ...task, status: "running" })); + const replacement = await session.commit(async (tx) => { + const consumer = await tx.getTask(ids.consumer); + if (consumer === undefined || consumer.status === "terminal") throw new Error("Consumer is not live"); + tx.setTask({ ...consumer, status: "terminal", outcome: { status: "completed", result: null } }); + return tx.task(taskSpec(ids.conversation, { output: { id: ids.producer, kind: "test.output" } })); + }, ctx); + expect(await session.readList(output, undefined, ctx)).toHaveLength(3); + await setTask(session, replacement, (task) => ({ ...task, status: "running" })); + await setTask(session, replacement, (task) => ({ + ...task, + 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([]); + }); + + it("filters and pages tasks", async () => { + const session = new Session(new MemoryStorage()); + const ids = await session.commit((tx) => { + const conversation = tx.createConversation({}); + const otherConversation = tx.createConversation({}); + return { + conversation, + otherConversation, + first: tx.task(taskSpec(conversation)), + second: tx.task( + taskSpec(conversation, { kind: "other", background: true, output: { kind: "test.output" } }), + ), + third: tx.task(taskSpec(otherConversation)), + }; + }, ctx); + await setTask(session, ids.first, (task) => ({ ...task, abort: true })); + await setTask(session, ids.first, (task) => ({ ...task, status: "running" })); + await setTask(session, ids.first, (task) => ({ + ...task, + status: "terminal", + outcome: { status: "aborted", result: null }, + })); + + const first = await session.scanTasks({ limit: 1 }, ctx); + expect(first.items.map(({ id }) => id)).toEqual([ids.first]); + expect((await session.scanTasks({ cursor: first.next, limit: 1 }, ctx)).items.map(({ id }) => id)).toEqual([ + ids.second, + ]); + expect((await session.getTasks([ids.first, 999_999], ctx)).has(ids.first)).toBe(true); + expect( + (await session.scanTasks({ conversationIds: [ids.conversation], limit: 10 }, ctx)).items.map(({ id }) => id), + ).toEqual([ids.first, ids.second]); + expect((await session.scanTasks({ kind: "other", limit: 10 }, ctx)).items.map(({ id }) => id)).toEqual([ + ids.second, + ]); + expect((await session.scanTasks({ outputId: ids.second, limit: 10 }, ctx)).items.map(({ id }) => id)).toEqual([ + ids.second, + ]); + expect( + ( + await session.scanTasks( + { + conversationIds: [ids.conversation], + statuses: ["terminal"], + kind: "test.task", + abort: true, + limit: 10, + }, + ctx, + ) + ).items.map(({ id }) => id), + ).toEqual([ids.first]); + }); +});