mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(durable): lifecycle conformance and task graph view (Packages 22, 23)
Package 22: runtime.now() and report() throw after the invocation ended; a migration-only commit is published; a failed Harness.open closes without the caller's context and rethrows the open error; ConversationWatch alias. Tests for close joining non-cooperative code, cancelled close, queued operations at the seal, progress calls and viewers on a paused Harness, the Chord guide, and the spec's usage examples. Package 23: Harness.taskGraph() and watchTaskGraph() mount every live task with its owner edge, committed status, flags, and owned conversations, advanced from commit publications. Example 24 prints the checkout's tree.
This commit is contained in:
Mario Zechner
parent
5b5ccddfac
commit
49683a3647
20 changed files with 2285 additions and 15 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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<Tool extends ToolRegistration> implements Conversation {
|
|||
return this.#host.views.state(this.id, context);
|
||||
}
|
||||
|
||||
watch(context: Context): Promise<WatchHandle<ConversationView>> {
|
||||
watch(context: Context): Promise<ConversationWatch> {
|
||||
return this.#host.views.watch(this.id, context);
|
||||
}
|
||||
}
|
||||
|
|
@ -167,6 +168,7 @@ class HarnessImpl<Tool extends ToolRegistration> extends SessionImpl implements
|
|||
readonly #host: ConversationHost<Tool>;
|
||||
readonly #tasks: TaskScheduler;
|
||||
readonly #submissions: Submissions;
|
||||
readonly #taskGraph: TaskGraphView;
|
||||
#closed = false;
|
||||
|
||||
constructor(storage: Storage, options: HarnessOptions<Tool>, context: Context) {
|
||||
|
|
@ -195,6 +197,7 @@ class HarnessImpl<Tool extends ToolRegistration> 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<Tool extends ToolRegistration> extends SessionImpl implements
|
|||
return total;
|
||||
}
|
||||
|
||||
taskGraph(context: Context): Promise<AttachedReplicatedState<TaskGraph>> {
|
||||
return this.#taskGraph.state(context);
|
||||
}
|
||||
|
||||
watchTaskGraph(context: Context): Promise<TaskGraphWatch> {
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
222
packages/durable/src/harness/task-graph.ts
Normal file
222
packages/durable/src/harness/task-graph.ts
Normal file
|
|
@ -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<JsonValue>["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<Record<string, TaskGraphNode>>;
|
||||
};
|
||||
|
||||
export type TaskGraphWatch = WatchHandle<TaskGraph>;
|
||||
|
||||
type AnyTaskRecord = TaskRecord<JsonValue, JsonValue, JsonValue>;
|
||||
type Observer = CommittedStateSource<TaskGraph> | CommittedWatch<TaskGraph>;
|
||||
type Mount = { value: TaskGraph; readonly observers: Set<Observer> };
|
||||
|
||||
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<AttachedReplicatedState<TaskGraph>> {
|
||||
const { observer, detach } = await this.#attach(
|
||||
(value, release) => new CommittedStateSource<TaskGraph>(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<TaskGraphWatch> {
|
||||
const { observer } = await this.#attach(
|
||||
(value, release) => new CommittedWatch<TaskGraph>(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<O extends Observer>(
|
||||
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<Observer>() };
|
||||
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<TaskGraph> {
|
||||
const tasks: Record<string, TaskGraphNode> = {};
|
||||
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<string, TaskGraphNode | undefined>();
|
||||
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<string, ConversationId[]>();
|
||||
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;
|
||||
}
|
||||
|
|
@ -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<AttachedReplicatedState<ConversationView>>;
|
||||
/** The structural view as a serialized exact-frame watch with bounded pending frames. */
|
||||
watch(context: Context): Promise<WatchHandle<ConversationView>>;
|
||||
watch(context: Context): Promise<ConversationWatch>;
|
||||
}
|
||||
|
||||
// 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<ConversationView>;
|
||||
|
||||
/** 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<void>;
|
||||
/** Session total: every conversation's `pi.usage` summed. */
|
||||
usage(context: Context): Promise<UsageState>;
|
||||
/** Every live task with its owner edge, status, and owned conversations (spec §9.5), as a disposable Chord state. */
|
||||
taskGraph(context: Context): Promise<AttachedReplicatedState<TaskGraph>>;
|
||||
/** The task graph as a serialized exact-frame watch with bounded pending frames. */
|
||||
watchTaskGraph(context: Context): Promise<TaskGraphWatch>;
|
||||
}
|
||||
|
||||
/** What a hook may use: committed reads and the asking task's memos, which hooks and the task share. */
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
394
packages/durable/test/chord-guide.test.ts
Normal file
394
packages/durable/test/chord-guide.test.ts
Normal file
|
|
@ -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<CanvasState>({
|
||||
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<CanvasState | null>;
|
||||
addStroke(stroke: Stroke, context: Context): Promise<void>;
|
||||
}
|
||||
const Canvas = defineService<CanvasService>("app.canvas");
|
||||
|
||||
async function createCanvasFacet(session: Session, context: Context): Promise<Facet> {
|
||||
// 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<void> {
|
||||
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<void>, 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<ReviewState, ReviewInput>({
|
||||
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<ReviewState | null>;
|
||||
identity(context: Context): Promise<{ conversationId: ConversationId; key: string }>;
|
||||
addComment(comment: ReviewComment, context: Context): Promise<void>;
|
||||
}
|
||||
const DiffReviews = defineService<DiffReviewService>("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<void> {
|
||||
sent.push({ value, ops });
|
||||
}
|
||||
async function render(text: string): Promise<void> {
|
||||
rendered.push(text);
|
||||
}
|
||||
|
||||
type JobInput = { command: string };
|
||||
type JobOutput = { stdout: string; chunks: number };
|
||||
const JobOutputDoc = defineDoc<JobOutput>({
|
||||
kind: "app.job-output",
|
||||
version: 1,
|
||||
scope: "task",
|
||||
initial: () => ({ stdout: "", chunks: 0 }),
|
||||
checkpointWhen: (_value, _ops, info) => info.deltasSinceBase >= 99,
|
||||
});
|
||||
async function appendJobOutput(
|
||||
runtime: TaskRuntime<JobInput, { phase: "running" }, null, object>,
|
||||
chunk: string,
|
||||
context: Context,
|
||||
): Promise<void> {
|
||||
// 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<void>,
|
||||
context: Context,
|
||||
): Promise<void> {
|
||||
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<JobInput, { phase: "running" }, null>({
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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}`);
|
||||
|
||||
|
|
|
|||
613
packages/durable/test/harness-lifecycle.test.ts
Normal file
613
packages/durable/test/harness-lifecycle.test.ts
Normal file
|
|
@ -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<void> = async () => {}) {
|
||||
return defineTask<null, Step, null>({
|
||||
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<typeof oneStep>, background = false): Promise<TaskId> {
|
||||
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<void> {
|
||||
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<boolean> {
|
||||
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<void> {
|
||||
const conversationId = await storage.mintId<ConversationId>();
|
||||
const id = await storage.mintId<TaskId<JsonValue>>();
|
||||
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<void> {
|
||||
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<void> => {
|
||||
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<string, () => Promise<unknown>> = {
|
||||
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<string, string> = {};
|
||||
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<unknown>): Promise<string> =>
|
||||
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<string, string> = {};
|
||||
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<string, (opened: Awaited<ReturnType<typeof paused>>) => Promise<unknown>> = {
|
||||
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);
|
||||
});
|
||||
});
|
||||
358
packages/durable/test/harness-task-graph.test.ts
Normal file
358
packages/durable/test/harness-task-graph.test.ts
Normal file
|
|
@ -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<void>; readonly late: Promise<void> }) {
|
||||
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<null, ParentState, null>({
|
||||
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<void>(() => {});
|
||||
return { child: never, late: never };
|
||||
}
|
||||
|
||||
/** Node states by task kind, for compact assertions. */
|
||||
function statuses(graph: TaskGraph): Record<string, string> {
|
||||
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<null, { phase: "work" }, null>({
|
||||
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<null, { phase: "work" }, null>({
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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<null, Memoed, string>({
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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) => {
|
||||
|
|
|
|||
|
|
@ -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<Readonly<State> | undefined>;
|
||||
type StateOf = Promise<DocumentState<State> | undefined>;
|
||||
type WatchOf = Promise<DocumentWatch<State> | undefined>;
|
||||
const check = async (session: Session, tx: Tx, conversationId: ConversationId, taskId: TaskId, at: EntryId) => {
|
||||
expectTypeOf(session.snapshotAsOf(RewindableDoc, conversationId, at, context)).toEqualTypeOf<Rewound>();
|
||||
expectTypeOf(
|
||||
session.snapshotAsOf(ConversationFamily, conversationId, "k", at, context),
|
||||
).toEqualTypeOf<Rewound>();
|
||||
// @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<StateOf>();
|
||||
expectTypeOf(session.documentState(LatestDoc, conversationId, context)).toEqualTypeOf<StateOf>();
|
||||
expectTypeOf(session.documentState(TaskDoc, taskId, context)).toEqualTypeOf<StateOf>();
|
||||
expectTypeOf(session.documentState(SessionFamily, "k", context)).toEqualTypeOf<StateOf>();
|
||||
expectTypeOf(session.documentState(ConversationFamily, conversationId, "k", context)).toEqualTypeOf<StateOf>();
|
||||
expectTypeOf(session.documentState(TaskFamily, taskId, "k", context)).toEqualTypeOf<StateOf>();
|
||||
expectTypeOf(session.watchDoc(SessionDoc, context)).toEqualTypeOf<WatchOf>();
|
||||
expectTypeOf(session.watchDoc(LatestDoc, conversationId, context)).toEqualTypeOf<WatchOf>();
|
||||
expectTypeOf(session.watchDoc(TaskDoc, taskId, context)).toEqualTypeOf<WatchOf>();
|
||||
expectTypeOf(session.watchDoc(SessionFamily, "k", context)).toEqualTypeOf<WatchOf>();
|
||||
expectTypeOf(session.watchDoc(ConversationFamily, conversationId, "k", context)).toEqualTypeOf<WatchOf>();
|
||||
expectTypeOf(session.watchDoc(TaskFamily, taskId, "k", context)).toEqualTypeOf<WatchOf>();
|
||||
// @ts-expect-error a watch of a Session document takes no owner
|
||||
void session.watchDoc(SessionDoc, conversationId, context);
|
||||
|
||||
expectTypeOf(await tx.entry(at)).toEqualTypeOf<EntryRecord | undefined>();
|
||||
expectTypeOf(await tx.entry(Note, at)).toEqualTypeOf<TypedEntry<{ text: string }> | undefined>();
|
||||
expectTypeOf(await tx.appendEntry(conversationId, { kind: "t.raw" })).toEqualTypeOf<EntryRecord>();
|
||||
const note = await tx.appendEntry(Note, conversationId, { data: { text: "x" } });
|
||||
expectTypeOf(note).toEqualTypeOf<TypedEntry<{ text: string }>>();
|
||||
expectTypeOf(note.data.text).toEqualTypeOf<string>();
|
||||
expectTypeOf(await tx.appendEntry(Marker, conversationId, {})).toEqualTypeOf<TypedEntry<never>>();
|
||||
// @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");
|
||||
});
|
||||
});
|
||||
|
|
|
|||
380
packages/durable/test/spec-usage.test.ts
Normal file
380
packages/durable/test/spec-usage.test.ts
Normal file
|
|
@ -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<ExecutionEnv> };
|
||||
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<string>;
|
||||
declare const payments: {
|
||||
charge(key: string): Promise<{ id: string }>;
|
||||
cancel(checkpoint: { phase: string }): Promise<void>;
|
||||
};
|
||||
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<typeof defineTask<null, { phase: "hold" }, null>>;
|
||||
declare const Reporter: ReturnType<typeof defineTask<null, { phase: "report" }, null>>;
|
||||
declare const subagentTool: ReturnType<typeof createReadTool>;
|
||||
|
||||
/** 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<null, Charge, { entryId: EntryId }>({
|
||||
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<object, { phase: "follow" }, null>({
|
||||
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<LiveState> | 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");
|
||||
});
|
||||
|
|
@ -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<ConversationId>(1);
|
||||
const entryId = idFromNumber<EntryId>(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<number>();
|
||||
expectTypeOf(task.input.steps).toEqualTypeOf<number>();
|
||||
await runtime.commit(() => ({ status: "running", checkpoint: { phase: "run", step: 0 } }), context);
|
||||
},
|
||||
run: async (task, runtime, context) => {
|
||||
expectTypeOf(task.state.checkpoint.step).toEqualTypeOf<number>();
|
||||
await runtime.hooks.each("beforeStep", (handler) => {
|
||||
expectTypeOf(handler).toEqualTypeOf<Hooks["beforeStep"]>();
|
||||
});
|
||||
await runtime.commit(
|
||||
() => ({ status: "terminal", outcome: { status: "completed", result: { ran: 1 } } }),
|
||||
context,
|
||||
);
|
||||
},
|
||||
},
|
||||
abort: async () => {},
|
||||
});
|
||||
expectTypeOf<HooksOf<typeof Stepper>>().toEqualTypeOf<Hooks>();
|
||||
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<null, Checkpoint, null, Hooks>,
|
||||
) => {
|
||||
const id = await conversation.commit(
|
||||
(tx) => tx.createTask(Stepper, { steps: 2 }, { ownership: { kind: "conversation" } }),
|
||||
callContext,
|
||||
);
|
||||
expectTypeOf(id).toEqualTypeOf<TaskId<{ ran: number }>>();
|
||||
const settled = await harness.waitForTask(id, callContext);
|
||||
expectTypeOf(settled).toEqualTypeOf<SettledTask<{ ran: number }>>();
|
||||
if (settled.state.outcome.status === "completed") {
|
||||
expectTypeOf(settled.state.outcome.result).toEqualTypeOf<{ ran: number }>();
|
||||
}
|
||||
expectTypeOf(await runtime.waitForTask(id, callContext)).toEqualTypeOf<SettledTask<{ ran: number }>>();
|
||||
const compaction = await conversation.compact(undefined, callContext);
|
||||
expectTypeOf(compaction).toEqualTypeOf<TaskId<CompactionResult>>();
|
||||
expectTypeOf(await harness.waitForTask(compaction, callContext)).toEqualTypeOf<SettledTask<CompactionResult>>();
|
||||
};
|
||||
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<null, Checkpoint, null>({
|
||||
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();
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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"],
|
||||
},
|
||||
},
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue