Durable: task, scheduler, generation và tool

/** Built-in task definitions every registry holds; they are not an extension and cannot be removed or replaced. */export const BUILTIN_TASKS: readonly AnyTask[] = [GenerationTask, ToolTask, CompactionTask];⋯ lược bớt dòng 11–17	constructor(extensions: readonly Extension<Tool>[]) {		this.#extensions = extensions;		this.#byName = new Map(extensions.map((extension) => [extension.name, extension]));		const tasks = new Map<string, AnyTask>();		for (const task of BUILTIN_TASKS) tasks.set(task.definition.name, task);		for (const extension of extensions) {			for (const task of extension.tasks ?? []) {				const name = task.definition.name;				if (tasks.has(name)) throw new Error(`Task ${name} of extension ${extension.name} is already installed`);				tasks.set(name, task);			}		}		this.#tasks = tasks;

### 5.2 Effect sandwich```textcommit intent phaseperform external effectcommit outcome or next phase```Reopening in an intent phase means the effect may have happened. The phasehandler retries safely, polls an external handle, or records interruption.Deferred providers are represented by a durable phase containing their handleand next poll time.

/** Durable reason and optional result recorded when a task becomes terminal. */export type TaskOutcome<R> =	| {			readonly status: "completed";			readonly result: R;			readonly error?: never;			readonly reason?: never;	  }	/** Expected task or domain failure explicitly committed by its implementation. */	| {			readonly status: "failed";			readonly error: TaskOutcomeError;			readonly result?: R;			readonly reason?: never;	  }	/** Explicit cancellation handled by the task's abort protocol. */	| {			readonly status: "aborted";			readonly reason?: string;			readonly result?: R;			readonly error?: never;	  }	/** Task that cannot resume because its definition or migration is unavailable. */	| {			readonly status: "orphaned";			readonly reason: string;			readonly result?: never;			readonly error?: never;	  }	/** Runtime-detected contract failure, such as an uncaught throw or no durable progress. */	| {			readonly status: "faulted";			readonly error: TaskOutcomeError;			readonly result?: never;			readonly reason?: never;	  };

/** Complete replacement record for one durable task state machine. */export type TaskRecord<I, S, R> = TaskRecordBase<I, R> &	(		| {				readonly state: Extract<					TaskState<S, R>,					{ readonly status: "pending" | "running" | "waiting" | "sleeping" }				>;				/** Small first-writer-wins values retained while the task can run. */				readonly memos?: Readonly<Record<string, JsonValue>>;		  }		| {				readonly state: Extract<TaskState<S, R>, { readonly status: "completing" | "terminal" }>;				readonly memos?: never;		  }	);

/** Define an executable task. Register it in the registry so a Harness can run tasks of its kind. */export function defineTask<I, S extends { phase: string }, R, H extends object = object>(	definition: TaskDefinition<I, S, R, H>,): Task<I, S, R, H> {	return { definition };}

/** Executable durable state machine definition, registered in the registry by `name`. */export type TaskDefinition<I, S extends { phase: string }, R, H extends object> = {	/** Registered task kind persisted in `TaskRecord.kind`. */	readonly name: string;	/** Definition version persisted with live input and checkpoints. */	readonly version: number;	/** First durable checkpoint for a newly created task. */	initial(input: I): S;	/** Exhaustive phase map; each handler receives the task narrowed to its phase. */	readonly phases: {		readonly [P in S["phase"]]: PhaseHandler<I, Extract<S, { phase: P }>, S, R, H>;	};	/** Runs in a fresh invocation after an abort mark and must commit a terminal outcome. */	abort(task: RunningTask<I, S, R>, runtime: TaskRuntime<I, S, R, H>, context: Context): Promise<void>;	/** Convert a record stored by any older supported version; runs at reservation. */	migrate?(		input: JsonValue,		checkpoint: JsonValue,		fromVersion: number,	): {		input: I;		checkpoint: S;	};	readonly hooks?: H;};/** Typed executable task definition. */export interface Task<I, S extends { phase: string }, R, H extends object> {	readonly definition: TaskDefinition<I, S, R, H>;}

/** Who owns a task: its conversation (a top-level task) or another task of the same conversation (a child task). */export type TaskOwnership = { readonly kind: "conversation" } | { readonly kind: "task"; readonly taskId: TaskId };/** How a waiting task treats the tasks it waits on (spec §5.5). */export type JoinPolicy = "failFast" | "allSettled";/** Creation options for a durable task. */export type TaskOptions = {	/** Omitted: inherit the creator's carrier. "root": start at the adapter root without a stored carrier. */	readonly trace?: TelemetryCarrier | "root";	/** Immutable creation-time links, attached to the spans of this task's first invocation. */	readonly links?: readonly TelemetryCarrier[];	/** Required: a task always names its owner (spec §5.5). */	readonly ownership: TaskOwnership;	/**	 * Default: the owner task's conversation, or the transaction's bound conversation; required for conversation-owned	 * tasks created by Session commits that are not bound to a conversation.	 */	readonly conversationId?: ConversationId;	/** Conversation-owned tasks only: excluded from ordinary idle waits, conversation aborts, and cascades. */	readonly background?: boolean;	/**	 * Its creator awaits it only in memory and does not resume after a restart, so it is pointless once the Harness that	 * created it is gone. When scheduling starts in a later Harness, the task, if still live, is abort-marked with reason	 * `restart`, and the mark cascades to its ordinary owned work, before any of it runs again.	 */	readonly abandonOnRestart?: boolean;};

const Ticker = defineTask<{ to: number }, { phase: "tick"; n: number }, string>({	name: "example.ticker",	version: 1,	initial: () => ({ phase: "tick", n: 1 }),	phases: {		tick: async (task, runtime, taskContext) => {			const n = task.state.checkpoint.n;			// Save the intent before the effect. A memo keeps the first value			// written under its name, so if the process dies after printing but			// before the next checkpoint is saved, the rerun sees the memo and			// does not print the same tick twice.			if ((await runtime.memo(`printed-${n}`, taskContext)) === undefined) {				await runtime.memo(`printed-${n}`, true, taskContext);				console.log(`tick ${n}`);			}			reachedTick(n);			// Save the outcome: the final result, or the next tick after a short			// sleep. The sleep is stored with the checkpoint, so the task holds no			// invocation meanwhile, and a Harness opened later continues it once			// its time has come.			if (n === task.input.to) {				await runtime.commit(					() => ({ status: "terminal", outcome: { status: "completed", result: `counted to ${n}` } }),					taskContext,				);				return;			}			await runtime.sleepUntil(runtime.now() + 50, { phase: "tick", n: n + 1 }, taskContext);		},	},	abort: async (_task, runtime, taskContext) => {		await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), taskContext);	},});

export interface TaskRuntime<I, S, R, H extends object> extends DocumentObserver, DocumentReader {	readonly taskId: TaskId<R>;	readonly conversationId: ConversationId;	/**	 * Aborted when the run is signalled by `abortTask()`, the Harness closes, or the invocation ends. Work still using it	 * after the invocation ended, such as a detached wait, is cancelled; it could not write anything anyway.	 */	readonly signal: AbortSignal;	/** Registry snapshot of the current phase; refreshed at every phase boundary. */	readonly registry: RegistrySnapshot;	/** The task's conversation's agent, resolved at most once per phase, at first use, and fixed for the phase. */	agent(context: Context): Promise<Agent>;	/** `HarnessOptions.settings`, resolved at each access. */	readonly settings: Settings;	/** `HarnessOptions.models`, resolved at each access. */	readonly models: Models;	/** Calls `HarnessOptions.env` for the task's conversation; rejects with its error. */	env(context: Context): Promise<ExecutionEnv | undefined>;	/** Handlers of this task's name from the extensions its conversation selects, in extension order. */	readonly hooks: HookRunner<H>;	/**	 * Commit on the Session line after rereading the task. Rejects when the task is terminal, the invocation ended, the	 * Harness is closing, or, in a run invocation, the task carries an abort mark. A returned state replaces the task's	 * state in the same commit; returning nothing leaves it unchanged. `tx.createTask()` defaults to the task's	 * conversation. Inside the callback read only through `tx`: `snapshot()`, `documentState()`, `watchDoc()`, `getTask()`, and the other Session and runtime reads may queue on the line the callback holds, and then never resolve.	 */	commit(		change: (			tx: Tx,			current: RunningTask<I, S, R>,		) => NextTaskState<S, R> | undefined | Promise<NextTaskState<S, R> | undefined>,		context: Context,	): Promise<void>;	/** Read a durable memo of this task. */	memo<T extends JsonValue>(name: string, context: Context): Promise<T | undefined>;	/** Store `candidate` unless a memo already exists; return the durable winner. */	memo<T extends JsonValue>(name: string, candidate: T, context: Context): Promise<T>;

/** Committed task record. */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>>;/** * Abort a task this task owns, as `Harness.abortTask()` does, and resolve once it is terminal. Rejects for a task * another owns; resolves at once for a terminal one. */abortOwned(id: TaskId, context: Context): Promise<void>;/** * Live tasks this task owns directly, in ID order: also those an earlier invocation created, before a restart. Tasks * in conversations it owns are not included. Copies of the committed records. */ownedTasks(context: Context): Promise<TaskRecord<JsonValue, JsonValue, JsonValue>[]>;/** Outcomes of terminal tasks, in order; rejects when one is missing or not terminal. Used after a wait. */outcomes<T>(ids: readonly TaskId<T>[], context: Context): Promise<TaskOutcome<T>[]>;⋯ lược bớt dòng 299–311/** * Call `callback` once after about `delayMs` on the Harness's timer (`HarnessOptions.setTimer`), for pacing work * inside a phase, such as throttled progress commits. Not durable: the invocation's end disarms it, and after its end * nothing is armed. Returns a function that disarms it. Wait for time between phases with `sleepUntil()`. */setTimer(callback: () => void, delayMs: number): () => void;⋯ lược bớt dòng 318–323/** * Commit a sleep: the task holds no invocation until the Harness clock reaches `until`, then resumes at `checkpoint`. * Resolves once the sleep is stored, not when the time comes; the phase returns right after. Returning * `{ status: "sleeping", until, checkpoint }` from `commit()` is the same sleep, for one that must commit together * with other writes. */sleepUntil(until: number, checkpoint: S, context: Context): Promise<void>;

#gated<T>(	invocation: Invocation,	change: (tx: Transaction, current: ErasedRunningTask) => T | Promise<T>,	context: Context,): Promise<T> {	if (invocation.ended) return Promise.reject(endedError(invocation));	return this.#session.commitWith(		async (tx) => {			if (invocation.ended) throw endedError(invocation);			if (this.#closing) throw closedError(this.#session);			const found = this.#live.get(invocation.taskId);			if (found === undefined) throw new Error(`Task ${invocation.taskId} is terminal`);			if (found.state.status !== "running") throw new Error(`Task ${invocation.taskId} is ${found.state.status}`);			const current = found as ErasedRunningTask;			if (invocation.mode === "run" && current.abortRequested) {				throw new Error(`Task ${invocation.taskId} has a durable abort mark`);			}			return change(tx, current);		},		context,		{ conversationId: invocation.conversationId, taskId: invocation.taskId },	);}

async #commitState(	tx: Transaction,	invocation: Invocation,	current: ErasedRunningTask,	next: NextTaskState<Checkpoint, JsonValue>,): Promise<void> {	if (next.status === "waiting") await this.#validateWait(tx, invocation, current, next.on, next.policy);	if (next.status === "sleeping") validateSleep(invocation, current.id, next.until);	if (next.status === "terminal") {		const overlay = overlayOf(tx);		await this.#loadScopes(false);		for (const record of overlay.tasks.values()) await this.#loadChain(parentOf(record), overlay);		if (this.#hasOwnedLive(current.id, overlay)) {			tx.setTask(withState(progressed(current), { status: "completing", outcome: next.outcome }));			return;		}	}	// A state the task committed that differs from its current one is progress: its restart count starts over. The	// same checkpoint committed again, for example as intent before an effect, is not.	const same = jsonEqual(current.state as unknown as JsonValue, next as unknown as JsonValue);	tx.setTask(withState(same ? current : progressed(current), next));}

memo: ((name: string, ...rest: readonly unknown[]) => {	if (rest.length === 1) {		return this.#read(invocation, async () => memoOf(this.#live.get(invocation.taskId), name));	}	const candidate = rest[0] as JsonValue;	return this.#gated(		invocation,		(tx, current) => {			const winner = memoOf(current, name);			if (winner !== undefined) return winner;			tx.setTask({ ...current, memos: { ...current.memos, [name]: candidate } } as AnyTaskRecord);			return candidate;		},		rest[1] as Context,	);}) as ErasedRuntime["memo"],sleepUntil: (until, checkpoint, context) =>	this.#gated(		invocation,		(tx, current) => this.#commitState(tx, invocation, current, { status: "sleeping", checkpoint, until }),		context,	),

const ownership = options.ownership;let owner: AnyTaskRecord | undefined;if (ownership.kind === "task") {	// Validated again against the owner's final candidate during assembly.	owner = await this.#currentTask(ownership.taskId);	this.#assertOpen();	if (owner === undefined) throw new Error(`Task owner ${ownership.taskId} does not exist`);	if (options.background === true) throw new TypeError("A child task cannot be background");	if (options.conversationId !== undefined && options.conversationId !== owner.conversationId) {		throw new Error(`A child task lives in its owner's conversation ${owner.conversationId}`);	}}const conversationId = owner?.conversationId ?? options.conversationId ?? this.#scope.conversationId;⋯ lược bớt dòng 421–424const checkpoint = definition.initial(input);const id = await this.#host.storage.mintId<TaskId<R>>();this.#assertOpen();const record = copyJson(	{		id,		conversationId,		kind: definition.name,		version: definition.version,		input,		trace:			options.trace === "root"				? undefined				: (options.trace ?? this.#context.telemetry.propagation?.inject(this.#context.telemetry)),		links: options.links,		...(owner === undefined ? {} : { owner: owner.id }),		background: options.background ?? false,		abortRequested: false,		...(options.abandonOnRestart === true ? { abandonOnRestart: true } : {}),		state: { status: "pending", checkpoint },	},	TABLE_JSON_COPY_OPTIONS,) as unknown as AnyTaskRecord;

async #validateWait(	tx: Transaction,	invocation: Invocation,	current: ErasedRunningTask,	on: readonly TaskId[],	policy: string,): Promise<void> {	if (invocation.mode === "abort") throw new Error(`Abort handler of task ${current.id} cannot wait`);	const overlay = overlayOf(tx);	const targets: TaskId[] = [];	for (const id of on) {		const member = overlay.tasks.get(id) ?? this.#live.get(id) ?? (await this.#storage.task(id, this.#context));		if (member === undefined) throw new Error(`Task ${id} does not exist`);		if (policy === "failFast" && member.owner !== current.id) {			throw new Error(`Task ${current.id} can wait failFast only on tasks it owns; ${id} is not one`);		}		// A live child already holds the waiter (task-owned tasks are never background): waiting on it adds no edge. A		// terminal task has no edges.		if (member.state.status !== "terminal" && member.owner !== current.id) targets.push(id);	}	if (targets.length === 0) return;	const cycle = await this.#waitCycle(current.id, targets, overlay);	if (cycle !== undefined) {		throw new Error(			`Task ${current.id} cannot wait on ${cycle[0]}: ${[current.id, ...cycle].join(" -> ")} is a wait cycle`,		);	}}

let round = [...this.#finalizeChecks];this.#finalizeChecks.clear();const done = new Set<TaskId>();const overlay = overlayOf(tx);while (round.length > 0) {	const next = new Set<TaskId>();	for (const id of round) {		const record = overlay.tasks.get(id) ?? this.#live.get(id);		if (record?.state.status !== "completing" || done.has(id)) continue;		if (this.#hasOwnedLive(id, overlay, done)) continue;		const outcome = record.state.outcome;		const terminal = withState(record, { status: "terminal", outcome });		tx.setTask(terminal);		(overlay.tasks as Map<TaskId, AnyTaskRecord>).set(id, terminal);		done.add(id);		// REMINDER: only the scheduler writes `faulted` and `orphaned` (spec §5.4); their cleanup waits for this commit.		if (outcome.status === "faulted" || outcome.status === "orphaned") {			await this.#settleOutcome(tx, record, outcome);		}		const above = this.#nearestLiveAbove(record, overlay);		if (above?.state.status === "completing") next.add(above.id);	}	round = [...next];}

await this.#session.commitWith(async (tx) => {	if (this.#closing) return;	const queued = await this.#loadScopes(cascade);	// A cascade from an abandoned owner passes its `restart` reason on; any other intent marks, or upgrades a	// `restart` mark, as a request. Collected first, so a request wins over a `restart` mark of the same pass.	const marks = new Map<TaskId, { readonly record: AnyTaskRecord; readonly reason: "restart" | undefined }>();	const mark = (record: AnyTaskRecord, reason: "restart" | undefined): void => {		const staged = marks.get(record.id);		if (staged !== undefined && staged.reason === undefined) return;		if (record.abortRequested && (record.abortReason === undefined || reason !== undefined)) return;		marks.set(record.id, { record, reason });	};	// Every live foreground task below a live owner with cancellation intent, which is its nearest one: a walk down	// from each owner with intent does not enter another, whose own walk covers what is below it. Derived again on	// every pass, as loading edges can reveal a cancelled owner.	for (const id of this.#intent) {		const owner = this.#live.get(id)!;		// Only an owner abandoned after a restart, and nothing else, passes its reason on.		const restart = owner.abortReason === "restart" && !failedOutcome(owner);		for (const { node, live } of this.#walkDown(id, { enter: (below) => !this.#intent.has(below) })) {			if (live !== undefined && !node.background) mark(live, restart ? "restart" : undefined);		}	}	for (const id of checks) {		const waiter = this.#live.get(id);		if (waiter?.state.status !== "waiting" || !(await this.#anyFailed(waiter.state.on))) continue;		// Every other live task: the failed one keeps its own outcome.		for (const member of waiter.state.on) {			const record = this.#live.get(member);			if (record !== undefined && !failedOutcome(record)) mark(record, undefined);		}	}	for (const { record, reason } of marks.values()) tx.setTask(withAbortMark(record, reason));	for (const id of queued) if (this.#belowCancelled({ conversation: id })) await this.#withdrawInputs(tx, id);	await this.#finalize(tx);

/** * Live tasks a task waits for before its next invocation: its live ordinary owned work when abort-marked, since * abort runs bottom-up, otherwise the live part of the `on` of a wait. */#waitingOn(record: AnyTaskRecord, first = false): readonly TaskId[] {	if (record.abortRequested) return this.#ownedLive(record.id, { first });	if (record.state.status !== "waiting") return [];	if (!first) return record.state.on.filter((id) => this.#live.has(id));	const live = record.state.on.find((id) => this.#live.has(id));	return live === undefined ? [] : [live];}

/** * Durable task scheduler of one Harness. * * `#live` mirrors every committed non-terminal task record: pending, running, waiting, sleeping, and completing. The * synchronous commit listener updates it on the Session line, so code running on the line reads exactly the committed * state from it. * * Tasks and conversations form one ownership tree (spec §5.5): a task's parent is its owner task, or its conversation; * a conversation's parent is its owner task, if any. Walks up that tree decide what reaches one task: a cancelling * owner, a scope. `#below` lists the tree downward around live work, so what lies below one node is a walk down from * it: a task's live ordinary owned work, which holds its outcome as `completing` and delays its abort handler, a * scope's tasks, a cascade's reach. No pass visits every live task; the sets below hold each pass's candidates. * * The ownership indexes are bounded by live work: an owner edge and an ended task's ownership fields are kept only while * something is listed below them in `#below`, so only the chains above live tasks are known, whole. Work that appears * later below a dropped node finds it unknown and loads its chain from storage again. * * Invariant: every task transition is decided and written by one callback serialized on the Session line. That covers * reservation, marks, runtime commits, finalization, and the synchronous step before each phase, which applies the * precedence rules and writes a fault or handover. Handlers and joins run off the line. An invocation ends inside the * step that decides its end, so a runtime commit it queued either lands before that decision or is rejected. */

async open(context: Context): Promise<void> {	this.#session.observeCommits((publication) => this.#observe(publication));	this.#session.subscribeClose(() => this.#seal());	this.#unsubscribeRegistry = this.#registry.subscribe(() => {		this.#notifyBlocked();		this.#kick();	});	const limit = this.#restartLimit();	const parked: AnyTaskRecord[] = [];	await this.#session.commitWith(async (tx) => {		// Every table read before the first write.		const scans = [];		for (const status of LIVE_STATUSES) {			scans.push(await scanAll((cursor) => tx.scanTasks({ status }, SCAN_PAGE_SIZE, cursor)));		}		for (const records of scans) {			for (const record of records) {				this.#track(undefined, record);				this.#resumed.add(record.id);				if (record.abandonOnRestart === true) this.#abandoned.add(record.id);				if (record.state.status === "running") {					const restarts = (record.restarts ?? 0) + 1;					const pending = withState(record, { status: "pending", checkpoint: record.state.checkpoint });					tx.setTask({ ...pending, restarts } as AnyTaskRecord);					if (restarts >= limit && !record.abortRequested) parked.push(record);				}				if (record.state.status === "waiting" && record.state.policy === "failFast") {					this.#failFastChecks.add(record.id);				}			}		}	}, context);⋯ lược bớt dòng 406–412	// Derive abort marks a crash left unapplied below cancelled owners, and finalize held outcomes.	this.#cascadePending = true;	this.#scheduleReconcile();	this.#scheduleWake();

#kick(): void {	this.#scheduleWake();	this.#dirty = true;	if (this.#draining || !this.#enabled || this.#closing) return;	this.#draining = true;	// Never commit synchronously from a commit or registry listener.	queueMicrotask(() => void this.#drain());}async #drain(): Promise<void> {	try {		while (this.#dirty && this.#enabled && !this.#closing) {			this.#dirty = false;			for (const reservation of await this.#reserve()) this.#start(reservation);		}	} catch (error) {		// As in `#reconcile()`: the reservation commit runs no extension code.		this.#failSession(error);	} finally {		this.#draining = false;		// A wakeup that arrived after the loop's last check still needs its pass.		if (this.#dirty) this.#kick();	}}

async #run(reservation: Reservation): Promise<void> {	const invocation = reservation.invocation;	const state = { task: reservation.task, snapshot: reservation.snapshot, reported: undefined as ReportedTask };	const phase: Phase = { snapshot: () => state.snapshot, task: () => state.task, agent: undefined };	const runtime = this.#runtime(invocation, phase);	let current = await this.#step(invocation, (tx, next) =>		this.#decide(tx, next, undefined, state, invocation.extension, invocation.context),	);	while (current !== undefined && !this.#closing) {		const task = current;		current = await this.#phase(invocation, task, async () => {			const checkpoint = task.state.checkpoint;			// Each phase handler resolves its agent afresh, at first use.			phase.agent = undefined;			let previous: PhaseResult;			try {				await erased(state.task).phases[checkpoint.phase]!(task, runtime, invocation.context);				previous = { checkpoint };			} catch (error) {				previous = { checkpoint, failure: { error } };			}			return this.#step(invocation, (tx, next) =>				this.#decide(tx, next, previous, state, invocation.extension, invocation.context),			);		});	}}

): Decision {	// 3. abort mark: end; a fresh abort invocation starts once the task's ordinary owned work is gone.	if (current.abortRequested) return false;	// An owner's cancellation intent ends it too, before its cascade marks it; reservation then holds it back.	if (!current.background && this.#belowCancelled(parentOf(current))) {		this.#cascadePending = true;		this.#scheduleReconcile();		return false;	}	if (previous === undefined) return true;	// 4. uncaught error.	if (previous.failure !== undefined) return { fault: previous.failure.error };	// 6. no durable progress.	if (jsonEqual(current.state.checkpoint, previous.checkpoint)) {		const message = `Task ${current.kind} phase ${previous.checkpoint.phase} returned without durable progress`;		return { fault: new Error(message) };	}	// 5. progress: refresh the snapshot; hand over to a replacement definition that can take the task.	state.snapshot = this.#registry.snapshot();	const next = state.snapshot.task(current.kind);	if (next !== state.task) {		if (next !== undefined && canReserve(next, current)) {			tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));			return false;		}		if (state.reported === undefined || state.reported.task !== next) {			state.reported = { task: next };			this.#report(				new Error(`Task ${current.id} keeps running under its old ${current.kind} definition`),				{					type: "task_outdated",					...this.#aboutTask({ ...recordTask(current), extension }),					reason: next === undefined ? "missing_task" : "incompatible_task",				},				context,			);		}	}	return true;}

async #step(	invocation: Invocation,	decide: (tx: Transaction, current: ErasedRunningTask) => Decision,): Promise<ErasedRunningTask | undefined> {	let fault: { readonly error: unknown } | undefined;	try {		const continued = await this.#session.commitWith(async (tx) => {			const found = this.#live.get(invocation.taskId);			const current = found?.state.status === "running" ? (found as ErasedRunningTask) : undefined;			const decision = current !== undefined && !this.#closing ? decide(tx, current) : false;			if (decision === true) return current;			this.#end(invocation);			if (decision !== false) {				fault = { error: decision.fault };				const message = errorMessage(decision.fault);				invocation.fault = { status: "faulted", error: { message } };				await this.#terminate(tx, current!, invocation.fault);			}			return undefined;		}, withoutAbortSignal(invocation.context));

async join(): Promise<void> {	const stopped = [...this.#invocations.values()];	for (const invocation of stopped) invocation.controller.abort();	await Promise.allSettled(stopped.map((invocation) => invocation.done));	if (stopped.length === 0) return;	// A close is not a crash: a task it stopped is `pending` again, so the next open counts no restart for it. Read	// on the line after commits admitted before the seal, such as an abort mark, so the write keeps them. A failure	// leaves them `running`, counted as if cut off; the Storage guard fails the Session and reports it.	await this.#writeAtClose(() =>		stopped.flatMap((invocation) => {			const record = this.#live.get(invocation.taskId);			if (record?.state.status !== "running") return [];			const value = withState(record, { status: "pending", checkpoint: record.state.checkpoint });			return [{ type: "task", value } as const];		}),	).catch(() => {});}

wake(): Wake {	const now = this.#now();	// A reconcile pass with work to do is work too: it finalizes held outcomes and marks cascades, at open as	// recovery. A process that dies before it lands must be woken again. A pass with nothing to do is not.	let active = this.#invocations.size > 0 || this.#reconcilingWork > 0 || this.#reconcileWork();	let at: number | undefined;	let snapshot: RegistrySnapshot | undefined;	let snapshotFailed = false;	for (const record of this.#live.values()) {		const state = record.state;		if (state.status === "completing") continue;		if (state.status === "waiting" && !record.abortRequested && state.on.some((id) => this.#live.has(id)))			continue;		if (state.status === "sleeping" && !record.abortRequested && state.until > now) {			at = at === undefined ? state.until : Math.min(at, state.until);			continue;		}		if (active) continue;		if (this.#parked(record)) continue;		// An abort invocation waits for the live work below it, which counts itself.		if (record.abortRequested && this.#waitingOn(record, true).length > 0) continue;		// A task no definition can take waits for a registry change, except an abort-marked one, which is orphaned.		if (!record.abortRequested || record.abortReason !== undefined) {			if (snapshot === undefined && !snapshotFailed) {				try {					snapshot = this.#registry.snapshot();				} catch (error) {					snapshotFailed = true;					this.#report(error, REGISTRY);				}			}			if (snapshot !== undefined && "reason" in this.#fit(record, snapshot.task(record.kind))) continue;		}		active = true;	}	return at === undefined ? { active } : { active, at };}

async retry(id: TaskId, context: Context): Promise<void> {	await this.#session.commitWith(async (tx) => {		const current = await tx.task(id);		if (current === undefined) throw new Error(`Task ${id} does not exist`);		const status = current.state.status;		if (status === "terminal" || status === "completing" || !this.#parked(current)) return;		tx.setTask(progressed(current));	}, context);	this.resume();}

/** Resolve the record's definition by kind, migrating an older stored version. */#resolve(record: RunnableTaskRecord, snapshot: RegistrySnapshot): Resolution {	const fit = this.#fit(record, snapshot.task(record.kind));	if ("reason" in fit) return { kind: "blocked", reason: fit.reason };	if (!fit.migrates) return { kind: "ready", task: fit.task, record };	const definition = erased(fit.task);	try {		if (definition.migrate === undefined) throw missingMigration(record, definition);		const migrated = definition.migrate(record.input, record.state.checkpoint, record.version);		const state = { ...record.state, checkpoint: copyJson(migrated.checkpoint) };		const migratedRecord = { ...record, version: definition.version, input: copyJson(migrated.input), state };		return { kind: "ready", task: fit.task, record: migratedRecord as RunnableTaskRecord };	} catch (error) {		this.#failedMigrations.set(record.id, { task: fit.task, error });		// Blocked now without a commit: the graph and the host's wake must hear it.		queueMicrotask(() => this.#notifyBlocked());		this.#scheduleWake();		const extension = installerOf(snapshot, fit.task);		this.#report(error, { type: "migration_failed", ...this.#aboutTask({ ...recordTask(record), extension }) });		return { kind: "blocked", reason: "migration_failed" };	}}#fit(record: Pick<AnyTaskRecord, "id" | "version">, task: AnyTask | undefined): Fit {	if (task === undefined) return { reason: "missing_task" };	const version = task.definition.version;	if (version === record.version) return { task, migrates: false };	if (version < record.version) return { reason: "task_too_old" };	const failed = this.#failedMigrations.get(record.id);	if (failed?.task === task) return { reason: "migration_failed", error: failed.error };	return { task, migrates: true };}

export const ToolTask = defineTask<ToolTaskInput, ToolTaskCheckpoint, ToolTaskResult, ToolHooks>({	name: "pi.tool",	version: 2,	initial: () => ({ phase: "call" }),	// Version 1 had only model-issued calls.	migrate: (input, checkpoint) => ({		input: { kind: "model", ...(input as { assistant: EntryId; callId: string }) },		checkpoint: checkpoint as ToolTaskCheckpoint,	}),	phases: {

abort: async (task, runtime, context) => {	const call = await readCall(runtime, task.input, context);	// Abandoned after a restart (`TaskOptions.abandonOnRestart`): say whether the call may have run.	const [code, message] =		task.abortReason !== "restart"			? ["aborted", `Tool ${call.name} was aborted`]			: task.state.checkpoint.phase === "execute"				? ["interrupted", `Tool ${call.name} was interrupted by a restart and may have partially run`]				: ["abandoned", `Tool ${call.name} was not started: its caller ended with a restart`];	const ending = { status: "aborted" } as const;	await settle(runtime, task.input, call, ending, (slot) => fromSlot(slot, code, message), context);},

/** Built-in live conversation state: run control and presentation of the current generation and tool round. */export type LiveState = {	/** Run control: the task that settles the run's inputs, and those inputs; present exactly while busy. */	run?: { taskId: TaskId; inputs: SubmissionId[] };	/** Presentation of the current generation attempt. */	generation?: {		attempt: number;		/** Committed throttled partial of the in-flight response. */		message?: JsonRepresentation<AssistantMessage>;		/** Durable backoff before the next attempt. */		retry?: { at: number; error: string };		/** Provider-side deferred response being polled. */		deferred?: { pollAt: number };	};	/** The current tool round in call order, from the tool-calling answer until the generation's `tools` phase ends it. */	tools?: ToolSlot[];	/**	 * Nested calls of running tool calls, in creation order, so a parent precedes its children and task IDs ascend. A	 * call's nested slots are removed when the call settles.	 */	nestedTools?: NestedToolSlot[];	/** Live compaction tasks in task ID order; absent when none. */	compactions?: CompactionStatus[];	/**	 * The newest compaction failure, so a client can show it after the task ended. A compaction that produces a summary	 * removes it; one that is aborted or finds nothing to compact leaves it.	 */	compactionFailure?: CompactionFailure;};

const entries = planSystemEntries(view, desired, agent.tools, runtime.now());const threshold =	compacted === undefined		? thresholdCompaction(view, entries, resolved.contextWindow, compactionPolicy(settings, model))		: undefined;if (threshold === "blocking") {	// Compact first and prepare again; the transcript is unchanged until the compaction appends.	await runtime.commit(async (tx): Promise<Next> => {		const child = await createCompaction(tx, conversationId, { reason: "threshold" }, runtime.taskId);		const checkpoint = { phase: "prepare", attempt, compacted: child, threshold: true } as const;		return { status: "waiting", checkpoint, on: [child], policy: "allSettled" };	}, context);	return;}await runtime.commit(async (tx) => {	let cutoff = (await tx.scanEntries({ conversationId }, 1)).items[0]?.id;	for (const entry of entries) cutoff = (await tx.appendEntry(SystemEntry, conversationId, entry)).id;	if (cutoff === undefined) throw new Error(`Conversation ${conversationId} has no entries to send`);	// Checked in this commit, so a compaction admitted during preparation counts.	if (threshold === "background" && (await tx.doc(LiveDoc, conversationId)).compactions === undefined) {		await createCompaction(tx, conversationId, { reason: "threshold" });	}	const request = {		attempt,		...(compacted === undefined ? {} : { compacted }),		model,		thinkingLevel,		streamOptions: settings.stream,		cutoff,	};	return { status: "running", checkpoint: { phase: "request", ...request } };}, context);

request: async (task, runtime, context) => {	const { attempt, compacted, model: ref, thinkingLevel, streamOptions, cutoff } = task.state.checkpoint;	const conversationId = runtime.conversationId;	await runtime.commit(async (tx) => {		const live = await tx.doc(LiveDoc, conversationId);		await convertPartial(tx, live, conversationId);		live.generation = { attempt };		return undefined;	}, context);	const model = runtime.models.getModel(ref.provider, ref.modelId);	if (model === undefined) return failNoModel(runtime, ref, context);	const view = await runtime.context(conversationId, context, { at: cutoff });	let messages = view.messages;	await runtime.hooks.each("beforeRequest", async (hook) => {		const replaced = await hook({ messages }, runtime, context);		if (replaced !== undefined) messages = replaced.messages;	});	const options: SimpleStreamOptions = {		...streamOptions,		signal: runtime.signal,		telemetry: context.telemetry,		sessionId: await ensureProviderSessionId(runtime, context),		...(thinkingLevel === "off" ? {} : { reasoning: thinkingLevel }),	};	const message = await streamResponse(runtime, model, messages, options, attempt, context);	const request = { attempt, compacted, model: ref, cutoff, messages: view.messages };	await classify(runtime, request, message, context);},

const flush = (): void => {	timer = undefined;	const partial = pending;	pending = undefined;	if (partial === undefined || stopped) return;	inFlight = (async () => {		// Copy synchronously: the provider keeps mutating its partial.		const message = copyJson(partial, { omitUndefinedProperties: true });		await runtime.commit(async (tx) => {			const live = await tx.doc(LiveDoc, runtime.conversationId);			live.generation ??= { attempt };			assignJson(live.generation as Draft<Record<string, JsonValue>>, "message", message);			return undefined;		}, context);	})()		.catch((error: unknown) => {			// Rejections after an abort mark or close are expected; the committed state stays consistent.			if (!runtime.signal.aborted) runtime.report(error);		})		.finally(() => {			inFlight = undefined;			if (pending !== undefined && !stopped) timer = runtime.setTimer(flush, interval);		});};try {	const events = runtime.models.streamSimple(model, { messages: [...messages] }, options);	for await (const event of events) {		// A partial without content, such as pi-ai's opening `start` event, shows nothing; a deferred response		// never gets past it, so it never leaves a partial.		if (event.type === "done" || event.type === "error" || event.partial.content.length === 0) continue;		pending = event.partial;		if (timer === undefined && inFlight === undefined) timer = runtime.setTimer(flush, interval);	}	return await events.result();} finally {	stopped = true;	timer?.();	await inFlight;}

async function answer(runtime: Runtime, message: AssistantMessage, context: Context): Promise<void> {	let continuation: UserInput | undefined;	await runtime.hooks.each("onYield", async (hook) => {		if (continuation !== undefined) return;		continuation = (await hook(message, runtime, context))?.continue;	});	const conversationId = runtime.conversationId;	await runtime.commit(async (tx): Promise<Next> => {		// Queue modes are read on the Session line, when the boundary is decided.		const boundary = await prepareBoundary(tx, conversationId, runtime.settings);		const live = await tx.doc(LiveDoc, conversationId);		const entry = await appendAssistant(tx, conversationId, message);		const result: Next = { status: "terminal", outcome: { status: "completed", result: { entryId: entry.id } } };		const { users, reset } = await applyBoundary(tx, boundary, "final", runtime.now());		if (continuation !== undefined && users.length === 0 && !reset) {			const user = { role: "user", content: continuation, timestamp: runtime.now() } as const;			await tx.appendEntry(UserEntry, conversationId, { model: [user] });			handOver(live, runtime.taskId, await createGeneration(tx, conversationId));			delete live.generation;			return result;		}		endRun(tx, live, runtime.taskId, { status: "done", answer: entry.id });		if (users.length > 0) await startRun(tx, conversationId, live, users);		return result;	}, context);}