feat(durable): conversation abort, ownership cascades, and subagent handles (Package 18)

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
 

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.

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<Tool extends ToolRegistration> implements Conversation {
 		);
 	}
 
+	abort(context: Context): Promise<void> {
+		this.#host.tasks.resume();
+		return this.#host.tasks.abortConversation(this.id, context);
+	}
+
 	waitForIdle(context: Context): Promise<void> {
 		this.#host.tasks.resume();
 		return this.#host.tasks.waitForIdle(this.id, context);
@@ -265,6 +272,11 @@ class HarnessImpl<Tool extends ToolRegistration> 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<Tool extends ToolRegistration> 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 = <T>(operation: (context: Context) => Promise<T>) => {
+		return async (context: Context): Promise<T> => {
+			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<Tool extends ToolRegistration>(
 	snapshot: RegistrySnapshot<Tool>,

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<void> {
+	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);
+	}
+}

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<TaskOutcome<JsonValue>, { 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<void>;
+	/** Withdraw a conversation's queued inputs, for conversation abort and abort cascades. */
+	readonly withdrawInputs: (tx: Transaction, conversationId: ConversationId) => Promise<void>;
+	/** Invocation-bound handle of an existing conversation, for task runtimes and tools. */
+	readonly conversation: (
+		id: ConversationId,
+		binding: InvocationBinding,
+		context: Context,
+	) => Promise<ConversationHandle | undefined>;
 	/** 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<TaskId, LiveTaskRecord>();
 	readonly #invocations = new Map<TaskId, Invocation>();
@@ -131,8 +152,14 @@ export class TaskScheduler {
 	readonly #idleWaiters = new Waiters<ConversationId | undefined, void>();
 	/** Definition whose migration failed per task; retried only once the registry resolves another definition. */
 	readonly #failedMigrations = new Map<TaskId, { readonly task: AnyTask; readonly error: unknown }>();
-	/** Owner conversations of each conversation, nearest first; owner edges never change. */
-	readonly #owners = new Map<ConversationId, readonly ConversationId[]>();
+	/** Owner edge of each loaded conversation, `null` when ownerless. */
+	readonly #edges = new Map<ConversationId, OwnerEdge | null>();
+	/** Tasks that own a loaded conversation. */
+	readonly #owners = new Set<TaskId>();
+	/** Whether each terminal owner recorded cancellation intent (see `cancellationIntent()`). */
+	readonly #cancelledTerminal = new Map<TaskId, boolean>();
+	#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<void> {
 		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<void> {
+		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<JsonValue>);
 		}
+		// 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<void> {
+		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<ConversationId[]> {
+		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<void> {
+		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<string, unknown>)[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<boolean> {
+	/** Whether a scoped hook registration matches the invocation's conversation: itself, or an owner for `subtree`. */
+	async #hookMatches(invocation: Invocation, scope: HookScope): Promise<boolean> {
 		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<readonly ConversationId[]> {
-		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;

packages/durable/src/harness/tool.ts

@@ -57,11 +57,11 @@ export const ToolTask = defineTask<ToolTaskInput, ToolTaskCheckpoint, ToolTaskRe
 			const tool = runtime.registry.tool(call.name);
 			if (tool === undefined) {
 				const error = harnessError("tool_unavailable", `Tool ${call.name} is not available`);
-				return settle(runtime, call, "completed", () => 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<ToolTaskInput, ToolTaskCheckpoint, ToolTaskRe
 			});
 			if (block !== undefined) {
 				const blocked = harnessError("blocked", `Tool call blocked: ${block}`);
-				return settle(runtime, call, "completed", () => 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<ToolTaskInput, ToolTaskCheckpoint, ToolTaskRe
 				return run(runtime, call, tool, args, context);
 			}
 			const message = `Tool ${call.name} was interrupted and may have partially run`;
-			await settle(runtime, call, "completed", (slot) => 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<ToolSlot> | undefined) => ToolExecutionResult,
 	context: Context,
 ): Promise<void> {
@@ -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<ToolSlot> | undefined, code: string, message: string): ToolExecutionResult {
 	const diagnostics = [...(slot?.diagnostics ?? [])];

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> = K extends Task<infer _I, infer _S, infer _R, infer H> ? 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<Submission>;
+	/** `Conversation.abort()`: withdraw queued inputs, abort the ordinary ownership scope, and wait until it is idle. */
 	abort(context: Context): Promise<void>;
+	/** Resolve when the conversation's ordinary ownership scope has no live non-background task. */
 	waitForIdle(context: Context): Promise<void>;
 }
 
@@ -168,7 +173,8 @@ export interface ToolExecutionApi extends DocumentObserver, DocumentReader {
 	): Promise<TaskId<R>>;
 	getTask<R>(id: TaskId<R>, context: Context): Promise<TaskRecord<JsonValue, JsonValue, R> | undefined>;
 	waitForTask<R>(id: TaskId<R>, context: Context): Promise<SettledTask<R>>;
-	// 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<ConversationHandle | undefined>;
 }
 
 /** Executable tool registered in a registry. Only pi-ai `Tool` fields enter the transcript. */
@@ -448,7 +454,15 @@ export interface Conversation {
 		context: Context,
 	): Promise<Page<EntryRecord, Cursor>>;
 	fork(at: EntryId, options: ConversationCreateOptions, context: Context): Promise<Conversation>;
-	/** 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<void>;
+	/**
+	 * 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<void>;
 	/** The structural view (spec §9.3) as a disposable read-only Chord state. */
 	viewState(context: Context): Promise<AttachedReplicatedState<ConversationView>>;
@@ -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<R>(id: TaskId<R>, context: Context): Promise<SettledTask<R>>;
-	/** 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<void>;
 	/** Session total: every conversation's `pi.usage` summed. */
 	usage(context: Context): Promise<UsageState>;

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<I, S, R, H extends object> extends DocumentObserver
 	getTask<T>(id: TaskId<T>, context: Context): Promise<TaskRecord<JsonValue, JsonValue, T> | undefined>;
 	/** Resolve with the task's terminal receipt; rejects when the invocation ends. */
 	waitForTask<T>(id: TaskId<T>, context: Context): Promise<SettledTask<T>>;
+	/**
+	 * 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<ConversationHandle | undefined>;
 	/** Committed entry visible from the task's conversation. */
 	entry(id: EntryId, context: Context): Promise<EntryRecord | undefined>;
 	/** Undefined when the entry is absent, not visible, or has another kind. */

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<ConversationId>();
+const attach = async (id: ConversationId, indent: string): Promise<void> => {
+	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);

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<string, Subagent> }>({
+	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<string> => {
+	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<void> => {
+	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);

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<string, Deferred<Ending>>();
+
+function gate(name: string): Deferred<Ending> {
+	let found = gates.get(name);
+	if (found === undefined) {
+		found = deferred<Ending>();
+		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<string>();
+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<Tree> {
+	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<void> {
+	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<MemoryStorage["commit"]>[0],
+				ctx: Parameters<MemoryStorage["commit"]>[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<string>();
+		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);
+	});
+});