diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 141c91b4a..62b49de5e 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -4,7 +4,7 @@ ### Breaking Changes -- `TaskRuntime` now requires `env`, `hooks`, `getTask()`, `waitForTask()`, and `entry()`; `ToolExecutionApi` requires `env` and `diagnostic()` and no longer declares `conversation()` until owned-conversation handles land. +- `TaskRuntime` now requires `env`, `hooks`, `getTask()`, `waitForTask()`, `entry()`, and `conversation()`; `ToolExecutionApi` requires `env`, `diagnostic()`, and `conversation()`. - `createRegistry()` also pre-registers the built-in `pi.tool` and `pi.post-tools` tasks. - `ToolResultEntry` now carries `data: { diagnostics }`. - `Tx` now requires `placeSubmission()` and `latestHeadMarker()`. @@ -30,6 +30,11 @@ - Added the `pi.usage` ledger (`UsageDoc`) of assistant usage per model and tool usage per tool, `ToolExecutionResult.usage`, and `Harness.usage()` for the Session total. - Added the structural `ConversationView` with `Conversation.viewState()` and `Conversation.watch()`. - Added the experimental `watchEvents()` agent event adapter with snapshot events, translated message and tool deltas, and overflow to a snapshot. +- Added `Conversation.abort()`, which withdraws queued inputs, aborts the conversation's ordinary ownership scope, and resolves once it is idle. +- Abort marks now cascade to work owned by the aborted task, or by a task that failed, faulted, or was orphaned: live foreground tasks below it are aborted and the queued inputs of their conversations withdrawn. Background tasks are boundaries. +- Conversation and Harness idle waits now include work in owned conversations and stop at background tasks. +- A tool task whose `execute()` throws, or that is interrupted by a restart without a safe rerun, now ends `failed` (still with its error result entry), which aborts the conversations the call owns. +- Added `TaskRuntime.conversation()` and `ToolExecutionApi.conversation()`: invocation-bound `ConversationHandle`s for submitting to, aborting, and waiting on existing conversations, such as the ones a task owns. ### Fixed diff --git a/packages/durable/README.md b/packages/durable/README.md index 72b3da5a4..e47a08a49 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -21,6 +21,7 @@ Built on [`@earendil-works/pi-ai`](../ai/README.md) for model access and `@earen - [Agent Events (Experimental)](#agent-events-experimental) - [Hooks](#hooks) - [More Conversations and Forks](#more-conversations-and-forks) +- [Abort and Subagents](#abort-and-subagents) - [Your Own State](#your-own-state) - [Usage and Cost](#usage-and-cost) - [Storage](#storage) @@ -279,6 +280,46 @@ const fork = await root.fork(entryId, { ownership: { kind: "ownerless" } }, cont A fork sees its parent's entries up to `entryId` and continues independently. It keeps the parent's settings as of that entry. +## Abort and Subagents + +`await root.abort(context)` stops a conversation: queued inputs are withdrawn (queued writes stay), every task of its current work is aborted, and the call resolves once the conversation is idle. + +A conversation can be **owned** by a task. A subagent tool creates its child inside `api.commit()` with `ownership: { kind: "task", taskId: api.taskId }`, then drives it through `api.conversation(id)`: + +```typescript +registry.tools.add({ + name: "subagent", + description: "Delegate a self-contained task to a subagent and get its answer back.", + parameters: Type.Object({ task: Type.String() }), + replay: "safe", // a rerun after a crash finds the same child and submission + execute: async (args, api, context) => { + const child = await api.commit(async (tx) => { + // The ownership index remembers the child, so a rerun reuses it. + const existing = (await tx.scanConversations({ ownerTaskId: api.taskId }, 1)).items[0]; + if (existing !== undefined) return existing.id; + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + (await tx.doc(ConversationConfig, created.id)).model = { provider: "openai", modelId: "gpt-6-sol" }; + return created.id; + }, context); + await api.details({ conversationId: child }, context); // lets a UI attach to the child + const request = { type: "input", content: (args as { task: string }).task, requestId: `subagent:${api.taskId}` } as const; + const settled = await (await (await api.conversation(child, context))!.submit(request, context)).wait(context); + return { content: [{ type: "text", text: settled.status }] }; + }, +}); +``` + +Owned work belongs to its owner: + +- Aborting the call aborts the child. So does the call failing: `execute()` throwing, or a crash that interrupts a call that is not replay-safe. +- The parent is idle only once the child is. +- A task created with `{ background: true }` is a boundary: work it owns survives the parent's abort and does not keep the parent busy. + +The examples show both patterns as product code: + +- [`22-subagent-foreground.ts`](test/examples/22-subagent-foreground.ts): the tool above, returning the child's answer. The UI finds the child through the call's `details` and prints the child's events indented under the call. +- [`23-subagent-background.ts`](test/examples/23-subagent-background.ts): the call returns at once. One commit creates a background supervisor task, the child it owns, and an entry in the parent's `app.subagents` document. The supervisor submits with a stable request ID, so a restart never submits twice. The UI lists subagents from that document and shows which ones are working. + ## Your Own State Documents are typed JSON objects committed together with entries. Define one, and edit it in a commit: @@ -356,6 +397,8 @@ node --conditions=source --experimental-strip-types test/examples/14-chat.ts | [19-json](test/examples/19-json.ts) | JSON mode: agent events or raw view operations, on SQLite, JSONL, or memory | | [20-inbox](test/examples/20-inbox.ts) | Steers, follow-ups, writes, and withdrawal while busy | | [21-late-join](test/examples/21-late-join.ts) | Attaching a view and an event stream mid-run | +| [22-subagent-foreground](test/examples/22-subagent-foreground.ts) | A replay-safe subagent tool whose child the call owns, with the child's events under the call | +| [23-subagent-background](test/examples/23-subagent-background.ts) | A background subagent: supervisor task, name registry, and restart-safe submission | | [00](test/examples/00-conversation.ts)–[13](test/examples/13-recovery.ts) | The layers underneath: sessions, documents, forks, watches, tasks, recovery | Examples that call OpenAI need `OPENAI_API_KEY`; most use the faux provider otherwise. diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index 08d201cb7..0b8e79117 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -1,5 +1,5 @@ import type { AttachedReplicatedState, Context, Draft, JsonValue } from "@earendil-works/chord"; -import { withoutAbortSignal } from "@earendil-works/chord/context"; +import { withAbortSignal, withoutAbortSignal } from "@earendil-works/chord/context"; import type { ModelThinkingLevel } from "@earendil-works/pi-ai"; import { ResetEntry } from "../entries.ts"; import { SessionImpl } from "../session/session.ts"; @@ -23,14 +23,16 @@ import type { import { ROOT_CONVERSATION_ID } from "../types.ts"; import { ConversationConfig, type ConversationConfigState, DEFAULT_RETRY_POLICY } from "./config.ts"; import { readContext } from "./context.ts"; +import { withdrawQueuedInputs } from "./inbox.ts"; import { settleSchedulerOutcome } from "./live.ts"; import { BUILTIN_SETUP_KEY, BUILTIN_TASKS } from "./registry.ts"; -import { TaskScheduler } from "./scheduler.ts"; +import { type InvocationBinding, TaskScheduler } from "./scheduler.ts"; import { Submissions } from "./submissions.ts"; import type { ContextView, Conversation, ConversationCreateOptions, + ConversationHandle, ConversationInit, ConversationRetryPolicy, ConversationStreamOptions, @@ -214,6 +216,11 @@ class ConversationImpl implements Conversation { ); } + abort(context: Context): Promise { + this.#host.tasks.resume(); + return this.#host.tasks.abortConversation(this.id, context); + } + waitForIdle(context: Context): Promise { this.#host.tasks.resume(); return this.#host.tasks.waitForIdle(this.id, context); @@ -265,6 +272,11 @@ class HarnessImpl extends SessionImpl implements now, report: options.onReport ?? (() => {}), settleOutcome: settleSchedulerOutcome, + withdrawInputs: withdrawQueuedInputs, + conversation: async (id, binding, callContext) => { + const record = await this.readOnLine(() => storage.conversation(id, callContext)); + return record === undefined ? undefined : boundConversation(id, binding, this.#submissions, this.#tasks); + }, context: withoutAbortSignal(context), }); this.#submissions = new Submissions(this, storage, now, () => this.#tasks.resume()); @@ -417,6 +429,40 @@ class HarnessImpl extends SessionImpl implements } } +/** + * Invocation-bound handle for tasks and tools. Every operation, and every operation of a submission it returns, first + * checks the invocation and runs under its signal, so it rejects once the invocation ends; admitted work stays durable. + */ +function boundConversation( + id: ConversationId, + binding: InvocationBinding, + submissions: Submissions, + tasks: TaskScheduler, +): ConversationHandle { + const bind = (context: Context): Context => withAbortSignal(binding.signal, context); + const bound = (operation: (context: Context) => Promise) => { + return async (context: Context): Promise => { + binding.check(); + return operation(bind(context)); + }; + }; + return { + id, + submit: async (draft, context) => { + binding.check(); + const submission = await submissions.submit(id, draft, bind(context)); + return { + id: submission.id, + status: bound((callContext) => submission.status(callContext)), + wait: bound((callContext) => submission.wait(callContext)), + abort: bound((callContext) => submission.abort(callContext)), + }; + }, + abort: bound((callContext) => tasks.abortConversation(id, callContext)), + waitForIdle: bound((callContext) => tasks.waitForIdle(id, callContext)), + }; +} + /** Reject names newly added relative to `previous` that `snapshot` does not register; existing names are never rechecked. */ function requireRegistered( snapshot: RegistrySnapshot, diff --git a/packages/durable/src/harness/inbox.ts b/packages/durable/src/harness/inbox.ts index 0cc05e3d6..b8627072e 100644 --- a/packages/durable/src/harness/inbox.ts +++ b/packages/durable/src/harness/inbox.ts @@ -115,3 +115,17 @@ export async function removeInboxItem(tx: Tx, conversationId: ConversationId, id const index = items.findIndex((item) => item.id === id); if (index >= 0) items.splice(index, 1); } + +/** + * Withdraw every queued input of a conversation, as `Conversation.abort()` and abort cascades do: each settles + * `unanswered` with `aborted` and leaves the inbox; queued writes stay for later placement. + */ +export async function withdrawQueuedInputs(tx: Tx, conversationId: ConversationId): Promise { + const items = (await tx.doc(InboxDoc, conversationId)).items; + for (let index = items.length - 1; index >= 0; index--) { + const item = items[index]!; + if (item.mode === "write") continue; + tx.settleSubmission(item.id, { status: "unanswered", reason: "aborted" }); + items.splice(index, 1); + } +} diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index 68b0f4794..cf7fb432d 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -7,6 +7,7 @@ import type { Transaction } from "../session/transaction.ts"; import type { CommitPublication, ConversationId, + ConversationRecord, DocumentWatch, EntryId, EntryRecord, @@ -24,6 +25,7 @@ import type { import { readContext } from "./context.ts"; import type { AnyTask, + ConversationHandle, HarnessInspection, HookScope, RegistryReader, @@ -89,6 +91,15 @@ type Decision = boolean | { readonly fault: unknown }; /** Terminal outcomes the scheduler writes without running task code. */ export type SchedulerOutcome = Extract, { readonly status: "faulted" | "orphaned" }>; +/** An invocation a conversation handle is bound to: its signal, and a check that throws once it ended. */ +export type InvocationBinding = { readonly signal: AbortSignal; check(): void }; + +/** A conversation's owner edge; the owner's `background` flag is loaded with it. Both never change. */ +type OwnerEdge = { readonly conversationId: ConversationId; readonly taskId: TaskId; background: boolean | undefined }; + +/** Where ordinary ownership traversal starts: one conversation, or every ownerless conversation. */ +type Scope = { readonly conversation: ConversationId } | { readonly roots: true }; + export type TaskSchedulerOptions = { readonly session: SessionImpl; readonly storage: Storage; @@ -99,6 +110,14 @@ export type TaskSchedulerOptions = { readonly report: (error: unknown) => void; /** Harness cleanup staged in the same commit as every terminal outcome the scheduler writes itself. */ readonly settleOutcome: (tx: Transaction, record: AnyTaskRecord, outcome: SchedulerOutcome) => Promise; + /** Withdraw a conversation's queued inputs, for conversation abort and abort cascades. */ + readonly withdrawInputs: (tx: Transaction, conversationId: ConversationId) => Promise; + /** Invocation-bound handle of an existing conversation, for task runtimes and tools. */ + readonly conversation: ( + id: ConversationId, + binding: InvocationBinding, + context: Context, + ) => Promise; /** Context for scheduler commits and invocations; carries no caller cancellation. */ readonly context: Context; }; @@ -123,6 +142,8 @@ export class TaskScheduler { readonly #now: () => number; readonly #report: (error: unknown) => void; readonly #settleOutcome: TaskSchedulerOptions["settleOutcome"]; + readonly #withdrawInputs: TaskSchedulerOptions["withdrawInputs"]; + readonly #conversation: TaskSchedulerOptions["conversation"]; readonly #context: Context; readonly #live = new Map(); readonly #invocations = new Map(); @@ -131,8 +152,14 @@ export class TaskScheduler { readonly #idleWaiters = new Waiters(); /** Definition whose migration failed per task; retried only once the registry resolves another definition. */ readonly #failedMigrations = new Map(); - /** Owner conversations of each conversation, nearest first; owner edges never change. */ - readonly #owners = new Map(); + /** Owner edge of each loaded conversation, `null` when ownerless. */ + readonly #edges = new Map(); + /** Tasks that own a loaded conversation. */ + readonly #owners = new Set(); + /** Whether each terminal owner recorded cancellation intent (see `cancellationIntent()`). */ + readonly #cancelledTerminal = new Map(); + #reconcileScheduled = false; + #cascadePending = false; #unsubscribeRegistry: () => void = () => {}; #enabled = false; #closing = false; @@ -148,6 +175,8 @@ export class TaskScheduler { this.#now = options.now; this.#report = options.report; this.#settleOutcome = options.settleOutcome; + this.#withdrawInputs = options.withdrawInputs; + this.#conversation = options.conversation; this.#context = options.context; } @@ -166,6 +195,9 @@ export class TaskScheduler { tx.setTask(withState(record, { status: "pending", checkpoint: record.state.checkpoint })); } }, context); + // Derive abort marks a crash left unapplied below cancelled owners. + this.#cascadePending = true; + this.#scheduleReconcile(); } /** Enable scheduling. Idempotent; the kick does nothing once closing. */ @@ -180,8 +212,8 @@ export class TaskScheduler { } /** - * 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. + * Commit the abort mark, or settle a task that no registered definition can take as `orphaned`, then join the run + * invocation seen on the line; the commit listener signalled it. The next drain starts the abort invocation. */ async abort(id: TaskId, context: Context): Promise<"marked" | "terminal"> { const marked = await this.#session.commitWith(async (tx) => { @@ -200,10 +232,8 @@ export class TaskScheduler { 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); - } + // The commit listener signalled the run; join it. + if (marked.run !== undefined) await awaitWithContext(marked.run.done, context); return marked.result; } @@ -219,34 +249,232 @@ export class TaskScheduler { return found.promise; } - /** Resolve when no live non-background task exists, optionally within one conversation. */ + /** + * Resolve when ordinary traversal from the conversation, or from every ownerless conversation, reaches no live + * non-background task. + */ waitForIdle(conversationId: ConversationId | undefined, context: Context): Promise { if (this.#closing) return Promise.reject(closedError()); if (this.#idle(conversationId)) return Promise.resolve(); + this.#scheduleReconcile(); return this.#idleWaiters.add(conversationId, context); } + /** + * `Conversation.abort()`: in one commit, withdraw the queued inputs and mark every live non-background task that + * ordinary traversal from the conversation reaches. The commit listener signals their run invocations; resolves once + * the scope is idle. + */ + async abortConversation(conversationId: ConversationId, context: Context): Promise { + await this.#session.commitWith(async (tx) => { + const queued = await this.#loadScopes(true); + const scope = { conversation: conversationId }; + for (const record of this.#live.values()) { + if (record.background || record.abortRequested || this.#inScope(record.conversationId, scope) !== true) { + continue; + } + tx.setTask({ ...record, abortRequested: true }); + } + for (const id of queued) if (this.#inScope(id, scope) === true) await this.#withdrawInputs(tx, id); + }, context); + await this.waitForIdle(conversationId, context); + } + // ─── Scheduling ──────────────────────────────────────────────────────── #observe(publication: CommitPublication): void { + const updated: LiveTaskRecord[] = []; 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); + const live = record as LiveTaskRecord; + if (live.abortRequested && this.#live.get(record.id)?.abortRequested !== true) { + this.#cascadePending = true; + // Signal a run invocation of the newly marked task; its next step ends it. + const invocation = this.#invocations.get(record.id); + if (invocation?.mode === "run") invocation.controller.abort(); + } + this.#live.set(record.id, live); + updated.push(live); continue; } + // Only owners' outcomes matter to traversal; `#loadChain` fills in owners loaded later. + if (this.#owners.has(record.id)) { + const cancelled = cancellationIntent(record); + this.#cancelledTerminal.set(record.id, cancelled); + if (cancelled) this.#cascadePending = true; + } this.#live.delete(record.id); this.#failedMigrations.delete(record.id); this.#taskWaiters.resolve(record.id, record as SettledTask); } + // Edges after the tasks, so an owner created in the same commit is live. + for (const change of publication.changes) { + if (change.type !== "conversation" || this.#edges.has(change.value.id)) continue; + const owner = change.value.owner; + this.#setEdge( + change.value.id, + owner === undefined ? null : { ...owner, background: this.#live.get(owner.taskId)?.background }, + ); + } + for (const change of publication.changes) { + // A queued input below a cancelled owner is withdrawn, even after its cascade. + if (change.type !== "submission" || change.value.status !== "queued" || change.value.type !== "input") + continue; + const id = change.value.conversationId; + if (!this.#chainKnown(id) || this.#belowCancelled(id)) this.#cascadePending = true; + } + for (const record of updated) { + // Work created below a cancelled owner, even after its cascade, is aborted too. + if (!this.#chainKnown(record.conversationId)) this.#scheduleReconcile(); + else if (!record.background && !record.abortRequested && this.#belowCancelled(record.conversationId)) { + this.#cascadePending = true; + } + } + // Also retries, with the next commit of any kind, a cascade whose commit failed. + if (this.#cascadePending) this.#scheduleReconcile(); if (!changed) return; + this.#resolveIdleWaiters(); + this.#kick(); + } + + #resolveIdleWaiters(): void { for (const conversationId of this.#idleWaiters.keys()) { if (this.#idle(conversationId)) this.#idleWaiters.resolve(conversationId); } - this.#kick(); + } + + // ─── Ownership ─────────────────────────────────────────────────────────── + + #scheduleReconcile(): void { + if (this.#reconcileScheduled || this.#closing) return; + this.#reconcileScheduled = true; + queueMicrotask(() => void this.#reconcile()); + } + + /** + * Load missing owner edges, then, when cancellation intent is pending, derive abort marks in one commit: every live + * non-background task below a cancelled owner, found by walking up without crossing a background owner that is not + * itself cancelled, is marked, and the queued inputs of such conversations are withdrawn. The cancelled owner's own + * record is the durable intent, so this also repairs marks a crash left unapplied. Resolves idle waiters that the + * loaded edges decide. + */ + async #reconcile(): Promise { + this.#reconcileScheduled = false; + const cascade = this.#cascadePending; + this.#cascadePending = false; + try { + await this.#session.commitWith(async (tx) => { + if (this.#closing) return; + const queued = await this.#loadScopes(cascade); + // Loading edges can reveal a cancelled owner, so marks are derived on every pass. + for (const record of this.#live.values()) { + if (record.background || record.abortRequested || !this.#belowCancelled(record.conversationId)) continue; + tx.setTask({ ...record, abortRequested: true }); + } + for (const id of queued) if (this.#belowCancelled(id)) await this.#withdrawInputs(tx, id); + }, this.#context); + } catch (error) { + // Any pass may have staged marks, so a failed one is retried with the next commit. + this.#cascadePending = true; + if (!this.#closing) this.#report(error); + } + this.#resolveIdleWaiters(); + } + + /** + * Load the owner chains of every live task's conversation and, with `queued`, of every conversation with queued + * submissions, on the Session line; returns the latter. Reads committed Storage directly, so it may run inside a + * commit callback. + */ + async #loadScopes(queued: boolean): Promise { + for (const record of [...this.#live.values()]) await this.#loadChain(record.conversationId); + if (!queued) return []; + const submissions = await scanAll((cursor) => + this.#storage.scanSubmissions({ status: "queued" }, SCAN_PAGE_SIZE, cursor, this.#context), + ); + const conversations = [...new Set(submissions.map((submission) => submission.conversationId))]; + for (const id of conversations) await this.#loadChain(id); + return conversations; + } + + /** Load the owner edges from `conversationId` up to its ownerless root. */ + async #loadChain(conversationId: ConversationId): Promise { + let id: ConversationId | undefined = conversationId; + while (id !== undefined) { + let edge: OwnerEdge | null | undefined = this.#edges.get(id); + if (edge === undefined) { + const record: ConversationRecord | undefined = await this.#storage.conversation(id, this.#context); + edge = record?.owner === undefined ? null : { ...record.owner, background: undefined }; + this.#setEdge(id, edge); + } + if (edge === null) return; + if (edge.background === undefined || this.#ownerCancelled(edge.taskId) === undefined) { + const owner: AnyTaskRecord | undefined = + this.#live.get(edge.taskId) ?? (await this.#storage.task(edge.taskId, this.#context)); + edge.background = owner?.background ?? false; + if (owner?.state.status === "terminal") this.#cancelledTerminal.set(owner.id, cancellationIntent(owner)); + } + id = edge.conversationId; + } + } + + #setEdge(conversationId: ConversationId, edge: OwnerEdge | null): void { + this.#edges.set(conversationId, edge); + if (edge !== null) this.#owners.add(edge.taskId); + } + + /** Whether every owner edge above `conversationId` is loaded. */ + #chainKnown(conversationId: ConversationId): boolean { + let id: ConversationId | undefined = conversationId; + while (id !== undefined) { + const edge = this.#edges.get(id); + if (edge === undefined || (edge !== null && edge.background === undefined)) return false; + id = edge?.conversationId; + } + return true; + } + + /** Cancellation intent of an owner task: its abort mark while live, or a terminal outcome other than `completed`. */ + #ownerCancelled(taskId: TaskId): boolean | undefined { + const live = this.#live.get(taskId); + return live !== undefined ? live.abortRequested : this.#cancelledTerminal.get(taskId); + } + + /** + * Whether ordinary traversal from `scope` reaches `conversationId`: walking up its owner edges reaches the scope's + * conversation, or an ownerless one for `roots`, without crossing a background owner. `undefined` while an edge is + * not loaded. + */ + #inScope(conversationId: ConversationId, scope: Scope): boolean | undefined { + let id = conversationId; + for (;;) { + if ("conversation" in scope && id === scope.conversation) return true; + const edge = this.#edges.get(id); + if (edge === undefined || (edge !== null && edge.background === undefined)) return undefined; + if (edge === null) return "roots" in scope; + if (edge.background) return false; + id = edge.conversationId; + } + } + + /** + * Whether a cancelled owner's abort reaches `conversationId`: walking up finds a cancelled owner before a background + * owner that is not cancelled. A cancelled background owner includes its ordinary subtree; nested background owners + * stay boundaries. + */ + #belowCancelled(conversationId: ConversationId): boolean { + let id = conversationId; + for (;;) { + const edge = this.#edges.get(id); + if (edge === undefined || edge === null) return false; + if (this.#ownerCancelled(edge.taskId) === true) return true; + if (edge.background !== false) return false; + id = edge.conversationId; + } } /** Close listener: runs synchronously once admission is sealed, before `join()`. */ @@ -543,10 +771,11 @@ export class TaskScheduler { invocation.controller.abort(endedError(invocation)); } + /** No live non-background task in the scope; a task whose owner edges are not loaded yet counts as inside. */ #idle(conversationId: ConversationId | undefined): boolean { + const scope: Scope = conversationId === undefined ? { roots: true } : { conversation: conversationId }; for (const record of this.#live.values()) { - if (record.background) continue; - if (conversationId === undefined || record.conversationId === conversationId) return false; + if (!record.background && this.#inScope(record.conversationId, scope) !== false) return false; } return true; } @@ -560,7 +789,7 @@ export class TaskScheduler { for (const { handlers, scope } of snapshot().hooks(task())) { const handler = (handlers as Record)[name]; if (typeof handler !== "function") continue; - if (scope !== undefined && !(await this.#inScope(invocation, scope))) continue; + if (scope !== undefined && !(await this.#hookMatches(invocation, scope))) continue; try { await invoke(handler.bind(handlers)); } catch (error) { @@ -623,6 +852,19 @@ export class TaskScheduler { this.#read(invocation, () => this.waitForTask(id, withAbortSignal(invocation.controller.signal, context)), )) as ErasedRuntime["waitForTask"], + conversation: (id, context) => + this.#read(invocation, () => + this.#conversation( + id, + { + signal: invocation.controller.signal, + check: () => { + if (invocation.ended) throw endedError(invocation); + }, + }, + context, + ), + ), entry: ((...args: readonly unknown[]) => { const [token, id, context] = args.length === 2 @@ -643,22 +885,19 @@ export class TaskScheduler { }; } - /** Whether a scoped hook registration matches the invocation's conversation. */ - async #inScope(invocation: Invocation, scope: HookScope): Promise { + /** Whether a scoped hook registration matches the invocation's conversation: itself, or an owner for `subtree`. */ + async #hookMatches(invocation: Invocation, scope: HookScope): Promise { if (scope.conversationId === invocation.conversationId) return true; if (scope.subtree !== true) return false; - return (await this.#ownersOf(invocation.conversationId)).includes(scope.conversationId); - } - - /** Conversations owning `conversationId` through tasks, transitively, nearest first. */ - async #ownersOf(conversationId: ConversationId): Promise { - const cached = this.#owners.get(conversationId); - if (cached !== undefined) return cached; - const record = await this.#session.readOnLine(() => this.#storage.conversation(conversationId, this.#context)); - const owner = record?.owner?.conversationId; - const owners = owner === undefined ? [] : [owner, ...(await this.#ownersOf(owner))]; - this.#owners.set(conversationId, owners); - return owners; + if (!this.#chainKnown(invocation.conversationId)) { + await this.#session.readOnLine(() => this.#loadChain(invocation.conversationId)); + } + let edge = this.#edges.get(invocation.conversationId); + while (edge) { + if (edge.conversationId === scope.conversationId) return true; + edge = this.#edges.get(edge.conversationId); + } + return false; } /** Run a committed-state read unless the invocation has ended. */ @@ -718,6 +957,11 @@ export class TaskScheduler { } } +/** A terminal task's durable cancellation intent: its abort mark, or any outcome but `completed`. */ +function cancellationIntent(record: AnyTaskRecord): boolean { + return record.abortRequested || (record.state.status === "terminal" && record.state.outcome.status !== "completed"); +} + /** 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; diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index b7029ca6c..4cd172127 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -57,11 +57,11 @@ export const ToolTask = defineTask error, context); + return settle(runtime, call, COMPLETED, () => error, context); } const prepared = prepare(tool, call.arguments as JsonObject); const checked = "error" in prepared ? prepared : validate(tool, call, prepared.args); - if ("error" in checked) return settle(runtime, call, "completed", () => invalid(checked.error), context); + if ("error" in checked) return settle(runtime, call, COMPLETED, () => invalid(checked.error), context); let args = checked.args; let block: string | undefined; await runtime.hooks.each("beforeTool", async (hook) => { @@ -77,10 +77,10 @@ export const ToolTask = defineTask blocked, context); + return settle(runtime, call, COMPLETED, () => blocked, context); } const validated = validate(tool, call, args); - if ("error" in validated) return settle(runtime, call, "completed", () => invalid(validated.error), context); + if ("error" in validated) return settle(runtime, call, COMPLETED, () => invalid(validated.error), context); const final = validated.args; await runtime.commit(async (tx) => { const slot = toolSlot(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId); @@ -105,13 +105,15 @@ export const ToolTask = defineTask fromSlot(slot, "interrupted", message), context); + // `failed` records cancellation intent, so the call's owned conversations, left unsupervised, are aborted. + const ending = { status: "failed", message } as const; + await settle(runtime, call, ending, (slot) => fromSlot(slot, "interrupted", message), context); }, }, abort: async (task, runtime, context) => { const call = await readCall(runtime, task.input, context); const message = `Tool ${call.name} was aborted`; - await settle(runtime, call, "aborted", (slot) => fromSlot(slot, "aborted", message), context); + await settle(runtime, call, { status: "aborted" }, (slot) => fromSlot(slot, "aborted", message), context); }, }); @@ -229,12 +231,14 @@ async function run( }, getTask: runtime.getTask, waitForTask: runtime.waitForTask, + conversation: runtime.conversation, snapshot: runtime.snapshot, snapshotAsOf: runtime.snapshotAsOf, watchDoc: runtime.watchDoc, }; let result: ToolExecutionResult; + let ending = COMPLETED; try { result = await tool.execute(args, api, context); } catch (error) { @@ -244,6 +248,9 @@ async function run( throw error; } result = { isError: true, diagnostics: [toolDiagnostic("tool_error", errorText(error))] }; + // A throw ends the task `failed`, which cancels what the call owned; it no longer supervises it. The error text + // is already in the result entry. + ending = { status: "failed", message: `Tool ${call.name} threw` }; } ended = true; reported.output.end(); @@ -251,7 +258,7 @@ async function run( const pending = await progress.stop(); try { const settled = await finalResult(runtime, call, result, reported, context); - await settle(runtime, call, "completed", () => settled, context); + await settle(runtime, call, ending, () => settled, context); } catch (error) { for (const waiter of pending) waiter.reject(error); throw error; @@ -352,7 +359,7 @@ async function finalResult( async function settle( runtime: Runtime, call: ToolCall, - status: "completed" | "aborted", + ending: Ending, build: (slot: Readonly | undefined) => ToolExecutionResult, context: Context, ): Promise { @@ -362,7 +369,12 @@ async function settle( const entry = await appendToolResult(tx, runtime.conversationId, call, result, runtime.now()); if (slot !== undefined) finishSlot(slot, entry.id); const entryId = entry.id; - if (status === "aborted") return { status: "terminal", outcome: { status: "aborted", result: { entryId } } }; + if (ending.status === "aborted") + return { status: "terminal", outcome: { status: "aborted", result: { entryId } } }; + if (ending.status === "failed") { + const error = { message: ending.message }; + return { status: "terminal", outcome: { status: "failed", error, result: { entryId } } }; + } // Tools build control objects freely; drop keys set to undefined so the task result is strict JSON. const control = result.control === undefined @@ -372,6 +384,14 @@ async function settle( }, context); } +/** + * How a tool task ends; the result entry is appended either way. `failed` (execution threw or was interrupted) + * records cancellation intent for the conversations the call owns; a result with `isError` still completes. + */ +type Ending = { readonly status: "completed" | "aborted" } | { readonly status: "failed"; readonly message: string }; + +const COMPLETED: Ending = { status: "completed" }; + /** An error result from the slot's durable partial output, details, and diagnostics. */ function fromSlot(slot: Readonly | undefined, code: string, message: string): ToolExecutionResult { const diagnostics = [...(slot?.diagnostics ?? [])]; diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 481cc18ab..695722d93 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -100,11 +100,16 @@ export type AnyTask = { /** Hook handler map declared by a task definition. */ export type HooksOf = K extends Task ? H : never; -/** Invocation-bound conversation operations available to tools. */ +/** + * Invocation-bound conversation operations for tasks and tools. Rejects after the invocation ends; passive entries are + * written with ordinary transaction writes instead. + */ export interface ConversationHandle { readonly id: ConversationId; submit(submission: InputSubmissionDraft, context: Context): Promise; + /** `Conversation.abort()`: withdraw queued inputs, abort the ordinary ownership scope, and wait until it is idle. */ abort(context: Context): Promise; + /** Resolve when the conversation's ordinary ownership scope has no live non-background task. */ waitForIdle(context: Context): Promise; } @@ -168,7 +173,8 @@ export interface ToolExecutionApi extends DocumentObserver, DocumentReader { ): Promise>; getTask(id: TaskId, context: Context): Promise | undefined>; waitForTask(id: TaskId, context: Context): Promise>; - // conversation() arrives with ConversationHandle in Package 18. + /** Invocation-bound handle of an existing conversation, such as one this tool created in `commit()`. */ + conversation(id: ConversationId, context: Context): Promise; } /** Executable tool registered in a registry. Only pi-ai `Tool` fields enter the transcript. */ @@ -448,7 +454,15 @@ 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. */ + /** + * Withdraw queued inputs (queued writes stay), mark every live non-background task of the ordinary ownership scope, + * signal them, and resolve once the scope is idle. Background subtrees survive. + */ + abort(context: Context): Promise; + /** + * Resolve when the ordinary ownership scope has no live non-background task: this conversation and the conversations + * owned, transitively, by its non-background tasks. + */ waitForIdle(context: Context): Promise; /** The structural view (spec §9.3) as a disposable read-only Chord state. */ viewState(context: Context): Promise>; @@ -489,7 +503,7 @@ export interface Harness extends Session { 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. */ + /** Resolve when the ordinary ownership scope of every ownerless conversation has no live non-background task. */ waitForIdle(context: Context): Promise; /** Session total: every conversation's `pi.usage` summed. */ usage(context: Context): Promise; diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index 7b4003fd4..cafbc277b 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -2,7 +2,7 @@ import type { AttachedReplicatedState, Context, Draft, JsonValue } from "@earend import type { Op } from "@earendil-works/chord/delta"; import type { Message, Models } from "@earendil-works/pi-ai"; import type { ExecutionEnv } from "./env/index.ts"; -import type { ContextView, RegistrySnapshot, SettledTask } from "./harness/types.ts"; +import type { ContextView, ConversationHandle, RegistrySnapshot, SettledTask } from "./harness/types.ts"; /** JSON object used as the root of every durable document. */ export type JsonObject = { [key: string]: JsonValue }; @@ -194,6 +194,11 @@ export interface TaskRuntime extends DocumentObserver getTask(id: TaskId, context: Context): Promise | undefined>; /** Resolve with the task's terminal receipt; rejects when the invocation ends. */ waitForTask(id: TaskId, context: Context): Promise>; + /** + * Invocation-bound handle of an existing conversation, for example one this task owns; `undefined` when absent. Its + * operations and the submissions it returns reject after the invocation ends; admitted work stays durable. + */ + conversation(id: ConversationId, context: Context): Promise; /** Committed entry visible from the task's conversation. */ entry(id: EntryId, context: Context): Promise; /** Undefined when the entry is absent, not visible, or has another kind. */ diff --git a/packages/durable/test/examples/22-subagent-foreground.ts b/packages/durable/test/examples/22-subagent-foreground.ts new file mode 100644 index 000000000..4a57b5ad9 --- /dev/null +++ b/packages/durable/test/examples/22-subagent-foreground.ts @@ -0,0 +1,122 @@ +// A foreground subagent tool: the parent's tool call creates a child conversation it owns, runs one task there, and +// returns the child's answer. Aborting the tool call aborts the child. A UI finds the child through the tool's running +// details and shows its events under the call. +// Uses OpenAI when OPENAI_API_KEY is set, and a scripted faux model otherwise. +// Run from packages/durable: +// node --conditions=source --experimental-strip-types test/examples/22-subagent-foreground.ts +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { type AssistantMessage, type FauxResponseStep, Type } from "@earendil-works/pi-ai"; +import { createModels } from "@earendil-works/pi-ai/models"; +import { fauxAssistantMessage, fauxProvider, fauxText, fauxToolCall } from "@earendil-works/pi-ai/providers/faux"; +import { openaiProvider } from "@earendil-works/pi-ai/providers/openai"; +import { + type AgentEvent, + AssistantEntry, + ConversationConfig, + type ConversationId, + createRegistry, + Harness, + MemoryStorage, + type ToolRegistration, + watchEvents, +} from "../../src/index.ts"; + +const context = BACKGROUND_CONTEXT; + +// ─── Product code: the subagent tool ──────────────────────────────────────── + +const subagent: ToolRegistration = { + name: "subagent", + description: "Delegate a self-contained task to a subagent and get its answer back.", + parameters: Type.Object({ task: Type.String({ description: "What the subagent should do" }) }), + // Safe to rerun after a crash: a rerun finds the child it already created and the submission it already made. + replay: "safe", + execute: async (args, api, callContext) => { + const { task } = args as { task: string }; + // The child is owned by this tool call's task, so aborting the call aborts the child. It starts with the + // parent's model and every tool except this one. + const parent = await api.snapshot(ConversationConfig, api.conversationId, callContext); + const child = await api.commit(async (tx) => { + // Ownership records the child: a rerun of this call finds it instead of creating another. + const existing = (await tx.scanConversations({ ownerTaskId: api.taskId }, 1)).items[0]; + if (existing !== undefined) return existing.id; + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + const config = await tx.doc(ConversationConfig, created.id); + if (parent?.model !== undefined) config.model = { ...parent.model }; + config.activeTools = config.activeTools.filter((name) => name !== "subagent"); + return created.id; + }, callContext); + // A UI watching the parent sees this and can attach to the child. + await api.details({ conversationId: child }, callContext); + + const conversation = (await api.conversation(child, callContext))!; + // The request ID makes a rerun get back the submission it made before the crash. + const request = { type: "input", content: task, requestId: `subagent:${api.taskId}` } as const; + const submission = await conversation.submit(request, callContext); + const settled = await submission.wait(callContext); + if (settled.status !== "done" || settled.type !== "input") throw new Error(`Subagent failed: ${settled.status}`); + const answer = await api.commit((tx) => tx.entry(AssistantEntry, settled.answer), callContext); + const text = (answer?.model?.[0] as AssistantMessage).content.flatMap((c) => (c.type === "text" ? [c.text] : [])); + return { content: [{ type: "text", text: text.join("") }], details: { conversationId: child } }; + }, +}; + +// ─── Host setup ───────────────────────────────────────────────────────────── + +const models = createModels(); +let model = { provider: "openai", modelId: "gpt-6-sol" }; +if (process.env.OPENAI_API_KEY !== undefined) { + models.setProvider(openaiProvider()); +} else { + // The parent delegates, the child answers, and the parent reports. + const faux = fauxProvider(); + models.setProvider(faux.provider); + model = { provider: "faux", modelId: "faux-1" }; + const delegate = fauxToolCall("subagent", { task: "Name three prime numbers." }, { id: "call-1" }); + faux.setResponses([ + fauxAssistantMessage([delegate], { stopReason: "toolUse" }), + fauxAssistantMessage([fauxText("2, 3, and 5.")]), + fauxAssistantMessage([fauxText("The subagent says: 2, 3, and 5.")]), + ] satisfies FauxResponseStep[]); +} +const registry = createRegistry(); +registry.tools.add(subagent); +const harness = await Harness.open(new MemoryStorage(), { models, registry }, context); +const root = await harness.root(context); +await root.setModel(model, context); + +// ─── UI: the parent's events, with each subagent's events indented under its call ─── + +const print = (indent: string, event: AgentEvent): void => { + if (event.type === "message_end" && event.entry.kind === "pi.assistant") { + const message = event.entry.model?.[0] as AssistantMessage; + const text = message.content.flatMap((c) => (c.type === "text" ? [c.text] : [])).join(""); + if (text !== "") console.log(`${indent}assistant: ${text}`); + } else if (event.type === "tool_execution_start") { + console.log(`${indent}tool ${event.toolName}(${JSON.stringify(event.args)})`); + } +}; +const attached = new Set(); +const attach = async (id: ConversationId, indent: string): Promise => { + attached.add(id); + const stream = await watchEvents(harness, id, context); + stream.start(async (events) => { + for (const event of events) { + print(indent, event); + if (event.type !== "tool_execution_update") continue; + const child = (event.details as { conversationId?: ConversationId } | undefined)?.conversationId; + if (child !== undefined && !attached.has(child)) await attach(child, `${indent} `); + } + }); +}; +await attach(root.id, ""); + +const submission = await root.submit( + { type: "input", content: "Use the subagent tool to find three prime numbers, then tell me what it said." }, + context, +); +await submission.wait(context); +await harness.waitForIdle(context); +// Event callbacks run after their commit; let the last ones print. +await new Promise((resolve) => setTimeout(resolve, 0)); +await harness.close(context); diff --git a/packages/durable/test/examples/23-subagent-background.ts b/packages/durable/test/examples/23-subagent-background.ts new file mode 100644 index 000000000..0aea5fda7 --- /dev/null +++ b/packages/durable/test/examples/23-subagent-background.ts @@ -0,0 +1,171 @@ +// A background subagent tool: the call starts a named subagent and returns at once. One commit creates a background +// supervisor task, a child conversation it owns, and an entry in the parent's `app.subagents` document. The supervisor +// submits the task with a stable request ID, so a crash before or after admission never submits twice. Abort and idle +// waits of the parent stop at the supervisor, so the parent answers while the subagent keeps working. +// Uses OpenAI when OPENAI_API_KEY is set, and a scripted faux model otherwise. +// Run from packages/durable: +// node --conditions=source --experimental-strip-types test/examples/23-subagent-background.ts +import type { Context } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { type AssistantMessage, type FauxResponseStep, Type } from "@earendil-works/pi-ai"; +import { createModels } from "@earendil-works/pi-ai/models"; +import { fauxAssistantMessage, fauxProvider, fauxText, fauxToolCall } from "@earendil-works/pi-ai/providers/faux"; +import { openaiProvider } from "@earendil-works/pi-ai/providers/openai"; +import { + AssistantEntry, + ConversationConfig, + type ConversationId, + createRegistry, + defineDoc, + defineTask, + type EntryId, + Harness, + LiveDoc, + MemoryStorage, + type TaskId, + type ToolRegistration, +} from "../../src/index.ts"; + +const context = BACKGROUND_CONTEXT; + +// ─── Product code ──────────────────────────────────────────────────────────── + +type Subagent = { conversationId: ConversationId; supervisor: TaskId<{ answer: EntryId }>; requestId: string }; + +/** The parent's subagents by name. A fork starts without them. */ +const Subagents = defineDoc<{ agents: Record }>({ + kind: "app.subagents", + version: 1, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({ agents: {} }), +}); + +/** Background supervisor: submits the subagent's task and waits for its answer. */ +const Supervisor = defineTask< + { parent: ConversationId; name: string; task: string }, + { phase: "run" }, + { answer: EntryId } +>({ + name: "app.subagent-supervisor", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (supervisor, runtime, taskContext) => { + const { parent, name, task } = supervisor.input; + const agent = (await runtime.snapshot(Subagents, parent, taskContext))!.agents[name]!; + const child = (await runtime.conversation(agent.conversationId, taskContext))!; + // After a restart, the same request ID returns the submission admitted before the crash. + const submission = await child.submit( + { type: "input", content: task, requestId: agent.requestId }, + taskContext, + ); + const settled = await submission.wait(taskContext); + if (settled.status !== "done" || settled.type !== "input") throw new Error(`Subagent ${name} failed`); + await runtime.commit( + () => ({ status: "terminal", outcome: { status: "completed", result: { answer: settled.answer } } }), + taskContext, + ); + }, + }, + abort: (_supervisor, runtime, taskContext) => + runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), taskContext), +}); + +const spawnSubagent: ToolRegistration = { + name: "spawn_subagent", + description: "Start a named subagent that works on a task in the background.", + parameters: Type.Object({ name: Type.String(), task: Type.String() }), + // Safe to rerun after a crash: the name registry deduplicates the spawn. + replay: "safe", + execute: async (args, api, callContext) => { + const { name, task } = args as { name: string; task: string }; + const parentConfig = await api.snapshot(ConversationConfig, api.conversationId, callContext); + const conversationId = await api.commit(async (tx) => { + const agents = (await tx.doc(Subagents, api.conversationId)).agents; + if (Object.hasOwn(agents, name)) return agents[name]!.conversationId; + const supervisor = await tx.createTask( + Supervisor, + { parent: api.conversationId, name, task }, + { background: true }, + ); + const child = await tx.createConversation({ ownership: { kind: "task", taskId: supervisor } }); + const config = await tx.doc(ConversationConfig, child.id); + if (parentConfig?.model !== undefined) config.model = { ...parentConfig.model }; + config.activeTools = []; + agents[name] = { conversationId: child.id, supervisor, requestId: `subagent:${name}` }; + return child.id; + }, callContext); + return { content: [{ type: "text", text: `Started subagent ${name}.` }], details: { conversationId } }; + }, +}; + +// ─── Host setup ───────────────────────────────────────────────────────────── + +const models = createModels(); +let model = { provider: "openai", modelId: "gpt-6-sol" }; +if (process.env.OPENAI_API_KEY !== undefined) { + models.setProvider(openaiProvider()); +} else { + // Parent and child requests interleave, so one router answers each by what it asks. + const faux = fauxProvider(); + models.setProvider(faux.provider); + model = { provider: "faux", modelId: "faux-1" }; + const route: FauxResponseStep = (request) => { + const last = request.messages.at(-1)!; + if (last.role === "toolResult") return fauxAssistantMessage([fauxText("The reader subagent is on it.")]); + if (JSON.stringify(last).includes("Summarize the plot")) { + // The subagent takes a while, so the parent answers first. + return new Promise((resolve) => + setTimeout(() => resolve(fauxAssistantMessage([fauxText("A whale, a captain, an obsession.")])), 300), + ); + } + const call = fauxToolCall("spawn_subagent", { name: "reader", task: "Summarize the plot of Moby Dick." }); + return fauxAssistantMessage([call], { stopReason: "toolUse" }); + }; + faux.setResponses([route, route, route]); +} +const registry = createRegistry(); +registry.tasks.add(Supervisor); +registry.tools.add(spawnSubagent); +const harness = await Harness.open(new MemoryStorage(), { models, registry }, context); +const root = await harness.root(context); +await root.setModel(model, context); + +// ─── A run that starts a subagent, and a UI listing subagents ─────────────── + +const answerText = async (id: EntryId, callContext: Context): Promise => { + const entry = await harness.commit((tx) => tx.entry(AssistantEntry, id), callContext); + return (entry?.model?.[0] as AssistantMessage).content.flatMap((c) => (c.type === "text" ? [c.text] : [])).join(""); +}; +const listSubagents = async (): Promise => { + const agents = (await harness.snapshot(Subagents, root.id, context))?.agents ?? {}; + for (const [name, agent] of Object.entries(agents)) { + const active = (await harness.snapshot(LiveDoc, agent.conversationId, context))?.run !== undefined; + console.log(` ${name}: conversation ${agent.conversationId}, ${active ? "working" : "idle"}`); + } +}; + +const settled = await ( + await root.submit( + { type: "input", content: "Start a subagent named reader that summarizes the plot of Moby Dick." }, + context, + ) +).wait(context); +if (settled.status === "done" && settled.type === "input") + console.log("parent:", await answerText(settled.answer, context)); +// The parent's idle wait stops at the background supervisor. +await root.waitForIdle(context); +console.log("subagents:"); +await listSubagents(); + +// Wait for the subagent's supervisor, then read the answer it recorded. +const reader = (await harness.snapshot(Subagents, root.id, context))!.agents.reader!; +const done = await harness.waitForTask(reader.supervisor, context); +if (done.state.outcome.status === "completed") { + console.log("reader:", await answerText(done.state.outcome.result.answer, context)); +} +console.log("subagents:"); +await listSubagents(); +await harness.close(context); diff --git a/packages/durable/test/harness-ownership.test.ts b/packages/durable/test/harness-ownership.test.ts new file mode 100644 index 000000000..0cdc9db3f --- /dev/null +++ b/packages/durable/test/harness-ownership.test.ts @@ -0,0 +1,661 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { type AssistantMessage, fauxAssistantMessage, fauxText, fauxToolCall, Type } from "@earendil-works/pi-ai"; +import { + type Conversation, + ConversationConfig, + type ConversationHandle, + type ConversationId, + createSession, + defineDoc, + defineTask, + type EntryId, + type Harness, + LiveDoc, + MemoryStorage, + type Storage, + StorageRejected, + type Submission, + type TaskId, +} from "@earendil-works/pi-durable"; +import { afterEach, describe, expect, it } from "vitest"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { chatSetup, openChat } from "./chat-support.ts"; +import { context } from "./session-support.ts"; +import { aborted, type Deferred, deferred, openTasks, settled } from "./task-support.ts"; + +/** Outcome a held task commits once its gate opens. */ +type Ending = "completed" | "failed"; +const gates = new Map>(); + +function gate(name: string): Deferred { + let found = gates.get(name); + if (found === undefined) { + found = deferred(); + gates.set(name, found); + } + return found; +} + +/** A task that holds until its named gate opens or it is aborted. */ +const Hold = defineTask<{ name: string }, { phase: "hold" }, null>({ + name: "test.hold", + version: 1, + initial: () => ({ phase: "hold" }), + phases: { + hold: async (task, runtime, ctx) => { + const ending = await Promise.race([gate(task.input.name).promise, aborted(runtime.signal)]); + await runtime.commit( + () => + ending === "completed" + ? { status: "terminal", outcome: { status: "completed", result: null } } + : { status: "terminal", outcome: { status: "failed", error: { message: "gate failed" } } }, + ctx, + ); + }, + }, + abort: (_task, runtime, ctx) => runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx), +}); + +function open(name: string, ending: Ending): void { + gate(name).resolve(ending); +} + +const directories = new Set(); +afterEach(async () => { + gates.clear(); + for (const directory of directories) await rm(directory, { recursive: true, force: true }); + directories.clear(); +}); + +type Tree = { owner: TaskId; child: ConversationId; inner: TaskId }; + +/** In `parent`, stage task `owner` and a conversation it owns holding task `inner`, in one commit. */ +async function ownedChild(parent: Conversation, name: string, options: { background?: boolean } = {}): Promise { + return parent.commit(async (tx) => { + const owner = await tx.createTask(Hold, { name }, { background: options.background ?? false }); + const child = await tx.createConversation({ ownership: { kind: "task", taskId: owner } }); + const inner = await tx.createTask(Hold, { name: `${name}.inner` }, { conversationId: child.id }); + return { owner, child: child.id, inner }; + }, context); +} + +/** A faux response held until `release` or cancellation; `reached` resolves when the request is sent. */ +function gated(message: AssistantMessage) { + const reached = deferred(); + const gate = deferred(); + const step = async (_request: unknown, options?: { signal?: AbortSignal }) => { + reached.resolve(); + await Promise.race([gate.promise, aborted(options!.signal!)]); + return message; + }; + return { step, reached: reached.promise, release: () => gate.resolve() }; +} + +/** Mark a conversation busy with `task` standing in for its run, so submissions queue. */ +async function busy(conversation: Conversation, task: TaskId): Promise { + await conversation.commit(async (tx) => { + (await tx.doc(LiveDoc, conversation.id)).run = { taskId: task, inputs: [] }; + }, context); +} + +async function status(harness: Harness, id: TaskId) { + return (await harness.getTask(id, context))!.state; +} + +async function openHarness(storage: Storage = new MemoryStorage()) { + const opened = await openTasks(storage, [Hold]); + const root = await opened.harness.root(context); + opened.harness.resume(); + return { ...opened, root }; +} + +describe("ownership", () => { + it("keeps a conversation busy while its owned foreground subtree has live work, even after the owner completed", async () => { + const { harness, root } = await openHarness(); + const tree = await ownedChild(root, "owner"); + open("owner", "completed"); + await harness.waitForTask(tree.owner, context); + const idle = root.waitForIdle(context); + expect(await settled(idle)).toBe(false); + expect(await settled(harness.waitForIdle(context))).toBe(false); + open("owner.inner", "completed"); + await idle; + await harness.waitForIdle(context); + await harness.close(context); + }); + + it("stops idle traversal at a background owner", async () => { + const { harness, root } = await openHarness(); + const tree = await ownedChild(root, "background", { background: true }); + await root.waitForIdle(context); + await harness.waitForIdle(context); + // The background child is its own scope. + const child = (await harness.conversation(tree.child, context))!; + expect(await settled(child.waitForIdle(context))).toBe(false); + await child.abort(context); + await harness.close(context); + }); + + it("cascades an abort mark to the owned foreground subtree and withdraws its queued inputs", async () => { + const { harness, root } = await openHarness(); + const tree = await ownedChild(root, "owner"); + const nested = await harness.conversation(tree.child, context); + const deeper = await ownedChild(nested!, "deeper"); + const shielded = await ownedChild(nested!, "shielded", { background: true }); + // Busy conversations queue submissions; the inner task stands in for the child's run. + await busy(nested!, tree.inner); + const queued = (await nested!.submit({ type: "input", content: "later" }, context)).id; + expect(await harness.abortTask(tree.owner, context)).toBe("marked"); + for (const id of [tree.owner, tree.inner, deeper.owner, deeper.inner]) { + expect((await harness.waitForTask(id, context)).state.outcome.status).toBe("aborted"); + } + // A nested background owner is a boundary. + expect((await status(harness, shielded.owner)).status).not.toBe("terminal"); + expect((await status(harness, shielded.inner)).status).not.toBe("terminal"); + expect(await (await harness.submission(queued, context))!.status(context)).toMatchObject({ + status: "unanswered", + reason: "aborted", + }); + await harness.abortTask(shielded.owner, context); + await harness.close(context); + }); + + it("cascades a failed owner but not a completed one", async () => { + const { harness, root } = await openHarness(); + const failed = await ownedChild(root, "failed"); + const completedTree = await ownedChild(root, "done"); + open("failed", "failed"); + open("done", "completed"); + expect((await harness.waitForTask(failed.inner, context)).state.outcome.status).toBe("aborted"); + await harness.waitForTask(completedTree.owner, context); + expect((await status(harness, completedTree.inner)).status).not.toBe("terminal"); + open("done.inner", "completed"); + await harness.close(context); + }); + + it("aborts a background task directly with its ordinary subtree", async () => { + const { harness, root } = await openHarness(); + const tree = await ownedChild(root, "background", { background: true }); + await harness.abortTask(tree.owner, context); + expect((await harness.waitForTask(tree.inner, context)).state.outcome.status).toBe("aborted"); + await harness.close(context); + }); + + it("aborts a conversation: queued inputs withdrawn, writes kept, foreground work aborted, background kept", async () => { + const { harness, root } = await openHarness(); + const foreground = await ownedChild(root, "foreground"); + const background = await ownedChild(root, "background", { background: true }); + await busy(root, foreground.owner); + const input = (await root.submit({ type: "input", content: "later" }, context)).id; + const write = (await root.submit({ type: "write", entry: { kind: "note" } }, context)).id; + await root.abort(context); + for (const id of [foreground.owner, foreground.inner]) { + expect((await status(harness, id)).status).toBe("terminal"); + } + expect((await status(harness, background.owner)).status).not.toBe("terminal"); + expect((await status(harness, background.inner)).status).not.toBe("terminal"); + expect(await (await harness.submission(input, context))!.status(context)).toMatchObject({ reason: "aborted" }); + expect((await (await harness.submission(write, context))!.status(context)).status).toBe("queued"); + await harness.abortTask(background.owner, context); + await harness.close(context); + }); + + it("aborts work created below a cancelled owner after its cascade", async () => { + const { harness, root } = await openHarness(); + const tree = await ownedChild(root, "owner"); + open("owner", "failed"); + await harness.waitForTask(tree.inner, context); + const child = (await harness.conversation(tree.child, context))!; + const late = await child.commit((tx) => tx.createTask(Hold, { name: "late" }), context); + expect((await harness.waitForTask(late, context)).state.outcome.status).toBe("aborted"); + await harness.close(context); + }); + + it("withdraws queued inputs below the aborted task but keeps its own conversation's queue and queued writes", async () => { + const { harness, root } = await openHarness(); + const tree = await ownedChild(root, "owner"); + const child = (await harness.conversation(tree.child, context))!; + await busy(root, tree.owner); + await busy(child, tree.inner); + const own = (await root.submit({ type: "input", content: "own" }, context)).id; + const below = (await child.submit({ type: "input", content: "below" }, context)).id; + const write = (await child.submit({ type: "write", entry: { kind: "note" } }, context)).id; + await harness.abortTask(tree.owner, context); + await harness.waitForTask(tree.inner, context); + const statusOf = async (id: typeof own) => + (await (await harness.submission(id, context))!.status(context)).status; + expect(await statusOf(own)).toBe("queued"); + expect(await statusOf(below)).toBe("unanswered"); + expect(await statusOf(write)).toBe("queued"); + await harness.close(context); + }); + + it("keeps a nested background owner's subtree when its cancelled background owner is aborted", async () => { + const { harness, root } = await openHarness(); + const outer = await ownedChild(root, "outer", { background: true }); + const inner = await ownedChild((await harness.conversation(outer.child, context))!, "inner", { + background: true, + }); + await harness.abortTask(outer.owner, context); + expect((await harness.waitForTask(outer.inner, context)).state.outcome.status).toBe("aborted"); + expect((await status(harness, inner.owner)).status).not.toBe("terminal"); + expect((await status(harness, inner.inner)).status).not.toBe("terminal"); + await harness.abortTask(inner.owner, context); + await harness.close(context); + }); + + it("decides idle after reopen from owner edges it has to load first", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-ownership-")); + directories.add(directory); + const path = join(directory, "session.sqlite"); + let opened = await openHarness(await openNodeSqliteStorage(path)); + const foreground = await ownedChild(opened.root, "fg"); + const deeper = await ownedChild((await opened.harness.conversation(foreground.child, context))!, "fg2"); + const background = await ownedChild(opened.root, "bg", { background: true }); + open("fg", "completed"); + open("fg2", "completed"); + await opened.harness.waitForTask(deeper.owner, context); + await opened.harness.close(context); + + opened = await openHarness(await openNodeSqliteStorage(path)); + // Two levels below completed foreground owners, the inner tasks keep the root busy. + expect(await settled(opened.root.waitForIdle(context))).toBe(false); + expect(await settled(opened.harness.waitForIdle(context))).toBe(false); + open("fg.inner", "completed"); + open("fg2.inner", "completed"); + // The background subtree does not count. + await opened.root.waitForIdle(context); + await opened.harness.waitForIdle(context); + expect((await status(opened.harness, background.inner)).status).not.toBe("terminal"); + await opened.harness.abortTask(background.owner, context); + await opened.harness.close(context); + }); + + for (const [label, abortRequested, outcome] of [ + ["a failed owner", false, { status: "failed", error: { message: "crash" } }], + ["an abort-marked owner that completed", true, { status: "completed", result: null }], + ] as const) { + it(`derives marks a crash left unapplied below ${label} at open`, async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-ownership-")); + directories.add(directory); + const path = join(directory, "session.sqlite"); + // Without a Harness, nothing derives marks: a cancelled owner with a live task below it. + const session = createSession(await openNodeSqliteStorage(path)); + const tree = await session.commit(async (tx) => { + const root = await tx.createConversation({ ownership: { kind: "ownerless" } }); + const owner = await tx.createTask(Hold, { name: "gone" }, { conversationId: root.id }); + const child = await tx.createConversation({ ownership: { kind: "task", taskId: owner } }); + return { owner, inner: await tx.createTask(Hold, { name: "orphan" }, { conversationId: child.id }) }; + }, context); + await session.commit(async (tx) => { + const record = (await tx.task(tree.owner))!; + const state = { status: "terminal", outcome }; + (tx as unknown as { setTask(value: unknown): void }).setTask({ ...record, abortRequested, state }); + }, context); + await session.close(context); + + const { harness } = await openHarness(await openNodeSqliteStorage(path)); + expect((await harness.waitForTask(tree.inner, context)).state.outcome.status).toBe("aborted"); + await harness.close(context); + }); + } + + it("withdraws an input queued below a cancelled owner after its cascade", async () => { + const { harness, root } = await openHarness(); + const tree = await ownedChild(root, "owner"); + const child = (await harness.conversation(tree.child, context))!; + // The inner task stands in for the child's run, which stays busy after the cascade. + await busy(child, tree.inner); + open("owner", "failed"); + await harness.waitForTask(tree.inner, context); + const late = await child.submit({ type: "input", content: "late" }, context); + expect(await late.wait(context)).toMatchObject({ status: "unanswered", reason: "aborted" }); + await harness.close(context); + }); + + it("marks work admitted after reopen below a cancelled owner whose edge was not loaded", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-ownership-")); + directories.add(directory); + const path = join(directory, "session.sqlite"); + let opened = await openHarness(await openNodeSqliteStorage(path)); + const tree = await ownedChild(opened.root, "owner"); + open("owner.inner", "completed"); + await opened.harness.waitForTask(tree.inner, context); + open("owner", "failed"); + await opened.harness.waitForTask(tree.owner, context); + await opened.harness.close(context); + + // The child is empty at open, so nothing loads its edge until new work arrives. The new task waits on a + // dependency, so it is never reserved and publishes nothing more. + opened = await openHarness(await openNodeSqliteStorage(path)); + const gatekeeper = await opened.root.commit((tx) => tx.createTask(Hold, { name: "gatekeeper" }), context); + const child = (await opened.harness.conversation(tree.child, context))!; + const blocked = await child.commit( + (tx) => tx.createTask(Hold, { name: "blocked" }, { after: [gatekeeper] }), + context, + ); + expect((await opened.harness.waitForTask(blocked, context)).state.outcome.status).toBe("aborted"); + open("gatekeeper", "completed"); + await opened.harness.close(context); + }); + + it("retries a cascade whose commit the Storage rejected", async () => { + let rejectMark: TaskId | undefined; + class Rejecting extends MemoryStorage { + override async commit( + writes: Parameters[0], + ctx: Parameters[1], + ) { + const marks = writes.some( + (write) => write.type === "task" && write.value.id === rejectMark && write.value.abortRequested, + ); + if (marks) { + rejectMark = undefined; + throw new StorageRejected("rejected once"); + } + return super.commit(writes, ctx); + } + } + const { harness, root, reports } = await openHarness(new Rejecting()); + const tree = await ownedChild(root, "owner"); + rejectMark = tree.inner; + open("owner", "failed"); + await harness.waitForTask(tree.owner, context); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(reports.some((error) => error instanceof StorageRejected)).toBe(true); + expect((await status(harness, tree.inner)).status).not.toBe("terminal"); + // The next commit retries the cascade. + await root.commit((tx) => tx.appendEntry(root.id, { kind: "note" }), context); + expect((await harness.waitForTask(tree.inner, context)).state.outcome.status).toBe("aborted"); + await harness.close(context); + }); +}); + +describe("owned conversations from tools and supervisors", () => { + it("gives a tool an invocation-bound handle whose submissions stay durable after the call ends", async () => { + const setup = chatSetup(); + let handle: ConversationHandle | undefined; + let missing: ConversationHandle | undefined = {} as ConversationHandle; + let submission: Submission | undefined; + setup.registry.tools.add({ + name: "delegate", + description: "Delegates", + parameters: Type.Object({}), + execute: async (_args, api, callContext) => { + const child = await api.commit(async (tx) => { + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + (await tx.doc(ConversationConfig, created.id)).model = { provider: "faux", modelId: "faux-1" }; + return created.id; + }, callContext); + missing = await api.conversation(99_999 as ConversationId, callContext); + handle = (await api.conversation(child, callContext))!; + submission = await handle.submit({ type: "input", content: "child task", requestId: "child" }, callContext); + const settled = await submission.wait(callContext); + return { content: [{ type: "text", text: settled.status }] }; + }, + }); + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("delegate", {}, { id: "c1" })], { stopReason: "toolUse" }), + fauxAssistantMessage([fauxText("child answer")]), + fauxAssistantMessage([fauxText("done")]), + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + expect(missing).toBeUndefined(); + // The call ended: the handle and its submission reject, while the submission record remains. + await expect(handle!.submit({ type: "input", content: "again" }, context)).rejects.toThrow( + "invocation has ended", + ); + await expect(handle!.waitForIdle(context)).rejects.toThrow("invocation has ended"); + await expect(submission!.wait(context)).rejects.toThrow("invocation has ended"); + expect((await (await harness.submission(submission!.id, context))!.status(context)).status).toBe("done"); + await harness.close(context); + }); + + it("aborts a subagent run with its parent's conversation, rejecting the waiting tool", async () => { + const setup = chatSetup(); + let child: ConversationId | undefined; + const toolWait = deferred(); + setup.registry.tools.add({ + name: "delegate", + description: "Delegates", + parameters: Type.Object({}), + execute: async (_args, api, callContext) => { + child = await api.commit(async (tx) => { + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + (await tx.doc(ConversationConfig, created.id)).model = { provider: "faux", modelId: "faux-1" }; + return created.id; + }, callContext); + const submission = await (await api.conversation(child, callContext))!.submit( + { type: "input", content: "child task" }, + callContext, + ); + try { + return { content: [{ type: "text", text: (await submission.wait(callContext)).status }] }; + } catch (error) { + toolWait.resolve((error as Error).message); + throw error; + } + }, + }); + const childRun = gated(fauxAssistantMessage([fauxText("never")])); + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("delegate", {}, { id: "c1" })], { stopReason: "toolUse" }), + childRun.step, + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const input = await root.submit({ type: "input", content: "go" }, context); + await childRun.reached; + await root.abort(context); + expect(await input.wait(context)).toMatchObject({ status: "unanswered", reason: "aborted" }); + expect(await toolWait.promise).toBeTruthy(); + const childConversation = (await harness.conversation(child!, context))!; + await childConversation.waitForIdle(context); + const childLive = await harness.snapshot(LiveDoc, child!, context); + expect(childLive).toEqual({}); + const childInputs = (await harness.inspect(context)).submissions.filter((s) => s.conversationId === child); + expect(childInputs).toEqual([]); + await harness.waitForIdle(context); + await harness.close(context); + }); + + it("lets a running tool abort its owned child and continue", async () => { + const setup = chatSetup(); + const childRun = gated(fauxAssistantMessage([fauxText("never")])); + setup.registry.tools.add({ + name: "delegate", + description: "Delegates and cancels", + parameters: Type.Object({}), + execute: async (_args, api, callContext) => { + const child = await api.commit(async (tx) => { + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + (await tx.doc(ConversationConfig, created.id)).model = { provider: "faux", modelId: "faux-1" }; + return created.id; + }, callContext); + const handle = (await api.conversation(child, callContext))!; + const submission = await handle.submit({ type: "input", content: "child task" }, callContext); + await childRun.reached; + await handle.abort(callContext); + const settledChild = await submission.wait(callContext); + return { content: [{ type: "text", text: `child ${settledChild.status}` }] }; + }, + }); + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("delegate", {}, { id: "c1" })], { stopReason: "toolUse" }), + childRun.step, + fauxAssistantMessage([fauxText("done")]), + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + expect((await (await root.submit({ type: "input", content: "go" }, context)).wait(context)).status).toBe("done"); + await harness.close(context); + }); + + it("aborts the children of a tool call that throws or is interrupted, while the run continues", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-tool-children-")); + directories.add(directory); + const path = join(directory, "session.sqlite"); + const children: TaskId[] = []; + const toolRunning = deferred(); + const setup = chatSetup(); + setup.registry.tasks.add(Hold); + setup.registry.tools.add({ + name: "spawn", + description: "Starts owned work, then throws or hangs", + parameters: Type.Object({ mode: Type.String() }), + execute: async (args, api, callContext) => { + const child = await api.commit(async (tx) => { + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + return tx.createTask(Hold, { name: `child.${api.callId}` }, { conversationId: created.id }); + }, callContext); + children.push(child); + if ((args as { mode: string }).mode === "throw") throw new Error("spawn failed"); + toolRunning.resolve(); + return aborted(callContext.abortSignal!); + }, + }); + const call = (mode: string, id: string) => + fauxAssistantMessage([fauxToolCall("spawn", { mode }, { id })], { stopReason: "toolUse" }); + setup.faux.setResponses([ + call("throw", "c1"), + fauxAssistantMessage([fauxText("after throw")]), + call("hang", "c2"), + ]); + let opened = await openChat(await openNodeSqliteStorage(path), setup); + expect((await (await opened.root.submit({ type: "input", content: "one" }, context)).wait(context)).status).toBe( + "done", + ); + expect((await opened.harness.waitForTask(children[0]!, context)).state.outcome.status).toBe("aborted"); + + // The second call hangs until the process stops; on reopen it is interrupted. + const second = (await opened.root.submit({ type: "input", content: "two" }, context)).id; + await toolRunning.promise; + await opened.harness.close(context); + setup.faux.setResponses([fauxAssistantMessage([fauxText("after interrupt")])]); + opened = await openChat(await openNodeSqliteStorage(path), setup); + opened.harness.resume(); + expect((await opened.harness.waitForTask(children[1]!, context)).state.outcome.status).toBe("aborted"); + expect((await (await opened.harness.submission(second, context))!.wait(context)).status).toBe("done"); + const tools = (await opened.harness.commit((tx) => tx.scanTasks({ kind: "pi.tool" }, 10), context)).items; + expect(tools.map((task) => task.state.status === "terminal" && task.state.outcome.status)).toEqual([ + "failed", + "failed", + ]); + await opened.harness.close(context); + }); + + it("reruns a replay-safe subagent tool after a restart with the same child and submission", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-safe-subagent-")); + directories.add(directory); + const path = join(directory, "session.sqlite"); + const children: ConversationId[] = []; + const setup = chatSetup(); + setup.registry.tools.add({ + name: "subagent", + description: "Delegates", + parameters: Type.Object({}), + replay: "safe", + execute: async (_args, api, callContext) => { + const child = await api.commit(async (tx) => { + const existing = (await tx.scanConversations({ ownerTaskId: api.taskId }, 1)).items[0]; + if (existing !== undefined) return existing.id; + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + (await tx.doc(ConversationConfig, created.id)).model = { provider: "faux", modelId: "faux-1" }; + return created.id; + }, callContext); + children.push(child); + const request = { type: "input", content: "child task", requestId: `subagent:${api.taskId}` } as const; + const submission = await (await api.conversation(child, callContext))!.submit(request, callContext); + const settled = await submission.wait(callContext); + return { content: [{ type: "text", text: settled.status }] }; + }, + }); + const childRun = gated(fauxAssistantMessage([fauxText("never")])); + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("subagent", {}, { id: "c1" })], { stopReason: "toolUse" }), + childRun.step, + ]); + let opened = await openChat(await openNodeSqliteStorage(path), setup); + const input = (await opened.root.submit({ type: "input", content: "go" }, context)).id; + await childRun.reached; + await opened.harness.close(context); + + setup.faux.setResponses([ + fauxAssistantMessage([fauxText("child answer")]), + fauxAssistantMessage([fauxText("done")]), + ]); + opened = await openChat(await openNodeSqliteStorage(path), setup); + expect((await (await opened.harness.submission(input, context))!.wait(context)).status).toBe("done"); + expect(children).toHaveLength(2); + expect(children[1]).toBe(children[0]); + const child = (await opened.harness.conversation(children[0]!, context))!; + const entries = (await child.entries({}, 100, undefined, context)).items; + expect(entries.filter((entry) => entry.kind === "pi.user")).toHaveLength(1); + const results = (await opened.root.context(context)).messages.filter((message) => message.role === "toolResult"); + expect(results.map((message) => message.isError)).toEqual([false]); + await opened.harness.close(context); + }); + + it("lets a background supervisor resubmit after a restart without submitting twice", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-supervisor-")); + directories.add(directory); + const path = join(directory, "session.sqlite"); + const Children = defineDoc<{ child?: ConversationId }>({ + kind: "test.children", + version: 1, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({}), + }); + const Supervisor = defineTask<{ parent: ConversationId }, { phase: "run" }, { answer: EntryId }>({ + name: "test.supervisor", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (task, runtime, ctx) => { + const id = (await runtime.snapshot(Children, task.input.parent, ctx))!.child!; + const child = (await runtime.conversation(id, ctx))!; + const submission = await child.submit({ type: "input", content: "work", requestId: "stable" }, ctx); + const settled = await submission.wait(ctx); + if (settled.status !== "done" || settled.type !== "input") throw new Error(settled.status); + await runtime.commit( + () => ({ status: "terminal", outcome: { status: "completed", result: { answer: settled.answer } } }), + ctx, + ); + }, + }, + abort: (_task, runtime, ctx) => + runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx), + }); + const setup = chatSetup(); + setup.registry.tasks.add(Supervisor); + const first = gated(fauxAssistantMessage([fauxText("never")])); + setup.faux.setResponses([first.step, fauxAssistantMessage([fauxText("answer")])]); + let opened = await openChat(await openNodeSqliteStorage(path), setup); + const root = opened.root; + const { supervisor, child } = await root.commit(async (tx) => { + const supervisor = await tx.createTask(Supervisor, { parent: root.id }, { background: true }); + const created = await tx.createConversation({ ownership: { kind: "task", taskId: supervisor } }); + (await tx.doc(ConversationConfig, created.id)).model = { provider: "faux", modelId: "faux-1" }; + (await tx.doc(Children, root.id)).child = created.id; + return { supervisor, child: created.id }; + }, context); + opened.harness.resume(); + // The submission is admitted and its generation requested; then the process stops. + await first.reached; + await opened.harness.close(context); + + opened = await openChat(await openNodeSqliteStorage(path), setup); + opened.harness.resume(); + const done = await opened.harness.waitForTask(supervisor, context); + expect(done.state.outcome.status).toBe("completed"); + const childConversation = (await opened.harness.conversation(child, context))!; + const entries = (await childConversation.entries({}, 100, undefined, context)).items; + expect(entries.filter((entry) => entry.kind === "pi.user")).toHaveLength(1); + // The root never waited for the background supervisor. + await opened.root.waitForIdle(context); + await opened.harness.close(context); + }); +});