diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index c559b489e..8a5360fe7 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -11,6 +11,10 @@ - Made task conversation membership immutable after task creation. - Added `ConversationQuery` to Storage and transaction conversation scans. - Added required `StoredDocument.deltasSinceBase` to Storage document reads. +- Task definitions now require an exhaustive `phases` map and an `abort` handler; define them with `defineTask()`. +- `RegistryReader` now requires `subscribe()`. +- Removed `Tx.setTask()`; a task changes its own state by returning the next state from its `runtime.commit()` callback. +- `Session.subscribeClose()` listeners now run synchronously when close begins, after admission is sealed. ### Added @@ -23,6 +27,7 @@ - Added `Harness.open()` with lazy root creation, atomic conversation creation and forks with `init`, conversation-bound commits, fork-aware entry pagination, model context derivation, and the built-in `ConversationConfig` document with model, thinking level, and active tool accessors. - Added `createRegistry()` for tools, tool wrappers, hooks, tasks, and system prompt sections with batched publication and stable keyed ordering. - Added `defineEntry()` typed entry kinds. +- Added the durable task runtime: `defineTask()`, registry-resolved phase handlers with checkpoint progress rules decided on the Session line, migration at reservation, typed runtime commits, memos, `sleep()`, and invocation-owned watches, plus `Harness.resume()`, `getTask()`, `waitForTask()`, `abortTask()`, and task-aware `waitForIdle()` on the Harness and conversations. Open reconciles running tasks to pending; tasks without a fitting definition stay blocked until registration, and aborting them settles them as `orphaned`. ### Fixed diff --git a/packages/durable/docs/pico-v5-handoff.md b/packages/durable/docs/pico-v5-handoff.md index 44351e780..5f28c14be 100644 --- a/packages/durable/docs/pico-v5-handoff.md +++ b/packages/durable/docs/pico-v5-handoff.md @@ -11,7 +11,7 @@ facades, membranes, document routing, view projection, events, or clone chains. - Obsolete `pico` and `pico4` prototypes were removed. - `pico3` remains. -- Packages 1–13 are implemented in `packages/durable`; Package 10 was already satisfied by Chord's canonical structural diff implementation. +- Packages 1–14 are implemented in `packages/durable`; Package 10 was already satisfied by Chord's canonical structural diff implementation. ## 1. Records, cursors, and memory tables @@ -334,9 +334,8 @@ and reopen. Implement `defineTask`, exhaustive phase maps, full checkpoint replacement, kind migration at reservation, runtime commits and memos, invocation lifetime gates, scheduler reservation, dependencies, terminal outcomes, typed waits, -holds, quiescence, and joins. Complete the task-facing §2.2 methods: `resume`, -`suspend`, `hold`, `getTask`, `waitForTask`, `markTask`, `abortTask`, and the -task-aware portion of idle waits. Task definitions come from the registry snapshot; +and joins. Complete the task-facing §2.2 methods: `resume`, `getTask`, +`waitForTask`, `abortTask`, and the task-aware portion of idle waits. Task definitions come from the registry snapshot; add `RegistryReader.subscribe()` so registry changes wake the scheduler. Include the execution-critical abort core: durable direct-task marks, @@ -350,6 +349,20 @@ orphans a task. Implement per-phase registry snapshots, refresh at every normal phase boundary, and hand over when the task definition object changed and the new definition can reserve the task. +Deferred to later packages; do not stub them in Package 14: +- Orphaning writes the terminal `orphaned` record and retires task documents. + Unanswered input submissions and clearing turn control are added in Package + 15, when submissions and turn control exist. +- `TaskRuntime.hooks` (the hook runner) is added in Package 16 with hook + dispatch. Package 14 adds `runtime.registry` (the phase's registry snapshot) + and `runtime.models`. +- `TaskRuntime.conversation()` returns a `ConversationHandle`, whose `submit` + needs Package 15 and whose `abort`/`waitForIdle` need ownership traversal; it + is added with the invocation-bound owned APIs in Package 18. +- Idle waits count live non-background tasks directly (Harness-wide or in the + addressed conversation). Traversal through owned conversations and background + boundaries replaces this in Package 18. + Use a fake two-phase external effect to test the real runtime. Acceptance is an intent/effect/outcome task interrupted after intent, closed, reopened on the same storage, and safely resumed to a durable terminal receipt. Also test unchanged- @@ -357,16 +370,24 @@ checkpoint faulting, same-phase progress, cancellation precedence, thrown handlers, dependencies, result values and entry IDs, first-writer-wins memos, terminal checkpoint/memo removal, task-document retirement, close/reopen without abort marks or fabricated outcomes, no fresh phase/abort dispatch while closing, -watch cleanup, holds and quiescence with eligible work, blocked tasks unblocked +watch cleanup, blocked tasks unblocked by later registration, migration failure leaving the record unchanged, handover after a same-name task replacement, no handover to a missing or incompatible definition, no overlap between predecessor and successor invocations, abort of a blocked task settling as `orphaned` with full cleanup, -mark-only versus signalling abort, and crashes at every direct-task abort stage. +run commits rejected after the mark, and crashes at every direct-task abort stage. + +Implemented design notes: every task transition is decided by one callback +serialized on the Session line (reservation, abort marks, runtime commits, and the +step before each phase). `Tx` has no task replacement; `runtime.commit()` +callbacks return the typed next state. `suspend()`, `hold()`, `quiescent()`, and +a public `markTask()` were dropped from §2.2. ## 15. First runnable no-tool chat turn -Implement the smallest real input-to-answer vertical path. Define the final +Implement the smallest real input-to-answer vertical path. Extend Package 14's +`orphaned` settlement to mark affected input submissions unanswered and clear +matching turn control in the same commit. Define the final inbox, turn-control, and generation presentation documents needed by this path; do not use provisional kinds or schemas. Add input `Submission` admission, request-ID deduplication, reacquisition and waiting, idle placement, active-turn @@ -406,8 +427,8 @@ awaits its own input submission rather than global idle. ## 16. First coding-agent tool turn -Implement hook dispatch (Session-wide and scoped to a conversation or its owned -subtree) and wire the real generation → tool tasks → post-tools → generation +Implement hook dispatch and `TaskRuntime.hooks`, deferred from Package 14 +(Session-wide and scoped to a conversation or its owned subtree) and wire the real generation → tool tasks → post-tools → generation chain. Implement offered-set checks, tool pinning from the phase snapshot, declaration and argument validation against both the offered declaration and the pinned implementation, hook composition, durable execution intent, stored @@ -441,9 +462,15 @@ self-head cuts, successor turns, queued reset/handoff, and every terminal cleanu Successful inputs still require an answer; writes settle on placement and never start generation. +Add `harness.blockedTasks()`, deferred from Package 14: the pending tasks this +process cannot run and why (`missing_task`, `task_too_old`, `migration_failed` +with the stored and registered versions and the migration error). It is derived +from task records, the current registry snapshot, and scheduler memory, and is +never persisted. + Specify and implement the Harness activity view: the active conversations, notifications when a conversation becomes active or -idle, and Harness quiescence, all derived from committed turn-control and task +idle, all derived from committed turn-control and task state. Define any remaining built-in preference/presentation documents once with final diff --git a/packages/durable/docs/pico-v5.md b/packages/durable/docs/pico-v5.md index 3b85adc16..adddf134c 100644 --- a/packages/durable/docs/pico-v5.md +++ b/packages/durable/docs/pico-v5.md @@ -411,9 +411,6 @@ interface Conversation { interface Harness extends Session { resume(): void; - suspend(context: Context): Promise; - quiescent(): boolean; - hold(): () => void; root( context: Context, @@ -433,7 +430,6 @@ interface Harness extends Session { conversationId?: ConversationId, ): Promise<"aborted" | "already_placed" | "settled" | "not_found">; abortTask(id: TaskId, context: Context): Promise<"marked" | "terminal">; - markTask(id: TaskId, context: Context): Promise<"marked" | "terminal">; waitForTask(id: TaskId, context: Context): Promise>; waitForIdle(context: Context): Promise; } @@ -473,11 +469,9 @@ write occurs. Reopen finds the same root by its reserved ID. A conversation with no configured model produces a durable `no_model` generation failure. `resume()` is idempotent while running and only enables scheduling. It does not -repeat open-time reconciliation. `suspend()` is terminal for that Harness -instance and follows the close semantics below; `resume()` after suspend/close -rejects. `quiescent()` means no task invocation is currently executing; eligible -or delayed durable tasks may still exist. `hold()` is available only while -quiescent and pauses reservation until its idempotent release function runs. +repeat open-time reconciliation, and it throws after close. Work that must happen +before any task runs, such as registration or seeding, happens before +`resume()`. `createConversation({ ownership, init, input })` and `fork(at, { ownership, init })` commit atomically: the conversation, its @@ -567,10 +561,10 @@ handoff write and then resolves; while busy, placement follows section 6 and may occur later. Observe its placement through the conversation watch. An idle wait does not guarantee placement of queued passive writes. -`markTask()` commits `abortRequested` and the durable foreground-subtree -cascade; it neither signals nor joins active invocations. The scheduler notices -the marks on its next drain. `abortTask()` also signals and joins the active run -before starting the abort invocation. `Conversation.abort()` withdraws queued +`abortTask()` commits `abortRequested` and the durable foreground-subtree +cascade, then signals and joins the active run; the scheduler then starts the +abort invocation. Marking without signalling is internal: the cascade marks +owned tasks in the same commit. `Conversation.abort()` withdraws queued input submissions, marks non-background tasks selected by ordinary ownership traversal, signals them, and resolves only after that scope is ordinarily idle. Passive writes and background subtrees survive. Conversation idle means no live @@ -581,9 +575,8 @@ wait aborts only that waiter. Conversation handles are stateless; compare them by `id`. Hosts discover conversations through lookups and scans. An activity view that lists active -conversations and reports conversations becoming active or idle, plus Harness -quiescence, is specified with the task runtime and turn control (Packages 14 -and 17); there is no creation listener. +conversations and reports conversations becoming active or idle is specified +with turn control (Package 17); there is no creation listener. `submit()` returns after durable admission, not settlement. An input submission creates a user message with the admission timestamp; `whenBusy` defaults to @@ -610,7 +603,7 @@ uses the same serialized asynchronous, bounded-buffer contract as `watchDoc()` in section 9.2. Neither carries semantic events or owns a second persistence authority. -`close()` is equivalent to `suspend()` for v1. It seals mutation admission and +`close()` seals mutation admission and task reservation, signals invocations, and stops future watch deliveries. Outside the Session line it lets already-admitted storage commits settle, joins task/tool/hook invocations, then closes states and storage. Already-running watch @@ -869,7 +862,6 @@ interface Tx { createTask( task: Task, input: I, options?: TaskOptions, ): Promise>; - setTask(value: TaskRecord): void; doc(token: SessionDocToken): Promise>; doc(token: ConversationDocToken, conversationId: ConversationId): Promise>; @@ -1205,23 +1197,24 @@ Session-line operation so it never exposes a mixture from one commit. A Session commit callback may be asynchronous. It owns the Session mutation line through callback execution, preparation, storage settlement, committed -baseline adoption, and publication enqueue. Commit/close observers run on the +baseline adoption, and publication enqueue. Commit observers run on the line; document-state and watch user callbacks run later. External model, process, tool, network, and human effects run outside it. -`subscribeCommits()` observes complete immutable publications synchronously -after adoption, and `subscribeClose()` observes Session closure synchronously. -Both run on the Session line and return idempotent disposers; their listeners -must not throw, block, or call Session APIs. Document-state subscribers and watch +`subscribeCommits()` observes complete immutable publications synchronously on +the line after adoption. `subscribeClose()` observes close synchronously when it +begins, after admission is sealed; the Harness stops watches and signals task +invocations there. Both return idempotent disposers; their listeners must not +throw, block, or call Session APIs. Document-state subscribers and watch listeners still run later, off the line. ```ts await session.commit(async tx => { - const task = await tx.task(taskId); // table read + 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.message; // document mutation remains valid - tx.setTask(nextTask(task)); + await tx.createTask(Follow, { after: message.id }); // further table writes are fine }, context); ``` @@ -1288,6 +1281,9 @@ type RunningTask = TaskRecord & { readonly state: Extract, { status: "running" }>; }; +/** State a task commits for itself: a replacement checkpoint or its terminal outcome. */ +type NextTaskState = Extract, { status: "running" | "terminal" }>; + interface HookRunner { each(name: K, invoke: (handler: H[K]) => void | Promise): Promise; } @@ -1302,10 +1298,16 @@ interface TaskRuntime extends DocumentObserver { readonly taskId: TaskId; readonly conversationId: ConversationId; readonly signal: AbortSignal; + /** Registry snapshot of the current phase; refreshed at every phase boundary. */ + readonly registry: RegistrySnapshot; + readonly models: Models; readonly hooks: HookRunner; commit( - change: (tx: Tx, current: RunningTask) => void | Promise, + change: ( + tx: Tx, + current: RunningTask, + ) => NextTaskState | undefined | Promise | undefined>, context: Context, ): Promise; @@ -1322,7 +1324,7 @@ type TaskDefinition = { readonly phases: { [P in S["phase"]]: PhaseHandler, S, R, H>; }; - abort(task: TaskRecord, runtime: TaskRuntime, context: Context): Promise; + abort(task: RunningTask, runtime: TaskRuntime, context: Context): Promise; migrate?(input: JsonValue, checkpoint: JsonValue, fromVersion: number): { input: I; checkpoint: S; @@ -1355,23 +1357,71 @@ Storage admission so persisted conversation-owner edges cannot become stale. The phase map is exhaustive and phase-narrowed. A handler may perform several commits around one effect, but each durable checkpoint is a full replacement. `TaskRuntime.commit()` rereads and gates the current durable task on the Session -line before invoking its callback. Transaction methods replace its checkpoint -or write its terminal outcome. +line before invoking its callback. It rejects when the invocation has ended, the +Harness is closing, the task is terminal, or a run invocation's task carries an +abort mark. Its `tx.createTask()` defaults to the task's conversation. When the +callback returns a state, the runtime replaces the task's state in the same +commit, so the checkpoint or outcome is atomic with the callback's entries, +documents, and child tasks and is type-checked against the task's checkpoint and +result types. Returning nothing leaves the state unchanged. A terminal state +drops the memos. `pending` is never returned; only reconciliation and handover +write it. + +`Tx` has no task replacement operation. A task changes only its own state, +through its runtime. The scheduler owns reservation, reconciliation, handover, +faults, and orphaning; `abortTask()` owns abort marks. Other code +stops a task with `abortTask()` and reads its result with `waitForTask()`. + +```ts +// 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); +}, +``` +`memo(name, candidate)` is one gated commit; `memo(name)` reads the committed +record. `sleep(until)` compares against the Harness `now` clock and rejects when +the invocation is signalled or its context is cancelled. Watches acquired through +the runtime stop when the invocation ends. Reservation durably changes `pending` to `running`. One invocation runs phase -handlers in sequence; checkpoint commits retain `running`. After a handler -settles, the scheduler rereads the task and applies the first matching rule: +handlers in sequence; checkpoint commits retain `running`. Before every phase, +including the first, the scheduler runs one synchronous step callback on the +Session line. It reads the committed task, applies the first matching rule +below to the phase that just returned, writes any fault or handover in the same +commit, and ends the invocation there when a rule stops it. A runtime commit the +invocation queued earlier therefore either lands before the step and counts, or +reaches the line after it and rejects. Before the first phase only rules 1–3 +apply: 1. Terminal: stop. 2. Session closing: stop; preserve the checkpoint and any abort mark for reopen. 3. Run mode with a durable abort mark: end and join the run invocation, then dispatch a fresh abort invocation. 4. Uncaught error: write terminal `faulted`. -5. Checkpoint changed, including progress within the same phase: invoke its - phase handler in the same task invocation. +5. Checkpoint changed, including progress within the same phase: refresh the + registry snapshot and either hand over (section 5.4) or invoke the phase + handler in the same task invocation. 6. Checkpoint unchanged: write terminal `faulted` because no durable progress was made. +An abort invocation runs its abort handler once. After it settles, a step +applies rules 1, 2, and 4; a handler that returns without a terminal outcome +faults the task. When Storage rejects a step's fault or handover write, the +invocation still ends; the task stays `running` and the next reservation runs it +again. + On open, running-task reconciliation changes surviving `running` tasks back to `pending`, preserving their checkpoint and abort mark. Task migration runs at reservation, atomically with `pending -> running`. One callback handles every @@ -1497,6 +1547,9 @@ An abort invocation resolves the definition the same way. When the definition can take the task, its abort handler runs; otherwise the task is orphaned as described below. +A failed migration is reported once through `onReport` and retried only after +the registry resolves a different definition object for the task's kind. + The Harness never terminalizes a task merely because registry code is missing or incompatible, neither at open nor later. A blocked task without an abort mark keeps its durable record unchanged, remains live, still blocks ordinary idle @@ -1508,7 +1561,11 @@ Aborting a blocked task cannot run its abort handler, because that code is missing or cannot take the task. The Harness therefore settles it as terminal `orphaned` instead of `aborted`: `aborted` means the task's own abort handler ran and decided the outcome, while `orphaned` means no task code ran, so external -effects the task started may remain uncleaned. Only an abort (direct, by +effects the task started may remain uncleaned. The orphaned `reason` is the +blocked reason (`missing_task`, `task_too_old`, or `migration_failed`). When +`abortTask()` finds no active invocation and the current snapshot +cannot take the task, the marking commit settles it directly; otherwise the +scheduler settles it when it would reserve the abort invocation. Only an abort (direct, by conversation, or by cascade) orphans a task; a missing definition alone never does. The orphaning commit performs the cleanup the task's code cannot: affected input submissions become unanswered with the reason, any matching active turn @@ -1517,16 +1574,16 @@ written; the terminal task record and unanswered submissions carry the reason. Faulting a turn task performs the same control/submission cleanup with a `faulted` outcome. -At every normal phase boundary (section 5.1, rule 5), after rereading the task -and applying the precedence rules, the scheduler refreshes the invocation's -registry snapshot. If the task definition resolved by name is a different object +At every normal phase boundary (section 5.1, rule 5), the step refreshes the +invocation's registry snapshot. If the task definition resolved by name is a different object than the one the invocation started with and the new definition can reserve the task (same version, or a higher version with `migrate`), the invocation hands -over: it commits the task back to `pending` with its checkpoint, memos, and abort -mark, ends, and the next reservation starts a fresh invocation under the new +over: the step commits the task back to `pending` with its checkpoint, memos, +and abort mark and ends the invocation, and the next reservation starts a fresh invocation under the new definition, applying the reservation rules above. When the definition is missing or cannot reserve the task, the invocation keeps running under its old -definition, reports once, and reconsiders at its next boundary. A handler that +definition, reports once through `onReport` per resolved definition, and +reconsiders at its next boundary. A handler that never settles never hands over. ## 6. Submissions and inbox diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index c4ff62d49..1f77c4e3b 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -1,4 +1,5 @@ -import type { Context, Draft } from "@earendil-works/chord"; +import type { Context, Draft, JsonValue } from "@earendil-works/chord"; +import { withoutAbortSignal } from "@earendil-works/chord/context"; import type { ModelThinkingLevel } from "@earendil-works/pi-ai"; import { SessionImpl } from "../session/session.ts"; import type { @@ -11,11 +12,14 @@ import type { EntryRecord, Page, Storage, + TaskId, + TaskRecord, Tx, } from "../types.ts"; import { ROOT_CONVERSATION_ID } from "../types.ts"; import { ConversationConfig, type ConversationConfigState } from "./config.ts"; import { captureContextBounds, deriveContext } from "./context.ts"; +import { TaskScheduler } from "./scheduler.ts"; import type { ContextView, Conversation, @@ -26,6 +30,7 @@ import type { ModelRef, RegistryReader, RegistrySnapshot, + SettledTask, ToolRegistration, } from "./types.ts"; @@ -44,6 +49,7 @@ type ConversationHost = { readonly harness: HarnessImpl; readonly storage: Storage; readonly registry: RegistryReader; + readonly tasks: TaskScheduler; create(target: CreateTarget, init: ConversationInit | undefined, context: Context): Promise; }; @@ -121,6 +127,10 @@ class ConversationImpl implements Conversation { ); } + waitForIdle(context: Context): Promise { + return this.#host.tasks.waitForIdle(this.id, context); + } + async #config(context: Context): Promise> { return ( (await this.#host.harness.snapshot(ConversationConfig, this.id, context)) ?? @@ -140,20 +150,59 @@ class HarnessImpl extends SessionImpl implements readonly #storage: Storage; readonly #registry: RegistryReader; readonly #host: ConversationHost; + readonly #tasks: TaskScheduler; #closed = false; - constructor(storage: Storage, options: HarnessOptions) { + constructor(storage: Storage, options: HarnessOptions, context: Context) { super(storage); this.#storage = storage; this.#registry = options.registry; + this.#tasks = new TaskScheduler({ + session: this, + storage, + registry: options.registry, + models: options.models, + now: options.now ?? Date.now, + report: options.onReport ?? (() => {}), + context: withoutAbortSignal(context), + }); this.#host = { harness: this, storage, registry: options.registry, + tasks: this.#tasks, create: (target, init, context) => this.#create(target, init, context), }; } + /** Reconcile surviving `running` tasks to `pending`; part of open. */ + openTasks(context: Context): Promise { + return this.#tasks.open(context); + } + + resume(): void { + this.#assertOpen(); + this.#tasks.resume(); + } + + getTask(id: TaskId, context: Context): Promise | undefined> { + return this.readOnLine(() => this.#storage.task(id, context)) as Promise< + TaskRecord | undefined + >; + } + + abortTask(id: TaskId, context: Context): Promise<"marked" | "terminal"> { + return this.#tasks.abort(id, context); + } + + waitForTask(id: TaskId, context: Context): Promise> { + return this.#tasks.waitForTask(id, context) as Promise>; + } + + waitForIdle(context: Context): Promise { + return this.#tasks.waitForIdle(undefined, context); + } + root(context: Context, options?: { readonly init?: ConversationInit }): Promise { return this.#create({ kind: "root" }, options?.init, context); } @@ -173,11 +222,16 @@ class HarnessImpl extends SessionImpl implements return super.close(context); } + /** Join task invocations after admission is sealed and before Storage closes; writes no task outcome. */ + protected override beforeClose(): Promise { + return this.#tasks.join(); + } + async #create(target: CreateTarget, init: ConversationInit | undefined, context: Context): Promise { this.#assertOpen(); const id = target.kind === "root" - ? await this.commitRoot(async (tx) => { + ? await this.commitWith(async (tx) => { if ((await tx.conversation(ROOT_CONVERSATION_ID)) !== undefined) return ROOT_CONVERSATION_ID; return this.#stageConfiguration(tx, await tx.createRootConversation(), false, init); }, context) @@ -239,6 +293,13 @@ export const Harness = { context: Context, ): Promise { context.abortSignal?.throwIfAborted(); - return new HarnessImpl(storage, options); + const harness = new HarnessImpl(storage, options, context); + try { + await harness.openTasks(context); + } catch (error) { + await harness.close(context); + throw error; + } + return harness; }, }; diff --git a/packages/durable/src/harness/registry.ts b/packages/durable/src/harness/registry.ts index 991b0ce78..f8f76f62d 100644 --- a/packages/durable/src/harness/registry.ts +++ b/packages/durable/src/harness/registry.ts @@ -158,6 +158,7 @@ class RegistryImpl implements Registry { #batch: Batch | undefined; /** First position of every key ever published; re-registered keys keep it. */ readonly #positions = new Map(); + readonly #listeners = new Set<() => void>(); #nextPosition = 0; readonly tools: Registry["tools"]; @@ -222,6 +223,11 @@ class RegistryImpl implements Registry { return this.#current; } + subscribe(listener: () => void): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + } + batch(register: () => void): Registration { if (this.#batch !== undefined) throw new Error("Registry batches cannot be nested"); const batch: Batch = { added: [], disposed: new Set() }; @@ -305,6 +311,7 @@ class RegistryImpl implements Registry { this.#current = new RegistryState(records); for (const record of batch.added) record.status = "published"; for (const record of batch.disposed) record.status = "disposed"; + for (const listener of [...this.#listeners]) listener(); } } diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts new file mode 100644 index 000000000..28eb5b75b --- /dev/null +++ b/packages/durable/src/harness/scheduler.ts @@ -0,0 +1,703 @@ +import { type Context, copyJson, type JsonValue } from "@earendil-works/chord"; +import { awaitWithContext, withAbortSignal } from "@earendil-works/chord/context"; +import type { Models } from "@earendil-works/pi-ai"; +import type { SessionImpl } from "../session/session.ts"; +import type { Transaction } from "../session/transaction.ts"; +import type { + CommitPublication, + ConversationId, + DocumentWatch, + JsonObject, + NextTaskState, + RunningTask, + Storage, + TaskDefinition, + TaskId, + TaskOutcome, + TaskRecord, + TaskRuntime, + TaskState, + Tx, +} from "../types.ts"; +import type { AnyTask, RegistryReader, RegistrySnapshot, SettledTask } from "./types.ts"; + +type AnyTaskRecord = TaskRecord; +type LiveTaskRecord = Extract; +type Checkpoint = { readonly phase: string }; +type ErasedDefinition = TaskDefinition; +type ErasedRuntime = TaskRuntime; +type ErasedRunningTask = RunningTask; + +const SCAN_PAGE_SIZE = 256; +/** Longest delay `setTimeout` supports; longer sleeps wait in several steps. */ +const MAX_TIMER_DELAY = 2_147_483_647; + +/** Why a pending task cannot be reserved under a registry snapshot. Derived, never persisted. */ +type BlockedReason = "missing_task" | "task_too_old" | "migration_failed"; + +type Resolution = + | { readonly kind: "ready"; readonly task: AnyTask; readonly record: LiveTaskRecord } + | { readonly kind: "blocked"; readonly reason: BlockedReason }; + +/** One in-memory execution of a task in run or abort mode. */ +type Invocation = { + readonly taskId: TaskId; + readonly conversationId: ConversationId; + readonly mode: "run" | "abort"; + readonly controller: AbortController; + /** Context passed to handlers; cancelled by `controller`. */ + readonly context: Context; + /** Watches acquired through the runtime; stopped at invocation end. */ + readonly watches: Set>; + ended: boolean; + readonly done: Promise; + readonly finish: () => void; +}; + +type Reservation = { + readonly invocation: Invocation; + readonly task: AnyTask; + readonly snapshot: RegistrySnapshot; +}; + +/** Replacement definition already reported as unable to take over, per invocation. */ +type ReportedTask = { readonly task: AnyTask | undefined } | undefined; + +/** Outcome of the phase that just returned, judged by the next step. */ +type PhaseResult = { readonly checkpoint: Checkpoint; readonly failure?: { readonly error: unknown } }; + +type Waiter = { readonly resolve: (value: T) => void; readonly reject: (error: unknown) => void }; + +export type TaskSchedulerOptions = { + readonly session: SessionImpl; + readonly storage: Storage; + readonly registry: RegistryReader; + readonly models: Models; + readonly now: () => number; + readonly report: (error: unknown) => void; + /** Context for scheduler commits and invocations; carries no caller cancellation. */ + readonly context: Context; +}; + +/** + * Durable task scheduler of one Harness. + * + * `#live` mirrors every committed pending/running task record. The synchronous commit listener updates it on the + * Session line, so code running on the line reads exactly the committed state from it. + * + * Invariant: every task transition is decided and written by one callback serialized on the Session line. That covers + * reservation, marks, runtime commits, and the synchronous step before each phase, which applies the precedence rules + * and writes a fault or handover. Handlers and joins run off the line. An invocation ends inside the step that decides its end, + * so a runtime commit it queued either lands before that decision or is rejected. + */ +export class TaskScheduler { + readonly #session: SessionImpl; + readonly #storage: Storage; + readonly #registry: RegistryReader; + readonly #models: Models; + readonly #now: () => number; + readonly #report: (error: unknown) => void; + readonly #context: Context; + readonly #live = new Map(); + readonly #invocations = new Map(); + readonly #taskWaiters = new Map>>>(); + /** Idle waiters by conversation; `undefined` waits for the whole Harness. */ + readonly #idleWaiters = new Map>>(); + /** Definition whose migration failed per task; retried only once the registry resolves another definition. */ + readonly #failedMigrations = new Map(); + #unsubscribeRegistry: () => void = () => {}; + #enabled = false; + #closing = false; + #dirty = false; + #draining = false; + + constructor(options: TaskSchedulerOptions) { + this.#session = options.session; + this.#storage = options.storage; + this.#registry = options.registry; + this.#models = options.models; + this.#now = options.now; + this.#report = options.report; + this.#context = options.context; + } + + /** Load live tasks and change surviving `running` tasks back to `pending`. Dispatches nothing. */ + async open(context: Context): Promise { + this.#session.subscribeCommits((publication) => this.#observe(publication)); + this.#session.subscribeClose(() => this.#seal()); + this.#unsubscribeRegistry = this.#registry.subscribe(() => this.#kick()); + await this.#session.commitWith(async (tx) => { + const running: LiveTaskRecord[] = []; + for (const status of ["pending", "running"] as const) { + let cursor: Parameters[2]; + do { + const page = await tx.scanTasks({ status }, SCAN_PAGE_SIZE, cursor); + for (const record of page.items as readonly LiveTaskRecord[]) { + this.#live.set(record.id, record); + if (status === "running") running.push(record); + } + cursor = page.next; + } while (cursor !== undefined); + } + for (const record of running) + tx.setTask(withState(record, { status: "pending", checkpoint: record.state.checkpoint })); + }, context); + } + + resume(): void { + this.#enabled = true; + this.#kick(); + } + + /** Wait for every invocation signalled by `#seal()`. Writes nothing. */ + async join(): Promise { + await Promise.allSettled([...this.#invocations.values()].map((invocation) => invocation.done)); + } + + /** + * Commit the abort mark, or settle a task that no registered definition can take as `orphaned`, then signal and + * join the run invocation seen on the line. The next drain starts the abort invocation. + */ + async abort(id: TaskId, context: Context): Promise<"marked" | "terminal"> { + const marked = await this.#session.commitWith(async (tx) => { + const current = await tx.task(id); + if (current === undefined) throw new Error(`Task ${id} does not exist`); + if (current.state.status === "terminal") return { result: "terminal" as const }; + const live = current as LiveTaskRecord; + const invocation = this.#invocations.get(id); + if (invocation === undefined) { + const resolution = this.#resolve(live, this.#registry.snapshot()); + if (resolution.kind === "blocked") { + tx.setTask(withState(live, terminal({ status: "orphaned", reason: resolution.reason }))); + return { result: "marked" as const }; + } + } + if (!live.abortRequested) tx.setTask({ ...live, abortRequested: true }); + return { result: "marked" as const, run: invocation?.mode === "run" ? invocation : undefined }; + }, context); + if (marked.run !== undefined) { + marked.run.controller.abort(); + await awaitWithContext(marked.run.done, context); + } + return marked.result; + } + + async waitForTask(id: TaskId, context: Context): Promise> { + // Check and register on the line so no terminal publication falls between them. + const found = await this.#session.readOnLine(async () => { + if (this.#closing) throw closedError(); + if (this.#live.has(id)) return { promise: addWaiter(this.#taskWaiters, id, context) }; + const record = await this.#storage.task(id, context); + if (record === undefined) throw new Error(`Task ${id} does not exist`); + return { promise: Promise.resolve(record as SettledTask) }; + }); + return found.promise; + } + + /** Resolve when no live non-background task exists, optionally within one conversation. */ + waitForIdle(conversationId: ConversationId | undefined, context: Context): Promise { + if (this.#closing) return Promise.reject(closedError()); + if (this.#idle(conversationId)) return Promise.resolve(); + return addWaiter(this.#idleWaiters, conversationId, context); + } + + // ─── Scheduling ──────────────────────────────────────────────────────── + + #observe(publication: CommitPublication): void { + let changed = false; + for (const change of publication.changes) { + if (change.type !== "task") continue; + changed = true; + const record = change.value; + if (record.state.status !== "terminal") { + this.#live.set(record.id, record as LiveTaskRecord); + continue; + } + this.#live.delete(record.id); + this.#failedMigrations.delete(record.id); + settleWaiters(this.#taskWaiters, record.id, (waiter) => waiter.resolve(record as SettledTask)); + } + if (!changed) return; + for (const conversationId of [...this.#idleWaiters.keys()]) { + if (this.#idle(conversationId)) settleWaiters(this.#idleWaiters, conversationId, (waiter) => waiter.resolve()); + } + this.#kick(); + } + + /** Close listener: runs synchronously once admission is sealed, before `join()`. */ + #seal(): void { + this.#closing = true; + this.#unsubscribeRegistry(); + const error = closedError(); + for (const id of [...this.#taskWaiters.keys()]) settleWaiters(this.#taskWaiters, id, (w) => w.reject(error)); + for (const id of [...this.#idleWaiters.keys()]) settleWaiters(this.#idleWaiters, id, (w) => w.reject(error)); + for (const invocation of this.#invocations.values()) invocation.controller.abort(); + } + + #kick(): void { + this.#dirty = true; + if (this.#draining || !this.#enabled || this.#closing) return; + this.#draining = true; + // Never commit synchronously from a commit or registry listener. + queueMicrotask(() => void this.#drain()); + } + + async #drain(): Promise { + try { + while (this.#dirty && this.#enabled && !this.#closing) { + this.#dirty = false; + for (const reservation of await this.#reserve()) this.#start(reservation); + } + } catch (error) { + if (!this.#closing) this.#report(error); + } finally { + this.#draining = false; + // A wakeup that arrived during a failed pass still needs its pass. + if (this.#dirty) this.#kick(); + } + } + + /** Reserve every eligible task in one commit; orphan abort-marked tasks no definition can take. */ + async #reserve(): Promise { + const reservations: Reservation[] = []; + try { + await this.#session.commitWith(async (tx) => { + if (!this.#enabled || this.#closing) return; + // Taken once per pass, and only when some task is a candidate. + let snapshot: RegistrySnapshot | undefined; + for (const record of [...this.#live.values()]) { + if (this.#invocations.has(record.id)) continue; + const mode = record.abortRequested ? "abort" : "run"; + // An abort mark bypasses dependencies; a task already running has passed them. + if ( + mode === "run" && + record.state.status === "pending" && + record.after.some((id) => this.#live.has(id)) + ) { + continue; + } + snapshot ??= this.#registry.snapshot(); + const resolution = this.#resolve(record, snapshot); + if (resolution.kind === "blocked") { + if (mode === "abort") { + tx.setTask(withState(record, terminal({ status: "orphaned", reason: resolution.reason }))); + } + continue; + } + if (resolution.record !== record || record.state.status !== "running") { + tx.setTask( + withState(resolution.record, { + status: "running", + checkpoint: resolution.record.state.checkpoint, + }), + ); + } + // Registered on the line, so marks and later reservations see it and close joins it. + const invocation = this.#createInvocation(record, mode); + reservations.push({ invocation, task: resolution.task, snapshot }); + } + }, this.#context); + } catch (error) { + for (const { invocation } of reservations) { + this.#invocations.delete(invocation.taskId); + invocation.finish(); + } + throw error; + } + return reservations; + } + + /** Resolve the record's definition by kind, migrating an older stored version. */ + #resolve(record: LiveTaskRecord, snapshot: RegistrySnapshot): Resolution { + const task = snapshot.task(record.kind); + if (task === undefined) return { kind: "blocked", reason: "missing_task" }; + const definition = erased(task); + if (definition.version === record.version) return { kind: "ready", task, record }; + if (definition.version < record.version) return { kind: "blocked", reason: "task_too_old" }; + if (this.#failedMigrations.get(record.id) === task) return { kind: "blocked", reason: "migration_failed" }; + try { + if (definition.migrate === undefined) { + throw new Error( + `Task ${record.kind} version ${definition.version} has no migration from ${record.version}`, + ); + } + const migrated = definition.migrate(record.input, record.state.checkpoint, record.version); + const migratedRecord = { + ...record, + version: definition.version, + input: copyJson(migrated.input), + state: { status: record.state.status, checkpoint: copyJson(migrated.checkpoint) }, + } as LiveTaskRecord; + return { kind: "ready", task, record: migratedRecord }; + } catch (error) { + this.#failedMigrations.set(record.id, task); + this.#report(error); + return { kind: "blocked", reason: "migration_failed" }; + } + } + + #createInvocation(record: LiveTaskRecord, mode: "run" | "abort"): Invocation { + const controller = new AbortController(); + let finish!: () => void; + const done = new Promise((resolve) => { + finish = resolve; + }); + const invocation: Invocation = { + taskId: record.id, + conversationId: record.conversationId, + mode, + controller, + context: withAbortSignal(controller.signal, this.#context), + watches: new Set(), + ended: false, + done, + finish, + }; + this.#invocations.set(record.id, invocation); + return invocation; + } + + #start(reservation: Reservation): void { + const invocation = reservation.invocation; + void (async () => { + try { + if (invocation.mode === "run") await this.#run(reservation); + else await this.#runAbort(reservation); + } catch (error) { + this.#report(error); + } finally { + this.#end(invocation); + invocation.finish(); + this.#kick(); + } + })(); + } + + /** Run phase handlers, each preceded by a step that decides on the line whether the invocation continues. */ + async #run(reservation: Reservation): Promise { + const invocation = reservation.invocation; + const state = { task: reservation.task, snapshot: reservation.snapshot, reported: undefined as ReportedTask }; + const runtime = this.#runtime(invocation, () => state.snapshot); + let previous: PhaseResult | undefined; + for (;;) { + const current = await this.#step(invocation, (tx, current) => this.#decide(tx, current, previous, state)); + // Close may seal between the decision and dispatch. + if (current === undefined || this.#closing) return; + const checkpoint = current.state.checkpoint; + try { + await erased(state.task).phases[checkpoint.phase]!(current, runtime, invocation.context); + previous = { checkpoint }; + } catch (error) { + previous = { checkpoint, failure: { error } }; + } + } + } + + /** + * Precedence rules for a run invocation, on the line. Rules 1 (terminal) and 2 (closing) are applied by `#step`. + * Returns whether the invocation continues with the next phase. + */ + #decide( + tx: Transaction, + current: ErasedRunningTask, + previous: PhaseResult | undefined, + state: { task: AnyTask; snapshot: RegistrySnapshot; reported: ReportedTask }, + ): boolean { + // 3. abort mark: end; the next drain starts a fresh abort invocation. + if (current.abortRequested) return false; + if (previous === undefined) return true; + // 4. uncaught error. + if (previous.failure !== undefined) { + tx.setTask(faulted(current, previous.failure.error)); + return false; + } + // 6. no durable progress. + if (jsonEqual(current.state.checkpoint, previous.checkpoint)) { + const message = `Task ${current.kind} phase ${previous.checkpoint.phase} returned without durable progress`; + tx.setTask(faulted(current, new Error(message))); + return false; + } + // 5. progress: refresh the snapshot; hand over to a replacement definition that can take the task. + state.snapshot = this.#registry.snapshot(); + const next = state.snapshot.task(current.kind); + if (next !== state.task) { + if (next !== undefined && canReserve(next, current)) { + tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint })); + return false; + } + if (state.reported === undefined || state.reported.task !== next) { + state.reported = { task: next }; + const cause = next === undefined ? "missing_task" : "incompatible_task"; + this.#report( + new Error(`Task ${current.id} keeps running under its old ${current.kind} definition`, { cause }), + ); + } + } + return true; + } + + /** Run the abort handler once; rules 1, 2, and 4 apply, and returning without an outcome faults. */ + async #runAbort(reservation: Reservation): Promise { + const invocation = reservation.invocation; + const current = this.#live.get(invocation.taskId) as ErasedRunningTask | undefined; + if (current === undefined || this.#closing) return; + let failure: { readonly error: unknown } | undefined; + try { + const runtime = this.#runtime(invocation, () => reservation.snapshot); + await erased(reservation.task).abort(current, runtime, invocation.context); + } catch (error) { + failure = { error }; + } + const message = `Abort handler of task ${invocation.taskId} returned without a terminal outcome`; + await this.#step(invocation, (tx, current) => { + tx.setTask(faulted(current, failure?.error ?? new Error(message))); + return false; + }); + } + + /** + * One synchronous decision on the Session line. A terminal task (rule 1) or a closing Harness (rule 2) ends the + * invocation without a write; otherwise `decide` may stage a write and returns whether the invocation continues. + * Ending happens inside the callback. A rejected step, such as admission after close, also ends the invocation. + */ + async #step( + invocation: Invocation, + decide: (tx: Transaction, current: ErasedRunningTask) => boolean, + ): Promise { + try { + return await this.#session.commitWith((tx) => { + const current = this.#live.get(invocation.taskId) as ErasedRunningTask | undefined; + if (current !== undefined && !this.#closing && decide(tx, current)) return current; + this.#end(invocation); + return undefined; + }, this.#context); + } catch (error) { + this.#end(invocation); + if (!this.#closing) this.#report(error); + return undefined; + } + } + + /** End an invocation: its runtime operations reject from now on, its watches stop, and its task is free. */ + #end(invocation: Invocation): void { + if (invocation.ended) return; + invocation.ended = true; + if (this.#invocations.get(invocation.taskId) === invocation) this.#invocations.delete(invocation.taskId); + for (const watch of invocation.watches) void watch.stop(); + } + + #idle(conversationId: ConversationId | undefined): boolean { + for (const record of this.#live.values()) { + if (record.background) continue; + if (conversationId === undefined || record.conversationId === conversationId) return false; + } + return true; + } + + // ─── Invocation runtime ────────────────────────────────────────────────── + + #runtime(invocation: Invocation, snapshot: () => RegistrySnapshot): ErasedRuntime { + return { + taskId: invocation.taskId as TaskId, + conversationId: invocation.conversationId, + signal: invocation.controller.signal, + models: this.#models, + get registry() { + return snapshot(); + }, + commit: (change, context) => + this.#gated( + invocation, + async (tx, current) => { + const next = await change(tx, current); + if (next !== undefined) tx.setTask(withState(current, next)); + }, + context, + ), + memo: ((name: string, ...rest: readonly unknown[]) => + rest.length === 1 + ? this.#readMemo(invocation, name) + : this.#writeMemo(invocation, name, rest[0] as JsonValue, rest[1] as Context)) as ErasedRuntime["memo"], + sleep: (until, context) => this.#sleep(invocation, until, context), + watchDoc: ((...args: readonly unknown[]) => this.#watchDoc(invocation, args)) as ErasedRuntime["watchDoc"], + }; + } + + /** Commit after rereading the task on the line and gating the invocation. */ + #gated( + invocation: Invocation, + change: (tx: Transaction, current: ErasedRunningTask) => T | Promise, + context: Context, + ): Promise { + if (invocation.ended) return Promise.reject(endedError(invocation)); + return this.#session.commitWith( + async (tx) => { + if (invocation.ended) throw endedError(invocation); + if (this.#closing) throw closedError(); + const current = this.#live.get(invocation.taskId) as ErasedRunningTask | undefined; + if (current === undefined) throw new Error(`Task ${invocation.taskId} is terminal`); + if (invocation.mode === "run" && current.abortRequested) { + throw new Error(`Task ${invocation.taskId} has a durable abort mark`); + } + return change(tx, current); + }, + context, + invocation.conversationId, + ); + } + + #readMemo(invocation: Invocation, name: string): Promise { + if (invocation.ended) return Promise.reject(endedError(invocation)); + return Promise.resolve(memoOf(this.#live.get(invocation.taskId), name)); + } + + #writeMemo(invocation: Invocation, name: string, candidate: JsonValue, context: Context): Promise { + return this.#gated( + invocation, + (tx, current) => { + const winner = memoOf(current, name); + if (winner !== undefined) return winner; + tx.setTask({ ...current, memos: { ...current.memos, [name]: candidate } } as AnyTaskRecord); + return candidate; + }, + context, + ); + } + + /** Wait until the Harness clock reaches `until`, rechecking it after every timer. */ + async #sleep(invocation: Invocation, until: number, context: Context): Promise { + if (invocation.ended) throw endedError(invocation); + const signals = [invocation.controller.signal]; + if (context.abortSignal !== undefined) signals.push(context.abortSignal); + const signal = AbortSignal.any(signals); + for (;;) { + signal.throwIfAborted(); + const remaining = until - this.#now(); + if (remaining <= 0) return; + await delay(Math.min(remaining, MAX_TIMER_DELAY), signal); + } + } + + async #watchDoc(invocation: Invocation, args: readonly unknown[]): Promise | undefined> { + if (invocation.ended) throw endedError(invocation); + const watchDoc = this.#session.watchDoc.bind(this.#session) as ( + ...args: readonly unknown[] + ) => Promise | undefined>; + const watch = await watchDoc(...args); + if (watch === undefined) return undefined; + if (invocation.ended) { + void watch.stop(); + throw endedError(invocation); + } + invocation.watches.add(watch); + void watch.closed.then(() => invocation.watches.delete(watch)); + return watch; + } +} + +/** Own memo entry only; memo names such as `toString` must not resolve to inherited properties. */ +function memoOf(record: LiveTaskRecord | undefined, name: string): JsonValue | undefined { + const memos = record?.memos; + return memos !== undefined && Object.hasOwn(memos, name) ? memos[name] : undefined; +} + +function erased(task: AnyTask): ErasedDefinition { + return task.definition as unknown as ErasedDefinition; +} + +function endedError(invocation: Invocation): Error { + return new Error(`Task ${invocation.taskId} invocation has ended`); +} + +function closedError(): Error { + return new Error("Harness is closed"); +} + +function terminal(outcome: TaskOutcome): NextTaskState { + return { status: "terminal", outcome }; +} + +function faulted(record: LiveTaskRecord, error: unknown): AnyTaskRecord { + const message = error instanceof Error ? error.message : String(error); + return withState(record, terminal({ status: "faulted", error: { message } })); +} + +/** Replace a live record's state; memos disappear in the terminal replacement. */ +function withState(record: LiveTaskRecord, state: TaskState): AnyTaskRecord { + if (state.status !== "terminal") return { ...record, state } as AnyTaskRecord; + const { memos: _memos, ...rest } = record; + return { ...rest, state }; +} + +/** Whether a definition can take the task at reservation: same version, or newer with a migration. */ +function canReserve(task: AnyTask, record: LiveTaskRecord): boolean { + const definition = task.definition; + return ( + definition.version === record.version || (definition.version > record.version && definition.migrate !== undefined) + ); +} + +/** Wait in `waiters[key]` until settled or `context` is cancelled. */ +function addWaiter(waiters: Map>>, key: K, context: Context): Promise { + return new Promise((resolve, reject) => { + const signal = context.abortSignal; + if (signal?.aborted) return reject(signal.reason); + let set = waiters.get(key); + if (set === undefined) { + set = new Set(); + waiters.set(key, set); + } + const own = set; + const remove = (): void => { + own.delete(waiter); + if (own.size === 0 && waiters.get(key) === own) waiters.delete(key); + signal?.removeEventListener("abort", onAbort); + }; + const waiter: Waiter = { + resolve: (value) => { + remove(); + resolve(value); + }, + reject: (error) => { + remove(); + reject(error); + }, + }; + const onAbort = (): void => waiter.reject(signal!.reason); + signal?.addEventListener("abort", onAbort, { once: true }); + own.add(waiter); + }); +} + +function settleWaiters(waiters: Map>>, key: K, settle: (waiter: Waiter) => void): void { + const set = waiters.get(key); + if (set === undefined) return; + waiters.delete(key); + for (const waiter of [...set]) settle(waiter); +} + +function delay(ms: number, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const onAbort = (): void => { + clearTimeout(timer); + reject(signal.reason); + }; + const timer = setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(); + }, ms); + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + +function jsonEqual(left: JsonValue | undefined, right: JsonValue | undefined): boolean { + if (left === right) return true; + if (typeof left !== "object" || typeof right !== "object" || left === null || right === null) return false; + if (Array.isArray(left) || Array.isArray(right)) { + if (!Array.isArray(left) || !Array.isArray(right) || left.length !== right.length) return false; + return left.every((value, index) => jsonEqual(value, right[index])); + } + const keys = Object.keys(left); + if (keys.length !== Object.keys(right).length) return false; + return keys.every((key) => Object.hasOwn(right, key) && jsonEqual(left[key], right[key])); +} diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index a49b5402a..f604976e9 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -215,6 +215,8 @@ export interface RegistrySnapshot { /** Immutable view of the whole current registry. */ snapshot(): RegistrySnapshot; + /** Called synchronously after every publication; wakes the scheduler to reconsider blocked tasks. */ + subscribe(listener: () => void): () => void; } /** Application-owned registry of tools, hooks, tasks, and the system prompt. */ @@ -305,16 +307,31 @@ export interface Conversation { context: Context, ): Promise>; fork(at: EntryId, options: ConversationCreateOptions, context: Context): Promise; + /** Resolve when no live non-background task belongs to this conversation. */ + waitForIdle(context: Context): Promise; } // TODO: decide how Harness exposes subscribeCommits() and subscribeClose(). Their listeners run on the Session line // and must not throw or call Session APIs, and Harness close will also join task invocations. /** Durable agent harness over one Session. */ export interface Harness extends Session { + /** Enable task scheduling. Idempotent; throws after close. */ + resume(): void; + /** Return the reserved root conversation, creating it with `init` in one commit when absent. */ root(context: Context, options?: { readonly init?: ConversationInit }): Promise; - // Conversation activity (active/idle notifications, quiescence) is specified with the task runtime and turn - // control in Packages 14 and 17. + // Conversation activity (active/idle notifications) is specified with turn control in Package 17. conversation(id: ConversationId, context: Context): Promise; createConversation(options: ConversationCreateOptions, context: Context): Promise; + + getTask(id: TaskId, context: Context): Promise | undefined>; + /** + * Commit the abort mark, signal and join an active run invocation, and schedule the abort invocation. A task whose + * definition cannot take it settles as `orphaned` instead. + */ + abortTask(id: TaskId, context: Context): Promise<"marked" | "terminal">; + /** Resolve with the terminal receipt; cancelling `context` cancels only this wait. */ + waitForTask(id: TaskId, context: Context): Promise>; + /** Resolve when no live non-background task exists. */ + waitForIdle(context: Context): Promise; } diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index 92924ba2e..008ea71f1 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -40,6 +40,7 @@ export type { } from "./harness/types.ts"; export { createSession } from "./session/session.ts"; export { MemoryStorage } from "./storage/memory.ts"; +export { defineTask } from "./tasks.ts"; export type { CheckpointInfo, CommitChange, @@ -77,10 +78,13 @@ export type { Id, JsonObject, LatestConversationSemantics, + NextTaskState, Page, + PhaseHandler, RewindableConversationDocFamilyToken, RewindableConversationDocToken, RewindableConversationSemantics, + RunningTask, Seq, Session, SessionDocFamilyToken, @@ -102,6 +106,7 @@ export type { TaskOutcomeError, TaskQuery, TaskRecord, + TaskRuntime, TaskState, Tx, WatchEnd, diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index 57a65183d..500d13c22 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -83,9 +83,12 @@ export class SessionImpl implements Session { return this.#enqueue(() => this.#runCommit(change, context)); } - /** Internal commit whose `tx.createTask()` defaults to `defaultConversationId`; used by Conversation handles. */ + /** + * Internal commit exposing the concrete transaction and its internal operations, such as the reserved-ID root + * bootstrap and task replacement. `tx.createTask()` defaults to `defaultConversationId`. + */ commitWith( - change: (tx: Tx) => T | Promise, + change: (tx: Transaction) => T | Promise, context: Context, defaultConversationId?: ConversationId, ): Promise { @@ -97,16 +100,6 @@ export class SessionImpl implements Session { return this.#enqueue(() => this.#runCommit(change, context, defaultConversationId)); } - /** Internal commit exposing the concrete transaction, whose reserved-ID root bootstrap is not on the public `Tx`. */ - commitRoot(change: (tx: Transaction) => T | Promise, context: Context): Promise { - try { - this.#assertUsable(); - } catch (error) { - return Promise.reject(error); - } - return this.#enqueue(() => this.#runCommit(change, context)); - } - /** Internal: run a read-only job on the mutation line so multi-read derivations observe one committed state. */ readOnLine(job: () => Promise): Promise { try { @@ -363,17 +356,28 @@ export class SessionImpl implements Session { close(context: Context): Promise { if (this.#closing === undefined) { const cleanup = withoutAbortSignal(context); - this.#closing = this.#enqueue(async () => { - for (const listener of [...this.#closeListeners]) listener(); - this.#closeListeners.clear(); - this.#commitListeners.clear(); - this.#documents.clear(); - await this.#storage.close(cleanup); - }); + // Seal admission before anything else runs, then stop observers; admitted work settles before Storage closes. + this.#closing = Promise.resolve() + .then(() => this.beforeClose()) + .then(() => + this.#enqueue(async () => { + this.#commitListeners.clear(); + this.#documents.clear(); + await this.#storage.close(cleanup); + }), + ); + const listeners = [...this.#closeListeners]; + this.#closeListeners.clear(); + for (const listener of listeners) listener(); } return awaitWithContext(this.#closing, context); } + /** Runs after close seals admission and before the line closes Storage; must not reject. */ + protected beforeClose(): Promise { + return Promise.resolve(); + } + /** Register a synchronous post-adoption listener. It must not throw, block, or call Session operations. */ subscribeCommits(listener: (publication: CommitPublication, context: Context) => void): () => void { this.#assertUsable(); @@ -381,7 +385,7 @@ export class SessionImpl implements Session { return () => this.#commitListeners.delete(listener); } - /** Register a synchronous close listener. It must not throw, block, or call Session operations. */ + /** Register a listener called synchronously when close begins. It must not throw, block, or call Session operations. */ subscribeClose(listener: () => void): () => void { this.#assertUsable(); this.#closeListeners.add(listener); diff --git a/packages/durable/src/session/transaction.ts b/packages/durable/src/session/transaction.ts index 258f8f623..d1ee6b1c8 100644 --- a/packages/durable/src/session/transaction.ts +++ b/packages/durable/src/session/transaction.ts @@ -308,6 +308,7 @@ export class Transaction implements Tx { }); } + /** Internal: replace one task record completely. Tasks change their own state through their runtime. */ setTask(value: AnyTaskRecord): void { this.#assertOpen(); this.#hasTableWrite = true; diff --git a/packages/durable/src/tasks.ts b/packages/durable/src/tasks.ts new file mode 100644 index 000000000..aa7a32c21 --- /dev/null +++ b/packages/durable/src/tasks.ts @@ -0,0 +1,8 @@ +import type { Task, TaskDefinition } from "./types.ts"; + +/** Define an executable task. Register it in the registry so a Harness can run tasks of its kind. */ +export function defineTask( + definition: TaskDefinition, +): Task { + return { definition }; +} diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index f353c235c..f617171d8 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -1,6 +1,7 @@ import type { AttachedReplicatedState, Context, Draft, JsonValue } from "@earendil-works/chord"; import type { Op } from "@earendil-works/chord/delta"; -import type { Message } from "@earendil-works/pi-ai"; +import type { Message, Models } from "@earendil-works/pi-ai"; +import type { RegistrySnapshot } from "./harness/types.ts"; /** JSON object used as the root of every durable document. */ export type JsonObject = { [key: string]: JsonValue }; @@ -124,9 +125,59 @@ export type TaskDocFamilyToken = DocF DocFamilyDefinition & { readonly scope: "task" } >; -declare const taskResultType: unique symbol; +/** Live task record reserved by one invocation. */ +export type RunningTask = TaskRecord & { + readonly state: Extract, { readonly status: "running" }>; +}; -/** Task definition fields currently supported by Session task creation. */ +/** Next state a task commits for itself: a replacement checkpoint or its terminal outcome. */ +export type NextTaskState = Extract, { readonly status: "running" | "terminal" }>; + +/** + * Runs one checkpoint phase. It must commit a changed checkpoint or a terminal outcome through `runtime.commit()`; + * returning without durable progress faults the task. + */ +export type PhaseHandler = ( + task: RunningTask, + runtime: TaskRuntime, + context: Context, +) => Promise; + +/** + * Operations of one task invocation. Every operation rejects after the invocation ends; watches acquired through it + * stop at invocation end. `_H` is the task's hook map, consumed once the runtime gains its hook runner. + */ +export interface TaskRuntime extends DocumentObserver { + readonly taskId: TaskId; + readonly conversationId: ConversationId; + /** Aborted when the run is signalled by `abortTask()` or the Harness closes. */ + readonly signal: AbortSignal; + /** Registry snapshot of the current phase; refreshed at every phase boundary. */ + readonly registry: RegistrySnapshot; + readonly models: Models; + + /** + * Commit on the Session line after rereading the task. Rejects when the task is terminal, the invocation ended, the + * Harness is closing, or, in a run invocation, the task carries an abort mark. A returned state replaces the task's + * state in the same commit; returning nothing leaves it unchanged. `tx.createTask()` defaults to the task's + * conversation. + */ + commit( + change: ( + tx: Tx, + current: RunningTask, + ) => NextTaskState | undefined | Promise | undefined>, + context: Context, + ): Promise; + /** Read a durable memo of this task. */ + memo(name: string, context: Context): Promise; + /** Store `candidate` unless a memo already exists; return the durable winner. */ + memo(name: string, candidate: T, context: Context): Promise; + /** Resolve once the Harness clock reaches `until`; rejects when the invocation or `context` is cancelled. */ + sleep(until: number, context: Context): Promise; +} + +/** Executable durable state machine definition, registered in the registry by `name`. */ export type TaskDefinition = { /** Registered task kind persisted in `TaskRecord.kind`. */ readonly name: string; @@ -134,9 +185,22 @@ export type TaskDefinition readonly version: number; /** First durable checkpoint for a newly created task. */ initial(input: I): S; + /** Exhaustive phase map; each handler receives the task narrowed to its phase. */ + readonly phases: { + readonly [P in S["phase"]]: PhaseHandler, S, R, H>; + }; + /** Runs in a fresh invocation after an abort mark and must commit a terminal outcome. */ + abort(task: RunningTask, runtime: TaskRuntime, context: Context): Promise; + /** Convert a record stored by any older supported version; runs at reservation. */ + migrate?( + input: JsonValue, + checkpoint: JsonValue, + fromVersion: number, + ): { + input: I; + checkpoint: S; + }; readonly hooks?: H; - /** Type-only result marker until phase handlers commit typed outcomes. */ - readonly [taskResultType]?: R; }; /** Typed executable task definition. */ @@ -608,8 +672,6 @@ export interface Tx { input: I, options?: TaskOptions, ): Promise>; - /** Replace one task record completely. */ - setTask(value: TaskRecord): void; doc(token: SessionDocToken): Promise>; doc(token: ConversationDocToken, conversationId: ConversationId): Promise>; @@ -710,7 +772,7 @@ export interface Session extends DocumentObserver { close(context: Context): Promise; /** Observe complete commits synchronously after adoption. The listener must not throw, block, or call Session APIs. */ subscribeCommits(listener: (publication: CommitPublication, context: Context) => void): () => void; - /** Observe close synchronously on the Session line. The listener must not throw, block, or call Session APIs. */ + /** Observe close synchronously when it begins. The listener must not throw, block, or call Session APIs. */ subscribeClose(listener: () => void): () => void; snapshot(token: SessionDocToken, context: Context): Promise | undefined>; diff --git a/packages/durable/test/harness-conversations.test.ts b/packages/durable/test/harness-conversations.test.ts index 4b1859570..036d980f9 100644 --- a/packages/durable/test/harness-conversations.test.ts +++ b/packages/durable/test/harness-conversations.test.ts @@ -7,10 +7,10 @@ import { type ConversationId, defineDoc, defineEntry, + defineTask, type EntryRecord, MemoryStorage, ROOT_CONVERSATION_ID, - type Task, } from "@earendil-works/pi-durable"; import { afterEach, describe, expect, it } from "vitest"; import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; @@ -281,9 +281,13 @@ describe("Harness root and conversations", () => { it("binds commits and task creation to the conversation", async () => { const { harness } = await openHarness(new MemoryStorage()); const conversation = await harness.createConversation({ ownership: { kind: "ownerless" } }, context); - const task: Task<{ n: number }, { phase: "run" }, void, object> = { - definition: { name: "test.work", version: 1, initial: () => ({ phase: "run" }) }, - }; + const task = defineTask<{ n: number }, { phase: "run" }, null>({ + name: "test.work", + version: 1, + initial: () => ({ phase: "run" }), + phases: { run: async () => {} }, + abort: async () => {}, + }); const taskId = await conversation.commit((tx) => tx.createTask(task, { n: 1 }), context); const record = await harness.commit((tx) => tx.task(taskId), context); expect(record?.conversationId).toBe(conversation.id); diff --git a/packages/durable/test/harness-registry.test.ts b/packages/durable/test/harness-registry.test.ts index e824a82b2..a9ab5c40e 100644 --- a/packages/durable/test/harness-registry.test.ts +++ b/packages/durable/test/harness-registry.test.ts @@ -1,6 +1,6 @@ import type { JsonValue } from "@earendil-works/chord"; import { Type } from "@earendil-works/pi-ai"; -import { type AnyTask, createRegistry, type ToolRegistration } from "@earendil-works/pi-durable"; +import { createRegistry, defineTask, type ToolRegistration } from "@earendil-works/pi-durable"; import { describe, expect, it } from "vitest"; type AppTool = ToolRegistration & { readonly snippet?: string }; @@ -17,16 +17,14 @@ function tool(name: string, extra: Partial = {}): AppTool { function task(name: string) { const hooks: { beforeRun(): void } = { beforeRun: () => {} }; - return { - definition: { - name, - version: 1, - initial: (_input: undefined) => ({ phase: "run" as const }), - phases: {}, - abort: () => {}, - hooks, - }, - } satisfies AnyTask; + return defineTask({ + name, + version: 1, + initial: () => ({ phase: "run" }), + phases: { run: async () => {} }, + abort: async () => {}, + hooks, + }); } function names(tools: readonly { readonly name: string }[]): string[] { diff --git a/packages/durable/test/harness-tasks-recovery.test.ts b/packages/durable/test/harness-tasks-recovery.test.ts new file mode 100644 index 000000000..80fe55902 --- /dev/null +++ b/packages/durable/test/harness-tasks-recovery.test.ts @@ -0,0 +1,822 @@ +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 } from "@earendil-works/pi-ai"; +import { + createRegistry, + defineDoc, + defineTask, + Harness, + MemoryStorage, + type Registry, + type Task, + type TaskId, + type TaskRuntime, +} from "@earendil-works/pi-durable"; +import { afterEach, describe, expect, it } from "vitest"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { ControlledStorage, context, flush } from "./session-support.ts"; +import { aborted, abortedWith, completed, countingReader, deferred, eventually, openTasks } from "./task-support.ts"; + +const directories = new Set(); + +async function sqlitePath(): Promise { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-tasks-")); + directories.add(directory); + return join(directory, "session.sqlite"); +} + +afterEach(async () => { + for (const directory of directories) await rm(directory, { recursive: true, force: true }); + directories.clear(); +}); + +type OpenHarness = Awaited>["harness"]; + +async function createIn( + harness: OpenHarness, + task: Task, +): Promise> { + const root = await harness.root(context); + return root.commit((tx) => tx.createTask(task, null), context); +} + +/** Fake external service whose operations are idempotent by request key. */ +class TransferService { + readonly applied = new Map(); + calls = 0; + + async apply(key: string, amount: number): Promise { + this.calls++; + if (!this.applied.has(key)) this.applied.set(key, amount * 10); + return this.applied.get(key)!; + } +} + +type TransferState = { phase: "prepare" } | { phase: "apply"; key: string }; + +/** + * Intent/effect/outcome task: `prepare` commits the intent, `apply` performs the effect and commits the outcome. + * While `interrupt.first` is set, the first `apply` blocks after the effect until the invocation is signalled. + */ +function transferTask(service: TransferService, interrupt: { first: boolean }) { + return defineTask<{ amount: number }, TransferState, { receipt: number }>({ + name: "test.transfer", + version: 1, + initial: () => ({ phase: "prepare" }), + phases: { + prepare: async (task, runtime, ctx) => { + await runtime.memo("requested", task.input.amount, ctx); + await runtime.commit( + () => ({ status: "running", checkpoint: { phase: "apply", key: `transfer-${task.id}` } }), + ctx, + ); + }, + apply: async (task, runtime, ctx) => { + const receipt = await service.apply(task.state.checkpoint.key, task.input.amount); + if (interrupt.first) { + interrupt.first = false; + await aborted(runtime.signal); + } + await runtime.commit(() => completed({ receipt }), ctx); + }, + }, + abort: async (_task, runtime, ctx) => { + await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx); + }, + }); +} + +type Step = { phase: "run" }; + +/** A versioned one-phase task that completes with `result`, optionally migrating older records. */ +function versioned( + version: number, + result: string, + migrate?: (input: JsonValue, checkpoint: JsonValue, fromVersion: number) => { input: null; checkpoint: Step }, +) { + return defineTask({ + name: "test.versioned", + version, + initial: () => ({ phase: "run" }), + phases: { run: (_task, runtime, ctx) => runtime.commit(() => completed(result), ctx) }, + abort: (_task, runtime, ctx) => runtime.commit(() => abortedWith(result), ctx), + ...(migrate === undefined ? {} : { migrate }), + }); +} + +describe("task recovery", () => { + it("resumes an intent/effect/outcome task interrupted after its intent across close and reopen", async () => { + const path = await sqlitePath(); + const service = new TransferService(); + const Transfer = transferTask(service, { first: true }); + + const first = await openTasks(await openNodeSqliteStorage(path), [Transfer]); + const root = await first.harness.root(context); + const id = await root.commit((tx) => tx.createTask(Transfer, { amount: 7 }), context); + first.harness.resume(); + await eventually(() => service.calls === 1); + await first.harness.close(context); + + const second = await openTasks(await openNodeSqliteStorage(path), [Transfer]); + // Open reconciled `running` to `pending` and kept the checkpoint and memos; nothing ran yet. + expect(await second.harness.getTask(id, context)).toMatchObject({ + abortRequested: false, + memos: { requested: 7 }, + state: { status: "pending", checkpoint: { phase: "apply", key: `transfer-${id}` } }, + }); + expect(service.calls).toBe(1); + second.harness.resume(); + const receipt = await second.harness.waitForTask(id, context); + expect(receipt.state.outcome).toEqual({ status: "completed", result: { receipt: 70 } }); + expect(service.calls).toBe(2); + expect(service.applied.size).toBe(1); + await second.harness.close(context); + + const third = await openTasks(await openNodeSqliteStorage(path), [Transfer]); + expect(await third.harness.getTask(id, context)).toEqual(receipt); + await third.harness.close(context); + }); + + it("resumes abort work after close at every direct-task abort stage", async () => { + const path = await sqlitePath(); + const log: string[] = []; + const abortGate = { block: true }; + const abortReached = deferred(); + const runRelease = deferred(); + const Abortable = defineTask({ + name: "test.abortable", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async () => { + log.push("run"); + // Ignores the abort signal until released, so the mark is durable while the run is active. + await runRelease.promise; + }, + }, + abort: async (_task, runtime, ctx) => { + log.push("abort"); + if (abortGate.block) { + abortReached.resolve(); + await aborted(runtime.signal); + } + await runtime.commit(() => abortedWith("stop"), ctx); + }, + }); + + // Stage 1: the mark is committed while the run invocation is still active. + let opened = await openTasks(await openNodeSqliteStorage(path), [Abortable]); + const id = await createIn(opened.harness, Abortable); + opened.harness.resume(); + await eventually(() => log.length === 1); + const aborting = opened.harness.abortTask(id, context); + while (!(await opened.harness.getTask(id, context))?.abortRequested) await flush(); + const closing = opened.harness.close(context); + runRelease.resolve(); + await closing; + expect(await aborting).toBe("marked"); + expect(log).toEqual(["run"]); + + // Stage 2: reopen dispatches the abort invocation, never the run; close while it is active. + opened = await openTasks(await openNodeSqliteStorage(path), [Abortable]); + expect(await opened.harness.getTask(id, context)).toMatchObject({ + abortRequested: true, + state: { status: "pending" }, + }); + opened.harness.resume(); + await abortReached.promise; + await opened.harness.close(context); + expect(log).toEqual(["run", "abort"]); + + // Stage 3: a fresh abort invocation settles the task. + abortGate.block = false; + opened = await openTasks(await openNodeSqliteStorage(path), [Abortable]); + opened.harness.resume(); + expect((await opened.harness.waitForTask(id, context)).state.outcome).toEqual({ + status: "aborted", + reason: "stop", + }); + await opened.harness.close(context); + expect(log).toEqual(["run", "abort", "abort"]); + + // Stage 4: the terminal receipt survives reopen and nothing runs again. + opened = await openTasks(await openNodeSqliteStorage(path), [Abortable]); + opened.harness.resume(); + await flush(); + expect(await opened.harness.abortTask(id, context)).toBe("terminal"); + await opened.harness.close(context); + expect(log).toEqual(["run", "abort", "abort"]); + }); +}); + +/** + * Crash simulation: the crashed Harness is abandoned without close, its held storage commit never lands, and its + * blocked handlers never return. A new Harness then opens the same storage. + */ +describe("task crash recovery", () => { + type Log = string[]; + + /** A task whose run ignores its signal and whose abort handler blocks while `blockAbort` is set. */ + function crashTask(log: Log, options: { blockAbort: boolean }) { + return defineTask({ + name: "test.crash", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async () => { + log.push("run"); + await new Promise(() => {}); + }, + }, + abort: async (_task, runtime, ctx) => { + log.push("abort"); + if (options.blockAbort) await new Promise(() => {}); + await runtime.commit(() => abortedWith("recovered"), ctx); + }, + }); + } + + async function recover(storage: ControlledStorage, log: Log): Promise<{ harness: OpenHarness; id: TaskId }> { + storage.crash(); + const { harness } = await openTasks(storage, [crashTask(log, { blockAbort: false })]); + const page = await harness.commit((tx) => tx.scanTasks({ kind: "test.crash" }, 1), context); + return { harness, id: page.items[0]!.id }; + } + + async function crashedRun(log: Log): Promise<{ storage: ControlledStorage; harness: OpenHarness; id: TaskId }> { + const storage = new ControlledStorage(); + const { harness } = await openTasks(storage, [crashTask(log, { blockAbort: true })]); + const id = await createIn(harness, crashTask(log, { blockAbort: true })); + harness.resume(); + await eventually(() => log.length === 1); + return { storage, harness, id }; + } + + it("crash while the mark commit is in storage: the run resumes", async () => { + const log: Log = []; + const { storage, harness, id } = await crashedRun(log); + const held = storage.holdCommits(); + void harness.abortTask(id, context); + await held.entered; + const recovered = await recover(storage, log); + expect(await recovered.harness.getTask(id, context)).toMatchObject({ + abortRequested: false, + state: { status: "pending" }, + }); + recovered.harness.resume(); + await eventually(() => log.length === 2); + expect(log).toEqual(["run", "run"]); + }); + + it("crash after the mark before the run joins: only the abort handler runs", async () => { + const log: Log = []; + const { storage, harness, id } = await crashedRun(log); + // The run ignores its signal, so abortTask never finishes joining it. + let joined = false; + void harness.abortTask(id, context).then(() => { + joined = true; + }); + while (!(await harness.getTask(id, context))?.abortRequested) await flush(); + await flush(); + expect(joined).toBe(false); + const recovered = await recover(storage, log); + expect(await recovered.harness.getTask(id, context)).toMatchObject({ + abortRequested: true, + state: { status: "pending" }, + }); + recovered.harness.resume(); + expect((await recovered.harness.waitForTask(id, context)).state.outcome).toEqual({ + status: "aborted", + reason: "recovered", + }); + expect(log).toEqual(["run", "abort"]); + await recovered.harness.close(context); + }); + + it("crash while the abort handler runs: a fresh abort invocation settles the task", async () => { + const log: Log = []; + const storage = new ControlledStorage(); + const Crash = crashTask(log, { blockAbort: true }); + const { harness } = await openTasks(storage, [Crash]); + const id = await createIn(harness, Crash); + await harness.abortTask(id, context); + harness.resume(); + await eventually(() => log.length === 1); + expect(log).toEqual(["abort"]); + expect(await harness.getTask(id, context)).toMatchObject({ state: { status: "running" } }); + const recovered = await recover(storage, log); + recovered.harness.resume(); + expect((await recovered.harness.waitForTask(id, context)).state.outcome).toEqual({ + status: "aborted", + reason: "recovered", + }); + expect(log).toEqual(["abort", "abort"]); + await recovered.harness.close(context); + }); + + it("crash while the abort outcome is in storage: the abort handler runs again", async () => { + const log: Log = []; + const storage = new ControlledStorage(); + const reached = deferred(); + const proceed = deferred(); + const Crash = defineTask({ + name: "test.crash", + version: 1, + initial: () => ({ phase: "run" }), + phases: { run: async () => {} }, + abort: async (_task, runtime, ctx) => { + log.push("abort"); + reached.resolve(); + await proceed.promise; + await runtime.commit(() => abortedWith("lost"), ctx); + }, + }); + const { harness } = await openTasks(storage, [Crash]); + const id = await createIn(harness, Crash); + await harness.abortTask(id, context); + harness.resume(); + await reached.promise; + const held = storage.holdCommits(); + proceed.resolve(); + await held.entered; + const recovered = await recover(storage, log); + recovered.harness.resume(); + expect((await recovered.harness.waitForTask(id, context)).state.outcome).toEqual({ + status: "aborted", + reason: "recovered", + }); + expect(log).toEqual(["abort", "abort"]); + await recovered.harness.close(context); + }); + + it("crash after the terminal outcome: nothing runs again", async () => { + const log: Log = []; + const storage = new ControlledStorage(); + const Crash = crashTask(log, { blockAbort: false }); + const { harness } = await openTasks(storage, [Crash]); + const id = await createIn(harness, Crash); + await harness.abortTask(id, context); + harness.resume(); + await harness.waitForTask(id, context); + const recovered = await recover(storage, log); + recovered.harness.resume(); + await flush(); + expect(log).toEqual(["abort"]); + expect((await recovered.harness.getTask(id, context))?.state).toEqual({ + status: "terminal", + outcome: { status: "aborted", reason: "recovered" }, + }); + await recovered.harness.close(context); + }); + + it("crash while the reservation commit is in storage: the task is still pending", async () => { + const log: Log = []; + const storage = new ControlledStorage(); + const Crash = crashTask(log, { blockAbort: false }); + const { harness } = await openTasks(storage, [Crash]); + const id = await createIn(harness, Crash); + const held = storage.holdCommits(); + harness.resume(); + await held.entered; + const recovered = await recover(storage, log); + expect((await recovered.harness.getTask(id, context))?.state.status).toBe("pending"); + expect(log).toEqual([]); + await recovered.harness.abortTask(id, context); + recovered.harness.resume(); + await recovered.harness.waitForTask(id, context); + expect(log).toEqual(["abort"]); + await recovered.harness.close(context); + }); +}); + +describe("blocked tasks", () => { + it("keeps a task with a missing definition pending, and live for idle waits, until registration", async () => { + const V1 = versioned(1, "v1"); + const { harness, registry } = await openTasks(new MemoryStorage(), []); + const id = await createIn(harness, V1); + harness.resume(); + await flush(); + expect((await harness.getTask(id, context))?.state.status).toBe("pending"); + const idle = harness.waitForIdle(context); + await flush(); + registry.tasks.add(V1); + await idle; + expect((await harness.getTask(id, context))?.state).toEqual({ + status: "terminal", + outcome: { status: "completed", result: "v1" }, + }); + await harness.close(context); + }); + + it("keeps a task stored by a newer version pending until a fitting definition is registered", async () => { + const registry = createRegistry(); + const old = registry.tasks.add(versioned(1, "old")); + const { harness } = await openTasks(new MemoryStorage(), [], { registry }); + const id = await createIn(harness, versioned(2, "new")); + harness.resume(); + await flush(); + expect((await harness.getTask(id, context))?.state.status).toBe("pending"); + registry.batch(() => { + old.dispose(); + registry.tasks.add(versioned(2, "new")); + }); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "completed", result: "new" }); + await harness.close(context); + }); + + it("migrates at reservation, leaves the record unchanged when migration fails, and retries only for a new definition", async () => { + const path = await sqlitePath(); + let opened = await openTasks(await openNodeSqliteStorage(path), []); + const id = await createIn(opened.harness, versioned(1, "v1")); + await opened.harness.close(context); + + const registry: Registry = createRegistry(); + let failures = 0; + const failing = registry.tasks.add( + versioned(2, "v2", () => { + failures++; + throw new Error("cannot migrate"); + }), + ); + opened = await openTasks(await openNodeSqliteStorage(path), [], { registry }); + opened.harness.resume(); + await eventually(() => opened.reports.length === 1); + // An unrelated registry change wakes the scheduler without retrying the same failed definition. + registry.tools.add({ + name: "unrelated", + description: "unrelated", + parameters: { type: "object", properties: {} } as never, + execute: async () => ({}), + }); + await flush(); + expect(failures).toBe(1); + expect(opened.reports).toHaveLength(1); + expect(String(opened.reports[0])).toContain("cannot migrate"); + expect(await opened.harness.getTask(id, context)).toMatchObject({ + version: 1, + state: { status: "pending", checkpoint: { phase: "run" } }, + }); + + const migrations: number[] = []; + registry.batch(() => { + failing.dispose(); + registry.tasks.add( + versioned(2, "v2", (_input, checkpoint, fromVersion) => { + migrations.push(fromVersion); + return { input: null, checkpoint: checkpoint as Step }; + }), + ); + }); + const receipt = await opened.harness.waitForTask(id, context); + expect(receipt).toMatchObject({ version: 2, state: { outcome: { status: "completed", result: "v2" } } }); + expect(migrations).toEqual([1]); + await opened.harness.close(context); + }); + + it("blocks an older record whose newer definition has no migration", async () => { + const path = await sqlitePath(); + let opened = await openTasks(await openNodeSqliteStorage(path), []); + const id = await createIn(opened.harness, versioned(1, "v1")); + await opened.harness.close(context); + opened = await openTasks(await openNodeSqliteStorage(path), [versioned(2, "v2")]); + opened.harness.resume(); + await eventually(() => opened.reports.length === 1); + expect(String(opened.reports[0])).toContain("has no migration from 1"); + expect(await opened.harness.getTask(id, context)).toMatchObject({ version: 1, state: { status: "pending" } }); + await opened.harness.close(context); + }); + + it("settles an aborted blocked task as orphaned and retires its documents", async () => { + const Scratch = defineDoc<{ n: number }>({ + kind: "test.orphan-scratch", + version: 1, + scope: "task", + initial: () => ({ n: 0 }), + }); + const { harness } = await openTasks(new MemoryStorage(), []); + const root = await harness.root(context); + const id = await root.commit(async (tx) => { + const created = await tx.createTask(versioned(1, "x"), null); + (await tx.doc(Scratch, created)).n = 1; + return created; + }, context); + // Before resume: the marking commit settles the blocked task directly. + expect(await harness.abortTask(id, context)).toBe("marked"); + expect((await harness.getTask(id, context))?.state).toEqual({ + status: "terminal", + outcome: { status: "orphaned", reason: "missing_task" }, + }); + expect(await harness.snapshot(Scratch, id, context)).toBeUndefined(); + await harness.close(context); + }); + + it("orphans a marked task whose definition disappeared while its run was active", async () => { + const reached = deferred(); + const Running = defineTask({ + name: "test.vanishing", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (_task, runtime) => { + reached.resolve(); + await aborted(runtime.signal); + }, + }, + abort: async () => { + throw new Error("must not run"); + }, + }); + const registry = createRegistry(); + const registration = registry.tasks.add(Running); + const { harness } = await openTasks(new MemoryStorage(), [], { registry }); + const id = await createIn(harness, Running); + harness.resume(); + await reached.promise; + registration.dispose(); + // An active run means the mark does not orphan directly; the scheduler orphans once the run has ended. + expect(await harness.abortTask(id, context)).toBe("marked"); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ + status: "orphaned", + reason: "missing_task", + }); + await harness.close(context); + }); + + it("orphans a reopened abort-marked task without a definition once scheduling resumes", async () => { + const path = await sqlitePath(); + let opened = await openTasks(await openNodeSqliteStorage(path), [versioned(1, "x")]); + const id = await createIn(opened.harness, versioned(1, "x")); + expect(await opened.harness.abortTask(id, context)).toBe("marked"); + await opened.harness.close(context); + + opened = await openTasks(await openNodeSqliteStorage(path), []); + expect(await opened.harness.getTask(id, context)).toMatchObject({ + abortRequested: true, + state: { status: "pending" }, + }); + const idle = opened.harness.waitForIdle(context); + opened.harness.resume(); + await idle; + expect((await opened.harness.getTask(id, context))?.state).toEqual({ + status: "terminal", + outcome: { status: "orphaned", reason: "missing_task" }, + }); + await opened.harness.close(context); + }); +}); + +describe("definition handover", () => { + type Handover = { phase: "a" } | { phase: "b" } | { phase: "c" }; + type Gates = { readonly a?: Promise; readonly b?: Promise; readonly onEnd?: () => void }; + + function handoverTask( + label: string, + version: number, + log: string[], + options: { readonly gates?: Gates; readonly migrate?: () => { input: null; checkpoint: Handover } } = {}, + ) { + const advance = + (phase: "a" | "b", next: Handover) => + async (_task: unknown, runtime: TaskRuntime, ctx: Context) => { + log.push(`${label}:${phase} start`); + await options.gates?.[phase]; + await runtime.commit(() => ({ status: "running", checkpoint: next }), ctx); + // Leave room for a wrongly dispatched successor before this invocation ends. + await flush(); + options.gates?.onEnd?.(); + log.push(`${label}:${phase} end`); + }; + return defineTask({ + name: "test.handover", + version, + initial: () => ({ phase: "a" }), + phases: { + a: advance("a", { phase: "b" }), + b: advance("b", { phase: "c" }), + c: async (_task, runtime, ctx) => { + log.push(`${label}:c`); + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async (_task, runtime, ctx) => { + log.push(`${label}:abort`); + await runtime.commit(() => abortedWith(label), ctx); + }, + ...(options.migrate === undefined ? {} : { migrate: options.migrate }), + }); + } + + async function startHandover(log: string[], gates: Gates, storage = new MemoryStorage()) { + const registry = createRegistry(); + const old = registry.tasks.add(handoverTask("old", 1, log, { gates })); + const opened = await openTasks(storage, [], { registry }); + const id = await createIn(opened.harness, handoverTask("old", 1, log)); + opened.harness.resume(); + await eventually(() => log.length === 1); + return { ...opened, old, id }; + } + + it("hands over at the next phase boundary to a same-version replacement without overlap", async () => { + const log: string[] = []; + const gate = deferred(); + const { harness, registry, old, id } = await startHandover(log, { a: gate.promise }); + registry.batch(() => { + old.dispose(); + registry.tasks.add(handoverTask("new", 1, log)); + }); + gate.resolve(); + await harness.waitForTask(id, context); + expect(log).toEqual(["old:a start", "old:a end", "new:b start", "new:b end", "new:c"]); + await harness.close(context); + }); + + it("hands over to a newer version with a migration", async () => { + const log: string[] = []; + const gate = deferred(); + const { harness, registry, old, id } = await startHandover(log, { a: gate.promise }); + registry.batch(() => { + old.dispose(); + registry.tasks.add( + handoverTask("v2", 2, log, { migrate: () => ({ input: null, checkpoint: { phase: "c" } }) }), + ); + }); + gate.resolve(); + const receipt = await harness.waitForTask(id, context); + expect(receipt.version).toBe(2); + expect(log).toEqual(["old:a start", "old:a end", "v2:c"]); + await harness.close(context); + }); + + it("hands over to a newer version whose migration fails and leaves the task blocked", async () => { + const log: string[] = []; + const gate = deferred(); + const { harness, registry, old, id, reports } = await startHandover(log, { a: gate.promise }); + registry.batch(() => { + old.dispose(); + registry.tasks.add( + handoverTask("broken", 2, log, { + migrate: () => { + throw new Error("broken migration"); + }, + }), + ); + }); + gate.resolve(); + await eventually(() => reports.length === 1); + await flush(); + expect(await harness.getTask(id, context)).toMatchObject({ + version: 1, + state: { status: "pending", checkpoint: { phase: "b" } }, + }); + expect(log).toEqual(["old:a start", "old:a end"]); + await harness.close(context); + }); + + it("keeps running under the old definition when the replacement is missing or cannot take the task", async () => { + const log: string[] = []; + const gateA = deferred(); + const gateB = deferred(); + const { harness, registry, old, id, reports } = await startHandover(log, { a: gateA.promise, b: gateB.promise }); + old.dispose(); + gateA.resolve(); + await eventually(() => log.includes("old:b start")); + // A newer definition without a migration cannot take the task either. + registry.tasks.add(handoverTask("incompatible", 2, log)); + gateB.resolve(); + await harness.waitForTask(id, context); + expect(log).toEqual(["old:a start", "old:a end", "old:b start", "old:b end", "old:c"]); + expect(reports.map((report) => (report as Error).cause)).toEqual(["missing_task", "incompatible_task"]); + await harness.close(context); + }); + + it("rejects a runtime commit of the old invocation queued behind its handover commit", async () => { + const log: string[] = []; + const gate = deferred(); + const storage = new ControlledStorage(); + let held: ReturnType | undefined; + let oldRuntime: TaskRuntime | undefined; + let harnessRef: OpenHarness | undefined; + const registry = createRegistry(); + const Old = defineTask({ + name: "test.handover", + version: 1, + initial: () => ({ phase: "a" }), + phases: { + a: async (task, runtime, ctx) => { + oldRuntime = runtime; + await gate.promise; + await runtime.commit(() => ({ status: "running", checkpoint: { phase: "b" } }), ctx); + // Hold the line with an unrelated commit so the handover commit queues behind it. + held = storage.holdCommits(); + void harnessRef!.commit(async (tx) => { + await tx.appendEntry(task.conversationId, { kind: "blocker" }); + }, ctx); + }, + b: async () => {}, + c: async () => {}, + }, + abort: async () => {}, + }); + const old = registry.tasks.add(Old); + const { harness } = await openTasks(storage, [], { registry }); + harnessRef = harness; + const id = await createIn(harness, Old); + harness.resume(); + await eventually(() => oldRuntime !== undefined); + registry.batch(() => { + old.dispose(); + registry.tasks.add(handoverTask("new", 1, log)); + }); + gate.resolve(); + await eventually(() => held !== undefined); + await held!.entered; + await flush(); + // Queued behind the handover commit while the invocation has not ended yet. + const late = oldRuntime!.commit(() => completed(null), context); + held!.release(); + await expect(late).rejects.toThrow("invocation has ended"); + await harness.waitForTask(id, context); + expect(log).toEqual(["new:b start", "new:b end", "new:c"]); + await harness.close(context); + }); + + it("preserves an abort mark that races the handover commit; the new definition aborts", async () => { + const log: string[] = []; + const gate = deferred(); + const storage = new ControlledStorage(); + let held: ReturnType | undefined; + // The progress commit landed; hold the next commit, the handover, and queue the abort mark behind it. + const gates: Gates = { + a: gate.promise, + onEnd: () => { + held ??= storage.holdCommits(); + }, + }; + const { harness, registry, old, id } = await startHandover(log, gates, storage); + registry.batch(() => { + old.dispose(); + registry.tasks.add(handoverTask("new", 1, log)); + }); + gate.resolve(); + await eventually(() => held !== undefined); + await held!.entered; + const aborting = harness.abortTask(id, context); + held!.release(); + expect(await aborting).toBe("marked"); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "aborted", reason: "new" }); + const states = storage.commits.flatMap((writes) => + writes.flatMap((write) => + write.type === "task" && write.value.id === id + ? [`${write.value.state.status}${write.value.abortRequested ? "+mark" : ""}`] + : [], + ), + ); + // created, reserved, progress, handover, mark, abort reservation, aborted + expect(states).toEqual([ + "pending", + "running", + "running", + "pending", + "pending+mark", + "running+mark", + "terminal+mark", + ]); + expect(log).toEqual(["old:a start", "old:a end", "new:abort"]); + await harness.close(context); + }); +}); + +describe("Harness open", () => { + it("releases its registry subscription and closes the Session when open fails", async () => { + const storage = new ControlledStorage(); + const log: string[] = []; + const Stuck = defineTask({ + name: "test.stuck", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async () => { + log.push("run"); + await new Promise(() => {}); + }, + }, + abort: async () => {}, + }); + // Leave a running task behind so open has a reconciliation commit to fail. + const first = await openTasks(storage, [Stuck]); + await createIn(first.harness, Stuck); + first.harness.resume(); + await eventually(() => log.length === 1); + + const registry = createRegistry(); + const reader = countingReader(registry); + storage.failNextCommit(new Error("disk full")); + await expect(Harness.open(storage, { models: createModels(), registry: reader }, context)).rejects.toThrow( + "disk full", + ); + expect(reader.subscriptions()).toBe(0); + await expect(storage.task(1 as TaskId, context)).rejects.toThrow(); + }); +}); diff --git a/packages/durable/test/harness-tasks.test.ts b/packages/durable/test/harness-tasks.test.ts new file mode 100644 index 000000000..087273808 --- /dev/null +++ b/packages/durable/test/harness-tasks.test.ts @@ -0,0 +1,1139 @@ +import type { Context } from "@earendil-works/chord"; +import { createModels } from "@earendil-works/pi-ai"; +import { + type Conversation, + createRegistry, + defineDoc, + defineEntry, + defineTask, + type EntryId, + type EntryRecord, + Harness, + MemoryStorage, + type RegistryReader, + type RegistrySnapshot, + StorageRejected, + type TaskId, + type TaskRuntime, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { user } from "./harness-support.ts"; +import { ControlledStorage, context, flush } from "./session-support.ts"; +import { aborted, abortedWith, completed, deferred, eventually, openTasks, settled } from "./task-support.ts"; + +type Step = { phase: "run" }; +type StepRuntime = TaskRuntime; + +/** A one-phase task; the default abort handler settles `aborted` with reason "test". */ +function oneStep( + name: string, + run: (task: { readonly id: TaskId }, runtime: StepRuntime, context: Context) => Promise, + abort: (runtime: StepRuntime, context: Context) => Promise = (runtime, ctx) => + runtime.commit(() => abortedWith("test"), ctx), +) { + return defineTask({ + name, + version: 1, + initial: () => ({ phase: "run" }), + phases: { run: (task, runtime, ctx) => run(task, runtime, ctx) }, + abort: (_task, runtime, ctx) => abort(runtime, ctx), + }); +} + +/** A one-phase task that waits for `gate` and completes with null. */ +function gated(name: string, gate: Promise) { + return oneStep(name, async (_task, runtime, ctx) => { + await gate; + await runtime.commit(() => completed(null), ctx); + }); +} + +async function start( + conversation: Conversation, + task: ReturnType>, + options: { readonly after?: readonly TaskId[]; readonly background?: boolean } = {}, +): Promise> { + return conversation.commit((tx) => tx.createTask(task, null, options), context); +} + +async function openRoot( + tasks: Parameters[1], + options: Parameters[2] & { readonly storage?: MemoryStorage } = {}, +) { + const opened = await openTasks(options.storage ?? new MemoryStorage(), tasks, options); + const root = await opened.harness.root(context); + return { ...opened, root }; +} + +/** Start `abortTask()` and resolve once its mark is durable, before it has joined the run. */ +async function markDurably( + harness: Awaited>["harness"], + id: TaskId, +): Promise<{ readonly aborting: Promise }> { + const aborting = harness.abortTask(id, context); + while (!(await harness.getTask(id, context))?.abortRequested) await flush(); + return { aborting }; +} + +function withSignal(signal: AbortSignal): Context { + return { ...context, abortSignal: signal }; +} + +describe("task phases", () => { + it("continues one invocation through checkpoint progress and completes with a typed result", async () => { + const seen: number[] = []; + const runtimes = new Set(); + const Counter = defineTask<{ to: number }, { phase: "count"; n: number }, number>({ + name: "test.counter", + version: 1, + initial: () => ({ phase: "count", n: 0 }), + phases: { + count: async (task, runtime, ctx) => { + seen.push(task.state.checkpoint.n); + runtimes.add(runtime); + await runtime.commit((_tx, current) => { + const n = current.state.checkpoint.n + 1; + return n === task.input.to ? completed(n) : { status: "running", checkpoint: { phase: "count", n } }; + }, ctx); + }, + }, + abort: async () => {}, + }); + const { harness, root } = await openRoot([Counter]); + const id = await root.commit((tx) => tx.createTask(Counter, { to: 3 }), context); + harness.resume(); + const receipt = await harness.waitForTask(id, context); + const result: number | undefined = + receipt.state.outcome.status === "completed" ? receipt.state.outcome.result : undefined; + expect(result).toBe(3); + expect(seen).toEqual([0, 1, 2]); + expect(runtimes.size).toBe(1); + await harness.close(context); + }); + + it("faults a phase without durable progress and a throwing phase", async () => { + const Idle = oneStep("test.idle", async () => {}); + const DocumentOnly = oneStep("test.document-only", async (_task, runtime, ctx) => { + // A commit that returns no state is not progress. + await runtime.commit(() => undefined, ctx); + }); + const Throws = oneStep("test.throws", async () => { + throw new Error("boom"); + }); + const { harness, root } = await openRoot([Idle, DocumentOnly, Throws]); + const ids = [await start(root, Idle), await start(root, DocumentOnly), await start(root, Throws)]; + harness.resume(); + const outcomes = await Promise.all(ids.map(async (id) => (await harness.waitForTask(id, context)).state.outcome)); + expect(outcomes).toEqual([ + { status: "faulted", error: { message: "Task test.idle phase run returned without durable progress" } }, + { + status: "faulted", + error: { message: "Task test.document-only phase run returned without durable progress" }, + }, + { status: "faulted", error: { message: "boom" } }, + ]); + await harness.close(context); + }); + + it("keeps a committed terminal outcome when the handler throws afterwards", async () => { + let lateCommit: unknown; + const Done = oneStep("test.done", async (_task, runtime, ctx) => { + await runtime.commit(() => completed("ok"), ctx); + lateCommit = await runtime.commit(() => undefined, ctx).catch((error: unknown) => error); + throw new Error("after terminal"); + }); + const { harness, root } = await openRoot([Done]); + const id = await start(root, Done); + harness.resume(); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "completed", result: "ok" }); + await eventually(() => lateCommit !== undefined); + expect(String(lateCommit)).toContain(`Task ${id} is terminal`); + await harness.close(context); + }); + + it("compares checkpoints by value, including arrays", async () => { + const Collect = defineTask({ + name: "test.collect", + version: 1, + initial: () => ({ phase: "collect", items: [] }), + phases: { + collect: async (task, runtime, ctx) => { + const items = task.state.checkpoint.items; + // Two rounds of progress, then an equal copy of the checkpoint, which is no progress. + const next = items.length < 2 ? [...items, `item${items.length}`] : [...items]; + await runtime.commit(() => ({ status: "running", checkpoint: { phase: "collect", items: next } }), ctx); + }, + }, + abort: async () => {}, + }); + const { harness, root } = await openRoot([Collect]); + const id = await root.commit((tx) => tx.createTask(Collect, null), context); + harness.resume(); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ + status: "faulted", + error: { message: "Task test.collect phase collect returned without durable progress" }, + }); + await harness.close(context); + }); + + it("commits results with entries atomically, keeps memos until terminal, and retires task documents", async () => { + const Progress = defineDoc<{ lines: string[] }>({ + kind: "test.task-progress", + version: 1, + scope: "task", + initial: () => ({ lines: [] }), + }); + const Answer = defineEntry("answer"); + const Child = oneStep("test.child", async (_task, runtime, ctx) => { + await runtime.commit(() => completed(null), ctx); + }); + let childId: TaskId | undefined; + let progressSeen: unknown; + const Writer = defineTask({ + name: "test.writer", + version: 1, + initial: () => ({ phase: "write" }), + phases: { + write: async (task, runtime, ctx) => { + expect(await runtime.memo("choice", ctx)).toBeUndefined(); + const winners = await Promise.all([runtime.memo("choice", "a", ctx), runtime.memo("choice", "b", ctx)]); + expect(winners).toEqual(["a", "a"]); + expect(await runtime.memo("choice", ctx)).toBe("a"); + // Memo names never resolve to inherited object properties. + expect(await runtime.memo("toString", ctx)).toBeUndefined(); + expect(await runtime.memo("toString", "own", ctx)).toBe("own"); + await runtime.commit(async (tx) => { + (await tx.doc(Progress, task.id)).lines.push("wrote"); + // Task creation defaults to the task's own conversation. + childId = await tx.createTask(Child, null); + return { status: "running", checkpoint: { phase: "answer" } }; + }, ctx); + }, + answer: async (task, runtime, ctx) => { + expect(task.memos).toEqual({ choice: "a", toString: "own" }); + await runtime.commit(async (tx) => { + progressSeen = [...(await tx.doc(Progress, task.id)).lines]; + }, ctx); + await runtime.commit(async (tx, current) => { + const entry = await tx.appendEntry(current.conversationId, { kind: "answer", model: [user("done")] }); + return completed({ entryId: entry.id }); + }, ctx); + }, + }, + abort: async () => {}, + }); + const { harness, root } = await openRoot([Writer, Child]); + const conversation = await harness.createConversation({ ownership: { kind: "ownerless" } }, context); + const id = await conversation.commit((tx) => tx.createTask(Writer, null), context); + harness.resume(); + const receipt = await harness.waitForTask(id, context); + expect(receipt.state).not.toHaveProperty("checkpoint"); + expect(receipt).not.toHaveProperty("memos"); + const outcome = receipt.state.outcome; + if (outcome.status !== "completed") throw new Error("expected completion"); + const entry = await harness.commit((tx) => tx.entry(outcome.result.entryId), context); + expect(Answer.is(entry)).toBe(true); + expect(entry?.conversationId).toBe(conversation.id); + expect(progressSeen).toEqual(["wrote"]); + expect(await harness.snapshot(Progress, id, context)).toBeUndefined(); + expect(await harness.getTask(id, context)).toEqual(receipt); + expect((await harness.waitForTask(childId!, context)).conversationId).toBe(conversation.id); + expect(root.id).not.toBe(conversation.id); + await harness.close(context); + }); + + it("waits for dependencies, which any terminal outcome satisfies", async () => { + const gate = deferred(); + const order: string[] = []; + const First = oneStep("test.first", async (_task, runtime, ctx) => { + order.push("first"); + await gate.promise; + await runtime.commit(() => completed(null), ctx); + }); + const Faulting = oneStep("test.faulting", async () => { + order.push("faulting"); + throw new Error("fails"); + }); + const Second = oneStep("test.second", async (_task, runtime, ctx) => { + order.push("second"); + await runtime.commit(() => completed(null), ctx); + }); + const { harness, root } = await openRoot([First, Faulting, Second]); + const first = await start(root, First); + const faulting = await start(root, Faulting, { after: [first] }); + const second = await start(root, Second, { after: [first, faulting] }); + harness.resume(); + await eventually(() => order.length === 1); + await flush(); + expect(order).toEqual(["first"]); + expect((await harness.getTask(second, context))?.state.status).toBe("pending"); + gate.resolve(); + await harness.waitForTask(second, context); + expect(order).toEqual(["first", "faulting", "second"]); + expect((await harness.getTask(faulting, context))?.state).toMatchObject({ outcome: { status: "faulted" } }); + await harness.close(context); + }); + + it("keeps one registry snapshot per phase and refreshes it at the phase boundary", async () => { + const seen: { phase: string; tools: string[]; same: boolean }[] = []; + const phaseGate = deferred(); + const entered = deferred(); + const tools = (snapshot: RegistrySnapshot) => snapshot.toolNames().slice(); + const Snapshots = defineTask({ + name: "test.snapshots", + version: 1, + initial: () => ({ phase: "a" }), + phases: { + a: async (_task, runtime, ctx) => { + const before = runtime.registry; + entered.resolve(); + await phaseGate.promise; + seen.push({ phase: "a", tools: tools(runtime.registry), same: runtime.registry === before }); + await runtime.commit(() => ({ status: "running", checkpoint: { phase: "b" } }), ctx); + }, + b: async (_task, runtime, ctx) => { + seen.push({ phase: "b", tools: tools(runtime.registry), same: true }); + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + const { harness, registry, root } = await openRoot([Snapshots]); + const id = await root.commit((tx) => tx.createTask(Snapshots, null), context); + harness.resume(); + await entered.promise; + registry.tools.add({ + name: "late", + description: "late", + parameters: { type: "object", properties: {} } as never, + execute: async () => ({}), + }); + phaseGate.resolve(); + await harness.waitForTask(id, context); + expect(seen).toEqual([ + { phase: "a", tools: [], same: true }, + { phase: "b", tools: ["late"], same: true }, + ]); + await harness.close(context); + }); +}); + +describe("task runtime", () => { + it("rejects runtime operations after the invocation ends and stops its watches", async () => { + const Notes = defineDoc<{ text: string }>({ + kind: "test.task-notes", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + }); + let captured: StepRuntime | undefined; + let watchClosed: Promise | undefined; + const delivered: string[] = []; + const Absent = defineDoc<{ text: string }>({ + kind: "test.task-absent", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + }); + let absent: unknown = "unset"; + const Watcher = oneStep("test.watcher", async (_task, runtime, ctx) => { + captured = runtime; + absent = await runtime.watchDoc(Absent, ctx); + const watch = await runtime.watchDoc(Notes, ctx); + watch!.start(async (value) => { + delivered.push(value?.text ?? "retired"); + }); + watchClosed = watch?.closed; + await runtime.commit(() => completed(null), ctx); + }); + const { harness, root } = await openRoot([Watcher]); + await harness.commit(async (tx) => { + (await tx.doc(Notes)).text = "hello"; + }, context); + const id = await start(root, Watcher); + harness.resume(); + await harness.waitForTask(id, context); + expect(absent).toBeUndefined(); + // The watch stops when the step after the phase ends the invocation; later commits deliver nothing. + expect(await watchClosed).toEqual({ reason: "stopped" }); + await harness.commit(async (tx) => { + (await tx.doc(Notes)).text = "after"; + }, context); + await flush(); + expect(delivered).toEqual([]); + const runtime = captured!; + await expect(runtime.commit(() => {}, context)).rejects.toThrow("invocation has ended"); + await expect(runtime.memo("x", context)).rejects.toThrow("invocation has ended"); + await expect(runtime.memo("x", 1, context)).rejects.toThrow("invocation has ended"); + await expect(runtime.sleep(0, context)).rejects.toThrow("invocation has ended"); + await expect(runtime.watchDoc(Notes, context)).rejects.toThrow("invocation has ended"); + await harness.close(context); + }); + + it("orders runtime commits against the step: one queued before it lands, one after it rejects", async () => { + const storage = new ControlledStorage(); + let before: Promise | undefined; + let after: Promise | undefined; + let held: ReturnType | undefined; + let harnessRef: Awaited>["harness"] | undefined; + const Detached = oneStep("test.detached", async (_task, runtime) => { + // Hold the line, queue a commit without awaiting it, and return; the step queues behind that commit. + held = storage.holdCommits(); + void harnessRef!.commit(async (tx) => { + await tx.appendEntry(runtime.conversationId, { kind: "blocker" }); + }, context); + before = runtime.commit(() => completed("before"), context); + // Queued after the handler returned, so after the step. + setTimeout(() => { + after = runtime.commit(() => completed("after"), context); + }, 0); + }); + const { harness, root } = await openRoot([Detached], { storage }); + harnessRef = harness; + const id = await start(root, Detached); + harness.resume(); + await eventually(() => held !== undefined); + await held!.entered; + await eventually(() => after !== undefined); + held!.release(); + await before; + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "completed", result: "before" }); + await expect(after).rejects.toThrow(/invocation has ended|is terminal/); + await harness.close(context); + }); + + it("stops a watch whose acquisition finishes after the invocation ended", async () => { + const Notes = defineDoc<{ text: string }>({ + kind: "test.late-watch", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + }); + const storage = new ControlledStorage(); + let watching: Promise | undefined; + let held: ReturnType | undefined; + let harnessRef: Awaited>["harness"] | undefined; + const Late = oneStep("test.late-watch", async (_task, runtime) => { + held = storage.holdCommits(); + void harnessRef!.commit(async (tx) => { + await tx.appendEntry(runtime.conversationId, { kind: "blocker" }); + }, context); + // Starts after the handler returned, so its line job queues behind the step that ends the invocation. + setTimeout(() => { + watching = runtime.watchDoc(Notes, context).catch((error: unknown) => error); + }, 0); + }); + const { harness, root } = await openRoot([Late], { storage }); + harnessRef = harness; + await harness.commit(async (tx) => { + (await tx.doc(Notes)).text = "x"; + }, context); + const id = await start(root, Late); + harness.resume(); + await eventually(() => held !== undefined); + await held!.entered; + await eventually(() => watching !== undefined); + held!.release(); + expect(String(await watching)).toContain("invocation has ended"); + expect((await harness.waitForTask(id, context)).state.outcome).toMatchObject({ status: "faulted" }); + await harness.close(context); + }); + + it("sleeps until the Harness clock reaches the deadline, rechecking after each timer", async () => { + let clock = 1_000; + const woke = deferred(); + const Sleeper = oneStep("test.sleeper", async (_task, runtime, ctx) => { + await runtime.sleep(900, ctx); + await runtime.sleep(1_005, ctx); + woke.resolve(); + await runtime.commit(() => completed(null), ctx); + }); + const { harness, root } = await openRoot([Sleeper], { now: () => clock }); + const id = await start(root, Sleeper); + harness.resume(); + // The clock stands still, so real timers keep firing without waking the task. + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(await settled(woke.promise)).toBe(false); + clock = 1_005; + await harness.waitForTask(id, context); + await harness.close(context); + }); + + it("rejects a sleep when the invocation is signalled or the sleep's own context is cancelled", async () => { + const results: string[] = []; + const sleeping = deferred(); + const Signalled = oneStep("test.sleep-signalled", async (_task, runtime, ctx) => { + sleeping.resolve(); + await runtime.sleep(Date.now() + 60_000, ctx).catch((error: unknown) => { + results.push(`signalled:${(error as Error).name}`); + throw error; + }); + }); + const Cancelled = oneStep("test.sleep-cancelled", async (_task, runtime, ctx) => { + const controller = new AbortController(); + const sleeping = runtime.sleep(Date.now() + 60_000, withSignal(controller.signal)); + controller.abort(new Error("stop sleeping")); + await sleeping.catch((error: unknown) => results.push(`cancelled:${(error as Error).message}`)); + await runtime.commit(() => completed(null), ctx); + }); + const { harness, root } = await openRoot([Signalled, Cancelled]); + const signalled = await start(root, Signalled); + const cancelled = await start(root, Cancelled); + harness.resume(); + await harness.waitForTask(cancelled, context); + await sleeping.promise; + await harness.abortTask(signalled, context); + expect((await harness.waitForTask(signalled, context)).state.outcome).toEqual({ + status: "aborted", + reason: "test", + }); + expect(results).toEqual(["cancelled:stop sleeping", "signalled:AbortError"]); + await harness.close(context); + }); +}); + +describe("task scheduling", () => { + it("retries reservation on the next wakeup after a rejected reservation commit", async () => { + let runs = 0; + const Once = oneStep("test.once", async (_task, runtime, ctx) => { + runs++; + await runtime.commit(() => completed(null), ctx); + }); + const storage = new ControlledStorage(); + const { harness, root, registry, reports } = await openRoot([Once], { storage }); + const id = await start(root, Once); + storage.failNextCommit(new StorageRejected("busy")); + harness.resume(); + await eventually(() => reports.length === 1); + expect((await harness.getTask(id, context))?.state.status).toBe("pending"); + // Any wakeup, here a registry change, reserves again. + registry.tools.add({ + name: "wake", + description: "wake", + parameters: { type: "object", properties: {} } as never, + execute: async () => ({}), + }); + await harness.waitForTask(id, context); + expect(runs).toBe(1); + await harness.close(context); + }); + + it("keeps a wakeup that arrives while a rejected reservation commit is in storage", async () => { + const First = oneStep("test.wake-first", async (_task, runtime, ctx) => { + await runtime.commit(() => completed(null), ctx); + }); + const Late = oneStep("test.wake-late", async (_task, runtime, ctx) => { + await runtime.commit(() => completed(null), ctx); + }); + const storage = new ControlledStorage(); + const { harness, root, registry } = await openRoot([First], { storage }); + const first = await start(root, First); + const late = await start(root, Late); + const held = storage.holdCommits(); + storage.failNextCommit(new StorageRejected("busy")); + harness.resume(); + await held.entered; + // Registering the missing definition wakes the scheduler while the doomed reservation is in storage. + registry.tasks.add(Late); + held.release(); + await harness.waitForTask(first, context); + await harness.waitForTask(late, context); + await harness.close(context); + }); + + it("reruns a task whose fault write was rejected", async () => { + let runs = 0; + const storage = new ControlledStorage(); + const Throws = oneStep("test.rejected-fault", async () => { + runs++; + // The next commit is the step's fault write. + if (runs === 1) storage.failNextCommit(new StorageRejected("busy")); + throw new Error("boom"); + }); + const { harness, root, reports } = await openRoot([Throws], { storage }); + const id = await start(root, Throws); + harness.resume(); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ + status: "faulted", + error: { message: "boom" }, + }); + expect(runs).toBe(2); + expect(reports.map(String)).toEqual(["StorageRejected: busy"]); + await harness.close(context); + }); + + it("waits for Harness and conversation idleness, counting blocked work and ignoring background tasks", async () => { + const gates = new Map void>(); + const Gated = oneStep("test.gated", async (task, runtime, ctx) => { + await new Promise((resolve) => gates.set(task.id, resolve)); + await runtime.commit(() => completed(null), ctx); + }); + const { harness, root } = await openRoot([Gated]); + await harness.waitForIdle(context); + const other = await harness.createConversation({ ownership: { kind: "ownerless" } }, context); + const foreground = await start(root, Gated); + const background = await start(root, Gated, { background: true }); + const elsewhere = await start(other, Gated); + // Work that has not started yet is live; cancelling a wait only rejects that wait. + const cancelled = new AbortController(); + const cancelledWait = harness.waitForIdle(withSignal(cancelled.signal)); + cancelled.abort(new Error("stop waiting")); + await expect(cancelledWait).rejects.toThrow("stop waiting"); + const aborted = new AbortController(); + aborted.abort(new Error("already cancelled")); + await expect(root.waitForIdle(withSignal(aborted.signal))).rejects.toThrow("already cancelled"); + harness.resume(); + await eventually(() => gates.size === 3); + const rootIdle = root.waitForIdle(context); + const harnessIdle = harness.waitForIdle(context); + gates.get(foreground)!(); + await rootIdle; + expect(await settled(harnessIdle)).toBe(false); + gates.get(elsewhere)!(); + await harnessIdle; + expect((await harness.getTask(background, context))?.state.status).toBe("running"); + gates.get(background)!(); + await harness.waitForTask(background, context); + await harness.close(context); + await expect(harness.waitForIdle(context)).rejects.toThrow("closed"); + await expect(root.waitForIdle(context)).rejects.toThrow("closed"); + }); + + it("rejects unknown tasks and reports terminal tasks", async () => { + const Done = oneStep("test.quick", async (_task, runtime, ctx) => { + await runtime.commit(() => completed(null), ctx); + }); + const { harness, root } = await openRoot([Done]); + const unknown = 999_999 as TaskId; + expect(await harness.getTask(unknown, context)).toBeUndefined(); + await expect(harness.waitForTask(unknown, context)).rejects.toThrow("does not exist"); + await expect(harness.abortTask(unknown, context)).rejects.toThrow("does not exist"); + const id = await start(root, Done); + harness.resume(); + await harness.waitForTask(id, context); + expect(await harness.waitForTask(id, context)).toMatchObject({ state: { status: "terminal" } }); + expect(await harness.abortTask(id, context)).toBe("terminal"); + await harness.close(context); + expect(() => harness.resume()).toThrow("closed"); + }); + + it("rejects task waits cancelled or closed while queued on the line, and pending waits on close", async () => { + const gate = deferred(); + const Blocking = gated("test.wait-close", gate.promise); + const storage = new ControlledStorage(); + const { harness, root } = await openRoot([Blocking], { storage }); + const id = await start(root, Blocking); + + // Hold the line with a commit, then queue waits behind it. + const held = storage.holdCommits(); + const blocker = harness.commit(async (tx) => { + await tx.appendEntry(root.id, { kind: "blocker" }); + }, context); + await held.entered; + const controller = new AbortController(); + const cancelledWhileQueued = harness.waitForTask(id, withSignal(controller.signal)); + controller.abort(new Error("wait cancelled")); + held.release(); + await blocker; + await expect(cancelledWhileQueued).rejects.toThrow("wait cancelled"); + + const pending = harness.waitForTask(id, context); + const idle = harness.waitForIdle(context); + const conversationIdle = root.waitForIdle(context); + await flush(); + const heldAgain = storage.holdCommits(); + const blockerAgain = harness.commit(async (tx) => { + await tx.appendEntry(root.id, { kind: "blocker" }); + }, context); + await heldAgain.entered; + const closedWhileQueued = harness.waitForTask(id, context); + const closing = harness.close(context); + heldAgain.release(); + await blockerAgain; + await expect(closedWhileQueued).rejects.toThrow("closed"); + await expect(pending).rejects.toThrow("closed"); + await expect(idle).rejects.toThrow("closed"); + await expect(conversationIdle).rejects.toThrow("closed"); + gate.resolve(); + await closing; + }); +}); + +describe("task abort", () => { + it("rejects the run's commits and memo writes after the mark, and settles through a fresh abort invocation", async () => { + const reached = deferred(); + const errors: string[] = []; + let memoRead: unknown; + const abortRuntimes: unknown[] = []; + let runRuntime: unknown; + const Marked = oneStep( + "test.marked", + async (_task, runtime, ctx) => { + runRuntime = runtime; + await runtime.memo("kept", 1, ctx); + reached.resolve(); + // Keep working after the signal: every later write of this run must reject. + await aborted(runtime.signal).catch(() => {}); + memoRead = await runtime.memo("kept", context); + await runtime.memo("late", 2, context).catch((error: unknown) => errors.push((error as Error).message)); + await runtime + .commit(() => completed(null), context) + .catch((error: unknown) => { + errors.push((error as Error).message); + }); + }, + async (runtime, ctx) => { + abortRuntimes.push(runtime); + await runtime.commit((_tx, current) => { + expect(current.abortRequested).toBe(true); + expect(current.memos).toEqual({ kept: 1 }); + return abortedWith("mark"); + }, ctx); + }, + ); + const { harness, root } = await openRoot([Marked]); + const id = await start(root, Marked); + harness.resume(); + await reached.promise; + expect(await harness.abortTask(id, context)).toBe("marked"); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "aborted", reason: "mark" }); + expect(memoRead).toBe(1); + expect(errors).toEqual([`Task ${id} has a durable abort mark`, `Task ${id} has a durable abort mark`]); + expect(abortRuntimes).toHaveLength(1); + expect(abortRuntimes[0]).not.toBe(runRuntime); + await harness.close(context); + }); + + it("starts no further phase after a mark that lands during a phase with progress", async () => { + const reached = deferred(); + const proceed = deferred(); + const phases: string[] = []; + const Two = defineTask({ + name: "test.mark-boundary", + version: 1, + initial: () => ({ phase: "one" }), + phases: { + one: async (_task, runtime, ctx) => { + phases.push("one"); + await runtime.commit(() => ({ status: "running", checkpoint: { phase: "two" } }), ctx); + reached.resolve(); + // Ignores the signal and returns normally after the mark. + await proceed.promise; + }, + two: async () => { + phases.push("two"); + }, + }, + abort: async (_task, runtime, ctx) => { + phases.push("abort"); + await runtime.commit(() => abortedWith("boundary"), ctx); + }, + }); + const { harness, root } = await openRoot([Two]); + const id = await root.commit((tx) => tx.createTask(Two, null), context); + harness.resume(); + await reached.promise; + const { aborting } = await markDurably(harness, id); + proceed.resolve(); + expect(await aborting).toBe("marked"); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "aborted", reason: "boundary" }); + expect(phases).toEqual(["one", "abort"]); + await harness.close(context); + }); + + it("signals and joins the run before returning, then runs the abort handler", async () => { + const reached = deferred(); + let runEnded = false; + const Signalled = oneStep("test.signalled", async (_task, runtime) => { + reached.resolve(); + try { + await aborted(runtime.signal); + } finally { + runEnded = true; + } + }); + const { harness, root } = await openRoot([Signalled]); + const id = await start(root, Signalled); + harness.resume(); + await reached.promise; + expect(await harness.abortTask(id, context)).toBe("marked"); + expect(runEnded).toBe(true); + expect(await harness.abortTask(id, context)).toBe("marked"); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "aborted", reason: "test" }); + await harness.close(context); + }); + + it("aborts pending work regardless of dependencies and faults abort handlers that throw or settle nothing", async () => { + const gate = deferred(); + const First = gated("test.dependency", gate.promise); + const Lazy = oneStep( + "test.lazy-abort", + async () => {}, + async () => {}, + ); + const Throwing = oneStep( + "test.throwing-abort", + async () => {}, + async () => { + throw new Error("abort failed"); + }, + ); + const { harness, root } = await openRoot([First, Lazy, Throwing]); + const first = await start(root, First); + const lazy = await start(root, Lazy, { after: [first] }); + const throwing = await start(root, Throwing, { after: [first] }); + harness.resume(); + await harness.abortTask(lazy, context); + await harness.abortTask(throwing, context); + expect((await harness.waitForTask(lazy, context)).state.outcome).toEqual({ + status: "faulted", + error: { message: `Abort handler of task ${lazy} returned without a terminal outcome` }, + }); + expect((await harness.waitForTask(throwing, context)).state.outcome).toEqual({ + status: "faulted", + error: { message: "abort failed" }, + }); + expect((await harness.getTask(first, context))?.state.status).toBe("running"); + gate.resolve(); + await harness.waitForTask(first, context); + await harness.close(context); + }); + + it("does not signal a running abort handler when aborted again, and a cancelled caller leaves the mark durable", async () => { + const reached = deferred(); + const proceed = deferred(); + const runRelease = deferred(); + let abortSignalled: boolean | undefined; + let aborts = 0; + const Run = oneStep( + "test.abort-again", + async () => { + // Ignores the signal, so the first caller is still joining when it gives up. + await runRelease.promise; + }, + async (runtime, ctx) => { + aborts++; + reached.resolve(); + await proceed.promise; + abortSignalled = runtime.signal.aborted; + await runtime.commit(() => abortedWith("once"), ctx); + }, + ); + const { harness, root } = await openRoot([Run]); + const id = await start(root, Run); + harness.resume(); + await flush(); + // This caller gives up while joining; the mark and the abort invocation are unaffected. + const controller = new AbortController(); + const cancelled = harness.abortTask(id, withSignal(controller.signal)); + while (!(await harness.getTask(id, context))?.abortRequested) await flush(); + controller.abort(new Error("caller gave up")); + await expect(cancelled).rejects.toThrow("caller gave up"); + runRelease.resolve(); + await reached.promise; + expect(await harness.abortTask(id, context)).toBe("marked"); + proceed.resolve(); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "aborted", reason: "once" }); + expect(abortSignalled).toBe(false); + expect(aborts).toBe(1); + await harness.close(context); + }); + + it("keeps a terminal outcome committed by an abort handler that throws afterwards", async () => { + const Run = oneStep( + "test.abort-then-throw", + async (_task, runtime) => { + await aborted(runtime.signal); + }, + async (runtime, ctx) => { + await runtime.commit(() => abortedWith("done"), ctx); + throw new Error("after terminal"); + }, + ); + const { harness, root } = await openRoot([Run]); + const id = await start(root, Run); + harness.resume(); + await flush(); + await harness.abortTask(id, context); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "aborted", reason: "done" }); + await harness.close(context); + }); + + it("never starts phase one when the abort lands while the reservation settles", async () => { + let ran = false; + const Reserved = oneStep("test.reserved", async (_task, runtime) => { + ran = true; + await aborted(runtime.signal); + }); + const storage = new ControlledStorage(); + const { harness, root } = await openRoot([Reserved], { storage }); + const id = await start(root, Reserved); + const held = storage.holdCommits(); + harness.resume(); + await held.entered; + // The reservation commit is in storage; the abort mark commit queues behind it. + const aborting = harness.abortTask(id, context); + held.release(); + expect(await aborting).toBe("marked"); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "aborted", reason: "test" }); + expect(ran).toBe(false); + await harness.close(context); + }); + + it("lets an abort mark win over a fault that races it", async () => { + const storage = new ControlledStorage(); + let marking: Promise | undefined; + let held: ReturnType | undefined; + let harnessRef: Awaited>["harness"] | undefined; + const Racing = oneStep("test.racing-fault", async (task) => { + held = storage.holdCommits(); + marking = harnessRef!.abortTask(task.id, context); + await held.entered; + // The mark is in storage but not committed when the handler throws, so the fault commit queues behind it. + throw new Error("would fault"); + }); + const { harness, root } = await openRoot([Racing], { storage }); + harnessRef = harness; + const id = await start(root, Racing); + harness.resume(); + await eventually(() => held !== undefined); + await held!.entered; + await flush(); + held!.release(); + expect((await harness.waitForTask(id, context)).state.outcome).toEqual({ status: "aborted", reason: "test" }); + expect(await marking).toBe("marked"); + await harness.close(context); + }); +}); + +describe("task close", () => { + it("stops without outcomes and starts no fresh phase or abort invocation while closing", async () => { + const reached = deferred(); + const proceed = deferred(); + const phases: string[] = []; + let abortRan = false; + const Two = defineTask({ + name: "test.two", + version: 1, + initial: () => ({ phase: "one" }), + phases: { + one: async (_task, runtime, ctx) => { + phases.push("one"); + await runtime.commit(() => ({ status: "running", checkpoint: { phase: "two" } }), ctx); + reached.resolve(); + // Ignores signals and returns normally once released; the closing rule wins over the next phase. + await proceed.promise; + }, + two: async () => { + phases.push("two"); + }, + }, + abort: async () => { + abortRan = true; + }, + }); + const storage = new ControlledStorage(); + const { harness, root } = await openRoot([Two], { storage }); + const id = await root.commit((tx) => tx.createTask(Two, null), context); + harness.resume(); + await reached.promise; + const { aborting } = await markDurably(harness, id); + const record = await harness.getTask(id, context); + const commits = storage.commits.length; + const closing = harness.close(context); + proceed.resolve(); + await closing; + expect(await aborting).toBe("marked"); + expect(storage.commits.length).toBe(commits); + expect(phases).toEqual(["one"]); + expect(abortRan).toBe(false); + expect(record).toMatchObject({ + abortRequested: true, + state: { status: "running", checkpoint: { phase: "two" } }, + }); + await expect(harness.commit(() => {}, context)).rejects.toThrow("closed"); + }); + + it("seals admission before signalling handlers and stops watches before joining them", async () => { + const Notes = defineDoc<{ text: string }>({ + kind: "test.close-notes", + version: 1, + scope: "session", + initial: () => ({ text: "" }), + }); + const reached = deferred(); + let fromListener: Promise | undefined; + let harnessRef: Awaited>["harness"] | undefined; + let watchEnd: unknown; + const Stubborn = oneStep("test.stubborn", async (_task, runtime) => { + // Uses a context close does not cancel, then waits for the watch to close. + const watch = await runtime.watchDoc(Notes, context); + runtime.signal.addEventListener("abort", () => { + fromListener = harnessRef!.commit(() => {}, context); + }); + reached.resolve(); + watchEnd = await watch!.closed; + }); + const { harness, root } = await openRoot([Stubborn]); + harnessRef = harness; + await harness.commit(async (tx) => { + (await tx.doc(Notes)).text = "x"; + }, context); + await start(root, Stubborn); + harness.resume(); + await reached.promise; + await harness.close(context); + await expect(fromListener).rejects.toThrow("closed"); + expect(watchEnd).toEqual({ reason: "session_closed" }); + }); + + it("writes no fault when close seals while the step after a failed phase is queued", async () => { + const storage = new ControlledStorage(); + let held: ReturnType | undefined; + let harnessRef: Awaited>["harness"] | undefined; + const Throws = oneStep("test.close-fault", async (_task, runtime) => { + held = storage.holdCommits(); + void harnessRef!.commit(async (tx) => { + await tx.appendEntry(runtime.conversationId, { kind: "blocker" }); + }, context); + throw new Error("would fault"); + }); + const { harness, root } = await openRoot([Throws], { storage }); + harnessRef = harness; + const id = await start(root, Throws); + harness.resume(); + await eventually(() => held !== undefined); + await held!.entered; + await flush(); + const closing = harness.close(context); + held!.release(); + await closing; + const taskWrites = storage.commits.flatMap((writes) => + writes.filter((write) => write.type === "task" && write.value.id === id), + ); + expect(taskWrites.map((write) => (write.type === "task" ? write.value.state.status : ""))).toEqual([ + "pending", + "running", + ]); + }); + + it("starts no abort handler whose reservation settles while closing", async () => { + let ran = false; + const Marked = oneStep( + "test.close-abort-reservation", + async () => {}, + async () => { + ran = true; + }, + ); + const storage = new ControlledStorage(); + const { harness, root } = await openRoot([Marked], { storage }); + const id = await start(root, Marked); + expect(await harness.abortTask(id, context)).toBe("marked"); + const held = storage.holdCommits(); + harness.resume(); + await held.entered; + const closing = harness.close(context); + held.release(); + await closing; + expect(ran).toBe(false); + const writes = storage.commits.at(-1)!.filter((write) => write.type === "task"); + expect(writes).toMatchObject([{ value: { id, abortRequested: true, state: { status: "running" } } }]); + }); + + it("rejects a runtime commit that was queued on the line when close sealed it", async () => { + const storage = new ControlledStorage(); + let queued: Promise | undefined; + let held: ReturnType | undefined; + let harnessRef: Awaited>["harness"] | undefined; + const proceed = deferred(); + const Queued = oneStep("test.close-queued-commit", async (_task, runtime) => { + held = storage.holdCommits(); + void harnessRef!.commit(async (tx) => { + await tx.appendEntry(runtime.conversationId, { kind: "blocker" }); + }, context); + queued = runtime.commit(() => completed(null), context).catch((error: unknown) => error); + // Ignores the close signal, so the invocation is still alive when the queued commit reaches the line. + await proceed.promise; + }); + const { harness, root } = await openRoot([Queued], { storage }); + harnessRef = harness; + const id = await start(root, Queued); + harness.resume(); + await eventually(() => held !== undefined); + await held!.entered; + await flush(); + const closing = harness.close(context); + held!.release(); + expect(String(await queued)).toContain("Harness is closed"); + proceed.resolve(); + await closing; + const taskWrites = storage.commits.flatMap((writes) => + writes.filter((write) => write.type === "task" && write.value.id === id), + ); + expect(taskWrites).toHaveLength(2); + }); + + it("starts no next phase when close seals during a step that decided to continue", async () => { + const phases: string[] = []; + const registry = createRegistry(); + let harnessRef: Awaited>["harness"] | undefined; + let closing: Promise | undefined; + let closeOnSnapshot = false; + // The step refreshes the snapshot after progress, inside its line callback and after its closing check. + const reader: RegistryReader = { + snapshot: () => { + if (closeOnSnapshot) { + closeOnSnapshot = false; + closing = harnessRef!.close(context); + } + return registry.snapshot(); + }, + subscribe: (listener) => registry.subscribe(listener), + }; + const Two = defineTask({ + name: "test.close-in-step", + version: 1, + initial: () => ({ phase: "one" }), + phases: { + one: async (_task, runtime, ctx) => { + phases.push("one"); + await runtime.commit(() => ({ status: "running", checkpoint: { phase: "two" } }), ctx); + closeOnSnapshot = true; + }, + two: async () => { + phases.push("two"); + }, + }, + abort: async () => {}, + }); + registry.tasks.add(Two); + const harness = await Harness.open(new MemoryStorage(), { models: createModels(), registry: reader }, context); + harnessRef = harness; + const root = await harness.root(context); + await root.commit((tx) => tx.createTask(Two, null), context); + harness.resume(); + await eventually(() => closing !== undefined); + await closing; + expect(phases).toEqual(["one"]); + }); + + it("joins a reservation that settles while closing without starting its handler", async () => { + let ran = false; + const Never = oneStep("test.close-reservation", async () => { + ran = true; + }); + const storage = new ControlledStorage(); + const { harness, root } = await openRoot([Never], { storage }); + const id = await start(root, Never); + const held = storage.holdCommits(); + harness.resume(); + await held.entered; + const closing = harness.close(context); + held.release(); + await closing; + expect(ran).toBe(false); + const writes = storage.commits.at(-1)!.filter((write) => write.type === "task"); + expect(writes).toMatchObject([{ type: "task", value: { id, state: { status: "running" } } }]); + }); +}); diff --git a/packages/durable/test/scratch.ts b/packages/durable/test/scratch.ts index 1992b8528..e6a7143da 100644 --- a/packages/durable/test/scratch.ts +++ b/packages/durable/test/scratch.ts @@ -1,6 +1,9 @@ -// A tour of the durable Session and Harness APIs in twelve small examples. +// A tour of the durable Session and Harness APIs in fourteen small examples. // Run from packages/durable: // node --conditions=source --experimental-strip-types test/scratch.ts +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import { type AssistantMessage, @@ -17,13 +20,14 @@ import { createSession, defineDoc, defineEntry, + defineTask, type EntryRecord, Harness, MemoryStorage, type PromptInput, - type Task, type ToolRegistration, } from "../src/index.ts"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; // A Session stores conversations, transcript entries, tasks, and documents. // MemoryStorage keeps everything in memory; other storage backends keep it on disk. @@ -109,15 +113,16 @@ console.log("3. parent notes after edit:", await session.snapshot(Notes, chat.id // conversations. All three are created in one commit, so after a crash either // all of them exist or none do. -// A task definition needs a name, a version, and the task's starting state. -// This example only creates the task record; nothing runs it yet. -const Supervisor: Task = { - definition: { - name: "example.supervisor", - version: 1, - initial: () => ({ phase: "ready" }), - }, -}; +// A task definition needs a name, a version, the task's starting state, a +// handler for every phase, and an abort handler (example 13 runs a task). +// This example only creates the task record; a plain Session never runs it. +const Supervisor = defineTask({ + name: "example.supervisor", + version: 1, + initial: () => ({ phase: "ready" }), + phases: { ready: async () => {} }, + abort: async () => {}, +}); // "latest" keeps only the current value. "initial" means forks of this // conversation start without a registry, so a child doesn't inherit its @@ -478,4 +483,141 @@ console.log(`12. rendered prompt:\n${rendered.join("\n")}`); prompt.dispose(); +// ─── 13. Run a durable task ───────────────────────────────────────────────── +// A task is a small state machine. Its state, the checkpoint, is saved after +// every step, so after a crash the next open continues from the last saved +// step. The usual pattern: save what you are about to do, do it, then save +// the result. A crash between doing and saving reruns that step, so the step +// must be safe to repeat; here the fake payment service ignores a repeated key. +const payments = new Map(); +type PaymentState = { phase: "prepare" } | { phase: "charge"; key: string }; +const Payment = defineTask<{ amount: number }, PaymentState, { receipt: number }>({ + name: "example.payment", + version: 1, + initial: () => ({ phase: "prepare" }), + // One handler per phase. Each must save progress through runtime.commit(): + // its callback returns the next checkpoint or the final outcome, and that + // state is saved in the same commit as everything else the callback wrote. + phases: { + prepare: async (task, runtime, taskContext) => { + await runtime.commit( + () => ({ status: "running", checkpoint: { phase: "charge", key: `payment-${task.id}` } }), + taskContext, + ); + }, + charge: async (task, runtime, taskContext) => { + const key = task.state.checkpoint.key; + if (!payments.has(key)) payments.set(key, task.input.amount * 100); + const receipt = payments.get(key)!; + await runtime.commit( + () => ({ status: "terminal", outcome: { status: "completed", result: { receipt } } }), + taskContext, + ); + }, + }, + // Runs instead of the phases after harness.abortTask(); it decides the outcome. + abort: async (_task, runtime, taskContext) => { + await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), taskContext); + }, +}); + +// The Harness finds task code by name in the registry. Nothing runs until +// resume(); a host calls it once it is ready for work to start. +registry.tasks.add(Payment); +const paymentId = await root.commit((tx) => tx.createTask(Payment, { amount: 5 }), context); +harness.resume(); +// The finished task record is the durable receipt; waitForTask() knows its result type. +const paid = await harness.waitForTask(paymentId, context); +console.log("13. payment outcome:", paid.state.outcome); + await harness.close(context); + +// ─── 14. Close, reopen, and continue where the task stopped ───────────────── +// Everything a task needs to continue is in storage, so a new Harness over the +// same storage picks up where the last one stopped. This example keeps its +// storage in a SQLite file so it survives closing. +const directory = await mkdtemp(join(tmpdir(), "pi-durable-scratch-")); +const databasePath = join(directory, "session.sqlite"); + +let reachedTick = (_n: number): void => {}; +const Ticker = defineTask<{ to: number }, { phase: "tick"; n: number }, string>({ + name: "example.ticker", + version: 1, + initial: () => ({ phase: "tick", n: 1 }), + phases: { + tick: async (task, runtime, taskContext) => { + const n = task.state.checkpoint.n; + // Save the intent before the effect. A memo keeps the first value + // written under its name, so if the process dies after printing but + // before the next checkpoint is saved, the rerun sees the memo and + // does not print the same tick twice. + if ((await runtime.memo(`printed-${n}`, taskContext)) === undefined) { + await runtime.memo(`printed-${n}`, true, taskContext); + console.log(`14. tick ${n}`); + } + reachedTick(n); + // Save the outcome: the next tick, or the final result. + await runtime.commit( + () => + n === task.input.to + ? { status: "terminal", outcome: { status: "completed", result: `counted to ${n}` } } + : { status: "running", checkpoint: { phase: "tick", n: n + 1 } }, + taskContext, + ); + // Wait a little between ticks. Closing the Harness cancels this wait; + // the checkpoint saved above is where the next Harness continues. + await runtime.sleep(Date.now() + 50, taskContext); + }, + }, + abort: async (_task, runtime, taskContext) => { + await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), taskContext); + }, +}); +registry.tasks.add(Ticker); + +// First run: start counting to 5, and close the Harness right after tick 2 is +// printed, before its next checkpoint is saved. That is the same situation as +// a crash between the effect and saving its outcome. +const firstRun = await Harness.open( + await openNodeSqliteStorage(databasePath), + { models: createModels(), registry }, + context, +); +const tickerId = await (await firstRun.root(context)).commit((tx) => tx.createTask(Ticker, { to: 5 }), context); +const tickTwo = new Promise((resolve) => { + reachedTick = (n) => { + if (n === 2) resolve(); + }; +}); +firstRun.resume(); +await tickTwo; +await firstRun.close(context); +reachedTick = () => {}; +const saved = await readTicker(); +console.log("14. closed; saved checkpoint:", saved.state, "memos:", saved.memos); + +// Second run: nothing to restart by hand. Opening the storage finds the +// unfinished task and resume() continues it. Tick 2 runs again because its +// outcome was never saved, but its memo says it was already printed. +const secondRun = await Harness.open( + await openNodeSqliteStorage(databasePath), + { models: createModels(), registry }, + context, +); +secondRun.resume(); +const counted = await secondRun.waitForTask(tickerId, context); +console.log("14. after reopen:", counted.state.outcome); +await secondRun.close(context); +await rm(directory, { recursive: true, force: true }); + +/** Read the ticker record through a short-lived Harness over the same file. */ +async function readTicker() { + const reader = await Harness.open( + await openNodeSqliteStorage(databasePath), + { models: createModels(), registry }, + context, + ); + const record = await reader.getTask(tickerId, context); + await reader.close(context); + return record!; +} diff --git a/packages/durable/test/session-documents.test.ts b/packages/durable/test/session-documents.test.ts index 4d3fa2608..4df98cd4b 100644 --- a/packages/durable/test/session-documents.test.ts +++ b/packages/durable/test/session-documents.test.ts @@ -351,8 +351,8 @@ describe("Session document transactions", () => { it("rejects Tx use after the callback settles", async () => { const { session, conversationId } = await setupLive(); - let captured: Parameters[0]>[0] | undefined; - await session.commit((tx) => { + let captured: Parameters[0]>[0] | undefined; + await session.commitWith((tx) => { captured = tx; }, context); await expect(captured!.doc(LiveDoc, conversationId)).rejects.toThrow("Transaction has settled"); @@ -411,7 +411,7 @@ describe("Session document transactions", () => { const counter = await session.snapshot(CounterDoc, context); const commits = storage.commits.length; await expect( - session.commit(async (tx) => { + session.commitWith(async (tx) => { (await tx.doc(LiveDoc, conversationId)).message = "lost"; (await tx.doc(CounterDoc)).count = 2; // Replacing a missing task fails during assembly, after every change was prepared. diff --git a/packages/durable/test/session-forks.test.ts b/packages/durable/test/session-forks.test.ts index 03140fdd7..e7ecf0133 100644 --- a/packages/durable/test/session-forks.test.ts +++ b/packages/durable/test/session-forks.test.ts @@ -2,6 +2,7 @@ import { type ConversationId, defineDoc, defineDocFamily, + defineTask, type EntryId, StorageRejected, type StorageWrite, @@ -627,13 +628,13 @@ describe("Session conversation document forks", () => { scope: "task", initial: () => ({ value: "task" }), }); - const Work = { - definition: { - name: "fork.scope.work", - version: 1, - initial: () => ({ phase: "start" }), - }, - }; + const Work = defineTask({ + name: "fork.scope.work", + version: 1, + initial: () => ({ phase: "start" }), + phases: { start: async () => {} }, + abort: async () => {}, + }); const { session, storage, publications } = openTestSession(); const parentId = await createConversation(session); let forkAt!: EntryId; diff --git a/packages/durable/test/session-support.ts b/packages/durable/test/session-support.ts index e5096c552..e8da108d8 100644 --- a/packages/durable/test/session-support.ts +++ b/packages/durable/test/session-support.ts @@ -57,6 +57,11 @@ export class ControlledStorage extends MemoryStorage { return { entered: held.entered.promise, release: () => this.#release("find", held) }; } + /** Simulate a crash during the held commit: it never reaches storage, and later commits proceed. */ + crash(): void { + this.#commitGate = undefined; + } + failNextCommit(error: Error): void { this.#commitFailure = error; } diff --git a/packages/durable/test/session-tables.test.ts b/packages/durable/test/session-tables.test.ts index 104441172..90623adb9 100644 --- a/packages/durable/test/session-tables.test.ts +++ b/packages/durable/test/session-tables.test.ts @@ -3,9 +3,9 @@ import { type ConversationId, defineDoc, defineDocFamily, + defineTask, type EntryId, ReadAfterWrite, - type Task, type TaskId, type TaskRecord, } from "@earendil-works/pi-durable"; @@ -14,9 +14,13 @@ import { idFromNumber } from "../src/ids.ts"; import { context, createConversation, documentChanges, flush, openTestSession } from "./session-support.ts"; type Checkpoint = { phase: "start" } | { phase: "next"; step: number }; -const WorkTask: Task<{ path: string }, Checkpoint, { ok: boolean }, object> = { - definition: { name: "test.work", version: 1, initial: () => ({ phase: "start" }) }, -}; +const WorkTask = defineTask<{ path: string }, Checkpoint, { ok: boolean }>({ + name: "test.work", + version: 1, + initial: () => ({ phase: "start" }), + phases: { start: async () => {}, next: async () => {} }, + abort: async () => {}, +}); type Progress = { lines: string[] }; const ProgressDoc = defineDoc({ @@ -115,7 +119,7 @@ describe("Session transaction tables", () => { const { session } = openTestSession(); const conversationId = await createConversation(session); const taskId = await createTask(session, conversationId); - await session.commit(async (tx) => { + await session.commitWith(async (tx) => { const task = (await tx.task(taskId))!; tx.setTask(task); await expect(tx.task(taskId)).rejects.toBeInstanceOf(ReadAfterWrite); @@ -193,7 +197,7 @@ describe("Session transaction tables", () => { expect(await storage.conversation(created.child.id, context)).toEqual(created.child); await expect( - session.commit(async (tx) => { + session.commitWith(async (tx) => { const supervisor = (await tx.task(created.supervisorId))!; tx.setTask({ ...supervisor, conversationId: created.child.id }); }, context), @@ -206,7 +210,7 @@ describe("Session transaction tables", () => { const parentId = await createConversation(session); let movedStagedTaskId: TaskId<{ ok: boolean }> | undefined; await expect( - session.commit(async (tx) => { + session.commitWith(async (tx) => { const taskId = await tx.createTask(WorkTask, { path: "move" }, { conversationId: parentId }); movedStagedTaskId = taskId; tx.setTask({ @@ -232,7 +236,7 @@ describe("Session transaction tables", () => { let rejectedChildId: ConversationId | undefined; await expect( - session.commit(async (tx) => { + session.commitWith(async (tx) => { const supervisorId = await tx.createTask(WorkTask, { path: "aborting" }, { conversationId: parentId }); rejectedChildId = (await tx.createConversation({ ownership: { kind: "task", taskId: supervisorId } })).id; tx.setTask({ @@ -253,7 +257,7 @@ describe("Session transaction tables", () => { expect(await storage.conversation(rejectedChildId, context)).toBeUndefined(); await expect( - session.commit(async (tx) => { + session.commitWith(async (tx) => { const supervisorId = await tx.createTask(WorkTask, { path: "terminal" }, { conversationId: parentId }); await tx.createConversation({ ownership: { kind: "task", taskId: supervisorId } }); tx.setTask({ @@ -271,7 +275,7 @@ describe("Session transaction tables", () => { ).rejects.toThrow("is terminal"); const terminalOwnerId = await createTask(session, parentId); - await session.commit(async (tx) => { + await session.commitWith(async (tx) => { tx.setTask(terminal((await tx.task(terminalOwnerId))!)); }, context); await expect( @@ -313,7 +317,7 @@ describe("Session transaction tables", () => { const { session, storage } = openTestSession(); const conversationId = await createConversation(session); const taskId = await createTask(session, conversationId); - await session.commit(async (tx) => { + await session.commitWith(async (tx) => { const task = (await tx.task(taskId))!; tx.setTask({ id: task.id, @@ -367,7 +371,7 @@ describe("Session transaction tables", () => { const { session } = openTestSession(); const conversationId = await createConversation(session); const taskId = await createTask(session, conversationId, true); - await session.commit(async (tx) => { + await session.commitWith(async (tx) => { const task = (await tx.task(taskId))!; const progress = await tx.doc(ProgressDoc, taskId); tx.setTask(terminal(task)); @@ -390,7 +394,7 @@ describe("Session transaction tables", () => { }, context); await flush(); const published = publications.length; - await session.commit(async (tx) => { + await session.commitWith(async (tx) => { const task = (await tx.task(taskId))!; (await tx.doc(StepDoc, taskId, "new", null)).lines.push("created then retired"); tx.setTask(terminal(task)); @@ -421,7 +425,7 @@ describe("Session transaction tables", () => { ); expect(alive.items).toEqual([]); await expect( - session.commit(async (tx) => { + session.commitWith(async (tx) => { tx.setTask(terminal((await tx.task(taskId))!)); }, context), ).rejects.toThrow(`Task ${taskId} is already terminal`); diff --git a/packages/durable/test/task-support.ts b/packages/durable/test/task-support.ts new file mode 100644 index 000000000..d4cd831c8 --- /dev/null +++ b/packages/durable/test/task-support.ts @@ -0,0 +1,110 @@ +import { createModels } from "@earendil-works/pi-ai"; +import { + type AnyTask, + createRegistry, + Harness, + type Registry, + type RegistryReader, + type Storage, +} from "@earendil-works/pi-durable"; +import { context, flush } from "./session-support.ts"; + +export type Deferred = { + readonly promise: Promise; + resolve(value: T): void; + reject(error: unknown): void; +}; + +export function deferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((done, fail) => { + resolve = done; + reject = fail; + }); + return { promise, resolve, reject }; +} + +/** Reject with the signal's reason once it aborts; for handlers that block until cancelled. */ +export function aborted(signal: AbortSignal): Promise { + return new Promise((_, reject) => { + if (signal.aborted) reject(signal.reason); + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); +} + +/** Flush macrotask turns until `check` holds. */ +export async function eventually(check: () => boolean): Promise { + for (let attempt = 0; attempt < 200; attempt++) { + if (check()) return; + await flush(); + } + throw new Error("Condition was not reached"); +} + +/** Whether `promise` has settled after pending work flushes. */ +export async function settled(promise: Promise): Promise { + let done = false; + promise.then( + () => { + done = true; + }, + () => { + done = true; + }, + ); + await flush(); + return done; +} + +/** Open a Harness whose registry holds `tasks`; failures passed to `onReport` are collected. */ +export async function openTasks( + storage: Storage, + tasks: readonly AnyTask[], + options: { readonly registry?: Registry; readonly now?: () => number } = {}, +): Promise<{ readonly harness: Harness; readonly registry: Registry; readonly reports: unknown[] }> { + const registry = options.registry ?? createRegistry(); + for (const task of tasks) registry.tasks.add(task); + const reports: unknown[] = []; + const harness = await Harness.open( + storage, + { + models: createModels(), + registry, + onReport: (error) => reports.push(error), + ...(options.now === undefined ? {} : { now: options.now }), + }, + context, + ); + return { harness, registry, reports }; +} + +/** Next state that completes a task with `result`. */ +export function completed(result: R) { + return { status: "terminal", outcome: { status: "completed", result } } as const; +} + +/** Next state that aborts a task. */ +export function abortedWith(reason: string) { + return { status: "terminal", outcome: { status: "aborted", reason } } as const; +} + +/** Registry reader that counts live subscriptions. */ +export function countingReader(registry: Registry): RegistryReader & { subscriptions(): number } { + let count = 0; + return { + snapshot: () => registry.snapshot(), + subscribe: (listener) => { + count++; + const unsubscribe = registry.subscribe(listener); + let active = true; + return () => { + if (!active) return; + active = false; + count--; + unsubscribe(); + }; + }, + subscriptions: () => count, + }; +}