pi-mono/packages/durable/src/session/transaction.ts

import { BACKGROUND_CONTEXT, type Context } from "@earendil-works/chord/context";
import {
	type Change,
	type Draft,
	type Op,
	type PendingOps,
	type Prepared,
	type Tracker,
	track,
} from "@earendil-works/chord/delta";
import { copyJson, type JsonValue } from "@earendil-works/chord/json";
import type { TelemetryCarrier } from "@earendil-works/pi-telemetry";
import {
	type AnyDocDefinition,
	type AnyDocToken,
	addressId,
	checkRecordScope,
	checkRecordVersion,
	documentCreate,
	materializeDocumentValue,
	type ResolvedAddress,
	resolveAddress,
} from "../documents.ts";
import { ReadAfterWrite } from "../errors.ts";
import type {
	ConversationDocFamilyToken,
	ConversationDocToken,
	ConversationId,
	ConversationOwnership,
	ConversationQuery,
	ConversationRecord,
	Cursor,
	DocumentAddress,
	DocumentCommitChange,
	DocumentContent,
	DocumentCopySource,
	DocumentCreate,
	DocumentId,
	DocumentRecord,
	Entry,
	EntryDraft,
	EntryId,
	EntryQuery,
	EntryRecord,
	JsonObject,
	Seq,
	SessionDocFamilyToken,
	SessionDocToken,
	Storage,
	StorageWrite,
	SubmissionCreate,
	SubmissionId,
	SubmissionRecord,
	SubmissionSettlement,
	Task,
	TaskDocFamilyToken,
	TaskDocToken,
	TaskId,
	TaskOptions,
	TaskQuery,
	TaskRecord,
	Tx,
	TypedEntry,
	TypedEntryDraft,
} from "../types.ts";
import { ROOT_CONVERSATION_ID } from "../types.ts";
import { prepareForkDocumentCopies } from "./forks.ts";
import { documentKey, incarnationKey, scopeKey, tableKeys } from "./keys.ts";

type AnyTaskRecord = TaskRecord<JsonValue, JsonValue, JsonValue>;

/** A staged submission change: a settlement, or the placement of a queued submission at its entry. */
type SubmissionChange = SubmissionSettlement | { readonly status: "placed"; readonly entry: EntryId };

/**
 * Complete record after applying one change. Placement turns a queued input `placed` and a queued write `done`; a
 * continue is never queued, so never placed. Only a placed input or continue can be answered. A settled record stays.
 */
function applySubmissionChange(current: SubmissionRecord, change: SubmissionChange): SubmissionRecord {
	if (current.status === "done" || current.status === "unanswered") return current;
	if (change.status === "placed") {
		if (current.status !== "queued") throw new Error(`Submission ${current.id} is not queued`);
		const status = current.type === "input" ? "placed" : "done";
		return { ...current, status, entry: change.entry } as SubmissionRecord;
	}
	if (change.status === "done" && current.status !== "placed") {
		throw new Error(`Submission ${current.id} is not a placed input or continue`);
	}
	// Queued and placed records carry no answer, reason, or detail; an unanswered input keeps its entry.
	return { ...current, ...change } as SubmissionRecord;
}

const INTERNAL_SCAN_PAGE_SIZE = 256;
const EMPTY_OPERATIONS: readonly Op[] = [];
const TABLE_JSON_COPY_OPTIONS = { omitUndefinedProperties: true } as const;

/** One committed document incarnation owned by the Session tracker cache. */
export type LoadedDocument = {
	readonly addressId: string;
	readonly record: DocumentRecord;
	/** Persisted definition version; older while the tracked value is migrated only in memory. */
	storedVersion: number;
	/** Definition version whose shape the tracked value has; access with another version reloads from Storage. */
	readonly valueVersion: number;
	/** Stored deltas after the newest base; advanced by adoption so the next predicate call needs no read. */
	deltasSinceBase: number;
	readonly tracker: Tracker<JsonObject>;
};

/**
 * The changes of a cached incarnation that deferred commits adopted in memory and the Session has not stored yet
 * (`CommitPersistence`), folded. The Session owns it; a commit that stores them takes them.
 */
export type HeldDocument = {
	readonly loaded: LoadedDocument;
	readonly pending: PendingOps;
	/** Set when a held commit selected a base (`checkpointWhen`): their store writes the whole value. */
	base: boolean;
};

/** What adoption publishes: to the Session's own observers, and the stored changes to `observeCommits()`. */
export type AdoptedDocuments = {
	/** This commit's own changes, as memory adopted them. */
	readonly observed: readonly DocumentCommitChange[];
	/** What Storage got: `observed`, unless the commit also stored held changes, which precede its own. */
	readonly stored: readonly DocumentCommitChange[];
};

/** Session services used by a transaction while it holds the mutation line. */
export interface TransactionHost {
	/** Reads wait while their key has an unpublished writer (`GatedStorage`). */
	readonly storage: Storage;
	/** Wait while `key` has an unpublished writer; `undefined` when it has none. */
	touch(key: string): Promise<void> | undefined;
	/** Wall clock for task lifecycle times. */
	now(): number;
	/** Return the cached current incarnation without loading. */
	cached(addressId: string): LoadedDocument | undefined;
	/** Return the cached current incarnation, cold-loading and migrating it when necessary. */
	load(
		definition: AnyDocDefinition,
		addressId: string,
		address: DocumentAddress,
		context: Context,
	): Promise<LoadedDocument | undefined>;
	/** Install a newly committed incarnation. */
	install(document: LoadedDocument): void;
	/** Remove a retired incarnation if it is still the cached occupant of its address. */
	evict(addressId: string, recordId: DocumentId): void;
	/** Stage writes that belong to every newly created or forked conversation, in its creating transaction. */
	conversationCreated(tx: Transaction, record: ConversationRecord): Promise<void>;
	/** The newest entry ID visible from a conversation, known from the Session's commits. */
	latestEntryId(conversationId: ConversationId, context: Context): Promise<EntryId | undefined>;
	/**
	 * The current documents of a live task when the Session knows all of them (it created the task), else `undefined`.
	 */
	taskDocuments(taskId: TaskId): Iterable<DocumentRecord> | undefined;
	/** The newest visible head marker of a conversation, known from the Session's commits. */
	latestHeadMarker(
		conversationId: ConversationId,
		context: Context,
	): Promise<(EntryRecord & { readonly head: EntryId }) | undefined>;
}

/** Committed and candidate state for one task touched by this transaction. */
type TransactionTask = {
	committedRead?: Promise<AnyTaskRecord | undefined>;
	write?: { readonly kind: "create" | "replace"; readonly record: AnyTaskRecord };
	publicationConversationId?: ConversationId;
};

/** Internal run creation resolves submission carriers against staged records before reading storage. */
export type TaskCreationOptions = TaskOptions & { readonly submissionIds?: readonly SubmissionId[] };

/** Defaults a commit binds to: `tx.createTask()` conversation and the task attributed to appended entries. */
export type TransactionScope = {
	readonly conversationId?: ConversationId;
	/** Task whose runtime commit this is; stamped as `byTaskId` on appended entries. */
	readonly taskId?: TaskId;
};

/** Storage/cache provenance of one staged document incarnation. */
type DocumentTarget =
	| { readonly kind: "loaded"; readonly document: LoadedDocument }
	| {
			readonly kind: "created";
			readonly record: DocumentCreate;
			readonly version: number;
			readonly tracker: Tracker<JsonObject>;
	  }
	| {
			readonly kind: "fork-copy";
			readonly record: DocumentCreate;
			readonly source: DocumentCopySource;
	  }
	| { readonly kind: "retire-only"; readonly record: DocumentRecord };

/** One document incarnation acquired, created, or retired by this transaction. */
type DocumentEntry = {
	readonly addressId: string;
	readonly address: DocumentAddress;
	/** Absent for definition-free fork copies and retirement entries discovered by a terminal-task scan. */
	definition?: AnyDocDefinition;
	/** Memoized public acquisition; absent for metadata-only retirement. */
	draftPromise?: Promise<Draft<JsonObject>>;
	/** Set after acquisition or retirement lookup finds the affected incarnation. */
	target?: DocumentTarget;
	change?: Change<JsonObject>;
	prepared?: Prepared<JsonObject>;
	retireOnCommit: boolean;
};

/**
 * What one staged incarnation writes and publishes, decided once before Storage admission so adoption only applies it.
 * `record` is a `DocumentRecord` for an incarnation that is already committed and a `DocumentCreate` for a new one.
 */
type DocumentPlan = {
	readonly addressId: string;
	readonly record: DocumentCreate | DocumentRecord;
	retire: boolean;
	/** Creation, copy, or change content; absent when only retirement is written. */
	content?: Extract<StorageWrite, { readonly type: "document.create" | "document.copy" | "document.change" }>;
	/** Prepared change of a tracked incarnation; absent for fork copies and retirement-only entries. */
	readonly change?: {
		readonly tracker: Tracker<JsonObject>;
		readonly prepared: Prepared<JsonObject>;
		readonly version: number;
		/** The cached incarnation this change updates; absent when it creates one. */
		readonly loaded?: LoadedDocument;
		readonly definition?: AnyDocDefinition;
	};
	/** Resolved before Storage admission so adoption performs no reads. */
	conversationId?: ConversationId;
	/**
	 * Set when this commit also stores held changes of the incarnation (`takeHeld()`): the operations its stored change
	 * publishes, held first, and whether it has a change of its own to show the Session's observers.
	 */
	held?: { readonly ops: readonly Op[]; readonly observed: boolean };
};

/**
 * Transaction for one Session commit callback.
 *
 * Every asynchronous operation is tracked so callback settlement can reject and drain unfinished work. Session calls
 * one settlement method, then either discards prepared changes or adopts them once after Storage succeeds.
 */
export class Transaction implements Tx {
	readonly #host: TransactionHost;
	#context: Context;
	readonly #scope: TransactionScope;
	readonly #pendingOperations = new Set<Promise<unknown>>();
	#sealed = false;
	#hasTableWrite = false;

	/** Atomic batch; conversation and entry writes stage eagerly, while task and document writes assemble later. */
	readonly #writes: StorageWrite[] = [];
	readonly #createdConversationIds = new Set<ConversationId>();
	readonly #forkSourceConversationIds = new Set<ConversationId>();
	readonly #forkSourceDocumentIds = new Set<DocumentId>();
	/** One entry per task touched by a public read, candidate write, or document-owner lookup. */
	readonly #tasksById = new Map<TaskId, TransactionTask>();
	/** Submissions created by this transaction, by ID. */
	readonly #submissions = new Map<SubmissionId, SubmissionRecord>();
	/** Submission settlements and placements in staging order; resolved against the latest candidate record during assembly. */
	readonly #submissionChanges: { readonly id: SubmissionId; readonly change: SubmissionChange }[] = [];

	/** Write and publication plans of every staged incarnation, built during assembly. */
	readonly #plans: DocumentPlan[] = [];
	/** Every document acquisition or retirement marker in staging order. */
	readonly #documents: DocumentEntry[] = [];
	/** Latest transaction-local incarnation or retirement marker at each logical address. */
	readonly #latestDocumentByAddress = new Map<string, DocumentEntry>();

	constructor(host: TransactionHost, context: Context, scope: TransactionScope = {}) {
		this.#host = host;
		this.#context = context;
		this.#scope = scope;
	}

	/** Internal: the context of the commit this transaction belongs to. */
	get context(): Context {
		return this.#context;
	}

	/** Internal: whether a table write was staged, after which table reads throw `ReadAfterWrite`. */
	get hasTableWrite(): boolean {
		return this.#hasTableWrite;
	}

	/**
	 * Internal: whether `tx` is a transaction of a commit of the Session behind `host`. A brand check, so a proxy or
	 * another wrapper of a transaction is not one.
	 */
	static ownedBy(tx: Tx, host: TransactionHost): tx is Transaction {
		return #host in tx && tx.#host === host;
	}

	// ─── Table reads ────────────────────────────────────────────────────────

	conversation(id: ConversationId): Promise<ConversationRecord | undefined> {
		return this.#read("conversation", () => this.#host.storage.conversation(id, this.#context));
	}

	entry(id: EntryId): Promise<EntryRecord | undefined>;
	entry<D extends JsonValue>(token: Entry<D>, id: EntryId): Promise<TypedEntry<D> | undefined>;
	entry(first: EntryId | { readonly kind: string }, second?: EntryId): Promise<EntryRecord | undefined> {
		const kind = typeof first === "number" ? undefined : first.kind;
		const id = typeof first === "number" ? first : second!;
		return this.#read("entry", async () => {
			const entry = (await this.#host.storage.entry(id, this.#context))?.entry;
			return kind === undefined || entry?.kind === kind ? entry : undefined;
		});
	}

	task(id: TaskId): Promise<AnyTaskRecord | undefined> {
		return this.#read("task", () => this.#committedTask(id));
	}

	scanConversations(query: ConversationQuery, limit: number, cursor?: Cursor) {
		return this.#read("scanConversations", () =>
			this.#host.storage.scanConversations(query, limit, cursor, this.#context),
		);
	}

	scanEntries(query: EntryQuery, limit: number, cursor?: Cursor) {
		return this.#read("scanEntries", () => this.#host.storage.scanEntries(query, limit, cursor, this.#context));
	}

	latestHeadMarker(conversationId: ConversationId) {
		return this.#read("latestHeadMarker", () => this.#host.latestHeadMarker(conversationId, this.#context));
	}

	latestEntryId(conversationId: ConversationId): Promise<EntryId | undefined> {
		return this.#read("latestEntryId", () => this.#host.latestEntryId(conversationId, this.#context));
	}

	scanTasks(query: TaskQuery, limit: number, cursor?: Cursor) {
		return this.#read("scanTasks", () => this.#host.storage.scanTasks(query, limit, cursor, this.#context));
	}

	/** Committed submission record. */
	submission(id: SubmissionId): Promise<SubmissionRecord | undefined> {
		return this.#read("submission", () => this.#host.storage.submission(id, this.#context));
	}

	/** Committed submission with a conversation-scoped request ID. */
	submissionByRequest(conversationId: ConversationId, requestId: string): Promise<SubmissionRecord | undefined> {
		return this.#read("submissionByRequest", () =>
			this.#host.storage.submissionByRequest(conversationId, requestId, this.#context),
		);
	}

	// ─── Table writes ───────────────────────────────────────────────────────

	createConversation(options: { readonly ownership: ConversationOwnership }): Promise<ConversationRecord> {
		return this.#write(() => this.#stageConversation(undefined, options.ownership));
	}

	/** Internal final-form bootstrap path for the reserved root identity. */
	createRootConversation(): Promise<ConversationRecord> {
		return this.#write(() => this.#stageConversation(undefined, { kind: "ownerless" }, ROOT_CONVERSATION_ID));
	}

	forkConversation(
		parentConversationId: ConversationId,
		at: EntryId,
		options: { readonly ownership: ConversationOwnership },
	): Promise<ConversationRecord> {
		return this.#write(() =>
			this.#stageConversation({ conversationId: parentConversationId, at }, options.ownership),
		);
	}

	async #stageConversation(
		parent: NonNullable<ConversationRecord["parent"]> | undefined,
		ownership: ConversationOwnership,
		reservedId?: ConversationId,
	): Promise<ConversationRecord> {
		const ownerTaskId = ownership.kind === "task" ? ownership.taskId : undefined;
		const id = reservedId ?? (await this.#host.storage.mintId<ConversationId>());
		this.#assertOpen();
		let owner: ConversationRecord["owner"];
		if (ownerTaskId !== undefined) {
			const task = await this.#currentTask(ownerTaskId);
			this.#assertOpen();
			if (task === undefined) throw new Error(`Conversation owner task ${ownerTaskId} does not exist`);
			owner = { conversationId: task.conversationId, taskId: ownerTaskId };
		}
		const record: ConversationRecord = {
			id,
			...(parent === undefined ? {} : { parent }),
			...(owner === undefined ? {} : { owner }),
		};
		const copies =
			parent === undefined
				? []
				: await prepareForkDocumentCopies(this.#host.storage, parent.conversationId, parent.at, id, this.#context);
		this.#assertOpen();
		for (const copy of copies) {
			this.#forkSourceDocumentIds.add(copy.source.id);
			const entry: DocumentEntry = {
				addressId: addressId(copy.record),
				address: copy.record,
				target: { kind: "fork-copy", ...copy },
				retireOnCommit: false,
			};
			this.#documents.push(entry);
			this.#latestDocumentByAddress.set(entry.addressId, entry);
		}
		if (parent !== undefined) this.#forkSourceConversationIds.add(parent.conversationId);
		this.#createdConversationIds.add(id);
		this.#writes.push({ type: "conversation", value: record });
		await this.#host.conversationCreated(this, record);
		this.#assertOpen();
		return record;
	}

	appendEntry(conversationId: ConversationId, value: EntryDraft): Promise<EntryRecord>;
	appendEntry<D extends JsonValue>(
		token: Entry<D>,
		conversationId: ConversationId,
		value: TypedEntryDraft<NoInfer<D>>,
	): Promise<TypedEntry<D>>;
	appendEntry(
		first: ConversationId | { readonly kind: string },
		second: ConversationId | EntryDraft,
		third?: object,
	): Promise<EntryRecord> {
		if (typeof first === "number") return this.#appendEntry(first, second as EntryDraft);
		return this.#appendEntry(second as ConversationId, { ...third, kind: first.kind } as EntryDraft);
	}

	#appendEntry(conversationId: ConversationId, value: EntryDraft): Promise<EntryRecord> {
		return this.#write(async () => {
			await this.#requireConversation(conversationId);
			this.#assertOpen();
			const id = await this.#host.storage.mintId<EntryId>();
			this.#assertOpen();
			const { head, ...rest } = value;
			// Undefined `head` and `byTaskId` are omitted by the copy.
			const record = copyJson(
				{
					...rest,
					id,
					conversationId,
					head: head === "self" ? id : head,
					byTaskId: this.#scope.taskId,
				},
				TABLE_JSON_COPY_OPTIONS,
			) as unknown as EntryRecord;
			this.#writes.push({ type: "entry", value: record });
			return record;
		});
	}

	createTask<I, S extends { phase: string }, R, H extends object>(
		task: Task<I, S, R, H>,
		input: I,
		options: TaskCreationOptions,
	): Promise<TaskId<R>> {
		return this.#write(async () => {
			if (options.submissionIds !== undefined) {
				let trace: TaskOptions["trace"];
				let links: TelemetryCarrier[] | undefined;
				for (let index = 0; index < options.submissionIds.length; index++) {
					const id = options.submissionIds[index]!;
					const submission = this.#submissions.get(id) ?? (await this.#host.storage.submission(id, this.#context));
					if (index === 0) trace = submission?.trace ?? "root";
					else if (submission?.trace !== undefined) {
						links ??= [];
						links.push(submission.trace);
					}
				}
				options = { ...options, trace, links };
			}
			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;
			if (conversationId === undefined) throw new TypeError("Tx.createTask() requires options.conversationId");
			await this.#requireConversation(conversationId);
			this.#assertOpen();
			const definition = task.definition;
			const 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;
			this.#tasksById.set(id, { write: { kind: "create", record } });
			return id;
		});
	}

	/** Create a raw submission record with a fresh ID; no admission rules apply. */
	createSubmission(create: SubmissionCreate): Promise<SubmissionRecord> {
		return this.#write(async () => {
			await this.#requireConversation(create.conversationId);
			this.#assertOpen();
			const id = await this.#host.storage.mintId<SubmissionId>();
			this.#assertOpen();
			const record = copyJson({ ...create, id }, TABLE_JSON_COPY_OPTIONS) as unknown as SubmissionRecord;
			this.#submissions.set(id, record);
			return record;
		});
	}

	/**
	 * Settle a submission. Resolved during assembly against the transaction's latest candidate record, falling back to
	 * committed state, so it is not a caller table read and works after the first table write.
	 */
	settleSubmission(id: SubmissionId, settlement: SubmissionSettlement): void {
		this.#assertOpen();
		this.#hasTableWrite = true;
		this.#submissionChanges.push({ id, change: copyJson(settlement) as SubmissionSettlement });
	}

	/** Place a queued submission at `entry`; resolved during assembly like `settleSubmission()`. */
	placeSubmission(id: SubmissionId, entry: EntryId): void {
		this.#assertOpen();
		this.#hasTableWrite = true;
		this.#submissionChanges.push({ id, change: { status: "placed", entry } });
	}

	/** Internal: replace one task record completely. Tasks change their own state through their runtime. */
	setTask(value: AnyTaskRecord): void {
		this.#assertOpen();
		this.#hasTableWrite = true;
		const task = this.#taskEntry(value.id);
		const candidate = task.write?.record;
		if (candidate?.state.status === "terminal") {
			throw new Error(`Task ${value.id} already has a terminal candidate`);
		}
		if (candidate !== undefined && candidate.conversationId !== value.conversationId) {
			throw new Error(`Task ${value.id} cannot change conversations`);
		}
		task.write = {
			kind: task.write?.kind === "create" ? "create" : "replace",
			record: copyJson(this.#stampTimes(value, candidate), TABLE_JSON_COPY_OPTIONS) as unknown as AnyTaskRecord,
		};
	}

	/** Internal: candidate records of the tasks this transaction created or replaced so far. */
	stagedTasks(): AnyTaskRecord[] {
		const records: AnyTaskRecord[] = [];
		for (const task of this.#tasksById.values()) if (task.write !== undefined) records.push(task.write.record);
		return records;
	}

	/** Internal: IDs of the tasks this transaction creates. */
	createdTaskIds(): TaskId[] {
		const ids: TaskId[] = [];
		for (const [id, task] of this.#tasksById) if (task.write?.kind === "create") ids.push(id);
		return ids;
	}

	/** Internal: conversations this transaction created or forked so far. */
	stagedConversations(): ConversationRecord[] {
		const records: ConversationRecord[] = [];
		for (const write of this.#writes) if (write.type === "conversation") records.push(write.value);
		return records;
	}

	// ─── Documents ──────────────────────────────────────────────────────────
	doc<T extends JsonObject>(token: SessionDocToken<T>): Promise<Draft<T>>;
	doc<T extends JsonObject>(token: ConversationDocToken<T>, conversationId: ConversationId): Promise<Draft<T>>;
	doc<T extends JsonObject>(token: TaskDocToken<T>, taskId: TaskId): Promise<Draft<T>>;
	doc<T extends JsonObject, I extends JsonValue>(
		token: SessionDocFamilyToken<T, I>,
		key: string,
		seed: I,
	): Promise<Draft<T>>;
	doc<T extends JsonObject, I extends JsonValue>(
		token: ConversationDocFamilyToken<T, I>,
		conversationId: ConversationId,
		key: string,
		seed: I,
	): Promise<Draft<T>>;
	doc<T extends JsonObject, I extends JsonValue>(
		token: TaskDocFamilyToken<T, I>,
		taskId: TaskId,
		key: string,
		seed: I,
	): Promise<Draft<T>>;
	doc(token: AnyDocToken, ...args: readonly unknown[]): Promise<Draft<JsonObject>> {
		try {
			this.#assertOpen();
			const definition = token.definition;
			const resolved = resolveAddress(definition, args);
			this.#assertTaskDocumentsOpen(resolved);
			const latest = this.#latestDocumentByAddress.get(resolved.id);
			if (latest !== undefined && !latest.retireOnCommit) {
				if (latest.draftPromise !== undefined) return latest.draftPromise;
				if (latest.target?.kind === "fork-copy") {
					latest.draftPromise = this.#track(this.#acquireForkCopy(latest, definition, latest.target));
					return latest.draftPromise;
				}
			}
			const seed = definition.family === true ? copyJson(args[resolved.nextArgument]) : undefined;
			const docEntry: DocumentEntry = {
				addressId: resolved.id,
				address: resolved.address,
				definition,
				retireOnCommit: false,
			};
			this.#documents.push(docEntry);
			this.#latestDocumentByAddress.set(docEntry.addressId, docEntry);
			// Capture retirement before awaiting so a pending old acquisition and its replacement stay distinct.
			docEntry.draftPromise = this.#track(this.#acquire(docEntry, seed, latest?.retireOnCommit === true));
			return docEntry.draftPromise;
		} catch (error) {
			return Promise.reject(error);
		}
	}

	retireDoc<T extends JsonObject>(token: SessionDocToken<T>): Promise<void>;
	retireDoc<T extends JsonObject>(token: ConversationDocToken<T>, conversationId: ConversationId): Promise<void>;
	retireDoc<T extends JsonObject>(token: TaskDocToken<T>, taskId: TaskId): Promise<void>;
	retireDoc<T extends JsonObject, I extends JsonValue>(token: SessionDocFamilyToken<T, I>, key: string): Promise<void>;
	retireDoc<T extends JsonObject, I extends JsonValue>(
		token: ConversationDocFamilyToken<T, I>,
		conversationId: ConversationId,
		key: string,
	): Promise<void>;
	retireDoc<T extends JsonObject, I extends JsonValue>(
		token: TaskDocFamilyToken<T, I>,
		taskId: TaskId,
		key: string,
	): Promise<void>;
	retireDoc(token: AnyDocToken, ...args: readonly unknown[]): Promise<void> {
		try {
			this.#assertOpen();
			const definition = token.definition;
			const resolved = resolveAddress(definition, args);
			const latest = this.#latestDocumentByAddress.get(resolved.id);
			if (latest?.retireOnCommit) return Promise.resolve();
			if (latest?.target?.kind === "fork-copy") {
				checkRecordScope(definition, latest.target.record);
				latest.retireOnCommit = true;
				return Promise.resolve();
			}
			if (latest?.draftPromise !== undefined) {
				// Retirement of an acquired draft persists its final content before retirement.
				latest.retireOnCommit = true;
				return this.#track(latest.draftPromise.then(() => undefined));
			}
			const entry: DocumentEntry = {
				addressId: resolved.id,
				address: resolved.address,
				definition,
				retireOnCommit: true,
			};
			this.#documents.push(entry);
			this.#latestDocumentByAddress.set(entry.addressId, entry);
			return this.#track(this.#findRetirement(entry));
		} catch (error) {
			return Promise.reject(error);
		}
	}

	async #acquire(entry: DocumentEntry, seed: JsonValue | undefined, skipLoad: boolean): Promise<Draft<JsonObject>> {
		const definition = entry.definition!;
		const loaded = skipLoad
			? undefined
			: await this.#host.load(definition, entry.addressId, entry.address, this.#context);
		this.#assertOpen();
		if (loaded !== undefined) {
			checkRecordScope(definition, loaded.record);
			checkRecordVersion(definition, loaded.record, loaded.storedVersion);
			entry.target = { kind: "loaded", document: loaded };
			entry.change = loaded.tracker.beginChange();
			return entry.change.state;
		}
		const scope = entry.address.scope;
		if (scope.kind === "conversation") await this.#requireConversation(scope.conversationId);
		if (scope.kind === "task") {
			const task = await this.#currentTask(scope.taskId);
			if (task === undefined) throw new Error(`Task ${scope.taskId} does not exist`);
			if (task.state.status === "terminal") throw new Error(`Task ${scope.taskId} is terminal`);
		}
		this.#assertOpen();
		const value = copyJson(
			definition.family === true ? definition.initial(seed) : definition.initial(),
		) as JsonObject;
		const id = await this.#host.storage.mintId<DocumentId>();
		this.#assertOpen();
		const tracker = track(value);
		entry.target = {
			kind: "created",
			record: documentCreate(definition, entry.address, id),
			version: definition.version,
			tracker,
		};
		entry.change = tracker.beginChange();
		return entry.change.state;
	}

	async #acquireForkCopy(
		entry: DocumentEntry,
		definition: AnyDocDefinition,
		target: Extract<DocumentTarget, { readonly kind: "fork-copy" }>,
	): Promise<Draft<JsonObject>> {
		const stored = await this.#host.storage.document(target.source.id, target.source.at, this.#context);
		this.#assertOpen();
		if (stored === undefined) {
			throw new Error(`Fork source document ${target.source.id} cannot be read`);
		}
		if (
			stored.record.scope.kind !== "conversation" ||
			stored.record.kind !== target.record.kind ||
			stored.record.key !== target.record.key ||
			stored.record.history !== target.record.history ||
			stored.record.fork !== target.record.fork
		) {
			throw new Error(`Fork source document ${target.source.id} does not match the copied record`);
		}
		const value = materializeDocumentValue(definition, target.record, stored.version, stored.value);
		const tracker = track(value);
		entry.definition = definition;
		entry.target = { kind: "created", record: target.record, version: definition.version, tracker };
		entry.change = tracker.beginChange();
		return entry.change.state;
	}

	async #findRetirement(entry: DocumentEntry): Promise<void> {
		const gate = this.#host.touch(documentKey(entry.addressId));
		if (gate !== undefined) await gate;
		const record =
			this.#host.cached(entry.addressId)?.record ??
			(await this.#host.storage.findDocument(entry.address, "current", this.#context));
		this.#assertOpen();
		if (record === undefined) return;
		checkRecordScope(entry.definition!, record);
		entry.target = { kind: "retire-only", record };
	}

	// ─── Settlement ─────────────────────────────────────────────────────────

	/** Seal after callback failure: abort every change and observe every pending operation. */
	async settleFailure(): Promise<void> {
		this.#sealed = true;
		this.#abortChanges();
		await Promise.allSettled(this.#pendingOperations);
		this.#context = BACKGROUND_CONTEXT;
	}

	/**
	 * Seal after callback success, prepare every open change, and assemble the atomic batch.
	 * Any failure aborts every change before Storage admission.
	 */
	async settleSuccess(): Promise<readonly StorageWrite[]> {
		this.#sealed = true;
		if (this.#pendingOperations.size > 0) {
			this.#abortChanges();
			await Promise.allSettled(this.#pendingOperations);
			this.#context = BACKGROUND_CONTEXT;
			throw new Error("Session commit callback settled before its pending Tx operations");
		}
		try {
			// Synchronously prepare every open change; this revokes every draft.
			for (const document of this.#documents) {
				if (document.change !== undefined) document.prepared = document.change.prepare();
			}
			return await this.#assemble();
		} catch (error) {
			this.#abortChanges();
			throw error;
		} finally {
			this.#context = BACKGROUND_CONTEXT;
		}
	}

	/** Abort every prepared change after Storage failure or when no write is required. */
	discard(): void {
		this.#abortChanges();
	}

	/**
	 * Internal: what a deferred commit holds in memory instead of storing (`CommitPersistence`), each changed incarnation
	 * with its operations and whether they selected a base; `undefined` when the commit must be stored. It may be held
	 * only when every write changes an existing `latest`, `initial`-fork conversation document at its stored version.
	 */
	holdable(): { readonly loaded: LoadedDocument; readonly ops: readonly Op[]; readonly base: boolean }[] | undefined {
		if (this.#writes.some((write) => write.type !== "document.change")) return undefined;
		const held = [];
		for (const plan of this.#plans) {
			const { content, change } = plan;
			if (content === undefined && !plan.retire) continue;
			if (plan.retire || change?.loaded === undefined || content?.type !== "document.change") return undefined;
			const { loaded, prepared, version } = change;
			const record = loaded.record;
			if (record.scope.kind !== "conversation" || record.history !== "latest" || record.fork !== "initial") {
				return undefined;
			}
			// A migration stores a base at once.
			if (loaded.storedVersion !== version) return undefined;
			held.push({ loaded, ops: prepared.ops, base: content.content.kind === "base" });
		}
		return held;
	}

	/**
	 * Internal: adopt a held commit (`holdable()`) in memory only. Trackers advance by pointer swap; stored versions and
	 * delta counts stay, as nothing was stored. Returns its publication for the Session's own observers.
	 */
	adoptHeld(): DocumentCommitChange[] {
		const publications: DocumentCommitChange[] = [];
		for (const plan of this.#plans) {
			// Every plan of a held commit changes or reads a loaded incarnation (`holdable()`).
			const { tracker, prepared, version, loaded } = plan.change!;
			if (prepared.ops.length === 0) {
				prepared.abort();
				continue;
			}
			tracker.adopt(prepared);
			const record = loaded!.record;
			const conversationId = plan.conversationId;
			publications.push({
				type: "document",
				kind: "change",
				record,
				conversationId,
				version,
				value: prepared.value,
				ops: prepared.ops,
			});
		}
		return publications;
	}

	/**
	 * Internal: store the held changes of every incarnation this commit reads or writes in front of its own, so a stored
	 * document never misses an operation its later ones build on. Called right before Storage admission, after every
	 * fallible step, because it empties the holds: a failure earlier keeps them. Returns the holds it took.
	 */
	takeHeld<H extends HeldDocument>(held: (id: DocumentId) => H | undefined): H[] {
		const taken: H[] = [];
		for (const plan of this.#plans) {
			// A new incarnation holds nothing yet.
			if (!("createdAt" in plan.record)) continue;
			const hold = held(plan.record.id);
			if (hold === undefined) continue;
			taken.push(hold);
			const own = plan.content?.type === "document.change" ? plan.content : undefined;
			const change = plan.change;
			const value = change?.prepared.value ?? hold.loaded.tracker.value;
			const stored = storeHeld(hold, change?.prepared.ops ?? EMPTY_OPERATIONS, value, hold.loaded.valueVersion);
			// A base this commit selected or must write covers what was held.
			const content = own?.content.kind === "base" ? own.content : stored.content;
			const write = { type: "document.change", id: plan.record.id, content } as const;
			if (own !== undefined) this.#writes[this.#writes.indexOf(own)] = write;
			else this.#insertBeforeRetirement(write);
			if (!plan.retire) plan.held = { ops: stored.ops, observed: own !== undefined };
			plan.content = write;
			// Held documents are conversation documents; one this commit only read has no owner resolved yet.
			if (plan.record.scope.kind === "conversation") plan.conversationId = plan.record.scope.conversationId;
		}
		return taken;
	}

	/**
	 * Internal: the keys this commit writes, once settled (and after `takeHeld()`), for the Session's unpublished write
	 * set: its table records, and the address and incarnation of every document it creates, changes, or retires, with
	 * the scope of every incarnation it creates or retires.
	 */
	writeKeys(): string[] {
		const keys: string[] = [];
		tableKeys(this.#writes, keys);
		for (const plan of this.#plans) {
			if (plan.content === undefined && !plan.retire) continue;
			keys.push(documentKey(plan.addressId), incarnationKey(plan.record.id));
			if (plan.retire || !("createdAt" in plan.record)) keys.push(scopeKey(plan.record.scope));
		}
		return keys;
	}

	/** Stage a content write for an incarnation, before its retirement when it retires. */
	#insertBeforeRetirement(write: Extract<StorageWrite, { readonly type: "document.change" }>): void {
		const at = this.#writes.findIndex(
			(candidate) => candidate.type === "document.retire" && candidate.id === write.id,
		);
		if (at < 0) this.#writes.push(write);
		else this.#writes.splice(at, 0, write);
	}

	/** Adopt every prepared change by pointer swap after Storage success and describe the publication. */
	adopt(seq: Seq): AdoptedDocuments {
		const publications: DocumentCommitChange[] = [];
		// Only when the commit stored held changes: then what Storage got differs from what memory adopted.
		let stored: DocumentCommitChange[] | undefined;
		for (const plan of this.#plans) {
			const committed = "createdAt" in plan.record;
			let record: DocumentRecord = committed ? (plan.record as DocumentRecord) : { ...plan.record, createdAt: seq };
			if (plan.retire) record = { ...record, retiredAt: seq };
			const change = plan.change;
			if (change !== undefined) {
				const { tracker, prepared, loaded, version } = change;
				// A new incarnation is adopted unless it retires in the same commit; a loaded one only when it changed.
				if (loaded === undefined ? !plan.retire : prepared.ops.length > 0) tracker.adopt(prepared);
				else prepared.abort();
				if (loaded !== undefined) {
					if (loaded.storedVersion < version) loaded.storedVersion = version;
					if (plan.content?.type === "document.change") {
						if (plan.content.content.kind === "base") loaded.deltasSinceBase = 0;
						else loaded.deltasSinceBase++;
					}
				} else if (!plan.retire) {
					this.#host.install({
						addressId: plan.addressId,
						record,
						storedVersion: version,
						valueVersion: version,
						deltasSinceBase: 0,
						tracker,
					});
				}
			}
			if (plan.retire && committed) this.#host.evict(plan.addressId, record.id);
			const publication = describe(plan, record);
			if (publication === undefined) continue;
			const held = plan.held;
			if (held === undefined) {
				publications.push(publication);
				stored?.push(publication);
				continue;
			}
			stored ??= [...publications];
			if (held.observed) publications.push(publication);
			// Only a document change of a loaded incarnation carries held operations; a retirement is published alone.
			stored.push(
				publication.type === "document" && publication.kind === "change"
					? { ...publication, ops: held.ops }
					: publication,
			);
		}
		return { observed: publications, stored: stored ?? publications };
	}

	async #assemble(): Promise<StorageWrite[]> {
		const storage = this.#host.storage;
		const plans = this.#plans;
		for (const document of this.#documents) {
			const plan = planDocument(document);
			if (plan !== undefined) plans.push(plan);
		}
		this.#rejectForkSourceWrites(plans);
		await this.#validateOwners();
		for (const [id, task] of this.#tasksById) {
			if (task.write?.kind !== "replace") continue;
			const committed = await this.#committedTask(id);
			if (committed === undefined) throw new Error(`Task ${id} does not exist`);
			if (committed.state.status === "terminal") throw new Error(`Task ${id} is already terminal`);
			if (committed.conversationId !== task.write.record.conversationId) {
				throw new Error(`Task ${id} cannot change conversations`);
			}
		}

		// Terminal settlement retires every task document, including ones created by this transaction.
		const terminalTaskIds = new Set<TaskId>();
		for (const task of this.#tasksById.values()) {
			if (task.write?.record.state.status === "terminal") terminalTaskIds.add(task.write.record.id);
		}
		if (terminalTaskIds.size > 0) {
			const retiring = new Set<DocumentId>();
			for (const plan of plans) {
				const scope = plan.record.scope;
				if (scope.kind !== "task" || !terminalTaskIds.has(scope.taskId)) continue;
				plan.retire = true;
				retiring.add(plan.record.id);
			}
			const retire = (record: DocumentRecord): void => {
				if (retiring.has(record.id)) return;
				plans.push({ addressId: addressId(record), record, retire: true });
				retiring.add(record.id);
			};
			for (const taskId of terminalTaskIds) {
				if (this.#tasksById.get(taskId)?.write?.kind === "create") continue;
				// The Session knows the documents of the tasks it created; any other task's are scanned.
				const gate = this.#host.touch(scopeKey({ kind: "task", taskId }));
				if (gate !== undefined) await gate;
				const known = this.#host.taskDocuments(taskId);
				if (known !== undefined) {
					for (const record of known) retire(record);
					continue;
				}
				let cursor: Cursor | undefined;
				do {
					const page = await storage.scanDocuments(
						{ scope: { kind: "task", taskId }, at: "current" },
						INTERNAL_SCAN_PAGE_SIZE,
						cursor,
						this.#context,
					);
					for (const record of page.items) retire(record);
					cursor = page.next;
				} while (cursor !== undefined);
			}
		}

		// Resolve publication ownership before Storage admission so adoption remains synchronous.
		for (const plan of plans) {
			if (!publishes(plan)) continue;
			const scope = plan.record.scope;
			if (scope.kind === "conversation") plan.conversationId = scope.conversationId;
			if (scope.kind !== "task") continue;
			const task = this.#taskEntry(scope.taskId);
			if (task.publicationConversationId === undefined) {
				const current = await this.#currentTask(scope.taskId);
				if (current !== undefined) task.publicationConversationId = current.conversationId;
			}
			plan.conversationId = task.publicationConversationId;
		}

		for (const { id, change } of this.#submissionChanges) {
			const current = this.#submissions.get(id) ?? (await storage.submission(id, this.#context));
			if (current === undefined) throw new Error(`Submission ${id} does not exist`);
			const next = applySubmissionChange(current, change);
			if (next !== current) this.#submissions.set(id, next);
		}

		const writes = this.#writes;
		for (const value of this.#submissions.values()) writes.push({ type: "submission", value });
		for (const task of this.#tasksById.values()) {
			if (task.write !== undefined) writes.push({ type: "task", value: task.write.record });
		}
		for (const plan of plans) {
			const change = plan.change;
			// Checkpoint predicates run last, after every validation.
			if (plan.content?.type === "document.change" && plan.content.content.kind === "delta" && change?.loaded) {
				const { prepared, version } = change;
				const info = { deltasSinceBase: change.loaded.deltasSinceBase };
				if (change.definition!.checkpointWhen(prepared.value, prepared.ops, info)) {
					plan.content = { ...plan.content, content: { version, kind: "base", value: prepared.value } };
				}
			}
			if (plan.content !== undefined) writes.push(plan.content);
			if (plan.retire) writes.push({ type: "document.retire", id: plan.record.id });
		}
		return writes;
	}

	// ─── Helpers ────────────────────────────────────────────────────────────

	/**
	 * New owned work needs a live owner, judged on the owner's final candidate: not `completing`, terminal, or
	 * abort-marked. A task therefore cannot create owned work in the commit that finishes it (spec §5.5).
	 */
	async #validateOwners(): Promise<void> {
		const owners: { readonly what: string; readonly taskId: TaskId }[] = [];
		for (const write of this.#writes) {
			if (write.type !== "conversation" || write.value.owner === undefined) continue;
			owners.push({ what: "Conversation owner task", taskId: write.value.owner.taskId });
		}
		for (const task of this.#tasksById.values()) {
			const record = task.write?.kind === "create" ? task.write.record : undefined;
			if (record?.owner !== undefined) owners.push({ what: "Task owner", taskId: record.owner });
		}
		for (const { what, taskId } of owners) {
			const task = await this.#currentTask(taskId);
			if (task === undefined) throw new Error(`${what} ${taskId} does not exist`);
			if (task.state.status === "terminal" || task.state.status === "completing") {
				throw new Error(`${what} ${taskId} is ${task.state.status}`);
			}
			if (task.abortRequested) throw new Error(`${what} ${taskId} is abort-marked`);
		}
	}

	#rejectForkSourceWrites(plans: readonly DocumentPlan[]): void {
		for (const plan of plans) {
			if (plan.content === undefined && !plan.retire) continue;
			const record = plan.record;
			if (this.#forkSourceDocumentIds.has(record.id)) {
				throw new Error(`Cannot change fork source document ${record.id} in the fork transaction`);
			}
			const scope = plan.record.scope;
			if (
				scope.kind === "conversation" &&
				this.#forkSourceConversationIds.has(scope.conversationId) &&
				record.scope.kind === "conversation" &&
				record.fork === "current"
			) {
				throw new Error(
					`Cannot fork conversation ${scope.conversationId} while changing its current-policy documents`,
				);
			}
		}
	}

	#abortChanges(): void {
		for (const document of this.#documents) document.change?.abort();
	}

	#assertOpen(): void {
		if (this.#sealed) throw new Error("Transaction has settled");
	}

	#assertTaskDocumentsOpen(resolved: ResolvedAddress): void {
		const scope = resolved.address.scope;
		if (scope.kind === "task" && this.#tasksById.get(scope.taskId)?.write?.record.state.status === "terminal") {
			throw new Error(`Task ${scope.taskId} is terminal`);
		}
	}

	/** Register an operation so callback settlement can reject and drain it. */
	#track<T>(operation: Promise<T>): Promise<T> {
		this.#pendingOperations.add(operation);
		const settle = (): void => {
			this.#pendingOperations.delete(operation);
		};
		operation.then(settle, settle);
		return operation;
	}

	#read<T>(method: string, read: () => Promise<T>): Promise<T> {
		try {
			this.#assertOpen();
			if (this.#hasTableWrite) throw new ReadAfterWrite(`Tx.${method}()`);
			return this.#track(read());
		} catch (error) {
			return Promise.reject(error);
		}
	}

	#write<T>(write: () => Promise<T>): Promise<T> {
		try {
			this.#assertOpen();
			this.#hasTableWrite = true;
			return this.#track(write());
		} catch (error) {
			return Promise.reject(error);
		}
	}

	async #requireConversation(id: ConversationId): Promise<void> {
		if (this.#createdConversationIds.has(id)) return;
		if ((await this.#host.storage.conversation(id, this.#context)) === undefined) {
			throw new Error(`Conversation ${id} does not exist`);
		}
	}

	/**
	 * Lifecycle times: `startedAt` on the first change to `running`, `endedAt` on the change to `terminal`. Once set,
	 * they carry over from the replaced record; records written before they existed lack them.
	 */
	#stampTimes(value: AnyTaskRecord, candidate: AnyTaskRecord | undefined): AnyTaskRecord {
		const status = value.state.status;
		const now = (stamps: boolean) => (stamps ? this.#host.now() : undefined);
		const startedAt = candidate?.startedAt ?? value.startedAt ?? now(status === "running");
		const endedAt = candidate?.endedAt ?? value.endedAt ?? now(status === "terminal");
		if (startedAt === value.startedAt && endedAt === value.endedAt) return value;
		return {
			...value,
			...(startedAt === undefined ? {} : { startedAt }),
			...(endedAt === undefined ? {} : { endedAt }),
		};
	}

	#taskEntry(id: TaskId): TransactionTask {
		let task = this.#tasksById.get(id);
		if (task === undefined) {
			task = {};
			this.#tasksById.set(id, task);
		}
		return task;
	}

	/** Latest candidate task record, falling back to committed state; not a caller table read. */
	async #currentTask(id: TaskId): Promise<AnyTaskRecord | undefined> {
		return this.#tasksById.get(id)?.write?.record ?? (await this.#committedTask(id));
	}

	#committedTask(id: TaskId): Promise<AnyTaskRecord | undefined> {
		const task = this.#taskEntry(id);
		task.committedRead ??= this.#host.storage.task(id, this.#context);
		return task.committedRead;
	}
}

/** Plan of one staged document: its record, content write, and prepared change. Retirement is decided later. */
function planDocument(document: DocumentEntry): DocumentPlan | undefined {
	const target = document.target;
	if (target === undefined) return undefined;
	const plan = { addressId: document.addressId, retire: document.retireOnCommit };
	switch (target.kind) {
		case "created": {
			const prepared = document.prepared!;
			const { record, version, tracker } = target;
			const content = { version, kind: "base", value: prepared.value } as const;
			return {
				...plan,
				record,
				content: { type: "document.create", record, content },
				change: { tracker, prepared, version },
			};
		}
		case "fork-copy":
			return {
				...plan,
				record: target.record,
				content: { type: "document.copy", record: target.record, source: target.source },
			};
		case "retire-only":
			return { ...plan, record: target.record };
		case "loaded": {
			const loaded = target.document;
			const definition = document.definition!;
			const prepared = document.prepared!;
			const version = definition.version;
			const id = loaded.record.id;
			// A version change stores a base even without operations; otherwise only a change stores a delta.
			const content =
				loaded.storedVersion < version
					? ({ type: "document.change", id, content: { version, kind: "base", value: prepared.value } } as const)
					: prepared.ops.length > 0
						? ({ type: "document.change", id, content: { version, kind: "delta", ops: prepared.ops } } as const)
						: undefined;
			const change = {
				tracker: loaded.tracker,
				prepared,
				version,
				loaded,
				definition,
			};
			return { ...plan, record: loaded.record, change, ...(content === undefined ? {} : { content }) };
		}
	}
}

