Pi Durable: mô hình khái niệm

/** Session kernel extended with conversation handles and a registry. */class HarnessImpl<Tool extends ToolRegistration> extends SessionImpl implements HarnessType {	readonly #options: HarnessOptions<Tool>;	readonly #hostSettings: () => HarnessSettings | undefined;	readonly #report: (error: unknown, info: ReportInfo, context?: Context) => void;	readonly #host: ConversationHost<Tool>;	readonly #tasks: TaskScheduler;	readonly #submissions: Submissions;	readonly #taskGraph: TaskGraphView;	constructor(rawStorage: Storage, options: HarnessOptions<Tool>, context: Context) {		const report = safeReport(options.onReport, options.telemetry ?? NOOP_TELEMETRY_CONTEXT);		const now = safeNow(options.now, report);		const hostSettings = currentSettings(options.settings);		// Read at each use, as all settings are, but only this field: it is read on every commit.		super(rawStorage, now, () => hostSettings()?.documentRetentionMs, options.setTimer);		// Every component reads and writes through the Session's failure guard: `rawStorage` serves only `super()`.		const storage = this.storage;

declare const idBrand: unique symbol;/** Erased nominal number identifying one durable record kind. */export type Id<Kind extends string, Type = unknown> = number & {	readonly [idBrand]: {		readonly kind: Kind;		readonly type: Type;	};};export type ConversationId = Id<"conversation">;export type EntryId = Id<"entry">;export type TaskId<Result = unknown> = Id<"task", Result>;export type SubmissionId = Id<"submission">;export type DocumentId = Id<"document">;declare const seqBrand: unique symbol;/** Strictly increasing sequence assigned to one atomic storage commit; gaps are permitted. */export type Seq = number & { readonly [seqBrand]: "sequence" };/** The root conversation always uses this reserved ID. */export const ROOT_CONVERSATION_ID = 1 as ConversationId;

/** Ownership selected explicitly whenever a conversation is created. */export type ConversationOwnership = { readonly kind: "ownerless" } | { readonly kind: "task"; readonly taskId: TaskId };/** Immutable identity, history ancestry, and task ownership of a transcript scope. */export type ConversationRecord = {	readonly id: ConversationId;	/** Fork source and inclusive parent entry through which history is inherited. */	readonly parent?: {		readonly conversationId: ConversationId;		readonly at: EntryId;	};	/** Creator edge used for attribution, subtree abort, and subtree idle waits. */	readonly owner?: {		readonly conversationId: ConversationId;		readonly taskId: TaskId;	};};

/** Immutable transcript event with separate model-facing and application-facing payloads. */export type EntryRecord = {	readonly id: EntryId;	readonly conversationId: ConversationId;	/** Application-defined entry discriminator. */	readonly kind: string;	/** Messages contributed to model context; absent for display or bookkeeping entries. */	readonly model?: readonly Message[];	/** JSON payload consumed by views, extensions, or bookkeeping logic. */	readonly data?: JsonValue;	/** First entry in the active context selected by this entry. */	readonly head?: EntryId;	/** Context-only overrides of earlier visible entries. */	readonly edits?: readonly ContextEdit[];	/** Task that appended this entry, when it was produced by durable work. */	readonly byTaskId?: TaskId;};/** Entry content supplied before the Session assigns identity and task attribution. */export type EntryDraft = Omit<EntryRecord, "id" | "conversationId" | "byTaskId" | "head"> & {	/** `"self"` starts active context at the newly assigned entry ID. */	readonly head?: EntryId | "self";};

/** Define a typed entry kind whose `is()` guard narrows by `EntryRecord.kind`. */export function defineEntry<D extends JsonValue = never>(kind: string): Entry<D> {	if (typeof kind !== "string" || kind.length === 0) throw new TypeError("Entry kind must be a non-empty string");	return {		kind,		is: (entry: EntryRecord | undefined): entry is TypedEntry<D> => entry !== undefined && entry.kind === kind,	};}/** User input: `model` is `[UserMessage]`. Written by submissions. */export const UserEntry = defineEntry("pi.user");/** Provider result with any stop reason: `model` is `[AssistantMessage]`. Written by generation. */export const AssistantEntry = defineEntry("pi.assistant");/** Positional prompt and tool change: `model` is `[SystemMessage]` with empty `content`. */export const SystemEntry = defineEntry("pi.system");/** * Tool result: `model` is `[ToolResultMessage]`, whose content ends with the rendered diagnostics block; `data` holds * the structured diagnostics, possibly none. Written by tool tasks, and by generation for calls it did not offer. */export const ToolResultEntry = defineEntry<{ diagnostics: ToolDiagnostic[] }>("pi.tool-result");/** * Start of a new context: always `head: "self"`, with `model` absent for a plain reset or `[UserMessage]` carrying the * handoff text. Written by `Conversation.reset()` and the `handoff` tool control. */export const ResetEntry = defineEntry("pi.reset");/** * Compaction summary: `model` is `[UserMessage]` with the wrapped summary, `head` the first kept entry. Written by * compaction tasks, directly or through a write submission. */export const CompactionEntry = defineEntry<{ reason: CompactionReason }>("pi.compaction");

/** Derive the context view from the entries scanned within `bounds`. */function deriveRange(bounds: ContextBounds, entries: readonly EntryRecord[]): ContextRange {	const edits = new Map<EntryId, ContextEdit>();	// Edits of every entry in the range count, including older head markers that `selectActive()` drops.	for (const entry of entries) for (const edit of entry.edits ?? []) edits.set(edit.target, edit);	const active = selectActive(bounds.head, entries);	const contributions = active.map((entry) => contribute(entry, edits.get(entry.id)));	const { settled, open } = settle([], contributions.flat());	const baseline = findBaseline(bounds.head, active);	const messages = leadWithSystem([...settled, ...orderToolResults(open)], baseline);	const view = { head: bounds.head, entries: active, contributions, messages };	return { bounds, entries, view, edited: new Set(edits.keys()), settled, open, baseline };}

/** Conversation document that retains only its current state. */export type LatestConversationSemantics = {	readonly scope: "conversation";	readonly history: "latest";	readonly fork: "current" | "initial";};/** Conversation document whose history remains addressable for as-of reads. */export type RewindableConversationSemantics = {	readonly scope: "conversation";	readonly history: "rewindable";	readonly fork: "asOf" | "current" | "initial";};/** Ownership and lifetime of a document; only conversation documents declare history and fork behavior. */export type DocumentSemantics =	| { readonly scope: "session"; readonly history?: never; readonly fork?: never }	| LatestConversationSemantics	| RewindableConversationSemantics	| { readonly scope: "task"; readonly history?: never; readonly fork?: never };/** Stored replay state supplied to a document's checkpoint predicate. */export type CheckpointInfo = {	/** Deltas already stored after the newest base, excluding the change being evaluated. */	readonly deltasSinceBase: number;};/** Definition fields shared by singleton documents and document families. */export type CommonDocDefinition<T extends JsonObject> = {	/** Stable persisted kind; part of the public protocol. */	readonly kind: string;	/** Positive integer version of the stored value shape. */	readonly version: number;	initial(): T;	migrate?(value: JsonObject, fromVersion: number): T;	/** Required policy: return true to store this ordinary change as a complete base instead of a delta. */	checkpointWhen(value: Readonly<T>, ops: readonly Op[], info: CheckpointInfo): boolean;};

/** Define a Session-scoped singleton document. */export function defineDoc<T extends JsonObject>(	definition: CommonDocDefinition<T> & { readonly scope: "session" },): SessionDocToken<T>;/** Define a latest-only conversation singleton document. */export function defineDoc<T extends JsonObject>(	definition: CommonDocDefinition<T> & LatestConversationSemantics,): ConversationDocToken<T>;/** Define a rewindable conversation singleton document. */export function defineDoc<T extends JsonObject>(	definition: CommonDocDefinition<T> & RewindableConversationSemantics,): RewindableConversationDocToken<T>;/** Define a task-scoped singleton document. */export function defineDoc<T extends JsonObject>(	definition: CommonDocDefinition<T> & { readonly scope: "task" },): TaskDocToken<T>;export function defineDoc<T extends JsonObject>(definition: DocDefinition<T>): DocToken<T, DocDefinition<T>> {	validateDefinition(definition);	return { definition };}⋯ lược bớt dòng 58–94function validateDefinition(definition: AnyDocDefinition): void {	if (!Number.isSafeInteger(definition.version) || definition.version < 1) {		throw new TypeError(`Document ${definition.kind} version must be a positive integer`);	}}

/** * The built-in creation hook, in every commit that creates or forks a conversation: empty `pi.live`, `pi.inbox`, and * `pi.usage`, a fresh `pi.provider`, the conversation's `pi.agent` (see `createAgent()`), then * `HarnessOptions.conversationCreated`. */protected override async conversationCreated(tx: Transaction, record: ConversationRecord): Promise<void> {	await tx.doc(LiveDoc, record.id);	await tx.doc(InboxDoc, record.id);	await tx.doc(UsageDoc, record.id);	await tx.doc(ProviderDoc, record.id);	await createAgent(tx, record);	await this.#options.conversationCreated?.(tx, record);}

/** * Built-in part of every Harness commit that creates or forks a conversation, for `pi.agent`: a fork keeps its `asOf` * copy; a new task-owned conversation copies the stored agent of its owner task's conversation; a new ownerless one * starts empty. */export async function createAgent(tx: Tx, conversation: ConversationRecord): Promise<void> {	if (conversation.parent !== undefined) return;	const agent = await tx.doc(AgentDoc, conversation.id);	if (conversation.owner === undefined) return;	const owner = await tx.doc(AgentDoc, conversation.owner.conversationId);	Object.assign(agent, copyJson(owner) as AgentState);}

/** 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;};

/** Create a conversation with explicitly selected ownership. */createConversation(options: { readonly ownership: ConversationOwnership }): Promise<ConversationRecord>;/** Create a history fork at one concrete visible entry with explicitly selected ownership. */forkConversation(	parentConversationId: ConversationId,	at: EntryId,	options: { readonly ownership: ConversationOwnership },): Promise<ConversationRecord>;/** Returned records are Session-owned immutable values and may be shared with commit listeners. */appendEntry(conversationId: ConversationId, value: EntryDraft): Promise<EntryRecord>;/** The token supplies `kind` and types `data`. */appendEntry<D extends JsonValue>(	token: Entry<D>,	conversationId: ConversationId,	value: TypedEntryDraft<NoInfer<D>>,): Promise<TypedEntry<D>>;createTask<I, S extends { phase: string }, R, H extends object>(	task: Task<I, S, R, H>,	input: I,	options: TaskOptions,): Promise<TaskId<R>>;

/** * The entry an input is placed as, instead of `pi.user`: an application kind, and data the transcript shows, such as its * author. Its model message is still the input's user message. */export type InputEntry = { readonly kind: string; readonly data?: JsonValue };/** Host submission: user input that may start a run, or a passive entry write. */export type SubmissionDraft = {	readonly requestId?: string;} & (	| {			readonly type: "input";			readonly content: UserInput;			readonly whenBusy?: "steer" | "followUp" | "reject";			/** Place the input as this entry; default `pi.user` without data. The kind must not start with `pi.`. */			readonly entry?: InputEntry;	  }	| {			readonly type: "write";			readonly entry: EntryDraft;			readonly content?: never;			readonly whenBusy?: never;	  });export type InputSubmissionDraft = Extract<SubmissionDraft, { readonly type: "input" }>;export type SettledSubmissionRecord = SubmissionRecord & {	readonly status: "done" | "unanswered";};/** Awaitable host object for one durably admitted submission. */export interface Submission {	readonly id: SubmissionId;	status(context: Context): Promise<SubmissionRecord>;	wait(context: Context): Promise<SettledSubmissionRecord>;	abort(context: Context): Promise<"aborted" | "already_placed" | "settled">;}

export type SubmissionRecord =	| (SubmissionRecordBase & {			readonly type: "input";	  } & (				| {						/** Admitted but not yet represented in the transcript. */						readonly status: "queued";						readonly entry?: never;						readonly answer?: never;						readonly reason?: never;						readonly detail?: never;				  }				| {						/** Added to the transcript and owned by an active run. */						readonly status: "placed";						readonly entry: EntryId;						readonly answer?: never;						readonly reason?: never;						readonly detail?: never;				  }				| {						/** Successfully answered user input. */						readonly status: "done";						readonly entry: EntryId;						readonly answer: EntryId;						readonly reason?: never;						readonly detail?: never;				  }				| {						/** Terminal input that can no longer receive an answer. */						readonly status: "unanswered";						readonly entry?: EntryId;						readonly answer?: never;						readonly reason: string;						readonly detail?: JsonValue;				  }			))

/** Admit a submission in one commit; see `admitSubmission()`. */async submit(conversationId: ConversationId, draft: SubmissionDraft, context: Context): Promise<Submission> {	this.#resume();	const parent = draft.type === "input" ? this.#telemetry : context.telemetry;	return createTypedSpanStarter(parent, [DURABLE_TELEMETRY_SCHEMA])(		"pi.durable.submit",		{			"submission.type": draft.type,			"conversation.id": conversationId,		},		async (span) => {			if (draft.type === "input") span.addLink({ context: context.telemetry });			try {				const { id, status } = await this.#session.commitWith(					(tx) => admitSubmission(tx, conversationId, draft, this.#now(), this.#queueModes(), span),					withTelemetry(span, context),				);				span.setAttributes({ "submission.id": id, result: status });				return new SubmissionHandle(id, this);

if (draft.type === "input" && draft.entry !== undefined) checkInputEntry(draft.entry);if (draft.requestId !== undefined) {	const existing = await tx.submissionByRequest(conversationId, draft.requestId);	if (existing !== undefined) {		if (existing.type !== draft.type) {			throw new Error(`Request ${draft.requestId} already identifies a submission of type ${existing.type}`);		}		return { id: existing.id, status: existing.status === "unanswered" ? "done" : existing.status };	}}const fields = { conversationId, requestId: draft.requestId, trace: span?.propagation?.inject(span) };if (	draft.type === "write" &&	isDisplayOnly(draft.entry) &&	!hasQueuedReset((await tx.doc(InboxDoc, conversationId)).items)) {	const entry = await tx.appendEntry(conversationId, draft.entry);	const write = { ...fields, type: "write", status: "done", entry: entry.id } as const;	return { id: (await tx.createSubmission(write)).id, status: "done" };}

const live = await tx.doc(LiveDoc, conversationId);const busy = live.run !== undefined;if (	draft.type === "input" &&	draft.whenBusy === "steer" &&	live.run !== undefined &&	span?.propagation !== undefined) {	const run = await tx.task(live.run.taskId);	if (run?.trace !== undefined) span.addLink({ context: span.propagation.extract(run.trace) });}if (busy && draft.type === "input" && draft.whenBusy === "reject") throw new ConversationBusy(conversationId);// A boundary reads the table, so it is prepared before the first table write; a busy one needs none.const boundary = busy ? undefined : await prepareBoundary(tx, conversationId, queueModes);if (boundary === undefined || boundary.inbox.items.length > 0) {	const { id } = await tx.createSubmission({		...fields,		type: draft.type,		status: "queued",	});	// Hosts may leave optional fields `undefined`; drafts take strict JSON.	const value = copyJson(draft.type === "write" ? draft.entry : draft.content, {		omitUndefinedProperties: true,	});	const items = (boundary?.inbox ?? (await tx.doc(InboxDoc, conversationId))).items;	if (draft.type === "write") items.push({ id, mode: "write", entry: value as JsonObject });	else {		const mode = draft.whenBusy === "steer" ? "steer" : "followUp";		const content = value as JsonRepresentation<UserInput>;		const entry = draft.entry === undefined ? {} : { entry: pickInputEntry(draft.entry) };		items.push({ id, mode, content, ...entry });	}	if (boundary === undefined) return { id, status: "queued" };	const { users } = await applyBoundary(tx, boundary, "final", now);	if (users.length > 0) await startRun(tx, conversationId, live, users);	return { id, status: draft.type === "write" ? "done" : users.at(-1) === id ? "placed" : "queued" };}

	if (draft.type === "write") {		if (isStale(boundary, draft.entry)) {			const stale = { status: "unanswered", reason: "stale" } as const;			return {				id: (await tx.createSubmission({ ...fields, type: "write", ...stale })).id,				status: "done",			};		}		const entry = await tx.appendEntry(conversationId, draft.entry);		const write = { ...fields, type: "write", status: "done", entry: entry.id } as const;		return { id: (await tx.createSubmission(write)).id, status: "done" };	}	const entry = await appendInput(tx, conversationId, draft.content, draft.entry, now);	const input = { ...fields, type: "input", status: "placed", entry: entry.id } as const;	const { id } = await tx.createSubmission(input);	await startRun(tx, conversationId, live, [id]);	return { id, status: "placed" };}

const { conversationId, inbox } = boundary;const items = inbox.items;const reset = hasQueuedReset(items);const final = at === "final" || reset;const pick = (mode: "steer" | "followUp", queueMode: QueueMode): number[] => {	const indexes = items.flatMap((item, index) => (item.mode === mode ? [index] : []));	return queueMode === "all" ? indexes : indexes.slice(0, 1);};const writes = items.flatMap((item, index) => (item.mode === "write" ? [index] : []));const users = [...pick("steer", boundary.steeringMode), ...(final ? pick("followUp", boundary.followUpMode) : [])];users.sort((a, b) => a - b);

async wait(id: SubmissionId, context: Context): Promise<SettledSubmissionRecord> {	this.#resume();	// Check and register on the line so no settling publication falls between them.	const found = await this.#session.readOnLine(async () => {		const record = await this.#storage.submission(id, context);		if (record === undefined) throw new Error(`Submission ${id} does not exist`);		if (isSettled(record)) return { promise: Promise.resolve(record) };		// Close rejects registered waiters synchronously and may begin during the read.		if (this.#closed) throw closedError(this.#session);		return { promise: this.#waiters.add(id, context) };	});	return found.promise;}/** Withdraw a queued submission and remove its inbox item; placed inputs and settled submissions are reported. */abort(id: SubmissionId, context: Context, conversationId?: ConversationId): Promise<AbortResult | "not_found"> {	return this.#session.commitWith(async (tx) => {		const record = await tx.submission(id);		if (record === undefined || (conversationId !== undefined && record.conversationId !== conversationId)) {			return "not_found";		}		if (record.status === "queued") {			tx.settleSubmission(id, { status: "unanswered", reason: "aborted" });			await removeInboxItem(tx, record.conversationId, id);			return "aborted";		}		return record.status === "placed" ? "already_placed" : "settled";	}, context);}#observe(publication: CommitPublication): void {	for (const change of publication.changes) {		if (change.type !== "submission" || !isSettled(change.value)) continue;		this.#waiters.resolve(change.value.id, change.value);	}}

/** Built by `hook()`; matches tasks by name. */export type HookRegistration = { readonly task: string; readonly handlers: object };/** Built by `wrapTool()` and `wrapSection()`; targets a tool name or a section key. Wrappers are pure. */export type Wrap<Tool extends ToolRegistration = ToolRegistration> =	| { readonly tool: string; wrap(tool: Tool): Tool }	| { readonly section: string; wrap(section: PromptSection<Tool>): PromptSection<Tool> };/** Named bundle of code; installed in a registry and selected by conversations by name. */export interface Extension<Tool extends ToolRegistration = ToolRegistration> {	readonly name: string;	readonly tools?: readonly Tool[];	readonly sections?: readonly PromptSection<Tool>[];	readonly hooks?: readonly HookRegistration[];	/** Apply where this extension is selected, in order. */	readonly wraps?: readonly Wrap<Tool>[];	/** Resolved by name for every task, whichever conversations select this extension. */	readonly tasks?: readonly AnyTask[];}

/** Identity function that types an extension. */export function defineExtension<Tool extends ToolRegistration = ToolRegistration>(	extension: Extension<Tool>,): Extension<Tool> {	return extension;}⋯ lược bớt dòng 11–27/** Hook handlers for tasks with `task`'s name. */export function hook<K extends AnyTask>(task: K, handlers: Partial<HooksOf<K>>): HookRegistration {	return { task: task.definition.name, handlers };}/** Wrap the tool named like `tool` wherever the wrapping extension is selected. */export function wrapTool<Tool extends ToolRegistration>(tool: Tool, wrapper: (tool: Tool) => Tool): Wrap<Tool> {	return { tool: tool.name, wrap: wrapper };}

class RegistryImpl<Tool extends ToolRegistration> implements Registry<Tool> {	#current = new RegistryState<Tool>([]);	readonly #listeners = new Set<() => void>();	snapshot(): RegistrySnapshot<Tool> {		return this.#current;	}	subscribe(listener: () => void): () => void {		this.#listeners.add(listener);		return () => this.#listeners.delete(listener);	}	install(extension: Extension<Tool>): void {		validateExtension(extension);		const current = this.#current.installed();		const index = current.findIndex((installed) => installed.name === extension.name);		const next =			index < 0 ? [...current, extension] : current.map((installed, at) => (at === index ? extension : installed));		this.#publish(next);	}	uninstall(extension: Extension): void {		const current = this.#current.installed();		if (!current.some((installed) => installed.name === extension.name)) return;		this.#publish(current.filter((installed) => installed.name !== extension.name));	}	/** Build and validate the next state, which throws on a task name collision, then publish it synchronously. */	#publish(extensions: readonly Extension<Tool>[]): void {		this.#current = new RegistryState(extensions);		for (const listener of [...this.#listeners]) listener();	}}

/** Stored choices of one conversation; names, not objects. Unset fields follow the host. */export type AgentState = {	model?: ModelRef;	thinkingLevel?: ModelThinkingLevel;	/** An array selects exactly these extensions, in order. An object edits the host default selection. */	extensions?: string[] | { add?: string[]; remove?: string[] };	/** Filters the selected extensions' tools: the enabled tools. An array enables exactly these, in order. */	tools?: string[] | { remove: string[] };	/**	 * Filters the enabled tools the model may call (`callers` includes `model`): the tools offered to it. Never widens:	 * other tools can still call the ones it leaves out.	 * - Unset: every such tool except deferred ones.	 * - An array: exactly these, in order, deferred ones too.	 * - An object: every such tool except deferred ones and those in `remove`, in registry order, followed by the	 *   deferred tools in `add` that are not in `remove`, in `add` order. Names of tools that are not deferred or not	 *   registered have no effect in `add`.	 */	modelTools?: string[] | { remove?: string[]; add?: string[] };	/** Rendered after every extension section, as the section `instructions`. */	instructions?: string;	/** Directory within the environment's file system, passed to `HarnessOptions.env`. */	cwd?: string;};

/** Apply one change to `pi.agent`: a given field replaces the stored one, `null` clears it, `undefined` changes nothing. */export async function configure(tx: Tx, conversationId: ConversationId, change: AgentChange): Promise<void> {	const state = await tx.doc(AgentDoc, conversationId);	applyChange(state, change);}⋯ lược bớt dòng 92–119function applyChange(state: Draft<AgentState>, change: AgentChange): void {	const set = <K extends keyof AgentState>(key: K, value: AgentState[K] | null | undefined) => {		if (value === undefined) return;		if (value === null) delete state[key];		else state[key] = value as Draft<AgentState>[K];	};	set("model", change.model === undefined || change.model === null ? change.model : { ...change.model });	set("thinkingLevel", change.thinkingLevel);

/** A conversation's agent resolved against a registry snapshot and the settings. */export type Agent<Tool extends ToolRegistration = ToolRegistration> = {	readonly model?: ModelRef;	readonly thinkingLevel: ModelThinkingLevel;	readonly extensions: readonly Extension<Tool>[];	/** The tools a request offers, in order: enabled, callable by the model, and selected by `modelTools`. */	readonly tools: readonly Tool[];	/** The tools nested calls (`executeTool()`) resolve among, in order: enabled and callable by tools. */	readonly callable: readonly Tool[];	/** Enabled deferred tools the model may call and is not offered yet, in registry order: what `tool_search` loads. */	readonly deferred: readonly Tool[];	/** Extension sections, then `instructions` when set. */	readonly sections: readonly PromptSection<Tool>[];	readonly instructions?: string;	readonly cwd?: string;};

/** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */async resolveAgent(	id: ConversationId,	snapshot: RegistrySnapshot<Tool> | undefined,	context: Context,): Promise<Agent<Tool>> {	const registry = snapshot ?? this.#options.registry.snapshot();	const state = await this.snapshot(AgentDoc, id, context);	return resolveAgent(state, registry, resolveSettings(this.#hostSettings()), (error, extension) =>		this.#report(error, { type: "extension_error", conversationId: id, extension }, context),	);}

const extensions = selectExtensions(state?.extensions, snapshot, settings);const composed = new Map<string, Tool>();for (const extension of extensions) for (const tool of extension.tools ?? []) composed.set(tool.name, tool);const sections = new Map<string, PromptSection<Tool>>();for (const extension of extensions) {	for (const section of extension.sections ?? []) sections.set(section.key, section);}⋯ lược bớt dòng 237–258const enabled = filterTools([...composed.values()], state?.tools);const callableBy = (caller: "model" | "tools") => (tool: Tool) =>	tool.callers === undefined || tool.callers.includes(caller);const modelCallable = enabled.filter(callableBy("model"));const tools = offeredTools(modelCallable, state?.modelTools);const offered = new Set(tools);const deferred = modelCallable.filter((tool) => tool.deferred === true && !offered.has(tool));const callable = enabled.filter(callableBy("tools"));const instructions = state?.instructions;const agentSections = [...sections.values()];if (instructions !== undefined) agentSections.push({ key: INSTRUCTIONS_KEY, render: () => instructions });

/** Input to system prompt section rendering for one request preparation. */export type PromptInput<Tool extends ToolRegistration = ToolRegistration> = {	readonly conversationId: ConversationId;	/** The request's resolution; `agent.tools` are the tools offered in this request. */	readonly agent: Agent<Tool>;	/** Built by `HarnessOptions.env` for this preparation; `undefined` without an environment. */	readonly env: ExecutionEnv | undefined;	/** Sections already in effect after replaying the active transcript. */	readonly shown: Readonly<Record<string, string>>;	/** Committed document reads. */	readonly read: DocumentReader;};/** One system prompt section; the agent's sections render in order before each request. */export type PromptSection<Tool extends ToolRegistration = ToolRegistration> = {	readonly key: string;	render(input: PromptInput<Tool>, context: Context): string | undefined | Promise<string | undefined>;	/** Default true: wrap the text as `<key>\n...\n</key>`. */	readonly tag?: boolean;};

/** * Render the agent's sections in order. `undefined` omits a section; tagged text is wrapped as `<key>\n...\n</key>`. A * section that throws keeps its shown text, if any, and is reported; errors after `context` is aborted propagate. */export async function renderSections<Tool extends ToolRegistration>(	sections: readonly PromptSection<Tool>[],	input: PromptInput<Tool>,	shown: ReadonlyMap<string, string>,	report: (error: unknown, section: PromptSection<Tool>) => void,	context: Context,): Promise<Map<string, string>> {	const desired = new Map<string, string>();	for (const section of sections) {		let text: string | undefined;		try {			text = await section.render(input, context);		} catch (error) {			if (context.abortSignal?.aborted) throw error;			report(error, section);			const kept = shown.get(section.key);			if (kept !== undefined) desired.set(section.key, kept);			continue;		}		if (text === undefined) continue;		desired.set(section.key, section.tag === false ? text : `<${section.key}>\n${text}\n</${section.key}>`);	}	return desired;}

/** * Plan the `pi.system` entries that make the replayed sections and tools of `view` equal `desired` and `tools` in values * and order. * * - A head marker with no later `pi.system` entry in context: one complete baseline that omits every retained earlier *   `pi.system` entry, written even when it restates the replayed values. * - Otherwise, when a minimal section patch would leave a different order: remove every shown section, then re-add *   every desired section in order. * - Otherwise the minimal patch of changed values and `null` removals, or nothing. * * Tool changes ride on the last planned entry, or on one entry of their own. */export function planSystemEntries(	view: ContextView,	desired: ReadonlyMap<string, string>,	tools: readonly Tool[],	timestamp: number,): SystemDraft[] {	const head = view.head;	if (head !== undefined && !view.entries.some((entry) => SystemEntry.is(entry) && entry.id > head.id)) {		const edits: ContextEdit[] = view.entries			.filter((entry) => SystemEntry.is(entry))			.map((entry) => ({ target: entry.id, action: "omit" }));		const baseline = systemEntry(			Object.fromEntries(desired),			{ toolsRemoved: [], toolsAdded: tools.map(toToolDeclaration) },			timestamp,		);		return [edits.length === 0 ? baseline : { ...baseline, edits }];	}	const sections = planSections(replaySections(view.messages), desired);	const changes = planTools(getCurrentTools(view.messages), tools);	if (changes.toolsRemoved.length === 0 && changes.toolsAdded.length === 0) {		return sections.map((patch) => systemEntry(patch, undefined, timestamp));	}	if (sections.length === 0) return [systemEntry(undefined, changes, timestamp)];	return sections.map((patch, index) =>		systemEntry(patch, index === sections.length - 1 ? changes : undefined, timestamp),	);}

export const Harness = {	/** Open a Harness over storage. The registry may keep changing while the Harness runs. */	async open<Tool extends ToolRegistration>(		storage: Storage,		options: HarnessOptions<Tool>,		context: Context,	): Promise<Harness> {		context.abortSignal?.throwIfAborted();		const snapshot = options.registry.snapshot();		const missing = BUILTIN_TASKS.filter((task) => snapshot.task(task.definition.name) === undefined);		if (missing.length > 0) {			const names = missing.map((task) => task.definition.name).join(", ");			throw new Error(`Registry lacks built-in tasks ${names}; create it with createRegistry()`);		}		const harness = new HarnessImpl(storage, options, context);		try {			await harness.openTasks(context);		} catch (error) {			// The caller's context may be what failed open: close without it, and rethrow the open error.			// A failing Storage close fails the Session, which reports it.			await harness.close(withoutAbortSignal(context)).catch(() => {});			throw error;		}		return harness;	},};

async #create(target: CreateTarget, options: CreateOptions, context: Context): Promise<Conversation> {	this.assertUsable();	const id = await this.commitWith(async (tx) => {		if (target.kind === "root" && (await tx.conversation(ROOT_CONVERSATION_ID)) !== undefined) {			return ROOT_CONVERSATION_ID;		}		const record =			target.kind === "root"				? await tx.createRootConversation()				: target.kind === "fork"					? await tx.forkConversation(target.parentId, target.at, { ownership: target.ownership })					: await tx.createConversation({ ownership: target.ownership });		if (options.agent !== undefined) await configure(tx, record.id, options.agent);		if (options.init !== undefined) await options.init(tx, record.id);		return record.id;	}, context);	return new ConversationImpl(id, this.#host);}

export type HarnessOptions<Tool extends ToolRegistration = ToolRegistration> = {	/** Host-owned adapter for content-free admission/phase telemetry; propagation and metrics are optional. */	readonly telemetry?: TelemetryContext;	/**	 * pi-ai model access used by generation, compaction, hooks and tools: a `Models`, or a function returning the current	 * one, called at each use, so a host can replace its model configuration without reopening the Harness.	 */	readonly models: Models | (() => Models);	/**	 * How model requests identify the app to providers, merged field by field over `DEFAULT_HARNESS_IDENTITY`	 * (`{ attribution: PI_ATTRIBUTION }`, provider attribution to pi). It applies to every request through the Harness's	 * models, including hooks and tools, and overrides the identity the `Models` was created with; a request's own	 * `identity` overrides it. Pass `{ attribution: null }` to send no attribution headers.	 */	readonly identity?: ClientIdentity;	readonly registry: RegistryReader<Tool>;	/** Run settings, or a function returning the current ones; read at each use either way. */	readonly settings?: HarnessSettings | (() => HarnessSettings | undefined);	/** Builds a conversation's environment at each use. Never called on the Session line; may be async. */	readonly env?: (target: EnvTarget, context: Context) => ExecutionEnv | undefined | Promise<ExecutionEnv | undefined>;	/**	 * Runs in every commit that creates or forks a conversation, raw `tx.createConversation()` included, after the	 * built-in `pi.*` documents and before the conveniences apply `agent` and run `init`. A fork already has its copies.	 * Table reads throw `ReadAfterWrite`, as in `init`; a throw fails the creating commit.	 */	readonly conversationCreated?: (tx: Tx, conversation: ConversationRecord) => void | Promise<void>;

/** Resolve the host settings: every field over its built-in default, object fields merged. */export function resolveSettings(settings: HarnessSettings | undefined): Settings {	const extensions = settings?.extensions;	return {		...(extensions === undefined ? {} : { extensions }),		stream: { ...settings?.stream },		retry: { ...DEFAULT_RETRY_POLICY, ...settings?.retry },		compaction: { ...DEFAULT_COMPACTION_POLICY, ...settings?.compaction },		// Field by field, so an explicitly undefined interval keeps its default.		progress: {			partialIntervalMs: settings?.progress?.partialIntervalMs ?? DEFAULT_PROGRESS_POLICY.partialIntervalMs,			outputIntervalMs: settings?.progress?.outputIntervalMs ?? DEFAULT_PROGRESS_POLICY.outputIntervalMs,		},		toolExecution: settings?.toolExecution ?? "parallel",		steeringMode: settings?.steeringMode ?? "one-at-a-time",		followUpMode: settings?.followUpMode ?? "one-at-a-time",		contextRetentionMs: settings?.contextRetentionMs ?? DEFAULT_CONTEXT_RETENTION_MS,		documentRetentionMs: documentRetention(settings?.documentRetentionMs),		afterFailedRun: settings?.afterFailedRun ?? "keep",		restartLimit: restartLimit(settings?.restartLimit),	};}

/** Structural mount of one conversation's active transcript and built-in documents (spec §9.3). */export type ConversationView = {	readonly conversation: ConversationRecord;	/** Raw active entries, as `ContextView.entries`: the head marker, then the non-head entries from its head. */	readonly entries: readonly EntryRecord[];	/** Built-in conversation documents keyed by kind; absent documents are absent. */	readonly docs: Readonly<Record<string, JsonObject>>;};

const docOps: Op[] = [];const entryOps: Op[] = [];let entries = mount.value.entries;// Entry writes are published in ID order.for (const change of publication.changes) {	if (change.type === "entry" && change.value.conversationId === id) {		const entry = change.value;		const value = entry as unknown as JsonValue;		if (entry.head === undefined) {			entryOps.push(["p", ["entries"], entries.length, 0, [value]]);			entries = [...entries, entry];			continue;		}		// A head marker keeps the non-head entries from its head, which are always a suffix, and goes in front.		const target = entry.head;		let kept = entries.findIndex((candidate) => candidate.head === undefined && candidate.id >= target);		if (kept < 0) kept = entries.length;		entryOps.push(["p", ["entries"], 0, kept, [value]]);		entries = [entry, ...entries.slice(kept)];		continue;	}	if (change.type !== "document" || change.conversationId !== id) continue;	const kind = change.record.kind;	if (!MOUNTED_KINDS.has(kind) || change.record.key !== undefined) continue;	const path: NonEmptyPath = ["docs", kind];	const mounted = mount.docs.get(kind);	if (change.value === null) {		if (mounted?.id !== change.record.id) continue;		mount.docs.delete(kind);		docOps.push(["d", path]);	} else if (mounted?.id === change.record.id && mounted.version === change.version) {		for (const op of change.ops) docOps.push(prefixed(op, path));	} else {		mount.docs.set(kind, { id: change.record.id, version: change.version! });		docOps.push(["s", path, change.value]);	}}

advance(value: T, ops: readonly Op[], context: Context): void {	if (this.#end !== undefined || this.#retired) return;	if (value === null) this.#retired = true;	if (this.#pending.length >= MAX_PENDING_WATCH_FRAMES) {		this.#pending.length = 0;		const replacement = this.#replace?.() ?? value;		this.#pending.push({ value: replacement, ops: [["r", replacement as JsonValue]], context });	} else {		this.#pending.push({ value, ops, context });	}	if (this.#started) this.#schedule();}

/** Experimental agent event, shaped like the coding agent's session events (spec §9.4). */export type AgentEvent =	| SnapshotEvent	| { type: "run_start"; inputs: readonly SubmissionId[] }	| { type: "run_end"; inputs: readonly SubmissionId[] }	| { type: "turn_start" }	| { type: "turn_end" }	| { type: "message_start"; message: Message }	/** `usage` is the partial's current usage, as in the coding agent's JSON mode. */	| { type: "message_update"; usage: Usage; changes: readonly MessageChange[] }	| { type: "message_end"; entry: EntryRecord }	| ({ type: "tool_execution_start"; args: JsonObject } & ToolEventCall)	| ({			type: "tool_execution_update";			/** A front trim and then an append of the retained window, or its replacement. */			output?: { trimStart?: number; append?: string } | { set: string };			details?: JsonValue;			diagnostics?: readonly ToolDiagnostic[];	  } & ToolEventCall)	/**	 * A model-issued call ends with its result `entry`, a nested call with its `result`. Both are absent when the tool	 * task faulted or was orphaned, or when the call's slot left `pi.live` unfinished: its run ended, or its parent	 * settled first.	 */	| ({ type: "tool_execution_end"; entry?: EntryRecord; result?: NestedToolExecutionResult } & ToolEventCall)	| { type: "inbox_update"; items: readonly QueuedItem[] }	| { type: "submission"; record: SubmissionRecord }	| { type: "auto_retry_start"; attempt: number; at: number; errorMessage: string }	| { type: "auto_retry_end"; attempt: number }	| { type: "deferred_poll"; pollAt: number }	| { type: "entry_appended"; entry: EntryRecord }	| { type: "agent_changed"; agent: AgentState }	| { type: "usage_changed"; usage: UsageState }	| { type: "task_failed"; taskId: TaskId; kind: string; message: string }	| { type: "compaction_start"; taskId: TaskId; reason: CompactionReason; blocking: boolean }	/**	 * The task's receipt tells whether it produced a summary; the summary entry has its own events. `error` is set when	 * the compaction failed, faulted, or was orphaned, as in `pi.live.compactionFailure`.	 */	| { type: "compaction_end"; taskId: TaskId; reason: CompactionReason; error?: string };

const views = conversationViews(harness);let watch!: CommittedWatch<readonly AgentEvent[]>;let snapshot!: SnapshotEvent;await views.attach(	conversationId,	async (initial, release, storage) => {		// Generations whose held outcome already ended their turn, read on the line with the snapshot.		const query = { conversationId, kind: "pi.generation", status: "completing" } as const;		const completing = await scanAll((cursor) => storage.scanTasks(query, 100, cursor, context));		const held = new Set<TaskId>(completing.map((record) => record.id));		let current = initial;		snapshot = snapshotOf(initial);		// Batches are the watch's values; an overflow delivers a snapshot of the newest view instead.		watch = new CommittedWatch<readonly AgentEvent[]>(			[],			release,			(error) => views.report(error),			() => [snapshotOf(current)],		);		return {			publication: (before, after, ops, publication, commitContext) => {				current = after;				const events = translate(conversationId, before, after, ops, publication, held);				if (events.length > 0) watch.advance(events, [], commitContext);			},			closeSession: (failure) => watch.closeSession(failure),		};	},	context,

One answered input, as entries and tasks:```textsubmit(input) → pi.user  pi.generation → pi.system (only if the prompt or tools changed), pi.assistant (tool calls)    pi.tool × n → pi.tool-result × n   (owned by the generation, which waits for them)  pi.generation → pi.assistant (answer) → submission done```

const view = await runtime.context(conversationId, context);const shown = replaySections(view.messages);let env: ExecutionEnv | undefined;try {	env = await runtime.env(context);} catch (error) {	if (context.abortSignal?.aborted) throw error;	runtime.report(error, { type: "host", callback: "env" });}const input: PromptInput = { conversationId, agent, env, shown: Object.fromEntries(shown), read: runtime };const desired = await renderSections(	agent.sections,	input,	shown,	(error, section) =>		runtime.report(error, {			type: "extension",			...optionalExtension(sectionExtension(agent, section.key)),		}),	context,);const entries = planSystemEntries(view, desired, agent.tools, runtime.now());⋯ lược bớt dòng 177–189await 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);

await runtime.commit(async (tx): Promise<Next> => {	const live = await tx.doc(LiveDoc, conversationId);	const entry = await appendAssistant(tx, conversationId, message);	const slots: ToolSlot[] = [];	const tools: TaskId<ToolTaskResult>[] = [];	const pending: string[] = [];	for (const call of calls) {		if (!offered.has(call.name)) {			const unavailable = harnessError("tool_unavailable", `Tool ${call.name} is not available`);			const result = await appendToolResult(tx, conversationId, call, unavailable, { timestamp: runtime.now() });			slots.push({ callId: call.id, name: call.name, status: "done", entry: result.id });			continue;		}		if (sequential && tools.length > 0) {			pending.push(call.id);			slots.push({ callId: call.id, name: call.name, status: "pending" });			continue;		}		const taskId = await createToolTask(tx, runtime, entry.id, call.id);		tools.push(taskId);		slots.push({ callId: call.id, name: call.name, taskId, status: "pending" });	}	delete live.generation;	live.tools = slots;	const checkpoint = { phase: "tools", assistant: entry.id, tools, pending } as const;	return { status: "waiting", checkpoint, on: tools, policy: "allSettled" };}, context);

await runtime.commit(async (tx) => {	const slot = toolSlot(await tx.doc(LiveDoc, runtime.conversationId), runtime.taskId);	if (slot !== undefined) slot.status = "running";	// A nested slot shows the arguments the call runs with, which repair, hooks, and coercion may have changed.	// Compared with the input, which the slot got at admission: reading the draft would track every leaf.	const changed = task.input.kind === "nested" && !jsonEqual(task.input.call.arguments as JsonValue, final);	if (changed && slot !== undefined && "parentTaskId" in slot) slot.arguments = final;	const intent = { phase: "execute", arguments: final, replay: tool.replay ?? "unsafe" } as const;	return { status: "running", checkpoint: intent };}, context);await run(runtime, task.input, call, tool, final, context);

	const timing = { timestamp: runtime.now(), durationMs };	const entry = await appendToolResult(tx, conversationId, call, result, timing);	if (slot !== undefined && !("parentTaskId" in slot)) {		finishSlot(slot);		slot.entry = entry.id;	}	// Tools build control objects freely; drop keys set to undefined so the task result is strict JSON.	const control =		result.control === undefined || ending.status !== "completed"			? {}			: { control: copyJson(result.control as JsonValue, { omitUndefinedProperties: true }) as ToolControl };	settled = { kind: "model", entryId: entry.id, ...control };⋯ lược bớt dòng 687–695return { status: "terminal", outcome: { status: "completed", result: settled } };

await runtime.commit(async (tx): Promise<Next> => {	const boundary = await prepareBoundary(tx, conversationId, runtime.settings);	if (added.length > 0) await addTools(tx, conversationId, added);	const live = await tx.doc(LiveDoc, conversationId);	const now = runtime.now();	if (terminate || handoff !== undefined) {⋯ lược bớt dòng 714–721	} else {		const { users, reset } = await applyBoundary(tx, boundary, "postTools", now);		if (reset) {			// The queued reset cut the run's context before an answer.			endRun(tx, live, runtime.taskId, { status: "unanswered", reason: "reset" });			if (users.length > 0) await startRun(tx, conversationId, live, users);		} else {			delete live.tools;			delete live.nestedTools;			if (live.run?.taskId === runtime.taskId) live.run.inputs.push(...users);			handOver(live, runtime.taskId, await createGeneration(tx, conversationId));		}	}	return { status: "terminal", outcome: { status: "completed", result: { entryId: assistant } } };}, context);