Durable: session, commit, document và storage

/** * Session kernel: one mutation line, the loaded document tracker cache, and committed publication. * * The cache is bounded by use: a document with no watch or document state attached leaves it once it was not read or * written for the retention period, and the next access loads it from Storage. Eviction runs as a job on the mutation * line, so no transaction ever holds a tracker that leaves. * * Only committed state is observable. Every commit callback, preparation, Storage settlement, adoption, and * publication enqueue runs while the line is held; listeners run later. * * Storage failures are final. Every Storage call goes through one guard: the first error a Storage method throws, other * than for a cancelled caller, fails the Session, as does a commit it cannot adopt. That call gets the error; every later * call and every Storage call still underway gets `SessionFailed`; close listeners run at once with the error, and the * Session then closes itself, Storage included, and `closed` settles. Nothing retries: transient errors are the * Storage's to retry. Reopening recovers from what was committed. */export class SessionImpl implements Session {

#enqueue<T>(job: () => Promise<T>): Promise<T> {	const run = this.#tail.then(job);	this.#tail = run.then(		() => undefined,		() => undefined,	);	return run;}

commitWith<T>(change: (tx: Transaction) => T | Promise<T>, context: Context, scope?: TransactionScope): Promise<T> {	try {		this.assertUsable();	} catch (error) {		return Promise.reject(error);	}	return this.#enqueue(() => this.#runCommit(change, context, scope));}/** Internal: run a read-only job on the mutation line so multi-read derivations observe one committed state. */readOnLine<T>(job: () => Promise<T>): Promise<T> {	try {		this.assertUsable();	} catch (error) {		return Promise.reject(error);	}	return this.#enqueue(async () => {		this.#assertHealthy();		return job();	});}

async #runCommit<T>(	change: (tx: Transaction) => T | Promise<T>,	context: Context,	scope?: TransactionScope,): Promise<T> {	this.#assertHealthy();	context.abortSignal?.throwIfAborted();	const tx = new Transaction(this.#host, context, scope);	let result: T;	try {		result = await change(tx);		// A callback that caught a failed read must not commit, nor succeed as if it had.		this.#assertHealthy();	} catch (error) {		await tx.settleFailure();		this.#sweepDocuments();		throw error;	}	let writes: readonly StorageWrite[];	try {		writes = await tx.settleSuccess();	} catch (error) {		// Settlement aborted every change already.		this.#sweepDocuments();		throw error;	}	if (writes.length === 0) {		tx.discard();		this.#sweepDocuments();		return result;	}

	let seq: Seq;	try {		// Once admitted, caller cancellation does not interrupt Storage settlement.		seq = await this.#storage.commit(writes, withoutAbortSignal(context));	} catch (error) {		// The guard has failed the Session.		tx.discard();		throw error;	}	let documents: DocumentCommitChange[];	try {		documents = tx.adopt(seq);	} catch (error) {		// Storage already committed; a failed adoption leaves memory behind durable state.		this.fail(error);		throw error;	}	for (const document of documents) this.#touchDocument(addressId(document.record));	this.#publish(seq, writes, documents, context);	this.#sweepDocuments();	return result;}

	if (changes === undefined) return;	for (const document of documents) changes.push(document);	const publication: CommitPublication = { seq, changes };	// The commit is durable: a failing listener neither fails the commit nor skips the others.	for (const listener of [...this.#internalListeners]) {		contained(			() => listener(publication, context),			(error) => this.fail(error, context),		);	}	for (const listener of [...this.#commitListeners]) {		contained(			() => listener(publication, context),			(error) => this.report(error, LISTENER, context),		);	}}

/** One record or document mutation in an atomic storage commit. */export type StorageWrite =	| { readonly type: "conversation"; readonly value: ConversationRecord }	| { readonly type: "entry"; readonly value: EntryRecord }	| { readonly type: "task"; readonly value: TaskRecord<JsonValue, JsonValue, JsonValue> }	| { readonly type: "submission"; readonly value: SubmissionRecord }	| {			readonly type: "document.create";			readonly record: DocumentCreate;			readonly content: Extract<DocumentContent, { readonly kind: "base" }>;	  }	| { readonly type: "document.copy"; readonly record: DocumentCreate; readonly source: DocumentCopySource }	| {			readonly type: "document.change";			readonly id: DocumentId;			readonly content: DocumentContent;	  }	| { readonly type: "document.retire"; readonly id: DocumentId };

/** 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>();

#read<T>(method: string, read: () => Promise<T>): Promise<T> {	try {		this.#assertOpen();		if (this.#hasTableWrite) throw new ReadAfterWrite(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);	}}

#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;	});}

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

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;

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

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

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 }) };}

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,		});	}}

} else if (action.content !== undefined) {	const revision = { ...action.content, seq } as DocumentRevision;	if (revision.kind === "base" && isCurrentOnly(stored!.record)) stored!.revisions = [revision];	else stored!.revisions.push(revision);}if (action.retire && action.create === undefined) stored!.record = { ...stored!.record, retiredAt: seq };if (action.retire && isCurrentOnly(stored!.record)) stored!.revisions = [];

return this.#enqueue(async () => {	this.#assertHealthy();	const storedEntry = await this.#storage.entry(conversationId, at, context);	if (storedEntry === undefined) {		throw new Error(`Entry ${at} is not visible from conversation ${conversationId}`);	}	const address: DocumentAddress = {		...resolved.address,		scope: { kind: "conversation", conversationId: storedEntry.entry.conversationId },	};	const record = await this.#storage.findDocument(address, storedEntry.commitSeq, context);	if (record === undefined) return undefined;	const stored = await this.#storage.document(record.id, storedEntry.commitSeq, context);	if (stored === undefined) {		throw new Error(`Historical document ${record.id} (${record.kind}) cannot be read`);	}	return materializeDocument(definition, stored);});

private materializeDocument(id: DocumentId, at: DocumentPoint): StoredDocument | undefined {	const stored = this.state.documents.get(id);	if (stored === undefined) return undefined;	if (at !== "current" && isCurrentOnly(stored.record)) {		throw new StorageRequestError(`Document ${id} does not retain historical content`);	}	if (!isAliveAt(stored.record, at)) return undefined;	const revisions = at === "current" ? stored.revisions : stored.revisions.filter((revision) => revision.seq <= at);	let baseIndex = revisions.length - 1;	while (baseIndex >= 0 && revisions[baseIndex]!.kind !== "base") baseIndex--;	const base = revisions[baseIndex];	if (base?.kind !== "base") throw new Error(`Document ${id} is missing a required base`);	const value = applyImmutableBatches(		base.value,		documentDeltaBatches(id, base.version, revisions, baseIndex + 1),	) as JsonObject;	return { record: stored.record, version: base.version, value, deltasSinceBase: revisions.length - baseIndex - 1 };}

private *visibleEntries(	conversationId: ConversationId,	minEntryId: number = Number.NEGATIVE_INFINITY,	maxEntryId: number = Number.POSITIVE_INFINITY,): Generator<EntryRecord> {	if (!this.state.conversations.has(conversationId)) {		throw new StorageRequestError(`Unknown conversation: ${conversationId}`);	}	let currentId = conversationId;	let upperEntryId = maxEntryId;	while (true) {		const ids = this.state.entryIds.get(currentId) ?? [];		for (let index = upperBound(ids, upperEntryId) - 1; index >= 0; index--) {			const id = ids[index];			if (id < minEntryId) break;			yield this.state.entries.get(id)!;		}		const conversation = this.state.conversations.get(currentId)!;		if (conversation.parent === undefined) break;		upperEntryId = Math.min(upperEntryId, conversation.parent.at);		if (upperEntryId < minEntryId) break;		currentId = conversation.parent.conversationId;	}}

/** Select every persisted conversation document copied by one fork. */export async function prepareForkDocumentCopies(	storage: Storage,	parentConversationId: ConversationId,	at: EntryId,	childConversationId: ConversationId,	context: Context,): Promise<readonly ForkDocumentCopy[]> {	const entry = await storage.entry(parentConversationId, at, context);	if (entry === undefined) throw new Error(`Entry ${at} is not visible from conversation ${parentConversationId}`);	const copies: ForkDocumentCopy[] = [];	const copiedAddresses = new Set<string>();	await collectCopies(		storage,		{ kind: "conversation", conversationId: entry.entry.conversationId },		entry.commitSeq,		"asOf",		childConversationId,		copies,		copiedAddresses,		context,	);	await collectCopies(		storage,		{ kind: "conversation", conversationId: parentConversationId },		"current",		"current",		childConversationId,		copies,		copiedAddresses,		context,	);	return copies;}

#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`,			);		}	}}

const observed = { version: loaded.valueVersion };unsubscribeCommit = this.observeCommits((publication, context) => {	for (const change of publication.changes) {		if (change.type !== "document" || change.record.id !== loaded.record.id) continue;		// A document state's frames carry no caller cancellation; a watch observes its own cancellation.		const frameContext = observer instanceof CommittedStateSource ? withoutAbortSignal(context) : context;		const ops = observedOperations(observed, change);		// A migration-only base changes nothing for an observer of the new version.		if (ops.length === 0) continue;		// A retired incarnation never changes again: its observers keep their last value but pin nothing.		if (change.value === null) unpin();		observer.advance(change.value, ops, frameContext);	}});unsubscribeClose = this.subscribeClose(() => observer.closeSession(this.#failure));return { observer, detach };

async #drain(): Promise<void> {	if (this.#running || this.#end !== undefined || !this.#started) return;	this.#running = true;	try {		while (this.#end === undefined) {			const frame = this.#pending.shift();			if (frame === undefined) break;			this.#value = frame.value;			const deliveryContext = withoutAbortSignal(frame.context);			try {				await this.#listener!(frame.value, frame.ops, deliveryContext);			} catch (error) {				if (this.#end === undefined) {					this.#report(error);					this.#terminate({ reason: "listener_error", error: toError(error) });				}				break;			}			if (frame.value === null) {				this.#terminate({ reason: "retired" });				break;			}		}	} finally {		this.#running = false;		if (this.#end === undefined && this.#pending.length > 0) this.#schedule();	}}

export interface Storage {	/**	 * Atomically persist one batch and return its sequence. Once resolved, later reads through this storage observe it.	 */	commit(writes: readonly StorageWrite[], context: Context): Promise<Seq>;	/**	 * Return a fresh branded candidate from the Session-global numeric ID namespace, greater than every ID minted or stored	 * before, also across reopen: scans order by ID as creation order, and the Harness relies on it.	 */	mintId<I extends Id<string>>(): Promise<I>;	/** Look up one conversation by exact ID. */	conversation(id: ConversationId, context: Context): Promise<ConversationRecord | undefined>;	/** Scan conversations in `query.order` (default ascending) by ID. */	scanConversations(		query: ConversationQuery,		limit: number,		cursor: Cursor | undefined,		context: Context,	): Promise<Page<ConversationRecord, Cursor>>;⋯ lược bớt dòng 1201–1257	/** Resolve the incarnation occupying one exact logical address at the selected point. */	findDocument(address: DocumentAddress, at: DocumentPoint, context: Context): Promise<DocumentRecord | undefined>;	/** Materialize one specific incarnation by ID at the selected point without following a replacement at its address. */	document(id: DocumentId, at: DocumentPoint, context: Context): Promise<StoredDocument | undefined>;	/** Scan incarnations alive in one exact scope at the selected point. */	scanDocuments(		query: DocumentQuery,		limit: number,		cursor: Cursor | undefined,		context: Context,	): Promise<Page<DocumentRecord, Cursor>>;

/** Complete checkpoint or Chord operation batch selected by the owning Session. */export type DocumentContent =	| {			readonly version: number;			readonly kind: "base";			readonly value: JsonObject;	  }	| {			readonly version: number;			readonly kind: "delta";			readonly ops: readonly Op[];	  };/** Exact persisted source selected for a definition-free document copy. */export type DocumentCopySource = {	readonly id: DocumentId;	readonly at: DocumentPoint;};/** Detached materialized value and stored definition version at a selected point. */export type StoredDocument = {	readonly record: DocumentRecord;	readonly version: number;	readonly value: JsonObject;	/** Deltas replayed after the selected base to materialize `value`. */	readonly deltasSinceBase: number;};

export function scanStart(	requested: ScanOrder | undefined,	cursor: Readonly<Record<string, unknown>> | undefined,	fallback: ScanOrder,): ScanStart {	if (requested !== undefined && requested !== "ascending" && requested !== "descending") {		throw new StorageRequestError(`Invalid scan order: ${String(requested)}`);	}	if (cursor === undefined) return { order: requested ?? fallback, after: undefined };	const { after, order: stored } = cursor;	if (typeof after !== "number" || !Number.isSafeInteger(after))		throw new StorageRequestError("Invalid storage cursor");	if (stored !== undefined && stored !== "ascending" && stored !== "descending") {		throw new StorageRequestError("Invalid storage cursor");	}	const order = stored ?? fallback;	if (requested !== undefined && requested !== order) {		throw new StorageRequestError(`The cursor continues a ${order} scan; the query asks for ${requested}`);	}	return { order, after };}/** The continuation of a scan whose last returned item has `id`. */export function nextCursor(id: Id<string>, order: ScanOrder): { readonly after: number; readonly order: ScanOrder } {	return { after: id, order };}

async commit(writes: readonly StorageWrite[], _context: Context): Promise<Seq> {	return this.prepareCommit(writes).apply();}/** Validate and detach one commit without changing observable state. */prepareCommit(writes: readonly StorageWrite[], seq: Seq = seqFromNumber(this.nextSeq)): PreparedMemoryCommit {	this.assertOpen();	if (!Number.isSafeInteger(seq) || seq < this.nextSeq) {		throw new Error(`Commit sequence ${seq} does not strictly increase`);	}	const detachedWrites = freeze(this.resolveDocumentCopies(writes.map((write) => clone(write))));	this.checkGlobalIds(detachedWrites);	const documentActions = this.prepareDocumentActions(detachedWrites);	this.checkDocumentActions(documentActions);	let applied = false;	return {		seq,		writes: detachedWrites,		apply: () => {			if (!applied) {				applied = true;				this.applyPreparedCommit(detachedWrites, documentActions, seq);			}			return seq;		},	};}

const initialSchema: readonly string[] = [	`CREATE TABLE ${tables.durableMetadata} (		singleton INTEGER PRIMARY KEY CHECK (singleton = 1),		next_id TEXT NOT NULL,		next_seq INTEGER NOT NULL	) STRICT`,	`INSERT INTO ${tables.durableMetadata} (singleton, next_id, next_seq) VALUES (1, '2', 1)`,	`CREATE TABLE ${tables.recordIds} (		id INTEGER PRIMARY KEY,		record_type TEXT NOT NULL CHECK (record_type IN ('conversation', 'entry', 'task', 'submission', 'document'))	) STRICT`,	`CREATE TABLE ${tables.conversations} (		id INTEGER PRIMARY KEY,		owner_conversation_id INTEGER,		owner_task_id INTEGER,		record TEXT NOT NULL CHECK (json_valid(record))	) STRICT`,	`CREATE INDEX ${indexes.conversationsByOwnerConversation} ON ${tables.conversations} (owner_conversation_id, id)`,	`CREATE INDEX ${indexes.conversationsByOwnerTask} ON ${tables.conversations} (owner_task_id, id)`,	`CREATE TABLE ${tables.entries} (		id INTEGER PRIMARY KEY,		conversation_id INTEGER NOT NULL,		head INTEGER,		commit_seq INTEGER NOT NULL,		record TEXT NOT NULL CHECK (json_valid(record))	) STRICT`,	`CREATE INDEX ${indexes.entriesByConversation} ON ${tables.entries} (conversation_id, id DESC)`,	`CREATE INDEX ${indexes.entryHeadsByConversation} ON ${tables.entries} (conversation_id, id DESC) WHERE head IS NOT NULL`,

`CREATE TABLE ${tables.tasks} (	id INTEGER PRIMARY KEY,	conversation_id INTEGER NOT NULL,	kind TEXT NOT NULL,	status TEXT NOT NULL CHECK (status IN ('pending', 'running', 'waiting', 'completing', 'terminal')),	abort_requested INTEGER NOT NULL CHECK (abort_requested IN (0, 1)),	background INTEGER NOT NULL CHECK (background IN (0, 1)),	record TEXT NOT NULL CHECK (json_valid(record))) STRICT`,`CREATE INDEX ${indexes.tasksByStatus} ON ${tables.tasks} (status, id)`,`CREATE INDEX ${indexes.tasksByConversation} ON ${tables.tasks} (conversation_id, id)`,`CREATE INDEX ${indexes.tasksByKind} ON ${tables.tasks} (kind, id)`,`CREATE INDEX ${indexes.tasksByAbortRequested} ON ${tables.tasks} (abort_requested, id)`,`CREATE INDEX ${indexes.tasksByBackground} ON ${tables.tasks} (background, id)`,`CREATE TABLE ${tables.submissions} (	id INTEGER PRIMARY KEY,	conversation_id INTEGER NOT NULL,	request_id TEXT,	status TEXT NOT NULL CHECK (status IN ('queued', 'placed', 'done', 'unanswered')),	record TEXT NOT NULL CHECK (json_valid(record))) STRICT`,`CREATE INDEX ${indexes.submissionsByRequest} ON ${tables.submissions} (conversation_id, request_id)`,`CREATE INDEX ${indexes.submissionsByConversation} ON ${tables.submissions} (conversation_id, id)`,`CREATE INDEX ${indexes.submissionsByStatus} ON ${tables.submissions} (status, id)`,

`CREATE TABLE ${tables.documents} (	id INTEGER PRIMARY KEY,	kind TEXT NOT NULL,	family INTEGER NOT NULL CHECK (family IN (0, 1)),	key_value TEXT NOT NULL,	scope_kind TEXT NOT NULL CHECK (scope_kind IN ('session', 'conversation', 'task')),	owner_id INTEGER NOT NULL,	created_at INTEGER NOT NULL,	retired_at INTEGER,	record TEXT NOT NULL CHECK (json_valid(record))) STRICT`,`CREATE INDEX ${indexes.documentsByAddress}	ON ${tables.documents} (kind, scope_kind, owner_id, family, key_value, created_at DESC, retired_at)`,`CREATE INDEX ${indexes.documentsByScope} ON ${tables.documents} (scope_kind, owner_id, id)`,`CREATE INDEX ${indexes.documentsByScopeKind} ON ${tables.documents} (scope_kind, owner_id, kind, id)`,`CREATE TABLE ${tables.documentRevisions} (	document_id INTEGER NOT NULL,	seq INTEGER NOT NULL,	kind TEXT NOT NULL CHECK (kind IN ('base', 'delta')),	version INTEGER NOT NULL,	content TEXT NOT NULL CHECK (json_valid(content)),	PRIMARY KEY (document_id, seq)) STRICT`,`CREATE INDEX ${indexes.documentRevisionsByKind} ON ${tables.documentRevisions} (document_id, kind, seq DESC)`,

await database.transaction(async (transaction) => {	await transaction.exec(`CREATE TABLE IF NOT EXISTS ${table} (		singleton INTEGER PRIMARY KEY CHECK (singleton = 1),		version INTEGER NOT NULL CHECK (version >= 0)	) STRICT`);	await transaction.run(`INSERT OR IGNORE INTO ${table} (singleton, version) VALUES (1, 0)`);	const row = await transaction.get<SchemaRow>(`SELECT version FROM ${table} WHERE singleton = 1`);	if (row === undefined) throw new Error("SQLite schema metadata is missing");	const currentVersion = migrations.at(-1)?.version ?? 0;	if (row.version > currentVersion) {		throw new Error(`SQLite schema version ${row.version} is newer than supported version ${currentVersion}`);	}	for (const migration of migrations) {		if (migration.version <= row.version) continue;		for (const statement of migration.statements) await transaction.exec(statement);		await transaction.run(`UPDATE ${table} SET version = ? WHERE singleton = 1`, migration.version);	}});

async commit(writes: readonly StorageWrite[], _context: Context): Promise<Seq> {	this.assertOpen();	const documentActions = this.prepareDocumentActions(writes);	const candidateNextId = this.candidateNextId(writes);	const seq = await this.db.transaction(async (transaction) => {		const metadata = await transaction.get<MetadataRow>(			`SELECT next_id, next_seq FROM ${this.schema.tables.durableMetadata} WHERE singleton = 1`,		);		if (metadata === undefined) throw new Error("Durable SQLite metadata is missing");		const committedSeq = seqFromNumber(metadata.next_seq);		await this.checkGlobalIds(transaction, writes);		await this.checkDocumentActions(transaction, documentActions);		for (const write of writes) await this.applyTableWrite(transaction, write, committedSeq);		await this.applyDocumentActions(transaction, documentActions, committedSeq);		await transaction.run(			`UPDATE ${this.schema.tables.durableMetadata} SET next_id = ?, next_seq = ? WHERE singleton = 1`,			String(Math.max(Number(metadata.next_id), candidateNextId)),			committedSeq + 1,		);		return committedSeq;	});	this.nextId = Math.max(this.nextId, candidateNextId);	return seq;}

type MainOperation =	| Extract<StorageWrite, { readonly type: "conversation" | "entry" | "submission" | "document.retire" }>	| { readonly type: "task"; readonly value: StoredTask }	| { readonly type: "task.sidecar"; readonly id: TaskId; readonly ordinal: number }	| { readonly type: "document.create"; readonly record: DocumentCreate; readonly ordinal: number }	| { readonly type: "document.change"; readonly id: DocumentId; readonly ordinal: number };type MainMarker = {	readonly format: typeof FORMAT_VERSION;	readonly type: "commit";	readonly seq: Seq;	readonly writes: readonly MainOperation[];};type SidecarPayload =	| { readonly type: "task"; readonly value: StoredTask }	| { readonly type: "document"; readonly id: DocumentId; readonly content: DocumentContent };type SidecarRecord = {	readonly format: typeof FORMAT_VERSION;	readonly type: "record";	readonly seq: Seq;	readonly ordinal: number;	readonly payload: SidecarPayload;};

async commit(writes: readonly StorageWrite[], context: Context): Promise<Seq> {	this.assertUsable();	const prepared = this.memory.prepareCommit(writes);	const encoded = this.encodeCommit(prepared.seq, prepared.writes);	const reclamations = this.planReclamations(prepared.writes, encoded);	const sidecars = await Promise.all(		[...encoded.sidecars].map(async ([file, content]) => ({			file,			content,			path: await this.resolveFile(file, context),		})),	);	for (const sidecar of sidecars) {		const result = await this.fs.appendFile(sidecar.path, sidecar.content, context);		if (!result.ok) throw this.poison(errorFromFile(`append to ${sidecar.file}`, result.error));	}	if (this.fsync) {		for (const sidecar of sidecars) {			const result = await this.fs.flushFile(sidecar.path, context);			if (!result.ok) throw this.poison(errorFromFile(`flush of ${sidecar.file}`, result.error));		}	}	const marker = await this.fs.appendFile(this.mainPath, encoded.marker, context);	if (!marker.ok) throw this.poison(errorFromFile(`append to ${MAIN_FILE}`, marker.error));	const seq = prepared.apply();	this.adoptSidecarState(prepared.writes);	await this.reclaimSidecars(reclamations, context);	return seq;}

async commit(writes: readonly StorageWrite[], _context: Context): Promise<Seq> {	this.assertOpen();	const documentActions = this.prepareDocumentActions(writes);	return this.transaction(ALL_STORES, "readwrite", async (transaction) => {		const metaStore = transaction.objectStore(STORES.meta);		const metadata = await requestResult(metaStore.get(META_KEY) as IDBRequest<MetaRow>);		if (metadata === undefined) throw new Error("Durable IndexedDB metadata is missing");		const seq = seqFromNumber(metadata.nextSeq);		await this.checkGlobalIds(transaction, writes);		await this.checkDocumentActions(transaction, documentActions);		for (const write of writes) await this.applyTableWrite(transaction, write, seq);		await this.applyDocumentActions(transaction, documentActions, seq);		let nextId = metadata.nextId;		for (const write of writes) {			const id = writeId(write);			if (id !== undefined) nextId = Math.max(nextId, id + 1);		}		await requestResult(metaStore.put({ key: META_KEY, nextId, nextSeq: seq + 1 } satisfies MetaRow));		return seq;	});}async mintId<I extends Id<string>>(): Promise<I> {	this.assertOpen();	return this.transaction(STORES.meta, "readwrite", async (transaction) => {		const store = transaction.objectStore(STORES.meta);		const metadata = await requestResult(store.get(META_KEY) as IDBRequest<MetaRow>);		if (metadata === undefined) throw new Error("Durable IndexedDB metadata is missing");		if (!Number.isSafeInteger(metadata.nextId)) throw new Error("ID space is exhausted");		await requestResult(store.put({ ...metadata, nextId: metadata.nextId + 1 }));		return idFromNumber<I>(metadata.nextId);	});}

/** Registers the runner-independent cases with a Vitest/Jest-compatible test runner. */export function registerStorageConformance(	runner: StorageConformanceRunner,	name: string,	withStorage: StorageConformanceProvider,): void {	const cases = createStorageConformance({		assertions: createExpectAssertions(runner.expect),		withStorage,	});	runner.describe(name, () => {		for (const testCase of cases) runner.it(testCase.name, testCase.run);	});}

/** A transaction read a table after its first table write. Read every required row before writing. */export class ReadAfterWrite extends Error {	constructor(method: string) {		super(`Tx.${method}() cannot read tables after the first table write`);		this.name = "ReadAfterWrite";	}}/** * A Storage read rejected an invalid request, such as an unknown conversation or a cursor from another scan, with no * durable effect. Unlike any other error a Storage throws, it fails only that read, not the Session; from `commit()` it is * fatal like any other. */export class StorageRequestError extends Error {	constructor(message: string) {		super(message);		this.name = "StorageRequestError";	}}/** * The Session failed, so nothing runs or commits any more. It fails when a Storage method throws, a commit cannot be * adopted, or the Harness's own bookkeeping throws (a scheduler commit, an internal commit listener, a registry whose * `snapshot()` throws). `cause` is that first error. Reopen the Session; reopening recovers from what was committed. */export class SessionFailed extends Error {	constructor(cause: unknown) {		super("Session failed after a storage error; close and reopen it", { cause });		this.name = "SessionFailed";	}}

/** * Run one backend call under the guard. `read` is a read's context; `commit()` and `mintId()` pass none, so nothing * exempts their errors: once a batch is admitted, whether it committed is unknown. A call still underway when another * fails the Session ends with `SessionFailed` too, whatever it returns: nothing it read or wrote is used. */async #call<T>(read: Context | undefined, run: () => Promise<T>): Promise<T> {	this.#assertHealthy();	if (this.#closing) throw new Error("Session is closed");	let result: T;	// An async wrapper, so a backend that throws synchronously is guarded too.	const call = (async () => run())();	this.#underway.add(call);	try {		result = await call;	} catch (error) {		this.#assertHealthy();		// An invalid read, or one its caller cancelled, says nothing about the Storage; it fails that call only.		const exempt =			read !== undefined && (error instanceof StorageRequestError || read.abortSignal?.aborted === true);		if (!exempt) this.#latch.fail(error);		throw error;	} finally {		this.#underway.delete(call);	}	this.#assertHealthy();	return result;}

/** * Internal: fail the Session with `error`, the first wins, and report it once. Admission ends, and close listeners run * now; they read the error from `failure`. The Storage guard calls this; the Harness also calls it for a throw in its * scheduler's own commits. */fail(error: unknown, context?: Context): void {	if (this.#failure !== undefined) return;	this.#failure = { error };	this.#failed.reject(new SessionFailed(error));	// Seal first, so a report handler that calls back finds the Session failed. Closing runs the close listeners	// synchronously; a failing backend close is the caller's to see, and here nobody waits.	this.close(BACKGROUND_CONTEXT).catch(() => {});	this.report(error, { type: "session_failed" }, context);}

`harness.closed` settles once the Harness has closed, by `close()` or after a storage error, so a host restarts on failure like this:```typescriptharness.closed.then((end) => {	if (end.reason === "failed") reopen(end.error);});```

/** * Open durable storage on a SQLite-backed Durable Object, optionally under a table prefix. Calls given the same * storage object share an operation queue. */export function openDurableObjectSqliteStorage(	storage: DurableObjectSqliteStorage,	options: SqliteStorageOptions = {},): Promise<SqliteStorage> {	return SqliteStorage.open(new DurableObjectSqliteDatabase(storage), options);}

const storageQueues = new WeakMap<DurableObjectSqliteStorage, SerialQueue>();function queueFor(storage: DurableObjectSqliteStorage): SerialQueue {	let queue = storageQueues.get(storage);	if (queue === undefined) {		queue = new SerialQueue();		storageQueues.set(storage, queue);	}	return queue;}⋯ lược bớt dòng 66–161	/** `ctx.storage.transaction()` commits when the callback resolves and rolls back when it rejects. */	transaction<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T> {		return this.queue.hold(() =>			this.storage.transaction(async () => {				const transaction = new DurableObjectSqliteTransaction(this.storage);				try {					return await callback(transaction);				} finally {					transaction.active = false;				}			}),		);	}

/** * When a Session next needs a running process (`Harness.wake()`). `active`: some task can run now or is running, or a * reconcile pass such as open's recovery has not landed yet, so a process that dies must be woken again soon, for * example by a recovery alarm. `at`: the earliest sleep not yet due, * when there is one. Neither: nothing needs a process until new work is submitted. Tasks blocked on a missing or * unmigratable definition, `completing` tasks, and waits on live tasks do not count; the work they wait for does. */export type Wake = { readonly active: boolean; readonly at?: number };

constructor(ctx: DurableObjectState, env: unknown) {	super(ctx, env);	this.#state = ctx;	void ctx.blockConcurrencyWhile(async () => {		// A fallback before anything can throw: an object whose open or recovery fails is started again.		const fallback = Date.now() + HEARTBEAT_MS;		const current = await ctx.storage.getAlarm();		if (current === null || current > fallback) await ctx.storage.setAlarm(fallback);		this.#offset = ctx.storage.kv.get<number>("offset") ?? 0;		const faux = fauxProvider();		faux.setResponses(Array.from({ length: 8 }, () => this.#step()));		const models = createModels();		models.setProvider(faux.provider);		const registry = createRegistry();		registry.install(defineExtension({ name: "test", tasks: [Reminder] }));		this.#harness = await Harness.open(			await openDurableObjectSqliteStorage(ctx.storage),			{				models,				registry,				now: () => this.now(),				onWake: (wake) => {					this.wakes.push(wake);					// A failed alarm write resets the object; it arms again when it next starts.					void this.arm(wake).catch(() => ctx.abort("Alarm write failed"));				},			},			context,		);		// Armed before `resume()` runs anything, as for any call that can start work.		await this.arm(this.#harness.wake());		this.#harness.resume();	});}

async alarm(): Promise<void> {	await this.arm({ ...this.#harness.wake(), active: true });	this.#harness.resume();}/** The alarm a host arms for `wake`: the earliest sleep, or a recovery heartbeat while work can run. */async arm(wake: Wake): Promise<void> {	const heartbeat = wake.active ? Date.now() + HEARTBEAT_MS : undefined;	const sleep = wake.at === undefined ? undefined : wake.at - this.#offset;	const at = heartbeat === undefined || sleep === undefined ? (heartbeat ?? sleep) : Math.min(heartbeat, sleep);	if (at === undefined) await this.#state.storage.deleteAlarm();	else await this.#state.storage.setAlarm(at);}async root(): Promise<Conversation> {	return this.#harness.root(context, { agent: { model: { provider: "faux", modelId: "faux-1" } } });}async submit(text: string, requestId: string): Promise<void> {	// Arm before admitting work: a process that dies after the commit is still woken.	await this.arm({ ...this.#harness.wake(), active: true });	await (await this.root()).submit({ type: "input", content: text, requestId }, context);}

/** * Cloudflare Workers: a pending timer keeps a Durable Object from hibernating or being evicted, even an unreferenced * one (`nodejs_compat` gives timers an `unref()` that changes nothing). */const inWorkers = typeof navigator !== "undefined" && navigator.userAgent === "Cloudflare-Workers";/** * The default `HarnessOptions.setTimer`: `setTimeout`, waiting out delays beyond its limit in steps. A background timer * is unreferenced, so it never keeps the process alive. It is declined in Cloudflare Workers and wherever timers cannot * be unreferenced. */export const nativeTimer: SetTimer = (callback, delayMs, options) => {	if (options.background && inWorkers) return undefined;