/** The publication of one adopted plan, if it publishes (`publishes()`), with `record` as committed. */
function describe(plan: DocumentPlan, record: DocumentRecord): DocumentCommitChange | undefined {
	const conversationId = plan.conversationId;
	if (plan.retire) return { type: "document", kind: "retire", record, conversationId };
	if (plan.content?.type === "document.copy") {
		return { type: "document.copy", record, conversationId: conversationId!, source: plan.content.source };
	}
	const change = plan.change;
	if (change === undefined || !publishes(plan)) return undefined;
	const ops = change.loaded === undefined ? EMPTY_OPERATIONS : change.prepared.ops;
	return {
		type: "document",
		kind: "change",
		record,
		conversationId,
		version: change.version,
		value: change.prepared.value,
		ops,
	};
}

/**
 * Take a hold's changes after `ops`, a commit's own on top of them, as one stored content for an incarnation now at
 * `value`, and the operations a reader of the stored state applies to reach it: the folded ones, or a root replacement
 * after the queue overflowed, which stores a base, as does a hold whose commits selected one.
 */
export function storeHeld(
	hold: HeldDocument,
	ops: readonly Op[],
	value: JsonObject,
	version: number,
): { readonly content: DocumentContent; readonly ops: readonly Op[] } {
	hold.pending.push(ops);
	const taken = hold.pending.take();
	if (taken.kind === "replace") return { content: { version, kind: "base", value }, ops: [["r", value]] };
	if (hold.base) return { content: { version, kind: "base", value }, ops: taken.ops };
	return { content: { version, kind: "delta", ops: taken.ops }, ops: taken.ops };
}

/**
 * Whether adoption publishes the plan: every creation, copy, and retirement, and a loaded incarnation that writes
 * content, which includes a migration-only base so observers of the older shape receive the new value.
 */
function publishes(plan: DocumentPlan): boolean {
	return plan.retire || plan.change?.loaded === undefined || plan.content !== undefined;
}