diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 530105f7e..5d587d0f4 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -20,6 +20,14 @@ - `AgentDoc`, `configure()`, `DEFAULT_RETRY_POLICY`, and `DEFAULT_COMPACTION_POLICY` exports. - `HarnessOptions.conversationCreated(tx, conversation)` runs in every commit that creates or forks a conversation, after the built-in documents, so applications can create their own documents in every conversation. - `Tx.submissionByRequest()` and `Tx.createSubmission()`; the latter writes a raw submission record without admission rules. +- `ConversationWatch` type alias for `Conversation.watch()`. +- `Harness.taskGraph()` and `Harness.watchTaskGraph()`: every live task with its owner edge, committed status, background flag, abort mark, and owned conversations, as a Chord state or watch that advances with each commit. + +### Fixed + +- `TaskRuntime.now()` and `report()` throw after the invocation ended, like every other runtime operation. +- A commit that only migrates a document to a newer version publishes it, so states and watches acquired with the older token receive the migrated value. +- A failed `Harness.open()` closes the Session without the caller's context and rethrows the original error; a cancelled context no longer masks it or leaves Storage open. ## [0.99.2] - 2026-09-30 diff --git a/packages/durable/README.md b/packages/durable/README.md index 04559c837..bf1d169d9 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -28,6 +28,7 @@ Built on [`@earendil-works/pi-ai`](../ai/README.md) for model access and `@earen - [More Conversations and Forks](#more-conversations-and-forks) - [Abort and Subagents](#abort-and-subagents) - [Child Tasks](#child-tasks) +- [Task Graph](#task-graph) - [Your Own State](#your-own-state) - [Usage and Cost](#usage-and-cost) - [Storage](#storage) @@ -467,6 +468,22 @@ decide: async (task, runtime, context) => { [`24-child-tasks.ts`](test/examples/24-child-tasks.ts) runs a checkout with four payments: a declined card, a cancelled checkout, and a restart while the payments run. +## Task Graph + +`harness.taskGraph(context)` shows every live task of the Session as one Chord state, for a task panel or debugging. Each node has its owner edge (`owner` task, or none for a task its conversation owns), its status, whether it is `background` or abort-marked, and the conversations it owns. `harness.watchTaskGraph(context)` delivers the same value as a watch, like a conversation's `watch()`. + +```typescript +const graph = await harness.taskGraph(context); +graph.subscribe((value) => { + for (const node of Object.values(value.tasks)) { + const status = node.state.status === "waiting" ? `waiting on ${node.state.on.join(", ")}` : node.state.status; + console.log(`${node.id} ${node.kind} ${status}`, node.owner ?? `conversation ${node.conversationId}`); + } +}); +``` + +A task appears with the commit that creates it and leaves with the commit that makes it terminal. Statuses are the committed ones: `pending`, `running`, `waiting` (with `on` and `policy`), and `completing` (with the held outcome's status). After a restart, tasks that were `running` show as `pending` until they run again. Whether a pending task is blocked by a missing definition is not part of the graph; `harness.inspect()` reports that. The graph lists live tasks only: once a subagent's owner task is terminal, a later task in its conversation is a top-level node, and the conversation's `ConversationRecord.owner` (also in its view's `conversation`) links it to its parent. [`24-child-tasks.ts`](test/examples/24-child-tasks.ts) prints the checkout's tree while its payments run. + ## Your Own State Documents are typed JSON objects committed together with entries. Define one, and edit it in a commit: diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index 8370613db..d3dd9e859 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -18,7 +18,6 @@ import type { TaskId, TaskRecord, Tx, - WatchHandle, } from "../types.ts"; import { ROOT_CONVERSATION_ID } from "../types.ts"; import { AgentDoc, configure, createAgent, resolveAgent, resolveSettings } from "./agent.ts"; @@ -29,6 +28,7 @@ import { LiveDoc, settleSchedulerOutcome } from "./live.ts"; import { BUILTIN_TASKS } from "./registry.ts"; import { type InvocationBinding, TaskScheduler } from "./scheduler.ts"; import { Submissions } from "./submissions.ts"; +import { type TaskGraph, TaskGraphView, type TaskGraphWatch } from "./task-graph.ts"; import type { Agent, AgentChange, @@ -39,6 +39,7 @@ import type { ConversationCreateOptions, ConversationHandle, ConversationInit, + ConversationWatch, HarnessInspection, HarnessOptions, Harness as HarnessType, @@ -154,7 +155,7 @@ class ConversationImpl implements Conversation { return this.#host.views.state(this.id, context); } - watch(context: Context): Promise> { + watch(context: Context): Promise { return this.#host.views.watch(this.id, context); } } @@ -167,6 +168,7 @@ class HarnessImpl extends SessionImpl implements readonly #host: ConversationHost; readonly #tasks: TaskScheduler; readonly #submissions: Submissions; + readonly #taskGraph: TaskGraphView; #closed = false; constructor(storage: Storage, options: HarnessOptions, context: Context) { @@ -195,6 +197,7 @@ class HarnessImpl extends SessionImpl implements context: withoutAbortSignal(context), }); this.#submissions = new Submissions(this, storage, now, settings, () => this.#tasks.resume()); + this.#taskGraph = new TaskGraphView(this, storage); this.#host = { harness: this, storage, @@ -290,6 +293,14 @@ class HarnessImpl extends SessionImpl implements return total; } + taskGraph(context: Context): Promise> { + return this.#taskGraph.state(context); + } + + watchTaskGraph(context: Context): Promise { + return this.#taskGraph.watch(context); + } + root( context: Context, options?: { readonly agent?: AgentChange; readonly init?: ConversationInit }, @@ -411,7 +422,10 @@ export const Harness = { try { await harness.openTasks(context); } catch (error) { - await harness.close(context); + // The caller's context may be what failed open: close without it, and rethrow the open error. + await harness + .close(withoutAbortSignal(context)) + .catch((closeError: unknown) => options.onReport?.(closeError)); throw error; } return harness; diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index c8e21c38d..7882a711b 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -1169,8 +1169,14 @@ export class TaskScheduler { }) as ErasedRuntime["entry"], context: (conversationId, context, at) => this.#read(invocation, () => readContext(this.#session, this.#storage, conversationId, context, at)), - now: () => this.#now(), - report: (error) => this.#report(error), + now: () => { + if (invocation.ended) throw endedError(invocation); + return this.#now(); + }, + report: (error) => { + if (invocation.ended) throw endedError(invocation); + this.#report(error); + }, }; } diff --git a/packages/durable/src/harness/task-graph.ts b/packages/durable/src/harness/task-graph.ts new file mode 100644 index 000000000..f2065d5d0 --- /dev/null +++ b/packages/durable/src/harness/task-graph.ts @@ -0,0 +1,222 @@ +import { type AttachedReplicatedState, type Context, type JsonValue, replicatedState } from "@earendil-works/chord"; +import { withoutAbortSignal } from "@earendil-works/chord/context"; +import { applyImmutable, type Op } from "@earendil-works/chord/delta"; +import { CommittedStateSource, CommittedWatch } from "../session/observation.ts"; +import type { SessionImpl } from "../session/session.ts"; +import type { + CommitPublication, + ConversationId, + JoinPolicy, + Storage, + TaskId, + TaskOutcome, + TaskRecord, + WatchHandle, +} from "../types.ts"; +import { closedError, scanAll } from "./util.ts"; + +/** A live task's durable status without its checkpoint and outcome payloads (spec §9.5). */ +export type TaskGraphState = + | { readonly status: "pending" | "running"; readonly phase: string } + | { + readonly status: "waiting"; + readonly phase: string; + readonly on: readonly TaskId[]; + readonly policy: JoinPolicy; + } + /** Outcome held until its ordinary owned work drains. */ + | { readonly status: "completing"; readonly outcome: TaskOutcome["status"] }; + +export type TaskGraphNode = { + readonly id: TaskId; + readonly kind: string; + readonly conversationId: ConversationId; + /** Owner task; absent for a conversation-owned task. */ + readonly owner?: TaskId; + readonly background: boolean; + readonly abortRequested: boolean; + readonly state: TaskGraphState; + /** Conversations this task owns, in ID order. */ + readonly conversations: readonly ConversationId[]; +}; + +/** Every live task of the Session (spec §9.5). */ +export type TaskGraph = { + /** Every live task, keyed by its decimal ID. */ + readonly tasks: Readonly>; +}; + +export type TaskGraphWatch = WatchHandle; + +type AnyTaskRecord = TaskRecord; +type Observer = CommittedStateSource | CommittedWatch; +type Mount = { value: TaskGraph; readonly observers: Set }; + +const LIVE_STATUSES = ["pending", "running", "waiting", "completing"] as const; +const SCAN_PAGE_SIZE = 256; + +/** + * The Harness's task graph mount: built on the Session line by its first observer and dropped with its last. It + * advances from the Session's commit publications, which are durable. + */ +export class TaskGraphView { + readonly #session: SessionImpl; + readonly #storage: Storage; + #mount: Mount | undefined; + #closed = false; + + constructor(session: SessionImpl, storage: Storage) { + this.#session = session; + this.#storage = storage; + session.subscribeCommits((publication, context) => { + if (this.#mount !== undefined) advance(this.#mount, publication, context); + }); + session.subscribeClose(() => { + this.#closed = true; + for (const observer of [...(this.#mount?.observers ?? [])]) observer.closeSession(); + this.#mount = undefined; + }); + } + + /** A disposable read-only Chord state of the graph. */ + async state(context: Context): Promise> { + const { observer, detach } = await this.#attach( + (value, release) => new CommittedStateSource(value, release), + context, + ); + try { + return replicatedState(observer); + } catch (error) { + detach(); + throw error; + } + } + + /** A serialized exact-frame watch of the graph; cancelling `context` stops it. */ + async watch(context: Context): Promise { + const { observer } = await this.#attach( + (value, release) => new CommittedWatch(value, release), + context, + ); + const signal = context.abortSignal; + if (signal?.aborted) { + observer.cancel(); + throw signal.reason; + } + if (signal !== undefined) observer.observeCancellation(signal); + return observer; + } + + /** Register an observer created from the current revision, atomically on the Session line. */ + #attach( + create: (value: TaskGraph, release: () => void) => O, + context: Context, + ): Promise<{ observer: O; detach: () => void }> { + return this.#session.readOnLine(async () => { + const mount = this.#mount ?? { value: await this.#build(context), observers: new Set() }; + const detach = (): void => { + mount.observers.delete(observer); + if (mount.observers.size === 0 && this.#mount === mount) this.#mount = undefined; + }; + const observer = create(mount.value, detach); + // Close or cancellation may begin while the mount builds; register nothing then. + if (this.#closed) throw closedError(); + context.abortSignal?.throwIfAborted(); + this.#mount = mount; + mount.observers.add(observer); + return { observer, detach }; + }); + } + + async #build(context: Context): Promise { + const tasks: Record = {}; + const records: AnyTaskRecord[] = []; + for (const status of LIVE_STATUSES) { + records.push( + ...(await scanAll((cursor) => this.#storage.scanTasks({ status }, SCAN_PAGE_SIZE, cursor, context))), + ); + } + records.sort((a, b) => a.id - b.id); + for (const record of records) { + const owned = await scanAll((cursor) => + this.#storage.scanConversations({ ownerTaskId: record.id }, SCAN_PAGE_SIZE, cursor, context), + ); + const conversations = owned.map((conversation) => conversation.id).sort((a, b) => a - b); + tasks[String(record.id)] = nodeOf(record, conversations); + } + return { tasks }; + } +} + +/** Derive the mount's operations from one publication, apply them, and hand the revision to every observer. */ +function advance(mount: Mount, publication: CommitPublication, context: Context): void { + const ops: Op[] = []; + // Nodes this publication set or deleted, over the mount's value. + const changed = new Map(); + const node = (key: string): TaskGraphNode | undefined => + changed.has(key) ? changed.get(key) : mount.value.tasks[key]; + for (const change of publication.changes) { + if (change.type !== "task") continue; + const record = change.value; + const key = String(record.id); + const previous = node(key); + if (record.state.status === "terminal") { + if (previous === undefined) continue; + ops.push(["d", ["tasks", key]]); + changed.set(key, undefined); + continue; + } + const next = nodeOf(record, previous?.conversations ?? []); + if (previous !== undefined && JSON.stringify(next) === JSON.stringify(previous)) continue; + ops.push(["s", ["tasks", key], next as unknown as JsonValue]); + changed.set(key, next); + } + // After the tasks, so a conversation created with its owner task in one commit finds the owner's node. Change order + // within a publication is unspecified, so each owner's list is sorted again. + const created = new Map(); + for (const change of publication.changes) { + if (change.type !== "conversation" || change.value.owner === undefined) continue; + const key = String(change.value.owner.taskId); + if (node(key) !== undefined) created.set(key, [...(created.get(key) ?? []), change.value.id]); + } + for (const [key, ids] of created) { + const conversations = [...node(key)!.conversations, ...ids].sort((a, b) => a - b); + ops.push(["s", ["tasks", key, "conversations"], conversations]); + } + if (ops.length === 0) return; + mount.value = applyImmutable(mount.value, ops); + const frameContext = withoutAbortSignal(context); + for (const observer of [...mount.observers]) observer.advance(mount.value, ops, frameContext); +} + +function nodeOf(record: AnyTaskRecord, conversations: readonly ConversationId[]): TaskGraphNode { + return { + id: record.id, + kind: record.kind, + conversationId: record.conversationId, + ...(record.owner === undefined ? {} : { owner: record.owner }), + background: record.background, + abortRequested: record.abortRequested, + state: stateOf(record), + conversations, + }; +} + +function stateOf(record: AnyTaskRecord): TaskGraphState { + const state = record.state; + switch (state.status) { + case "pending": + case "running": + return { status: state.status, phase: phaseOf(state.checkpoint) }; + case "waiting": + return { status: "waiting", phase: phaseOf(state.checkpoint), on: [...state.on], policy: state.policy }; + // Terminal records never reach here: they leave the graph. + case "completing": + case "terminal": + return { status: "completing", outcome: state.outcome.status }; + } +} + +function phaseOf(checkpoint: JsonValue): string { + return (checkpoint as { readonly phase: string }).phase; +} diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index abd499f33..b3b50827f 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -39,6 +39,7 @@ import type { Tx, WatchHandle, } from "../types.ts"; +import type { TaskGraph, TaskGraphWatch } from "./task-graph.ts"; import type { UsageState } from "./usage.ts"; import type { ConversationView } from "./view.ts"; @@ -528,16 +529,17 @@ export interface Conversation { /** The structural view (spec §9.3) as a disposable read-only Chord state. */ viewState(context: Context): Promise>; /** The structural view as a serialized exact-frame watch with bounded pending frames. */ - watch(context: Context): Promise>; + watch(context: Context): Promise; } -// TODO: decide how Harness exposes subscribeCommits() and subscribeClose(). Their listeners run on the Session line -// and must not throw or call Session APIs, and Harness close will also join task invocations. +export type ConversationWatch = WatchHandle; + /** Durable agent harness over one Session. */ export interface Harness extends Session { /** - * Enable task scheduling. Idempotent; throws after close. Calls that wait for progress (`Conversation.submit()`, - * `Submission.wait()`, `waitForTask()`, `waitForIdle()`) enable it too. + * Enable task scheduling. Idempotent; throws after close. Calls that ask for progress enable it too: + * `Conversation.submit()`, `Conversation.compact()`, `Conversation.abort()`, `Submission.wait()`, `waitForTask()`, + * `Harness.waitForIdle()`, and `Conversation.waitForIdle()`. Read-only viewers never do. */ resume(): void; @@ -571,6 +573,10 @@ export interface Harness extends Session { waitForIdle(context: Context): Promise; /** Session total: every conversation's `pi.usage` summed. */ usage(context: Context): Promise; + /** Every live task with its owner edge, status, and owned conversations (spec §9.5), as a disposable Chord state. */ + taskGraph(context: Context): Promise>; + /** The task graph as a serialized exact-frame watch with bounded pending frames. */ + watchTaskGraph(context: Context): Promise; } /** What a hook may use: committed reads and the asking task's memos, which hooks and the task share. */ diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index ccd22761d..5f2cbb88a 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -39,6 +39,7 @@ export { Harness } from "./harness/harness.ts"; export { InboxDoc, type InboxItem, type InboxState } from "./harness/inbox.ts"; export { type CompactionStatus, LiveDoc, type LiveState, type ToolSlot } from "./harness/live.ts"; export { createRegistry } from "./harness/registry.ts"; +export type { TaskGraph, TaskGraphNode, TaskGraphState, TaskGraphWatch } from "./harness/task-graph.ts"; export { ToolTask, type ToolTaskCheckpoint, type ToolTaskInput, type ToolTaskResult } from "./harness/tool.ts"; export type { Agent, @@ -57,6 +58,7 @@ export type { ConversationInit, ConversationRetryPolicy, ConversationStreamOptions, + ConversationWatch, EnvTarget, Extension, GenerationHooks, diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index 960b4d8c0..f1d8f1137 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -489,7 +489,10 @@ export class SessionImpl implements Session { if (change.type !== "document" || change.record.id !== loaded.record.id) continue; // A document state's frames carry no caller cancellation; a watch observes its own cancellation. const frameContext = observer instanceof CommittedStateSource ? withoutAbortSignal(context) : context; - observer.advance(change.value, observedOperations(observed, change), frameContext); + const ops = observedOperations(observed, change); + // A migration-only base changes nothing for an observer of the new version. + if (ops.length === 0) continue; + observer.advance(change.value, ops, frameContext); } }); unsubscribeClose = this.subscribeClose(() => observer.closeSession()); diff --git a/packages/durable/src/session/transaction.ts b/packages/durable/src/session/transaction.ts index 450abde1b..f4186a279 100644 --- a/packages/durable/src/session/transaction.ts +++ b/packages/durable/src/session/transaction.ts @@ -1014,7 +1014,10 @@ function planDocument(document: DocumentEntry): DocumentPlan | undefined { } } -/** Whether adoption publishes the plan: every creation, copy, and retirement, and a loaded incarnation that changed. */ +/** + * Whether adoption publishes the plan: every creation, copy, and retirement, and a loaded incarnation that writes + * content, which includes a migration-only base so observers of the older shape receive the new value. + */ function publishes(plan: DocumentPlan): boolean { - return plan.retire || plan.change?.loaded === undefined || plan.change.prepared.ops.length > 0; + return plan.retire || plan.change?.loaded === undefined || plan.content !== undefined; } diff --git a/packages/durable/test/chord-guide.test.ts b/packages/durable/test/chord-guide.test.ts new file mode 100644 index 000000000..4d4aa8455 --- /dev/null +++ b/packages/durable/test/chord-guide.test.ts @@ -0,0 +1,394 @@ +/** + * The code of docs/pico-v5-chord-usage.md, compiled and run against a real Harness. Keep the two in sync: the guide's + * blocks are copied here verbatim apart from formatting and the output sinks, which record instead of printing. + */ +import { + type Context, + createFacetHost, + createRemoteServiceBinding, + defineFacet, + defineService, + type Facet, + type FacetHost, + type RemoteServiceTransport, + type ReplicatedState, +} from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import type { Op } from "@earendil-works/chord/delta"; +import { + type ConversationId, + type DocumentObserver, + defineDoc, + defineDocFamily, + defineTask, + type Harness, + MemoryStorage, + type Session, + type TaskId, + type TaskRuntime, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { context } from "./session-support.ts"; +import { completed, deferred, eventually, openTasks } from "./task-support.ts"; + +const logged: unknown[][] = []; +const console = { log: (...values: unknown[]) => void logged.push(values) }; + +// ─── 1. A Session-wide canvas ──────────────────────────────────────────────── + +type Stroke = { color: string; points: { x: number; y: number }[] }; +type CanvasState = { strokes: Stroke[] }; + +const CanvasDoc = defineDoc({ + kind: "app.canvas", + version: 1, + scope: "session", + initial: () => ({ strokes: [] }), + // Store a complete base after at most 99 replayed deltas. + checkpointWhen: (_value, _ops, info) => info.deltasSinceBase >= 99, +}); + +interface CanvasService { + readonly state: ReplicatedState; + addStroke(stroke: Stroke, context: Context): Promise; +} +const Canvas = defineService("app.canvas"); + +async function createCanvasFacet(session: Session, context: Context): Promise { + // Creation is explicit; observation never writes. + await session.commit(async (tx) => { + await tx.doc(CanvasDoc); + }, context); + const state = await session.documentState(CanvasDoc, context); + if (state === undefined) throw new Error("canvas was retired during setup"); + return defineFacet({ + id: "app.canvas/session", + setup(env) { + env.own(() => state.dispose()); + env.provide(Canvas, { + state, + async addStroke(stroke, context) { + await session.commit(async (tx) => { + const draft = await tx.doc(CanvasDoc); + draft.strokes.push(stroke); // Chord copies the assigned stroke by value. + }, context); + }, + }); + }, + }); +} + +const CanvasConsumer = defineFacet({ + id: "app.canvas/consumer", + setup(env) { + const canvas = env.use(Canvas); // Declare now; access only after activation. + env.onActivate(() => { + env.own( + canvas.state.subscribe((value, _context, delivery) => { + console.log(delivery.kind, delivery.sequence, value?.strokes.length ?? "retired"); + }), + ); // subscribe delivers the current hydrated value, then updates. + }); + }, +}); + +async function runCanvasExample(session: Session): Promise { + const context = BACKGROUND_CONTEXT; + const provider = await createCanvasFacet(session, context); + const host = await createFacetHost({ facets: [provider, CanvasConsumer] }); + try { + await host.services.use(Canvas).addStroke( + { + color: "black", + points: [ + { x: 10, y: 20 }, + { x: 30, y: 40 }, + ], + }, + context, + ); + } finally { + await host.dispose(); // Unsubscribes; does not delete the canvas or close Session. + } +} + +async function connectCanvas(transport: RemoteServiceTransport, context: Context) { + const services = createRemoteServiceBinding({ services: [Canvas], transport }); + const canvas = services.use(Canvas); + const stop = canvas.state.subscribe((value, _context, delivery) => { + console.log(delivery.kind, delivery.sequence, value?.strokes ?? "retired"); + }); + try { + await services.ready(context); // Initial snapshot installed; not all future updates. + } catch (error) { + stop(); + await services.dispose(BACKGROUND_CONTEXT); + throw error; + } + return async () => { + stop(); + await services.dispose(BACKGROUND_CONTEXT); + }; +} + +async function shutdown(host: FacetHost, detachClients: () => Promise, harness: Harness, context: Context) { + await detachClients(); + await host.dispose(); + await harness.close(context); +} + +// ─── 2. Conversation-scoped diff reviews ───────────────────────────────────── + +type ReviewInput = { path: string; patch: string }; +type ReviewComment = { id: string; line: number; text: string }; +type ReviewState = ReviewInput & { comments: ReviewComment[] }; +const ReviewDoc = defineDocFamily({ + kind: "app.diff-review", + version: 1, + family: true, + scope: "conversation", + history: "latest", + fork: "current", + initial: (seed) => ({ path: seed.path, patch: seed.patch, comments: [] }), + checkpointWhen: (_value, _ops, info) => info.deltasSinceBase >= 49, +}); +interface DiffReviewService { + readonly state: ReplicatedState; + identity(context: Context): Promise<{ conversationId: ConversationId; key: string }>; + addComment(comment: ReviewComment, context: Context): Promise; +} +const DiffReviews = defineService("app.diff-reviews"); + +function reviewFacet( + session: Session, + conversationId: ConversationId, + reviews: readonly { key: string; seed: ReviewInput }[], + context: Context, +): Facet { + return defineFacet({ + id: "app.diff-reviews/session", + setup(env) { + const instances = env.provideMany(DiffReviews); + env.onActivate(async () => { + for (const review of reviews) { + await session.commit(async (tx) => { + await tx.doc(ReviewDoc, conversationId, review.key, review.seed); + }, context); + const state = await session.documentState(ReviewDoc, conversationId, review.key, context); + if (state === undefined) throw new Error("review was retired during setup"); + env.own(() => state.dispose()); + // Chord instance keys route services; they are not numeric document incarnation IDs. + instances.spawn(JSON.stringify([conversationId, review.key]), { + state, + async identity() { + return { conversationId, key: review.key }; + }, + async addComment(comment, context) { + await session.commit(async (tx) => { + const draft = await tx.doc(ReviewDoc, conversationId, review.key, review.seed); + draft.comments.push(comment); // Chord copies the assigned comment by value. + }, context); + }, + }); // The facet owns spawned service lifetimes automatically. + } + }); + }, + }); +} + +// ─── 3. Task-scoped output and a tool/task watch ───────────────────────────── + +const sent: unknown[] = []; +const rendered: string[] = []; +async function sendCommittedFrame(value: JobOutput | null, ops: readonly Op[]): Promise { + sent.push({ value, ops }); +} +async function render(text: string): Promise { + rendered.push(text); +} + +type JobInput = { command: string }; +type JobOutput = { stdout: string; chunks: number }; +const JobOutputDoc = defineDoc({ + kind: "app.job-output", + version: 1, + scope: "task", + initial: () => ({ stdout: "", chunks: 0 }), + checkpointWhen: (_value, _ops, info) => info.deltasSinceBase >= 99, +}); +async function appendJobOutput( + runtime: TaskRuntime, + chunk: string, + context: Context, +): Promise { + // Read process output outside this callback. The runtime gates the live task. + await runtime.commit(async (tx) => { + const draft = await tx.doc(JobOutputDoc, runtime.taskId); + draft.stdout = (draft.stdout + chunk).slice(-50_000); + draft.chunks += 1; + }, context); +} +async function observeJob( + api: DocumentObserver, + producerTaskId: TaskId, + finished: Promise, + context: Context, +): Promise { + const watch = await api.watchDoc(JobOutputDoc, producerTaskId, context); + if (watch === undefined) return; + console.log(watch.value === null ? "retired" : watch.value.stdout); + try { + watch.start(async (value, ops, _context) => { + await sendCommittedFrame(value, ops); + await render(value === null ? "retired" : value.stdout); + }); // Serialized callbacks never overlap. + await finished; // Caller-supplied observation lifetime; outside any commit. + } finally { + await watch.stop(); // Idempotent; prevents another callback from starting. + } +} + +// ─── Runs ──────────────────────────────────────────────────────────────────── + +/** In-process transport to a host's service provider; a real client puts a socket between the two. */ +function inProcess(host: FacetHost): RemoteServiceTransport { + return { + invoke: (call, callContext) => host.services.invoke(call, callContext), + subscribe: async (serviceId, mode, listener) => { + const subscription = host.services.subscribe(serviceId, mode, listener); + return { + snapshot: subscription.snapshot, + activate: () => subscription.activate(), + close: () => subscription.close(), + }; + }, + }; +} + +describe("Chord usage guide", () => { + it("runs the canvas: facet host, a remote client, withdrawal and detach before Harness close", async () => { + logged.length = 0; + const { harness } = await openTasks(new MemoryStorage(), []); + await runCanvasExample(harness); + expect(logged).toEqual([ + ["hydrate", 0, 0], + ["update", 1, 1], + ]); + expect((await harness.snapshot(CanvasDoc, context))?.strokes).toHaveLength(1); + + // A worker installs the canvas facet again; a late remote client hydrates the stroke, then follows updates. + logged.length = 0; + const host = await createFacetHost({ facets: [await createCanvasFacet(harness, context)] }); + const detach = await connectCanvas(inProcess(host), context); + await host.services.use(Canvas).addStroke({ color: "red", points: [] }, context); + await eventually(() => logged.length === 2); + expect(logged.map(([kind, , strokes]) => [kind, (strokes as Stroke[]).length])).toEqual([ + ["hydrate", 1], + ["update", 2], + ]); + // Record the shutdown order: clients detach and services withdraw while the Harness is still open. + const order: string[] = []; + const trackedHarness = new Proxy(harness, { + get: (target, key) => { + const value = Reflect.get(target, key, target); + if (typeof value !== "function") return value; + if (key !== "close") return value.bind(target); + return async (closeContext: Context) => { + order.push("close start"); + await target.close(closeContext); + order.push("close end"); + }; + }, + }); + const trackedHost: FacetHost = { + services: host.services, + reload: (facets) => host.reload(facets), + dispose: async () => { + order.push("dispose start"); + await host.dispose(); + order.push("dispose end"); + }, + }; + await shutdown( + trackedHost, + async () => { + order.push("detach start"); + await detach(); + order.push("detach end"); + }, + trackedHarness, + context, + ); + expect(order).toEqual(["detach start", "detach end", "dispose start", "dispose end", "close start", "close end"]); + expect(() => host.services.use(Canvas)).toThrow("disposed"); + await expect(harness.commit(() => {}, context)).rejects.toThrow("closed"); + }); + + it("runs the diff reviews with a keyed consumer", async () => { + const { harness } = await openTasks(new MemoryStorage(), []); + const root = await harness.root(context); + const reviews = [{ key: "review-7", seed: { path: "a.ts", patch: "-old\n+new" } }]; + const seen: { key: string; comments: number }[] = []; + const consumer = defineFacet({ + id: "app.diff-reviews/consumer", + setup(env) { + env.observe(DiffReviews, async (review, callContext) => { + const { key } = await review.identity(callContext); + review.state.subscribe((value) => void seen.push({ key, comments: value?.comments.length ?? -1 })); + await review.addComment({ id: "c1", line: 1, text: "why?" }, callContext); + }); + }, + }); + const host = await createFacetHost({ facets: [reviewFacet(harness, root.id, reviews, context), consumer] }); + await eventually(() => seen.some((frame) => frame.comments === 1)); + expect(seen).toEqual([ + { key: "review-7", comments: 0 }, + { key: "review-7", comments: 1 }, + ]); + expect(await harness.snapshot(ReviewDoc, root.id, "review-7", context)).toMatchObject({ + path: "a.ts", + comments: [{ id: "c1" }], + }); + await host.dispose(); + await harness.close(context); + }); + + it("runs the job output watch until the producer retires its document", async () => { + logged.length = 0; + sent.length = 0; + rendered.length = 0; + const gate = deferred(); + const Job = defineTask({ + name: "app.job", + version: 1, + initial: () => ({ phase: "running" }), + phases: { + running: async (task, runtime, ctx) => { + await appendJobOutput(runtime, `$ ${task.input.command}\n`, ctx); + await gate.promise; + await appendJobOutput(runtime, "ok\n", ctx); + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + const { harness } = await openTasks(new MemoryStorage(), [Job]); + const root = await harness.root(context); + const id = await root.commit( + (tx) => tx.createTask(Job, { command: "make" }, { ownership: { kind: "conversation" } }), + context, + ); + harness.resume(); + await eventually(async () => (await harness.snapshot(JobOutputDoc, id, context))?.chunks === 1); + const finished = harness.waitForTask(id, context).then(() => {}); + const observing = observeJob(harness, id, finished, context); + await eventually(() => logged.length === 1); + gate.resolve(); + await observing; + expect(logged).toEqual([["$ make\n"]]); + // Stopped once the task finished; the retirement frame may or may not have been delivered before. + expect(rendered.slice(0, 1)).toEqual(["$ make\nok\n"]); + expect(await harness.watchDoc(JobOutputDoc, id, context)).toBeUndefined(); + await harness.close(context); + }); +}); diff --git a/packages/durable/test/examples/24-child-tasks.ts b/packages/durable/test/examples/24-child-tasks.ts index 887e2f50d..e1b49defe 100644 --- a/packages/durable/test/examples/24-child-tasks.ts +++ b/packages/durable/test/examples/24-child-tasks.ts @@ -1,4 +1,5 @@ // A task that owns child tasks: a checkout charges four payments at once and waits for them. +// The task graph view prints the checkout's tree while its payments run. // Run from packages/durable: // node --conditions=source --experimental-strip-types test/examples/24-child-tasks.ts import { mkdtemp, rm } from "node:fs/promises"; @@ -12,6 +13,8 @@ import { defineExtension, defineTask, Harness, + type TaskGraph, + type TaskGraphNode, type TaskId, } from "../../src/index.ts"; import { openNodeSqliteStorage } from "../../src/storage/sqlite/node.ts"; @@ -95,6 +98,17 @@ const directory = await mkdtemp(join(tmpdir(), "pi-durable-example-")); const databasePath = join(directory, "session.sqlite"); const open = async () => Harness.open(await openNodeSqliteStorage(databasePath), { models: createModels(), registry }, context); +/** Print the live tasks as a tree along their owner edges. */ +function printGraph(graph: TaskGraph): void { + const nodes = Object.values(graph.tasks); + const print = (node: TaskGraphNode, depth: number): void => { + const state = node.state; + const status = state.status === "waiting" ? `waiting on ${state.on.join(", ")}` : state.status; + console.log(` ${" ".repeat(depth)}${node.kind} ${node.id}: ${status}`); + for (const child of nodes.filter((candidate) => candidate.owner === node.id)) print(child, depth + 1); + }; + for (const node of nodes.filter((candidate) => candidate.owner === undefined)) print(node, 0); +} const checkout = (root: Conversation, cards: string[]) => root.commit((tx) => tx.createTask(Checkout, { cards }, { ownership: { kind: "conversation" } }), context); @@ -109,6 +123,9 @@ console.log("The customer cancels:"); id = await checkout(root, ["visa-5", "visa-6", "visa-7", "visa-8"]); harness.resume(); await new Promise((resolve) => setTimeout(resolve, 20)); +const graph = await harness.taskGraph(context); +printGraph(graph.value); +graph.dispose(); await harness.abortTask(id, context); console.log(` checkout: ${(await harness.waitForTask(id, context)).state.outcome.status}`); diff --git a/packages/durable/test/harness-lifecycle.test.ts b/packages/durable/test/harness-lifecycle.test.ts new file mode 100644 index 000000000..425a5f81e --- /dev/null +++ b/packages/durable/test/harness-lifecycle.test.ts @@ -0,0 +1,613 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { Context, JsonValue } from "@earendil-works/chord"; +import { createModels, fauxAssistantMessage, fauxText, fauxToolCall, Type } from "@earendil-works/pi-ai"; +import { + type Conversation, + type ConversationId, + createRegistry, + defineDoc, + defineTask, + defineTool, + GenerationTask, + Harness, + MemoryStorage, + type Storage, + type TaskId, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import type { SessionImpl } from "../src/session/session.ts"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { chatSetup, openChat } from "./chat-support.ts"; +import { addHooks, addTask, addTool } from "./harness-support.ts"; +import { ControlledStorage, context, flush } from "./session-support.ts"; +import { completed, countingReader, deferred, eventually, openTasks, settled } from "./task-support.ts"; + +type Step = { phase: "run" }; + +/** A one-phase task running `run` and completing with null. */ +function oneStep(name: string, run: () => Promise = async () => {}) { + return defineTask({ + name, + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (_task, runtime, ctx) => { + await run(); + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); +} + +function start(conversation: Conversation, task: ReturnType, background = false): Promise { + return conversation.commit( + (tx) => tx.createTask(task, null, { ownership: { kind: "conversation" }, background }), + context, + ); +} + +/** Drop the loaded trackers, so the next acquisition reads Storage on the line. */ +function unloadDocuments(harness: Harness): Promise { + return (harness as unknown as SessionImpl).unloadDocuments(); +} + +function withSignal(signal: AbortSignal): Context { + return { ...context, abortSignal: signal }; +} + +/** Whether a raw Storage read still succeeds, as code of a joined invocation relies on. */ +async function storageOpen(storage: Storage): Promise { + return storage.task(1 as TaskId, context).then( + () => true, + () => false, + ); +} + +/** Commit a conversation with a `running` task, as a crash leaves it, so open has a reconciliation commit to fail. */ +async function seedRunningTask(storage: Storage): Promise { + const conversationId = await storage.mintId(); + const id = await storage.mintId>(); + const state = { status: "running", checkpoint: { phase: "run" } } as const; + const task = { id, conversationId, kind: "test.seeded", version: 1, input: null, background: false, state }; + await storage.commit( + [ + { type: "conversation", value: { id: conversationId } }, + { type: "task", value: { ...task, abortRequested: false } }, + ], + context, + ); +} + +describe("Harness open", () => { + it("closes without a cancelled caller context and rethrows the original error when open fails", async () => { + const storage = new ControlledStorage(); + await seedRunningTask(storage); + const reader = countingReader(createRegistry()); + const held = storage.holdCommits(); + storage.failNextCommit(new Error("disk full")); + const controller = new AbortController(); + const opening = Harness.open( + storage, + { models: createModels(), registry: reader }, + withSignal(controller.signal), + ); + await held.entered; + controller.abort(new Error("caller gave up")); + held.release(); + await expect(opening).rejects.toThrow("disk full"); + expect(reader.subscriptions()).toBe(0); + expect(await storageOpen(storage)).toBe(false); + }); + + it("reports a failing close and still rethrows the open error", async () => { + class FailingClose extends ControlledStorage { + override async close(closeContext: Context): Promise { + await super.close(closeContext); + throw new Error("close failed"); + } + } + const storage = new FailingClose(); + await seedRunningTask(storage); + storage.failNextCommit(new Error("disk full")); + const reports: unknown[] = []; + const opening = Harness.open( + storage, + { models: createModels(), registry: createRegistry(), onReport: (error) => reports.push(error) }, + context, + ); + await expect(opening).rejects.toThrow("disk full"); + expect(reports).toEqual([new Error("close failed")]); + }); +}); + +describe("Harness close", () => { + it("joins a task handler that ignores its signal before closing Storage", async () => { + const storage = new MemoryStorage(); + const reached = deferred(); + const gate = deferred(); + let readAfterRelease: boolean | undefined; + const Stubborn = oneStep("test.close-stubborn", async () => { + reached.resolve(); + await gate.promise; + readAfterRelease = await storageOpen(storage); + }); + const { harness } = await openTasks(storage, [Stubborn]); + await start(await harness.root(context), Stubborn); + harness.resume(); + await reached.promise; + const closing = harness.close(context); + expect(await settled(closing)).toBe(false); + expect(await storageOpen(storage)).toBe(true); + gate.resolve(); + await closing; + expect(readAfterRelease).toBe(true); + expect(await storageOpen(storage)).toBe(false); + }); + + it("joins a tool execute and a hook that ignore their signal before closing Storage", async () => { + for (const where of ["tool", "hook"] as const) { + const setup = chatSetup(); + const storage = new MemoryStorage(); + const reached = deferred(); + const gate = deferred(); + let readAfterRelease: boolean | undefined; + const stubborn = async (): Promise => { + reached.resolve(); + await gate.promise; + readAfterRelease = await storageOpen(storage); + }; + addTool( + setup.registry, + defineTool({ + name: "wait", + description: "wait", + parameters: Type.Object({}), + execute: async () => { + if (where === "tool") await stubborn(); + return { content: [] }; + }, + }), + ); + addHooks(setup.registry, GenerationTask, { + beforeRequest: async () => { + if (where === "hook") await stubborn(); + return undefined; + }, + }); + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("wait", {}, { id: "c1" })], { stopReason: "toolUse" }), + fauxAssistantMessage([fauxText("done")]), + ]); + const { harness, root } = await openChat(storage, setup); + await root.submit({ type: "input", content: "go" }, context); + await reached.promise; + const closing = harness.close(context); + expect(await settled(closing)).toBe(false); + expect(await storageOpen(storage)).toBe(true); + gate.resolve(); + await closing; + expect(readAfterRelease).toBe(true); + expect(await storageOpen(storage)).toBe(false); + } + }); + + it("lets a new Harness open the same Storage once close resolved: no old invocation code runs", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-lifecycle-")); + try { + const path = join(directory, "session.sqlite"); + const log: string[] = []; + const gate = deferred(); + let generation = 1; + const Stubborn = oneStep("test.generations", async () => { + const mine = generation; + log.push(`start ${mine}`); + if (mine === 1) await gate.promise; + log.push(`end ${mine}`); + }); + const first = await openTasks(await openNodeSqliteStorage(path), [Stubborn]); + const id = await start(await first.harness.root(context), Stubborn); + first.harness.resume(); + await eventually(() => log.length === 1); + const closing = first.harness.close(context).then(() => void log.push("closed")); + expect(await settled(closing)).toBe(false); + gate.resolve(); + await closing; + expect(log).toEqual(["start 1", "end 1", "closed"]); + generation = 2; + const second = await openTasks(await openNodeSqliteStorage(path), [Stubborn]); + second.harness.resume(); + await second.harness.waitForTask(id, context); + expect(log).toEqual(["start 1", "end 1", "closed", "start 2", "end 2"]); + await second.harness.close(context); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it("keeps shutting down after a cancelled close, and a second close awaits the same shutdown", async () => { + const storage = new MemoryStorage(); + const reached = deferred(); + const gate = deferred(); + const Stubborn = oneStep("test.close-cancelled", async () => { + reached.resolve(); + await gate.promise; + }); + const { harness } = await openTasks(storage, [Stubborn]); + await start(await harness.root(context), Stubborn); + harness.resume(); + await reached.promise; + const controller = new AbortController(); + const cancelled = harness.close(withSignal(controller.signal)); + controller.abort(new Error("stop waiting")); + await expect(cancelled).rejects.toThrow("stop waiting"); + // Admission stays sealed and the invocation still holds Storage open. + await expect(harness.commit(() => {}, context)).rejects.toThrow("closed"); + expect(await storageOpen(storage)).toBe(true); + const second = harness.close(context); + expect(await settled(second)).toBe(false); + gate.resolve(); + await second; + expect(await storageOpen(storage)).toBe(false); + }); + + it("settles durably a commit whose committer was cancelled while it was in Storage", async () => { + const storage = new ControlledStorage(); + const { harness } = await openTasks(storage, []); + const root = await harness.root(context); + const held = storage.holdCommits(); + const controller = new AbortController(); + const committing = root.commit( + (tx) => tx.appendEntry(root.id, { kind: "note" }).then((entry) => entry.id), + withSignal(controller.signal), + ); + await held.entered; + controller.abort(new Error("committer gave up")); + held.release(); + const id = await committing; + const page = await root.entries({}, 10, undefined, context); + expect(page.items.map((entry) => entry.id)).toEqual([id]); + await harness.close(context); + }); + + it("publishes no frame to states and watches from a commit that settles during close", async () => { + const Notes = defineDoc<{ text: string }>({ + kind: "test.close-frames", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + }); + const storage = new ControlledStorage(); + const { harness } = await openTasks(storage, []); + const root = await harness.root(context); + await harness.commit(async (tx) => { + (await tx.doc(Notes)).text = "before"; + }, context); + const docState = (await harness.documentState(Notes, context))!; + const docWatch = (await harness.watchDoc(Notes, context))!; + const viewState = await root.viewState(context); + const viewWatch = await root.watch(context); + const graphState = await harness.taskGraph(context); + const graphWatch = await harness.watchTaskGraph(context); + const frames: string[] = []; + graphState.subscribe((_value, _context, delivery) => { + if (delivery.kind === "update") frames.push("graphState"); + }); + graphWatch.start(async () => void frames.push("graphWatch")); + docState.subscribe((_value, _context, delivery) => { + if (delivery.kind === "update") frames.push("docState"); + }); + viewState.subscribe((_value, _context, delivery) => { + if (delivery.kind === "update") frames.push("viewState"); + }); + docWatch.start(async () => void frames.push("docWatch")); + viewWatch.start(async () => void frames.push("viewWatch")); + const docValue = docState.value; + const viewValue = viewState.value; + + const held = storage.holdCommits(); + const committing = harness.commit(async (tx) => { + (await tx.doc(Notes)).text = "during close"; + await tx.appendEntry(root.id, { kind: "note" }); + await tx.createTask(oneStep("test.close-graph"), null, { + ownership: { kind: "conversation" }, + conversationId: root.id, + }); + }, context); + await held.entered; + const closing = harness.close(context); + held.release(); + await committing; + await closing; + await flush(); + expect(frames).toEqual([]); + expect(docState.value).toBe(docValue); + expect(viewState.value).toBe(viewValue); + expect(graphState.value).toEqual({ tasks: {} }); + expect(await graphWatch.closed).toEqual({ reason: "session_closed" }); + expect(await docWatch.closed).toEqual({ reason: "session_closed" }); + expect(await viewWatch.closed).toEqual({ reason: "session_closed" }); + // The commit itself settled. + expect(storage.commits.at(-1)).toContainEqual(expect.objectContaining({ type: "entry" })); + }); + + it("leaves no subscription behind a watch acquisition cancelled on the line", async () => { + const Notes = defineDoc<{ text: string }>({ + kind: "test.cancelled-watch", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + }); + const storage = new ControlledStorage(); + const { harness } = await openTasks(storage, []); + const root = await harness.root(context); + await harness.commit(async (tx) => { + (await tx.doc(Notes)).text = "x"; + }, context); + // Count commit subscriptions of document observers. + let subscriptions = 0; + const subscribe = harness.subscribeCommits.bind(harness); + harness.subscribeCommits = (listener) => { + subscriptions++; + const unsubscribe = subscribe(listener); + let active = true; + return () => { + if (active) subscriptions--; + active = false; + unsubscribe(); + }; + }; + + // Cancel a document watch while its acquisition loads the document on the line. + await unloadDocuments(harness); + let find = storage.holdFindDocument(); + let controller = new AbortController(); + const docWatch = harness.watchDoc(Notes, withSignal(controller.signal)); + await find.entered; + controller.abort(new Error("cancelled")); + find.release(); + await expect(docWatch).rejects.toThrow("cancelled"); + expect(subscriptions).toBe(0); + + // Cancel a view watch while it builds its mount on the line. + await unloadDocuments(harness); + find = storage.holdFindDocument(); + controller = new AbortController(); + const viewWatch = root.watch(withSignal(controller.signal)); + await find.entered; + controller.abort(new Error("cancelled")); + find.release(); + await expect(viewWatch).rejects.toThrow("cancelled"); + // No observer kept the mount: the next observer builds a new one, and the one after that another. + const first = await root.viewState(context); + const firstValue = first.value; + first.dispose(); + const second = await root.viewState(context); + expect(second.value).not.toBe(firstValue); + second.dispose(); + await harness.close(context); + }); + + it("rejects conversation and Harness operations once close begins; inspect queued before reports closing", async () => { + const storage = new ControlledStorage(); + const { harness } = await openTasks(storage, []); + const root = await harness.root(context); + const entry = await root.commit((tx) => tx.appendEntry(root.id, { kind: "note" }), context); + const submission = await root.submit({ type: "write", entry: { kind: "note" } }, context); + + const held = storage.holdCommits(); + const blocking = harness.commit((tx) => tx.appendEntry(root.id, { kind: "blocker" }).then(() => {}), context); + await held.entered; + const queuedInspect = harness.inspect(context); + const closing = harness.close(context); + const operations: Record Promise> = { + submit: () => root.submit({ type: "input", content: "x" }, context), + agent: () => root.agent(context), + configure: () => root.configure({ thinkingLevel: "low" }, context), + commit: () => root.commit(() => {}, context), + context: () => root.context(context), + entries: () => root.entries({}, 10, undefined, context), + fork: () => root.fork(entry.id, { ownership: { kind: "ownerless" } }, context), + compact: () => root.compact(undefined, context), + reset: () => root.reset(undefined, context), + abort: () => root.abort(context), + conversationIdle: () => root.waitForIdle(context), + viewState: () => root.viewState(context), + watch: () => root.watch(context), + taskGraph: () => harness.taskGraph(context), + watchTaskGraph: () => harness.watchTaskGraph(context), + status: () => submission.status(context), + wait: () => submission.wait(context), + abortSubmission: () => submission.abort(context), + root: () => harness.root(context), + conversation: () => harness.conversation(root.id, context), + createConversation: () => harness.createConversation({ ownership: { kind: "ownerless" } }, context), + getTask: () => harness.getTask(1 as TaskId, context), + inspect: () => harness.inspect(context), + submission: () => harness.submission(submission.id, context), + abortTask: () => harness.abortTask(1 as TaskId, context), + waitForTask: () => harness.waitForTask(1 as TaskId, context), + harnessIdle: () => harness.waitForIdle(context), + usage: () => harness.usage(context), + }; + const outcomes: Record = {}; + for (const [name, operation] of Object.entries(operations)) { + outcomes[name] = await operation().then( + () => "resolved", + (error: unknown) => (/closed/.test(String(error)) ? "closed" : String(error)), + ); + } + expect(outcomes).toEqual(Object.fromEntries(Object.keys(operations).map((name) => [name, "closed"]))); + expect(() => harness.resume()).toThrow("closed"); + held.release(); + await blocking; + expect((await queuedInspect).scheduling).toBe("closing"); + await closing; + }); + + it("completes reads queued at the seal and rejects queued waits and acquisitions that would follow commits", async () => { + const Notes = defineDoc<{ text: string }>({ + kind: "test.queued-at-seal", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + }); + const Absent = defineDoc<{ text: string }>({ + kind: "test.queued-absent", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + }); + const Pending = oneStep("test.queued-pending"); + const storage = new ControlledStorage(); + const { harness } = await openTasks(storage, []); + const root = await harness.root(context); + await harness.commit((tx) => tx.doc(Notes).then(() => {}), context); + const taskId = await start(root, Pending); + const write = await root.submit({ type: "write", entry: { kind: "note" } }, context); + await write.wait(context); + + const held = storage.holdCommits(); + const blocking = harness.commit((tx) => tx.appendEntry(root.id, { kind: "blocker" }).then(() => {}), context); + await held.entered; + const settle = (promise: Promise): Promise => + promise.then( + (value) => (value === undefined ? "undefined" : "resolved"), + (error: unknown) => (/closed/.test(String(error)) ? "closed" : String(error)), + ); + const queued = { + context: settle(root.context(context)), + snapshot: settle(harness.snapshot(Notes, context)), + settledWait: settle(write.wait(context)), + absentWatch: settle(harness.watchDoc(Absent, context)), + absentState: settle(harness.documentState(Absent, context)), + waitForTask: settle(harness.waitForTask(taskId, context)), + documentState: settle(harness.documentState(Notes, context)), + watchDoc: settle(harness.watchDoc(Notes, context)), + viewState: settle(root.viewState(context)), + watch: settle(root.watch(context)), + taskGraph: settle(harness.taskGraph(context)), + }; + const closing = harness.close(context); + held.release(); + await blocking; + const outcomes: Record = {}; + for (const [name, outcome] of Object.entries(queued)) outcomes[name] = await outcome; + expect(outcomes).toEqual({ + context: "resolved", + snapshot: "resolved", + settledWait: "resolved", + absentWatch: "undefined", + absentState: "undefined", + waitForTask: "closed", + documentState: "closed", + watchDoc: "closed", + viewState: "closed", + watch: "closed", + taskGraph: "closed", + }); + await closing; + }); +}); + +describe("scheduling on a paused Harness", () => { + /** Open a paused Harness with one background task pending and a queued write submission. */ + async function paused() { + let ran = false; + const Marker = oneStep("test.marker", async () => { + ran = true; + }); + const { harness } = await openTasks(new MemoryStorage(), [Marker]); + const root = await harness.root(context); + // Background, so idle waits and conversation abort leave it alone. + const id = await start(root, Marker, true); + const submissionId = await harness.commit( + async (tx) => (await tx.createSubmission({ conversationId: root.id, type: "write", status: "queued" })).id, + context, + ); + return { harness, root, id, submissionId, ran: () => ran }; + } + + it("never schedules from a read-only viewer", async () => { + const { harness, root, id, submissionId, ran } = await paused(); + const Notes = defineDoc<{ text: string }>({ + kind: "test.viewer-notes", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + }); + await harness.commit((tx) => tx.doc(Notes).then(() => {}), context); + await harness.inspect(context); + await harness.getTask(id, context); + await harness.usage(context); + await harness.conversation(root.id, context); + await harness.snapshot(Notes, context); + (await harness.documentState(Notes, context))!.dispose(); + await (await harness.watchDoc(Notes, context))!.stop(); + const submission = (await harness.submission(submissionId, context))!; + await submission.status(context); + await root.agent(context); + await root.context(context); + await root.entries({}, 10, undefined, context); + (await root.viewState(context)).dispose(); + await (await root.watch(context)).stop(); + (await harness.taskGraph(context)).dispose(); + await (await harness.watchTaskGraph(context)).stop(); + await flush(); + await flush(); + expect(ran()).toBe(false); + expect((await harness.inspect(context)).scheduling).toBe("paused"); + await harness.close(context); + }); + + it("schedules from every progress call", async () => { + const calls: Record>) => Promise> = { + submit: ({ root }) => root.submit({ type: "write", entry: { kind: "note" } }, context), + compact: ({ root }) => root.compact(undefined, context), + abort: ({ root }) => root.abort(context), + conversationIdle: ({ root }) => root.waitForIdle(context), + submissionWait: async ({ harness, submissionId }) => + (await harness.submission(submissionId, context))!.wait(context), + waitForTask: ({ harness, id }) => harness.waitForTask(id, context), + harnessIdle: ({ harness }) => harness.waitForIdle(context), + }; + for (const [name, call] of Object.entries(calls)) { + const opened = await paused(); + // A queued write's wait settles only on placement; close rejects it. + const pending = call(opened).catch(() => {}); + await eventually(async () => opened.ran()).catch(() => { + throw new Error(`${name} did not enable scheduling`); + }); + await opened.harness.close(context); + await pending; + } + }); +}); + +describe("registry changes before resume", () => { + it("runs what the registry holds at resume: a definition installed or replaced after open", async () => { + const log: string[] = []; + const define = (label: string) => + oneStep("test.late-definition", async () => { + log.push(label); + }); + const storage = new MemoryStorage(); + const first = await openTasks(storage, []); + const root = await first.harness.root(context); + const missing = await start(root, define("unused")); + expect((await first.harness.inspect(context)).tasks).toMatchObject([ + { state: { kind: "blocked", reason: "missing_task" } }, + ]); + const installed = addTask(first.registry, define("v1")); + expect((await first.harness.inspect(context)).tasks).toMatchObject([{ state: { kind: "ready" } }]); + // The same extension name replaces the definition in place, still before resume. + installed.dispose(); + addTask(first.registry, define("v2")); + first.harness.resume(); + await first.harness.waitForTask(missing, context); + expect(log).toEqual(["v2"]); + await first.harness.close(context); + }); +}); diff --git a/packages/durable/test/harness-task-graph.test.ts b/packages/durable/test/harness-task-graph.test.ts new file mode 100644 index 000000000..3beb45e01 --- /dev/null +++ b/packages/durable/test/harness-task-graph.test.ts @@ -0,0 +1,358 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { JsonValue } from "@earendil-works/chord"; +import { applyImmutable, type Op } from "@earendil-works/chord/delta"; +import { + type ConversationId, + defineTask, + type EntryId, + MemoryStorage, + type TaskGraph, + type TaskId, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { ControlledStorage, context, flush } from "./session-support.ts"; +import { aborted, completed, deferred, eventually, openTasks } from "./task-support.ts"; + +type ParentState = { phase: "spawn" } | { phase: "join"; child: TaskId } | { phase: "finish" }; + +/** + * A parent that creates a child task and a conversation it owns, waits for the child, creates a second child, and + * completes while that child is still live, so its outcome is held as `completing`. + */ +function family(gates: { readonly child: Promise; readonly late: Promise }) { + const Child = defineTask<{ late: boolean }, { phase: "work" }, null>({ + name: "test.graph-child", + version: 1, + initial: () => ({ phase: "work" }), + phases: { + work: async (task, runtime, ctx) => { + await (task.input.late ? gates.late : gates.child); + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + const Parent = defineTask({ + name: "test.graph-parent", + version: 1, + initial: () => ({ phase: "spawn" }), + phases: { + spawn: async (task, runtime, ctx) => { + const ownership = { kind: "task", taskId: task.id } as const; + // Two conversations in a commit that leaves the parent's record unchanged. + await runtime.commit(async (tx) => { + await tx.createConversation({ ownership }); + await tx.createConversation({ ownership }); + return undefined; + }, ctx); + await runtime.commit(async (tx) => { + const child = await tx.createTask(Child, { late: false }, { ownership }); + return { status: "waiting", checkpoint: { phase: "join", child }, on: [child], policy: "allSettled" }; + }, ctx); + }, + join: async (task, runtime, ctx) => { + await runtime.commit(async (tx) => { + await tx.createTask(Child, { late: true }, { ownership: { kind: "task", taskId: task.id } }); + return { status: "running", checkpoint: { phase: "finish" } }; + }, ctx); + }, + finish: async (_task, runtime, ctx) => { + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + return { Parent, Child }; +} + +function gatesNever() { + const never = new Promise(() => {}); + return { child: never, late: never }; +} + +/** Node states by task kind, for compact assertions. */ +function statuses(graph: TaskGraph): Record { + return Object.fromEntries( + Object.values(graph.tasks).map((node) => [`${node.kind}#${node.id}`, node.state.status] as const), + ); +} + +describe("task graph view", () => { + it("follows every live task through its statuses, owner edges, and owned conversations", async () => { + const child = deferred(); + const late = deferred(); + const { Parent, Child } = family({ child: child.promise, late: late.promise }); + const { harness } = await openTasks(new MemoryStorage(), [Parent, Child]); + const root = await harness.root(context); + const seen: TaskGraph[] = []; + const observe = async () => { + const opened = await harness.taskGraph(context); + opened.subscribe((value, _context, delivery) => { + if (delivery.kind === "update") seen.push(value); + }); + return opened; + }; + let graph = await observe(); + expect(graph.value).toEqual({ tasks: {} }); + /** The advanced value equals a fresh build from Storage once the last observer left. */ + const rebuild = async () => { + const advanced = graph.value; + graph.dispose(); + graph = await observe(); + expect(graph.value).not.toBe(advanced); + expect(graph.value).toEqual(advanced); + }; + + const parent = await root.commit( + (tx) => tx.createTask(Parent, null, { ownership: { kind: "conversation" } }), + context, + ); + await flush(); + expect(graph.value.tasks[String(parent)]).toEqual({ + id: parent, + kind: "test.graph-parent", + conversationId: root.id, + background: false, + abortRequested: false, + state: { status: "pending", phase: "spawn" }, + conversations: [], + }); + + harness.resume(); + await eventually( + () => + Object.keys(graph.value.tasks).length === 2 && + statuses(graph.value)[`test.graph-child#${childId()}`] === "running", + ); + function childId(): string { + return ( + Object.values(graph.value.tasks) + .find((node) => node.kind === "test.graph-child") + ?.id.toString() ?? "" + ); + } + const first = Number(childId()) as TaskId; + const parentNode = graph.value.tasks[String(parent)]!; + expect(parentNode.state).toEqual({ status: "waiting", phase: "join", on: [first], policy: "allSettled" }); + expect(parentNode.conversations).toHaveLength(2); + expect([...parentNode.conversations].sort((a, b) => a - b)).toEqual(parentNode.conversations); + const owned = parentNode.conversations[0]!; + expect(graph.value.tasks[String(first)]).toMatchObject({ owner: parent, conversationId: root.id }); + expect((await harness.conversation(owned, context)) !== undefined).toBe(true); + await rebuild(); + + child.resolve(); + // The parent completes while the late child lives: its outcome is held. + await eventually(() => graph.value.tasks[String(parent)]?.state.status === "completing"); + expect(graph.value.tasks[String(parent)]!.state).toEqual({ status: "completing", outcome: "completed" }); + expect(graph.value.tasks[String(first)]).toBeUndefined(); + // Owned conversations stay listed while the owner lives. + expect(graph.value.tasks[String(parent)]!.conversations).toEqual(parentNode.conversations); + await rebuild(); + + late.resolve(); + await harness.waitForTask(parent, context); + await flush(); + expect(graph.value).toEqual({ tasks: {} }); + // Every revision is one commit that changed a node; none repeats its predecessor. + for (let index = 1; index < seen.length; index++) expect(seen[index]).not.toEqual(seen[index - 1]); + // The commit that created the two conversations published one revision setting them. + expect(seen.some((value) => value.tasks[String(parent)]?.conversations.length === 2)).toBe(true); + graph.dispose(); + await harness.close(context); + }); + + it("builds from committed tasks, shows surviving tasks as pending after reopen, and marks aborts", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-graph-")); + const path = join(directory, "session.sqlite"); + const gate = deferred(); + const Work = defineTask({ + name: "test.graph-work", + version: 1, + initial: () => ({ phase: "work" }), + phases: { + work: async (_task, runtime, ctx) => { + await Promise.race([gate.promise, aborted(runtime.signal)]); + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async (_task, runtime, ctx) => { + await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted", reason: "test" } }), ctx); + }, + }); + const first = await openTasks(await openNodeSqliteStorage(path), [Work]); + const firstRoot = await first.harness.root(context); + const [foreground, background, owned] = await firstRoot.commit(async (tx) => { + const foreground = await tx.createTask(Work, null, { ownership: { kind: "conversation" } }); + const background = await tx.createTask(Work, null, { ownership: { kind: "conversation" }, background: true }); + const owned = await tx.createConversation({ ownership: { kind: "task", taskId: foreground } }); + return [foreground, background, owned.id] as const; + }, context); + first.harness.resume(); + await eventually(async () => + (await first.harness.inspect(context)).tasks.every((task) => task.state.kind === "running"), + ); + const running = await first.harness.taskGraph(context); + expect(statuses(running.value)).toEqual({ + [`test.graph-work#${foreground}`]: "running", + [`test.graph-work#${background}`]: "running", + }); + await first.harness.close(context); + + // Acquired after reopen: built from the committed records and owner edges; open reconciled running to pending. + const { harness } = await openTasks(await openNodeSqliteStorage(path), [Work]); + const watch = await harness.watchTaskGraph(context); + expect(statuses(watch.value)).toEqual({ + [`test.graph-work#${foreground}`]: "pending", + [`test.graph-work#${background}`]: "pending", + }); + expect(watch.value.tasks[String(foreground)]!.conversations).toEqual([owned as ConversationId]); + expect(watch.value.tasks[String(background)]!.background).toBe(true); + + // Exact frames: replaying their operations from the acquisition revision gives each delivered value. + let replica: JsonValue = watch.value as unknown as JsonValue; + const frames: (readonly Op[])[] = []; + watch.start(async (value, ops) => { + replica = applyImmutable(replica, ops); + expect(replica).toEqual(value); + frames.push(ops); + }); + harness.resume(); + await eventually(() => watch.value.tasks[String(background)]?.state.status === "running"); + expect(await harness.abortTask(background, context)).toBe("marked"); + await harness.waitForTask(background, context); + await flush(); + expect(frames).toContainEqual([ + ["s", ["tasks", String(background)], expect.objectContaining({ abortRequested: true })], + ]); + expect(frames.at(-1)).toEqual([["d", ["tasks", String(background)]]]); + expect(Object.keys((replica as unknown as TaskGraph).tasks)).toEqual([String(foreground)]); + await watch.stop(); + gate.resolve(); + await harness.waitForTask(foreground, context); + await harness.close(context); + await rm(directory, { recursive: true, force: true }); + }); + + it("lists owned conversations in ID order whatever order one commit creates them in", async () => { + const gate = deferred(); + const created: number[] = []; + const Spawner = defineTask<{ at: EntryId }, { phase: "spawn" }, null>({ + name: "test.graph-spawner", + version: 1, + initial: () => ({ phase: "spawn" }), + phases: { + spawn: async (task, runtime, ctx) => { + await runtime.commit(async (tx) => { + const ownership = { kind: "task", taskId: task.id } as const; + const [forked, fresh] = await Promise.all([ + tx.forkConversation(runtime.conversationId, task.input.at, { ownership }), + tx.createConversation({ ownership }), + ]); + created.push(forked.id, fresh.id); + return undefined; + }, ctx); + await gate.promise; + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + const { harness } = await openTasks(new MemoryStorage(), [Spawner]); + const root = await harness.root(context); + const graph = await harness.taskGraph(context); + const id = await root.commit(async (tx) => { + const entry = await tx.appendEntry(root.id, { kind: "note" }); + return tx.createTask(Spawner, { at: entry.id }, { ownership: { kind: "conversation" } }); + }, context); + harness.resume(); + await eventually(() => (graph.value.tasks[String(id)]?.conversations.length ?? 0) === 2); + const advanced = graph.value; + expect(advanced.tasks[String(id)]!.conversations).toEqual([...created].sort((a, b) => a - b)); + graph.dispose(); + const rebuilt = await harness.taskGraph(context); + expect(rebuilt.value).toEqual(advanced); + rebuilt.dispose(); + gate.resolve(); + await harness.waitForTask(id, context); + await harness.close(context); + }); + + it("registers nothing for an acquisition cancelled while it waits for the line", async () => { + const storage = new ControlledStorage(); + const { harness } = await openTasks(storage, []); + const root = await harness.root(context); + await root.commit( + (tx) => tx.createTask(family(gatesNever()).Child, { late: false }, { ownership: { kind: "conversation" } }), + context, + ); + const held = storage.holdCommits(); + const blocking = harness.commit((tx) => tx.appendEntry(root.id, { kind: "blocker" }).then(() => {}), context); + await held.entered; + const controller = new AbortController(); + const cancelled = harness.watchTaskGraph({ ...context, abortSignal: controller.signal }); + controller.abort(new Error("cancelled")); + held.release(); + await blocking; + await expect(cancelled).rejects.toThrow("cancelled"); + // No observer kept the mount: each new observer builds a new revision. + const first = await harness.taskGraph(context); + const value = first.value; + first.dispose(); + const second = await harness.taskGraph(context); + expect(second.value).not.toBe(value); + expect(second.value).toEqual(value); + second.dispose(); + await harness.close(context); + }); + + it("publishes no revision for a commit that changes no node, and shares one mount between observers", async () => { + const gate = deferred(); + const reached = deferred(); + const Memo = defineTask({ + name: "test.graph-memo", + version: 1, + initial: () => ({ phase: "work" }), + phases: { + work: async (_task, runtime, ctx) => { + await runtime.memo("seen", true, ctx); + reached.resolve(); + await gate.promise; + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + const { harness } = await openTasks(new MemoryStorage(), [Memo]); + const root = await harness.root(context); + const id = await root.commit((tx) => tx.createTask(Memo, null, { ownership: { kind: "conversation" } }), context); + const first = await harness.taskGraph(context); + const second = await harness.taskGraph(context); + expect(second.value).toBe(first.value); + const updates: string[] = []; + first.subscribe((value, _context, delivery) => { + if (delivery.kind === "update") updates.push(value.tasks[String(id)]?.state.status ?? "gone"); + }); + harness.resume(); + await reached.promise; + await flush(); + // Reservation changed the node; the memo commit did not. + expect(updates).toEqual(["running"]); + gate.resolve(); + await harness.waitForTask(id, context); + await flush(); + expect(updates).toEqual(["running", "gone"]); + const last = first.value; + first.dispose(); + second.dispose(); + // No observer is left, so the mount was dropped: a new observer builds a new revision. + const rebuilt = await harness.taskGraph(context); + expect(rebuilt.value).toEqual(last); + expect(rebuilt.value).not.toBe(last); + rebuilt.dispose(); + await harness.close(context); + }); +}); diff --git a/packages/durable/test/harness-tasks-recovery.test.ts b/packages/durable/test/harness-tasks-recovery.test.ts index 574e5679e..986103dec 100644 --- a/packages/durable/test/harness-tasks-recovery.test.ts +++ b/packages/durable/test/harness-tasks-recovery.test.ts @@ -636,6 +636,46 @@ describe("definition handover", () => { await harness.close(context); }); + it("keeps the memos across a handover", async () => { + type Memoed = { phase: "a" } | { phase: "b" }; + const log: string[] = []; + const gate = deferred(); + const reached = deferred(); + const memoTask = (label: string) => + defineTask({ + name: "test.handover-memo", + version: 1, + initial: () => ({ phase: "a" }), + phases: { + a: async (_task, runtime, ctx) => { + await runtime.memo("picked", label, ctx); + reached.resolve(); + await gate.promise; + await runtime.commit(() => ({ status: "running", checkpoint: { phase: "b" } }), ctx); + }, + b: async (_task, runtime, ctx) => { + // The candidate loses to the memo the old definition stored. + const picked = await runtime.memo("picked", label, ctx); + log.push(`${label}:b ${picked}`); + await runtime.commit(() => completed(picked), ctx); + }, + }, + abort: async () => {}, + }); + const registry = createRegistry(); + addTask(registry, memoTask("old")); + const { harness } = await openTasks(new MemoryStorage(), [], { registry }); + const id = await createIn(harness, memoTask("old")); + harness.resume(); + await reached.promise; + addTask(registry, memoTask("new")); + gate.resolve(); + const receipt = await harness.waitForTask(id, context); + expect(receipt.state.outcome).toEqual({ status: "completed", result: "old" }); + expect(log).toEqual(["new:b old"]); + await harness.close(context); + }); + it("hands over to a newer version with a migration", async () => { const log: string[] = []; const gate = deferred(); diff --git a/packages/durable/test/harness-tasks.test.ts b/packages/durable/test/harness-tasks.test.ts index dd2f5947b..8cf843cb8 100644 --- a/packages/durable/test/harness-tasks.test.ts +++ b/packages/durable/test/harness-tasks.test.ts @@ -521,6 +521,8 @@ describe("task runtime", () => { await expect(runtime.watchDoc(Notes, context)).rejects.toThrow("invocation has ended"); // The handler's own context is cancelled by now; the ended invocation still wins. await expect(runtime.agent(handlerContext!)).rejects.toThrow("invocation has ended"); + expect(() => runtime.now()).toThrow("invocation has ended"); + expect(() => runtime.report(new Error("late"))).toThrow("invocation has ended"); await harness.close(context); }); diff --git a/packages/durable/test/session-checkpoints-migrations.test.ts b/packages/durable/test/session-checkpoints-migrations.test.ts index c13769381..375287979 100644 --- a/packages/durable/test/session-checkpoints-migrations.test.ts +++ b/packages/durable/test/session-checkpoints-migrations.test.ts @@ -395,8 +395,20 @@ describe("Session document migrations", () => { const { session, storage, publications } = openTestSession(); await session.commit((tx) => tx.doc(Old).then(() => undefined), context); await session.unloadDocuments(); + // An observer of the older shape. + const watch = (await session.watchDoc(Old, context))!; + const frames: unknown[] = []; + watch.start(async (value, ops) => { + frames.push({ value, ops }); + }); expect(await session.snapshot(Current, context)).toEqual({ count: 4 }); const snapshot = await session.snapshot(Current, context); + // An observer of the new shape: the migration changes nothing it sees. + const current = (await session.watchDoc(Current, context))!; + const currentFrames: unknown[] = []; + current.start(async (value) => { + currentFrames.push(value); + }); await session.commit((tx) => tx.doc(Current).then(() => undefined), context); await flush(); @@ -405,7 +417,12 @@ describe("Session document migrations", () => { content: { kind: "base", version: 3, value: { count: 4 } }, }); expect(checkpoints).toBe(0); - expect(documentChanges(publications.at(-1)!)).toHaveLength(0); + // The migration-only base is published, so the older-shape watch receives the new value as a root replacement. + expect(documentChanges(publications.at(-1)!)).toMatchObject([{ version: 3, value: { count: 4 }, ops: [] }]); + expect(frames).toEqual([{ value: { count: 4 }, ops: [["r", { count: 4 }]] }]); + expect(currentFrames).toEqual([]); + await watch.stop(); + await current.stop(); expect(await session.snapshot(Current, context)).toBe(snapshot); await session.commit(async (tx) => { diff --git a/packages/durable/test/session-definitions.test.ts b/packages/durable/test/session-definitions.test.ts index 789be0ada..ddc791187 100644 --- a/packages/durable/test/session-definitions.test.ts +++ b/packages/durable/test/session-definitions.test.ts @@ -2,13 +2,19 @@ import type { Draft } from "@earendil-works/chord"; import { type ConversationId, createSession, + type DocumentState, + type DocumentWatch, defineDoc, defineDocFamily, + defineEntry, + type EntryId, + type EntryRecord, MemoryStorage, ROOT_CONVERSATION_ID, type Session, type TaskId, type Tx, + type TypedEntry, } from "@earendil-works/pi-durable"; import { describe, expect, expectTypeOf, it } from "vitest"; import { idFromNumber } from "../src/ids.ts"; @@ -116,4 +122,56 @@ describe("document definitions", () => { await session.snapshot(SessionFamily, "k", 1, context).catch(() => undefined); await session.close(context); }); + + it("types historical reads, states, watches, and typed entries", () => { + const Note = defineEntry<{ text: string }>("t.note"); + const Marker = defineEntry("t.marker"); + type Rewound = Promise | undefined>; + type StateOf = Promise | undefined>; + type WatchOf = Promise | undefined>; + const check = async (session: Session, tx: Tx, conversationId: ConversationId, taskId: TaskId, at: EntryId) => { + expectTypeOf(session.snapshotAsOf(RewindableDoc, conversationId, at, context)).toEqualTypeOf(); + expectTypeOf( + session.snapshotAsOf(ConversationFamily, conversationId, "k", at, context), + ).toEqualTypeOf(); + // @ts-expect-error latest conversation documents keep no history + void session.snapshotAsOf(LatestDoc, conversationId, at, context); + // @ts-expect-error Session documents keep no history + void session.snapshotAsOf(SessionDoc, conversationId, at, context); + // @ts-expect-error task documents keep no history + void session.snapshotAsOf(TaskDoc, taskId, at, context); + + expectTypeOf(session.documentState(SessionDoc, context)).toEqualTypeOf(); + expectTypeOf(session.documentState(LatestDoc, conversationId, context)).toEqualTypeOf(); + expectTypeOf(session.documentState(TaskDoc, taskId, context)).toEqualTypeOf(); + expectTypeOf(session.documentState(SessionFamily, "k", context)).toEqualTypeOf(); + expectTypeOf(session.documentState(ConversationFamily, conversationId, "k", context)).toEqualTypeOf(); + expectTypeOf(session.documentState(TaskFamily, taskId, "k", context)).toEqualTypeOf(); + expectTypeOf(session.watchDoc(SessionDoc, context)).toEqualTypeOf(); + expectTypeOf(session.watchDoc(LatestDoc, conversationId, context)).toEqualTypeOf(); + expectTypeOf(session.watchDoc(TaskDoc, taskId, context)).toEqualTypeOf(); + expectTypeOf(session.watchDoc(SessionFamily, "k", context)).toEqualTypeOf(); + expectTypeOf(session.watchDoc(ConversationFamily, conversationId, "k", context)).toEqualTypeOf(); + expectTypeOf(session.watchDoc(TaskFamily, taskId, "k", context)).toEqualTypeOf(); + // @ts-expect-error a watch of a Session document takes no owner + void session.watchDoc(SessionDoc, conversationId, context); + + expectTypeOf(await tx.entry(at)).toEqualTypeOf(); + expectTypeOf(await tx.entry(Note, at)).toEqualTypeOf | undefined>(); + expectTypeOf(await tx.appendEntry(conversationId, { kind: "t.raw" })).toEqualTypeOf(); + const note = await tx.appendEntry(Note, conversationId, { data: { text: "x" } }); + expectTypeOf(note).toEqualTypeOf>(); + expectTypeOf(note.data.text).toEqualTypeOf(); + expectTypeOf(await tx.appendEntry(Marker, conversationId, {})).toEqualTypeOf>(); + // @ts-expect-error a data entry requires its data + await tx.appendEntry(Note, conversationId, {}); + // @ts-expect-error data is typed by the token + await tx.appendEntry(Note, conversationId, { data: { text: 1 } }); + // @ts-expect-error the token supplies the kind + await tx.appendEntry(Note, conversationId, { kind: "t.note", data: { text: "x" } }); + // @ts-expect-error an entry kind without data takes none + await tx.appendEntry(Marker, conversationId, { data: 1 }); + }; + expect(check).toBeTypeOf("function"); + }); }); diff --git a/packages/durable/test/spec-usage.test.ts b/packages/durable/test/spec-usage.test.ts new file mode 100644 index 000000000..70b199b79 --- /dev/null +++ b/packages/durable/test/spec-usage.test.ts @@ -0,0 +1,380 @@ +/** + * The usage examples of docs/pico-v5.md, compile-checked. Each block is copied as written apart from formatting, with + * the names the spec leaves to the application declared below. Keep the two in sync; `test/examples/` runs the same + * patterns end to end. + */ +import type { Context, Draft } from "@earendil-works/chord"; +import { type AssistantMessage, type Models, type ToolCall, Type } from "@earendil-works/pi-ai"; +import { + type ConversationId, + configure, + createRegistry, + defineDoc, + defineExtension, + defineTask, + defineTool, + type EntryId, + GenerationTask, + Harness, + type HarnessSettings, + type HookApi, + hook, + LiveDoc, + type LiveState, + type Session, + type Storage, + section, + type TaskId, + type ToolExecutionApi, + ToolTask, + type Tx, + type UserInput, + wrapTool, +} from "@earendil-works/pi-durable"; +import { expect, expectTypeOf, it } from "vitest"; +import type { ExecutionEnv } from "../src/env/index.ts"; +import { CodingTools, createBashTool, createEditTool, createReadTool } from "../src/tools/index.ts"; + +// ─── Names the spec leaves to the application ──────────────────────────────── + +declare const context: Context; +declare const storage: Storage; +declare const models: Models; +declare const session: Session; +declare const conversationId: ConversationId; +declare const message: { readonly kind: string }; +declare const haiku: { readonly provider: string; readonly modelId: string }; +declare const sonnet: { readonly provider: string; readonly modelId: string }; +declare const worktree: string; +declare const name: string; +declare function localEnv(cwd: string): ExecutionEnv; +declare const containers: { env(image: string, cwd: string, context: Context): Promise }; +declare function renderAgentsMd(): string | undefined; +declare function renderSkills(): string | undefined; +declare function isDangerous(call: ToolCall): boolean; +declare function writes(call: ToolCall): boolean; +declare function requestSecondPass( + answer: AssistantMessage, + api: HookApi, + context: Context, +): Promise<{ readonly continue: UserInput } | undefined>; +declare const metrics: { record(name: string, ms: number): void }; +declare function answerText(api: ToolExecutionApi, entry: EntryId, context: Context): Promise; +declare const payments: { + charge(key: string): Promise<{ id: string }>; + cancel(checkpoint: { phase: string }): Promise; +}; +declare function newKey(): string; +declare function receiptEntry(receipt: { id: string }): { kind: string; data: { id: string } }; +interface SettingsManager { + get(key: "timeoutMs"): number; + get(key: "autoCompact"): boolean; + set(key: "autoCompact", value: boolean): void; +} +declare const manager: SettingsManager; +declare const Anchor: ReturnType>; +declare const Reporter: ReturnType>; +declare const subagentTool: ReturnType; + +/** Never called: the declared names above have no values. */ +function examples() { + const readTool = createReadTool(); + const editTool = createEditTool(); + const bashTool = createBashTool(); + + // ─── Section 7.1: extensions and host setup ────────────────────────────────── + + const ContextFiles = defineExtension({ name: "context-files", sections: [section("agents-md", renderAgentsMd)] }); + const Skills = defineExtension({ name: "skills", sections: [section("skills", renderSkills)] }); + const SkillsV2 = defineExtension({ name: "skills", sections: [section("skills", renderSkills)] }); + const Coding = defineExtension({ + name: "coding", + sections: [ + section("preamble", () => "You are an expert coding assistant.", { tag: false }), + // The environment the host built for this conversation, in the conversation's directory. + section("cwd", (input) => input.env && `Working directory: ${input.env.cwd}`), + ], + }); + const Permissions = defineExtension({ + name: "permissions", + hooks: [ + hook(ToolTask, { beforeTool: async (call) => (isDangerous(call) ? { block: "Needs approval" } : undefined) }), + ], + }); + // A role and a review loop, for conversations that select it. + const Reviewer = defineExtension({ + name: "reviewer", + sections: [section("role", () => "You review diffs. Report problems as a list. Never edit files.")], + hooks: [hook(GenerationTask, { onYield: requestSecondPass })], + }); + + const Timing = defineExtension({ + name: "timing", + wraps: [ + wrapTool(bashTool, (tool) => ({ + ...tool, + execute: async (args, api, context) => { + const start = Date.now(); + try { + return await tool.execute(args, api, context); + } finally { + metrics.record("bash", Date.now() - start); + } + }, + })), + ], + }); + // A bash inside a Python virtualenv for one conversation: it replaces CodingTools' bash in place, + // and Timing, if selected, wraps it. + const Venv = defineExtension({ + name: "venv", + tools: [createBashTool({ commandPrefix: "source .venv/bin/activate" })], + }); + + // ─── Section 7.2: hooks reading extension state ────────────────────────────── + + const PlanModeDoc = defineDoc<{ enabled: boolean }>({ + kind: "app.plan-mode", + version: 1, + scope: "conversation", + history: "latest", + fork: "current", + initial: () => ({ enabled: false }), + }); + const PlanMode = defineExtension({ + name: "plan-mode", + hooks: [ + hook(ToolTask, { + // An absent document means plan mode is off. + beforeTool: async (call, api, context) => + (await api.snapshot(PlanModeDoc, api.conversationId, context))?.enabled && writes(call) + ? { block: "Plan mode: read-only" } + : undefined, + }), + ], + }); + + // ─── Section 7.3: subagents ────────────────────────────────────────────────── + + const Subagent = defineExtension({ + name: "subagent", + tools: [ + defineTool({ + name: "subagent", + description: "Delegate a self-contained task to a subagent and get its answer back.", + parameters: Type.Object({ task: Type.String() }), + replay: "safe", + execute: async (args, api, context) => { + const { task } = args; + const child = await api.commit(async (tx) => { + const existing = (await tx.scanConversations({ ownerTaskId: api.taskId }, 1)).items[0]; + if (existing !== undefined) return existing.id; + // Starts as a copy of this conversation's agent: model, thinking level, cwd, extensions, tools. + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + // Without this extension, the child is not offered this tool. + await configure(tx, created.id, { extensions: { remove: [Subagent] } }); + return created.id; + }, context); + await api.details({ conversationId: child }, context); + const handle = (await api.conversation(child, context))!; + const request = { type: "input", content: task, requestId: `subagent:${api.taskId}` } as const; + const settled = await (await handle.submit(request, context)).wait(context); + if (settled.status !== "done" || settled.type !== "input") + throw new Error(`Subagent failed: ${settled.status}`); + return { content: [{ type: "text", text: await answerText(api, settled.answer, context) }] }; + }, + }), + ], + }); + + const SubagentTools = defineExtension({ name: "subagent-tools", tasks: [Anchor, Reporter], tools: [subagentTool] }); + + // ─── Section 7.4 ───────────────────────────────────────────────────────────── + + const Chat = defineExtension({ + name: "chat", + sections: [section("preamble", () => "You are a helpful assistant.", { tag: false })], + }); + + // ─── Section 5.1: a task with intent, effect, outcome ──────────────────────── + + type Charge = { phase: "prepare" } | { phase: "charge"; key: string }; + const Payment = defineTask({ + name: "app.payment", + version: 1, + initial: () => ({ phase: "prepare" }), + phases: { + // Intent, effect, outcome. + prepare: async (_task, runtime, context) => { + await runtime.commit( + () => ({ status: "running", checkpoint: { phase: "charge", key: newKey() } }), + context, + ); + }, + charge: async (task, runtime, context) => { + const receipt = await payments.charge(task.state.checkpoint.key); // idempotent by key + await runtime.commit(async (tx, current) => { + const entry = await tx.appendEntry(current.conversationId, receiptEntry(receipt)); + return { status: "terminal", outcome: { status: "completed", result: { entryId: entry.id } } }; + }, context); + }, + }, + // The abort handler decides the outcome; returning without one faults the task. + abort: async (task, runtime, context) => { + await payments.cancel(task.state.checkpoint); + await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted", reason: "user" } }), context); + }, + }); + + const Follow = defineTask({ + name: "app.follow", + version: 1, + initial: () => ({ phase: "follow" }), + phases: { follow: async () => {} }, + abort: async () => {}, + }); + + // ─── Sequences ─────────────────────────────────────────────────────────────── + + const sequences = { + // Section 2.2: a tool's commit creates a configured child. + childInToolCommit: async (tx: Tx, api: { readonly taskId: TaskId }) => { + // In a tool's commit. The child starts as a copy of this conversation's agent: model, extensions, tools, cwd. + const child = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + // A cheaper model, only the read tool, and its own worktree; everything else stays as copied. + await configure(tx, child.id, { model: haiku, tools: [readTool], cwd: worktree }); + }, + + // Section 2.2: settings read live through getters. + liveSettings: () => { + // The user's settings, read live through getters. + class UserSettings implements HarnessSettings { + readonly manager: SettingsManager; + constructor(manager: SettingsManager) { + this.manager = manager; + } + get stream() { + return { timeoutMs: this.manager.get("timeoutMs") }; + } + get compaction() { + return { enabled: this.manager.get("autoCompact") }; + } + } + manager.set("autoCompact", false); // no Session write; every conversation follows at its next threshold check + return new UserSettings(manager); + }, + + // Section 2.2: an environment per conversation. + containerEnv: async () => { + const registry = createRegistry(); + // Absent: the conversation runs locally. Only conversations with this document run in a container. + // Subagents do not copy it: their creator writes it too when they should run in the container. + const ContainerDoc = defineDoc<{ image: string }>({ + kind: "app.container", + version: 1, + scope: "conversation", + history: "latest", + fork: "current", + initial: () => ({ image: "node:22" }), + }); + const harness = await Harness.open( + storage, + { + models, + registry, + env: async ({ conversationId, cwd, read }, context) => { + const container = await read.snapshot(ContainerDoc, conversationId, context); + return container !== undefined + ? containers.env(container.image, cwd ?? "/work", context) + : localEnv(cwd ?? process.cwd()); // cached NodeExecutionEnv per directory + }, + }, + context, + ); + return harness; + }, + + // Sections 2.2 and 7.1: host setup, tool filters, extension selection, plan mode, and reload. + host: async () => { + const registry = createRegistry(); + for (const extension of [CodingTools, Coding, ContextFiles, Skills, Permissions, Reviewer]) { + registry.install(extension); + } + const harness = await Harness.open( + storage, + { + models, + registry, + // Reviewer is installed but not selected by default: only conversations that select it get its role and hooks. + settings: { extensions: [CodingTools, Coding, ContextFiles, Skills, Permissions] }, + env: ({ cwd }) => localEnv(cwd ?? process.cwd()), // cached NodeExecutionEnv per directory + }, + context, + ); + // The conversation remembers its model and directory; a restart elsewhere keeps both. + const root = await harness.root(context, { agent: { model: sonnet, cwd: process.cwd() } }); + + await root.configure({ tools: { remove: [editTool] } }, context); + await root.configure({ tools: { remove: [bashTool] } }, context); // edit is offered again + await root.configure({ tools: null }, context); // every tool of the selected extensions again + + registry.install(Timing); + registry.install(Venv); // installed, but not in the default selection + const conversation = root; + await conversation.configure({ extensions: { add: [Venv] } }, context); + + registry.install(PlanMode); + // PlanMode is in the default selection; /plan toggles this conversation's state. + await root.commit(async (tx) => { + (await tx.doc(PlanModeDoc, root.id)).enabled = true; + }, context); + + registry.install(Subagent); + registry.install(Chat); + registry.install(SkillsV2); // same name: conversations selecting skills render v2 at their next request + registry.uninstall(Skills); // selecting conversations get a system delta removing its section; nothing is rewritten + // Restart: the host installs its extensions again; stored names resolve against them. + return harness; + }, + + // Section 7.3: a named subagent's spawn commit. + spawn: async (tx: Tx) => { + // In subagentTool's spawn commit, after the name checks: + const anchor = await tx.createTask(Anchor, null, { ownership: { kind: "conversation" }, background: true }); + // Owned by a task of the parent: starts as a copy of the parent's agent. + const child = await tx.createConversation({ ownership: { kind: "task", taskId: anchor } }); + await configure(tx, child.id, { + extensions: { remove: [SubagentTools] }, + instructions: `You are the subagent "${name}". Answer the main agent's requests.`, + }); + }, + + // Section 4: table reads before the first table write; documents stay usable. + tableRules: async () => { + await session.commit(async (tx) => { + const conversation = await tx.conversation(conversationId); // table read + const live = await tx.doc(LiveDoc, conversationId); + + await tx.appendEntry(conversationId, message); // first table write + delete live.generation; // document mutation remains valid + await tx.createTask(Follow, {}, { ownership: { kind: "conversation" }, conversationId }); // further table writes are fine + void conversation; + }, context); + }, + + // Section 3.4: drafts are revoked after their commit. + revokedDraft: async () => { + let escaped: Draft | undefined; + await session.commit(async (tx) => { + escaped = await tx.doc(LiveDoc, conversationId); + }, context); + escaped!.generation = undefined; // throws: the draft was revoked + }, + }; + return { sequences, Payment }; +} + +it("compiles the spec's usage examples", () => { + expectTypeOf(examples).returns.toHaveProperty("sequences"); + expect(examples).toBeTypeOf("function"); +}); diff --git a/packages/durable/test/types.test.ts b/packages/durable/test/types.test.ts index 83f900071..8a20bc07f 100644 --- a/packages/durable/test/types.test.ts +++ b/packages/durable/test/types.test.ts @@ -1,3 +1,4 @@ +import type { Context } from "@earendil-works/chord"; import { expect, expectTypeOf, it } from "vitest"; import { idFromNumber, seqFromNumber } from "../src/ids.ts"; import type { @@ -17,6 +18,23 @@ import type { TaskRecord, TaskState, } from "../src/index.ts"; +import { + type AnyTask, + type CompactionResult, + type Conversation, + defineExtension, + defineTask, + type Extension, + type Harness, + type HookApi, + type HooksOf, + hook, + type SettledTask, + type SubmissionDraft, + type TaskRuntime, +} from "../src/index.ts"; + +declare const callContext: Context; const conversationId = idFromNumber(1); const entryId = idFromNumber(2); @@ -234,3 +252,95 @@ it("encodes discriminator-dependent fields", () => { expectTypeOf(compileTimeFailures).toBeFunction(); }); + +it("types submissions, task waits, compaction, and tasks erased into extensions", () => { + const input = { type: "input", content: "hi", whenBusy: "steer" } satisfies SubmissionDraft; + const write = { type: "write", entry: { kind: "note" } } satisfies SubmissionDraft; + expectTypeOf(input.type).toEqualTypeOf<"input">(); + expectTypeOf(write.type).toEqualTypeOf<"write">(); + + // A task with narrowed input, several phases, and custom hooks. + type Hooks = { + beforeStep(step: number, api: HookApi, context: Context): { readonly skip: boolean } | undefined; + }; + type Checkpoint = { phase: "plan"; steps: number } | { phase: "run"; step: number }; + const Stepper = defineTask<{ steps: number }, Checkpoint, { ran: number }, Hooks>({ + name: "test.stepper", + version: 1, + initial: (input) => ({ phase: "plan", steps: input.steps }), + phases: { + plan: async (task, runtime, context) => { + expectTypeOf(task.state.checkpoint.steps).toEqualTypeOf(); + expectTypeOf(task.input.steps).toEqualTypeOf(); + await runtime.commit(() => ({ status: "running", checkpoint: { phase: "run", step: 0 } }), context); + }, + run: async (task, runtime, context) => { + expectTypeOf(task.state.checkpoint.step).toEqualTypeOf(); + await runtime.hooks.each("beforeStep", (handler) => { + expectTypeOf(handler).toEqualTypeOf(); + }); + await runtime.commit( + () => ({ status: "terminal", outcome: { status: "completed", result: { ran: 1 } } }), + context, + ); + }, + }, + abort: async () => {}, + }); + expectTypeOf>().toEqualTypeOf(); + const registration = hook(Stepper, { beforeStep: (step) => (step > 1 ? { skip: true } : undefined) }); + // Erased into an extension, whatever its input, phases, and hooks. + const erased: AnyTask = Stepper; + const extension: Extension = defineExtension({ name: "stepper", tasks: [Stepper], hooks: [registration] }); + void [erased, extension]; + + const typedWaits = async ( + harness: Harness, + conversation: Conversation, + runtime: TaskRuntime, + ) => { + const id = await conversation.commit( + (tx) => tx.createTask(Stepper, { steps: 2 }, { ownership: { kind: "conversation" } }), + callContext, + ); + expectTypeOf(id).toEqualTypeOf>(); + const settled = await harness.waitForTask(id, callContext); + expectTypeOf(settled).toEqualTypeOf>(); + if (settled.state.outcome.status === "completed") { + expectTypeOf(settled.state.outcome.result).toEqualTypeOf<{ ran: number }>(); + } + expectTypeOf(await runtime.waitForTask(id, callContext)).toEqualTypeOf>(); + const compaction = await conversation.compact(undefined, callContext); + expectTypeOf(compaction).toEqualTypeOf>(); + expectTypeOf(await harness.waitForTask(compaction, callContext)).toEqualTypeOf>(); + }; + expectTypeOf(typedWaits).toBeFunction(); + + const compileTimeFailures = (conversation: Conversation) => { + // @ts-expect-error an input submission carries no entry + const inputWithEntry: SubmissionDraft = { type: "input", content: "hi", entry: { kind: "note" } }; + // @ts-expect-error a write submission carries no content + const writeWithContent: SubmissionDraft = { type: "write", entry: { kind: "note" }, content: "hi" }; + // @ts-expect-error a write submission never generates, so it has no busy policy + const writeWhenBusy: SubmissionDraft = { type: "write", entry: { kind: "note" }, whenBusy: "steer" }; + // @ts-expect-error hook handlers are typed by the task's hooks + const wrongHook = hook(Stepper, { beforeStep: (_step: string) => undefined }); + // @ts-expect-error hook names come from the task's hooks + const unknownHook = hook(Stepper, { afterStep: () => undefined }); + const wrongInput = conversation.commit( + // @ts-expect-error task input is typed by the definition + (tx) => tx.createTask(Stepper, { steps: "two" }, { ownership: { kind: "conversation" } }), + callContext, + ); + const missingPhase = defineTask({ + name: "test.missing-phase", + version: 1, + initial: () => ({ phase: "plan", steps: 1 }), + // @ts-expect-error the phase map is exhaustive + phases: { plan: async () => {} }, + abort: async () => {}, + }); + void [inputWithEntry, writeWithContent, writeWhenBusy, wrongHook, unknownHook, wrongInput, missingPhase]; + }; + expectTypeOf(compileTimeFailures).toBeFunction(); +}); diff --git a/scripts/check-entry-graphs.mjs b/scripts/check-entry-graphs.mjs index 7154c5cf5..b52431f29 100644 --- a/scripts/check-entry-graphs.mjs +++ b/scripts/check-entry-graphs.mjs @@ -44,7 +44,7 @@ const BUDGETS = { "packages/durable": { ".": { // The built-in tool task validates arguments with pi-ai's TypeBox-based validation, so TypeBox is allowed. - maxFiles: 59, + maxFiles: 60, forbid: ["packages/ai/src/index.ts", "packages/ai/src/utils/typebox-helpers.ts"], }, },