Chord: state, delta, contract và RPC

- An **object** is state plus code at an address. It is created on demand and lives while someone references it.- Everyone gets a **reference**, never the object. References work the same locally and across processes, and survive reloads.- Every object knows its **principal**: an owner and a guest get different objects, with different methods.- **State** is observable JSON. Remote viewers get a snapshot followed by deltas.- **Hot reload** swaps code while it runs. Work in flight finishes on the code it started with, state is kept, and incompatible changes migrate at a safe point.Contracts are runtime values described with [TypeBox](https://github.com/sinclairzx81/typebox) schemas. Hosts exchange JSON frames over links: in memory, a Node IPC channel, or a WebSocket.

/** Typed identity for one value carried by a {@link Context}. */export interface ContextKey<T> {	readonly token: symbol;	/** Type-only marker that prevents keys with different value types from being interchangeable. */	readonly valueType?: (value: T) => T;}/** Immutable invocation-scoped values passed explicitly through operations. */export interface Context {	readonly abortSignal: AbortSignal | undefined;	readonly telemetry: TelemetrySpan;	value<T>(key: ContextKey<T>): T | undefined;	toString(): string;}

abstract class BaseContext implements Context {	abstract value<T>(key: ContextKey<T>): T | undefined;	abstract toString(): string;	get abortSignal(): AbortSignal | undefined {		return this.value(ABORT_SIGNAL_CONTEXT_KEY);	}	get telemetry(): TelemetrySpan {		return this.value(TELEMETRY_CONTEXT_KEY) ?? NOOP_TELEMETRY_CONTEXT;	}}⋯ lược bớt dòng 38–55class ContextValue<T> extends BaseContext {	readonly #parent: Context;	readonly #key: ContextKey<T>;	readonly #value: T;	constructor(parent: Context, key: ContextKey<T>, value: T) {		super();		this.#parent = parent;		this.#key = key;		this.#value = value;	}	value<Value>(key: ContextKey<Value>): Value | undefined {		if (key.token === this.#key.token) return this.#value as unknown as Value;		return this.#parent.value(key);	}	toString(): string {		return `${this.#parent}.WithValue(${this.#key.token.description ?? "anonymous"})`;	}}export const BACKGROUND_CONTEXT: Context = new EmptyContext("[Context BACKGROUND_CONTEXT]");export const TODO_CONTEXT: Context = new EmptyContext("[Context TODO_CONTEXT]");

/** * The explicit call context of one root (a short Chord call tree). Calls made with it carry its root id and code * version; `get` references are released when the call that received this ctx settles. */export interface Ctx<P = unknown> {	readonly signal: AbortSignal;	readonly rootId: string;	readonly codeVersion: number;	/** Who the root acts as: the caller at ingress, the host's own principal inside objects. */	readonly principal: P;	/**	 * A reference for this call, released when the call settles. Resolves once it is ready: live, with every state	 * member of its variant (unless `state: false`). Rejects with the activation error, `aborted` when the ctx is	 * aborted or the call settles first, or `unavailable/ready_timeout` after the host's `readyTimeout`; the reference	 * is released then.	 */	get<C extends AnyContract>(type: C, input: InputOf<C>, options?: GetOptions<P>): Promise<Ref<C>>;}

export type Seg = string | number;export type Path = readonly Seg[];export type NonEmptyPath = readonly [Seg, ...Seg[]];/** A path inline, or an id assigned by the encoder on second use. */export type PathRef<P extends Path = Path> = P | number;/** * Tuples are the form — in memory, on the wire, on disk. * * `r` is the ONLY op that replaces a whole value. `s`/`d`/`a`/`t` cannot target * the root: the type forbids it. `p` and `m` may, because a tracked value can * itself be an array. Operation shape is not canonical: an array may be emptied by either * a replacement or a root splice. * * `Op` knows nothing about the path dictionary. Interning, id references and * omitted paths live in `WireOp` and exist only between `encode` and `decode`. */export type Op =	| readonly ["r", JsonValue]	| readonly ["s", NonEmptyPath, JsonValue]	| readonly ["d", NonEmptyPath]	| readonly ["a", NonEmptyPath, string]	| readonly ["t", NonEmptyPath, number]	| readonly ["p", Path, number, number, JsonValue[]]	/** Reorder an array in place: `new[i] = old[permutation[i]]`. */	| readonly ["m", Path, number[]];

export interface Prepared<T extends object> {	readonly base: T;	readonly value: T;	readonly ops: readonly Op[];	readonly baseRevision: number;	abort(): void;}export interface Change<T extends object> {	readonly state: Draft<T>;	prepare(): Prepared<T>;	abort(): void;}export interface Tracker<T extends object> {	readonly value: T;	readonly revision: number;	beginChange(): Change<T>;	prepareReplace(value: T): Prepared<T>;	adopt(prepared: Prepared<T>): void;}

function setProperty(node: OverlayNode, property: PropertyKey, supplied: unknown): boolean {	assertWritable(node.context);	if (typeof property === "symbol") throw new TypeError("Symbol writes are not supported");	if (Array.isArray(nodeBase(node))) {		if (property === "length") {			setArrayLength(node, toArrayLength(supplied));			return true;		}		const index = arrayIndex(property);		if (index === undefined) throw new TypeError("Only array indices and length can be written");		if (index > arrayLength(arrayOverlay(node))) throw new TypeError("Overlay arrays cannot contain holes");		setArrayIndex(node, index, clonePlacement(supplied));		return true;	}	const key = String(property);	if (supplied === undefined) return deleteProperty(node, key);	const stored = clonePlacement(supplied);	const current = objectValue(node, key);	const wasDeleted = isObjectDeleted(node, key);	if (!wasDeleted && !isContainer(stored) && current === stored) return true;	setObjectWrite(node, key, stored);	if (wasDeleted && Object.hasOwn(nodeBase(node), key)) {		if (node.readded === undefined) node.readded = new Set();		node.readded.add(key);	}	deleteObjectDeletion(node, key);	markDirty(node);	return true;}

prepare(): Prepared<T> {	if (this.#settled) throw new Error("Change has already been settled");	const context = this.#context!;	assertWritable(context);	context.status.value = "prepared";	try {		if (!context.replacement) context.ops = context.dirty.length === 0 ? [] : emitOperations(context);		const prepared = materializePrepared<T>(context);		this.#preparedStatus = context.status;		this.#settled = true;		return prepared;	} catch (error) {		context.status.value = "aborted";		clearContext(context);		this.#settled = true;		throw error;	} finally {		this.#context = undefined;	}}

function emitChangedValue(	operations: Op[],	path: NonEmptyPath,	before: JsonValue | undefined,	after: JsonValue,): boolean {	if (!isContainer(after) && before === after) return false;	if (isContainer(before) && isContainer(after) && equalTrustedJson(before, after)) return false;	if (typeof before === "string" && typeof after === "string") {		// Slice/equality stays on V8's rope fast path; startsWith scans the entire		// accumulated prefix on every append.		if (after.length > before.length && after.slice(0, before.length) === before) {			operations.push(["a", path, after.slice(before.length)]);			return true;		}		const shared = overlap(before, after, 65_536);		if (shared > 0) {			operations.push(["t", path, before.length - shared]);			if (after.length > shared) operations.push(["a", path, after.slice(shared)]);			return true;		}	}	operations.push(["s", path, after]);	return true;}

/** * The plan when the retained base elements keep their order, the common case of inserts and removals: computed from the * pieces alone, so appending to a long array costs no walk over its elements. `undefined` when base elements moved. */function orderedArrayPlan(baseLength: number, pieces: readonly Piece[]): ArrayPlan | undefined {	// Retained base ranges in piece order, `[first, last]` each; ordered means strictly ascending without overlap.	const ranges: number[] = [];	for (const piece of pieces) {		if (piece.kind !== "base" || piece.length === 0) continue;		if (piece.step !== 1 && piece.length > 1) return undefined;		if (ranges.length > 0 && piece.start <= ranges[ranges.length - 1]!) return undefined;		ranges.push(piece.start, piece.start + piece.length - 1);	}	// Unretained runs from the end backwards, as `start, length` pairs: after the last range, between ranges, before the first.	const removeRuns: number[] = [];	let end = baseLength;	for (let index = ranges.length - 2; index >= -2; index -= 2) {		const start = index >= 0 ? ranges[index + 1]! + 1 : 0;		if (end > start) removeRuns.push(start, end - start);		if (index >= 0) end = ranges[index]!;	}	return { removeRuns, permutation: undefined, insertRuns: insertRunsOf(pieces) };}

beginChange(): Change<T> {	const context = createContext(this.#selfRef, this.#owner, this.#revision, this.#value, false, this.#value);	this.#register(context);	return new ChangeImpl(context);}⋯ lược bớt dòng 240–253adopt(prepared: Prepared<T>): void {	const context = PREPARED.get(prepared as object);	if (context?.owner !== this.#owner) throw new Error("Prepared change belongs to a different tracker");	if (context.status.value === "consumed") throw new Error("Prepared change has already been used");	if (context.status.value === "aborted") throw new Error("Prepared change has been aborted");	if (context.status.value === "stale") throw new Error("Prepared change is stale");	if (context.status.value !== "prepared") throw new Error("Prepared change is not ready");	if (context.baseRevision !== this.#revision) {		context.status.value = "stale";		clearContext(context);		throw new Error("Prepared change is stale");	}	if (this.#value !== prepared.base) {		context.status.value = "stale";		throw new Error("Prepared change is stale");	}	// Materialization happened during prepare. Adoption is an infallible pointer	// swap so storage failure can discard the candidate without touching authority.	this.#value = prepared.value;	context.status.value = "consumed";	this.#revision += 1;	this.#invalidate(context);}

/** Compute a compact operation batch from two immutable JSON revisions. */export function diffRevisions(before: JsonValue, after: JsonValue): Op[] {	const operations: Op[] = [];	diffValue(before, after, [], operations);	if (overflowedBatches.has(operations)) return [["r", after]];	if (operations.length === 0 || operations[0]?.[0] === "r") return operations;	let deltaCost = 2;	for (const operation of operations) deltaCost += operationCost(operation) + 1;	if (deltaCost < 65_536) return operations;	const snapshotCost = jsonCost(after) + 6;	return deltaCost >= snapshotCost ? [["r", after]] : operations;}

/** Apply one decoded operation batch without mutating the previous immutable value. */export function applyImmutable<T>(target: T | undefined, ops: readonly Op[]): T {	return applyImmutableBatches(target, [ops]);}/** * Apply decoded operation batches as one final-result-only replay. * * Containers copied for an earlier batch may be mutated while applying a later * batch, so this deliberately exposes no intermediate revisions. */export function applyImmutableBatches<T>(target: T | undefined, batches: Iterable<readonly Op[]>): T {	let root = target as unknown as JsonValue;	const owned = new WeakSet<object>();	for (const ops of batches) {		for (const op of ops) {			assertValidOp(op);			if (op[0] === "r") {				root = op[1];				continue;			}			if (op[0] === "p") {				const appended = appendCopy(root, op, owned);				if (appended !== undefined) {					root = appended;					continue;				}			}			root = copyContainers(root, op[0] === "p" || op[0] === "m" ? op[1] : op[1].slice(0, -1), owned);			root = applyOps(root, [op]);		}	}	return root as unknown as T;}

export interface ReplicatedStateDelivery {	readonly kind: "hydrate" | "update";	readonly sequence: number;}export interface ReplicatedState<T> {	/**	 * Contract-immutable value, or undefined until hydration. It is not frozen and may share containers with an	 * in-process provider; consumers must not mutate it. Later updates do not mutate previously returned values.	 */	readonly value: T | undefined;	/**	 * Each subscription serializes callbacks, awaiting hydration before updates. Values are immutable and may	 * share unchanged revision data. At most 100 deliveries wait behind the running callback; overflow keeps	 * only the newest pending value/context/delivery, so update sequences may skip. Failures are reported in	 * isolation and delivery continues. Unsubscribe discards pending work without aborting or joining a callback.	 */	subscribe(listener: (value: T, context: Context, delivery: ReplicatedStateDelivery) => Promise<void>): () => void;	subscribe(listener: (value: T, context: Context, delivery: ReplicatedStateDelivery) => void): () => void;}export interface MutableReplicatedState<T extends object> extends ReplicatedState<T> {	readonly value: T;	/**	 * Atomically publish one synchronous overlay mutation. Draft handles are unusable after the callback returns.	 * Values placed through the draft are cloned by value; assigning `undefined` to an object property deletes it.	 */	change(context: Context, mutate: (draft: Draft<T>) => void): void;	/**	 * Atomically take immutable ownership of an alias-free strict-JSON replacement.	 * The caller must not mutate the transferred root after this call.	 */	replace(context: Context, value: T): void;}

change(context: Context, mutate: Parameters<MutableReplicatedState<T>["change"]>[1]): void {	if (this.#changing) throw new Error("Replicated state cannot be changed reentrantly from a change callback");	this.#changing = true;	let prepared: Prepared<T>;	try {		const change = this.#tracker.beginChange();		try {			const outcome = (mutate as (draft: typeof change.state) => unknown)(change.state);			if (isPromiseLike(outcome)) {				void Promise.resolve(outcome).catch(() => undefined);				throw new TypeError("Replicated state change callbacks must be synchronous");			}			prepared = change.prepare();		} catch (error) {			change.abort();			throw error;		}	} finally {		this.#changing = false;	}	this.#tracker.adopt(prepared);	if (prepared.ops.length > 0) this.#publisher.publish(prepared.value, context);}replace(context: Context, value: T): void {	if (this.#changing) throw new Error("Replicated state cannot be replaced from a change callback");	const prepared = this.#tracker.prepareReplace(value);	this.#tracker.adopt(prepared);	if (prepared.ops.length > 0) this.#publisher.publish(prepared.value, context);}

push(frame: StateDelivery<T>): void {	if (this.#closed) return;	if (this.#pending.length === 100) {		// A cold replica can queue updates reentrantly before this subscriber's first hydration starts.		const hydration = this.#started ? undefined : this.#pending[0];		this.#pending.length = 0;		if (hydration !== undefined) this.#pending.push(hydration);	}	this.#pending.push(frame);}drain(): void {	if (this.#running || this.#closed) return;	this.#running = true;	for (let frame = this.#pending.shift(); frame !== undefined; frame = this.#pending.shift()) {		this.#started = true;		try {			const result: unknown = this.#listener(frame.value, frame.context, frame.delivery);			if (isPromiseLike(result)) {				void Promise.resolve(result).then(					() => this.#resume(),					(error: unknown) => {						this.#report(error);						this.#resume();					},				);				return;			}		} catch (error) {			this.#report(error);		}	}	this.#running = false;}

/** One immutable authoritative revision committed after an attachment snapshot. */export interface ReplicatedStateSourceFrame<T> {	/** Monotonic source cursor. The first frame after a snapshot must be `snapshot.cursor + 1`. */	readonly cursor: number;	/** The exact immutable value produced by this commit. */	readonly value: T;	/** The exact immutable operation batch that produced `value` from the preceding source revision. */	readonly ops: readonly Op[];	readonly context: Context;}export interface ReplicatedStateSourceAttachment<T> {	/** Fixed immutable snapshot captured at the atomic attachment boundary. */	readonly snapshot: { readonly value: T; readonly cursor: number };	/**	 * Install the sole listener and synchronously drain every buffered frame in source commit order.	 * This method is single-use. After it begins, every new committed frame must also be delivered in order	 * until disposal, including commits made reentrantly while a prior frame is being delivered.	 */	activate(listener: (frame: ReplicatedStateSourceFrame<T>) => void): void;	/** Stop delivery and release source resources. Disposal must be idempotent. */	dispose(): void;}/** * An authoritative immutable revision source. * * `attach()` must synchronously and atomically capture one snapshot and register the returned attachment to buffer * every later committed frame. The snapshot must include every commit before that boundary; buffered frames must * include every commit after it, with no overlap or gap. Snapshot values, frame values, and operation batches are * immutable and remain valid after delivery. Chord only publishes these references; it never applies or re-diffs them. */export interface ReplicatedStateSource<T> {	attach(): ReplicatedStateSourceAttachment<T>;}

export type StateEvent<T> =	| { readonly type: "snap"; readonly value: T }	| { readonly type: "ops"; readonly ops: readonly Op[]; readonly value: T }	| { readonly type: "reset"; readonly value: T };/** A readable state source: a cell, a derivation, or a reference's state member. Values are immutable by contract. */export interface Readable<T> {	readonly value: T;	/** Called with the current value immediately, then with every new value. */	subscribe(listener: (value: T) => void): () => void;	/** Delta stream: a snapshot immediately, then ops, or a reset after a migration or rebind. */	events(listener: (event: StateEvent<T>) => void): () => void;}

/** The tracker tracks containers, so a cell tracks `{ v: value }` and strips the box from published ops. */interface Box<T> {	v: T;}function unbox(ops: readonly Op[], value: unknown): Op[] {	const result: Op[] = [];	for (const op of ops) {		if (op[0] === "r" || op[1].length === 0) return [["r", value as JsonValue]];		const path = op[1].slice(1);		switch (op[0]) {			case "p":				result.push(["p", path, op[2], op[3], op[4]]);				break;			case "m":				result.push(["m", path, op[2]]);				break;			default:				// s/d/a/t on the value itself replace it.				if (path.length === 0) return [["r", value as JsonValue]];				if (op[0] === "d") result.push(["d", path as [string | number, ...(string | number)[]]]);				else result.push([op[0], path as [string | number, ...(string | number)[]], op[2]] as Op);		}	}	return result;}

/** * Mutates a draft (only mutated paths are published), or returns a new value (`n => n + 1`). A returned value is * always validated against the schema. The draft is unusable after `fn` returns. */change(ctx: Ctx, fn: (draft: Draft<T>) => T): void;change(ctx: Ctx, fn: (draft: Draft<T>) => void): void;change(ctx: Ctx, fn: (draft: Draft<T>) => unknown): void {	this.#guard(ctx);	if (!this.#initialized) throw notReady(`state ${this.key} (its init has not run)`);	const change = this.#tracker.beginChange();	let prepared: ReturnType<typeof change.prepare>;	try {		const draft = change.state as Draft<Box<T>>;		const result: unknown = fn(draft.v as Draft<T>);		if (result instanceof Promise) throw new TypeError("cell.change callbacks must be synchronous");		if (result !== undefined) {			if (result !== draft.v) assertValid("state", `state ${this.key}`, this.schema, result);			draft.v = result as Draft<T>;		}		prepared = change.prepare();	} catch (error) {		change.abort();		throw error;	}	this.#publish(prepared, "ops");}

/** Adopts this revision and every derived one, then queues all notifications (cell first, then derivations). */#publish(	prepared: ReturnType<Tracker<Box<T>>["prepareReplace"]>,	type: "ops" | "reset",	validate = true,	ops?: readonly Op[],): void {	if (prepared.ops.length === 0 && type === "ops") {		this.#tracker.adopt(prepared);		return;	}	const value = prepared.value.v;	let candidates: Map<Derived<unknown>, unknown>;	try {		if (validate) this.validate(value);		candidates = computeDependents(this as Source<unknown>, value, validate);	} catch (error) {		prepared.abort();		throw error;	}	holdPublications(() => {		this.#tracker.adopt(prepared);		const published = ops !== undefined && ops.length > 0 ? ops : unbox(prepared.ops, value);		this.fanout.emit(type === "reset" ? { type, value } : { type, ops: published, value });		for (const [derived, next] of candidates) derived.install(next);	});}

// One FIFO queue for every observer notification (state events, shape pushes, registry tables, listener// invocations).//// Problem: notifying observers synchronously from inside a write lets a re-entrant write overtake the write being// published. Example: subscriber A reacts to n = 1 by writing n = 2; subscriber B then sees 2 before 1.// Solution: a publication first adopts every revision it makes (cells and derivations), then enqueues its// notifications. Notifications run in enqueue order, so every observer sees revisions in adoption order.// `holdPublications` defers delivery until a multi-step change (a reload commit) is complete.const queue: (() => void)[] = [];let held = 0;let draining = false;/** Runs `task` after everything enqueued before it. Tasks must not throw (report instead). */export function enqueue(task: () => void): void {	queue.push(task);	drain();}/** Delivers nothing until `fn` returns, then everything it enqueued, in order. */export function holdPublications<T>(fn: () => T): T {	held++;	try {		return fn();	} finally {		held--;		drain();	}}

// contract.ts: shared by provider and consumersexport const Counter = remote("demo.counter", {	version: 1,	input: Type.Object({ name: Type.String() }),	host: hostNamed("server"), // where its objects live; callers never name a host	variants: {		default: {			state: { count: Type.Number() },			methods: { increment: method(Type.Object({ by: Type.Number() })) },		},	},});const host = createHost({ name: "server", self: { kind: "server" } });host.serve(Counter, ({ name }, _principal, self) => {	const count = self.state("count", Type.Number(), 0);	return {		variant: "default",		count,		increment: async ({ by }, ctx) => {			count.change(ctx, (n) => n + by);		},	};});await host.root(async (ctx) => {	const clicks = await ctx.get(Counter, { name: "clicks" }); // ready, with its state; released when this root ends	clicks.count.subscribe((n) => console.log("count", n));	await clicks.increment({ by: 2 }, ctx); // logs 0, then 2});

export function remote<const D extends RemoteDef>(	name: string,	def: D & { readonly host: PlacementFor<Static<D["input"]>> },): RemoteContract<D> {	checkHeader(name, def.version);	const placement = def.host as Placement | undefined;	if (placement === undefined || !Value.Check(PlacementSchema, placement))		throw new TypeError(`remote contract ${name} needs \`host\`: hostNamed(…) or hostFamily(…)`);	// A JSON copy: the description travels, and a placement object may carry nothing else.	const host: Placement =		"name" in placement ? { name: placement.name } : { prefix: placement.prefix, key: placement.key };	const variants: Record<string, Record<string, MemberDescription>> = {};	if (Object.keys(def.variants).length === 0) throw new TypeError(`remote contract ${name} has no variants`);	for (const [kind, variant] of Object.entries(def.variants)) {		const members: Record<string, MemberDescription> = {};		for (const [key, schema] of Object.entries(variant.state ?? {}))			members[key] = describeMember(key, schema, "remote");		for (const [key, spec] of Object.entries(variant.methods ?? {})) {			if (key in members) throw new TypeError(`member ${key} is both state and method in ${name}.${kind}`);			members[key] = describeMember(key, spec, "remote");			if (members[key].kind !== "method") throw new TypeError(`methods.${key} is not a method()`);		}		variants[kind] = members;	}	const description: Description = { name, scope: "remote", version: def.version, input: def.input, host, variants };	return finish<RemoteContract<D>>({		scope: "remote",		name,		version: def.version,		input: def.input,		def,		description,	});}

/** How a consumer token binds to a provider: directly or through an `accepts` adapter. Throws contract_mismatch. */export function bind(provider: AnyContract, accepts: Accepts | undefined, consumer: AnyContract): Adapter | undefined {	if (provider.fingerprint === consumer.fingerprint) return undefined;	// A moved type is another type for its consumers: no adapter reaches across hosts.	if (!samePlacement(provider.description, consumer.description)) {		throw new ChordError(			"contract_mismatch",			`${provider.name} lives on ${placementOf(provider.description)}, the consumer expects ${placementOf(consumer.description)}`,		);	}	if (provider.version === consumer.version) {		const reason = coverage(provider.description, consumer.description);		if (reason === undefined) return undefined;		throw new ChordError(			"contract_mismatch",			`${provider.name}@${provider.version} does not cover the consumer's description (${reason}); forgotten version bump?`,		);	}	const adapter = accepts?.[consumer.version];	if (adapter === undefined) {		throw new ChordError(			"contract_mismatch",			`${provider.name}: provider version ${provider.version}, consumer version ${consumer.version}, no accepts adapter`,		);	}	return adapter;}

acquire(	type: AnyContract,	input: unknown,	options: GetOptions | undefined,	owner: RefOwner,	from?: ObjectImpl,	candidate = false,): RefCore {	const elsewhere = this.#elsewhere(type, input);	if (elsewhere !== undefined) return this.#acquireRemote(type, elsewhere.at, elsewhere.input, options, owner);	const service = this.#bindingService(type.name, candidate);	if (service === undefined) throw unknownType(type.name, this.name);	bind(service.contract, service.options.accepts, type);	const { address, normalized, principal } = this.#locate(service, input, options?.principal ?? this.principal);⋯ lược bớt dòng 558–565	let object = this.objects.get(address);	if (from !== undefined && object !== undefined && (object === from || reaches(object, from))) {		throw invalid("cycle", `${from.address} -> ${address} would create a ref cycle`);	}	if (object === undefined) {		if (this.gating?.has(type.name)) {			throw new ChordError("upgrade_conflict", `${type.name} is being upgraded; retry`, { reason: "gate" });		}		const dying = this.#dying.get(address);		if (dying?.status === "disposed") this.#dying.delete(address);		object = new ObjectImpl(this, {			address,			typeName: type.name,			input: normalized,			principal,			hostLifetime: hostLifetime(service),		});		this.objects.set(address, object);		// A new object for an address still being disposed waits for that cleanup (exclusive resources).		object.start(dying?.status === "disposing" ? dying.dispose(false) : undefined);	}	const ref = new RefImpl(object, type, owner);	object.addRef(ref);	return ref;}

async #activate(after: Promise<void> | undefined): Promise<void> {	await after?.catch(noop);	const signal = this.activationAbort.signal;	const abandoned = (): ChordError => new ChordError("aborted", `${this.address} abandoned`);	if (signal.aborted) throw abandoned();	const service = this.host.service(this.typeName);	if (service === undefined)		throw new ChordError("unavailable", `${this.typeName} is not served`, { reason: "unknown_type" });	this.service = service;	const run = this.run(service, undefined);	try {		await this.open(run.created.resources, run.created.refs, signal);		if (signal.aborted) throw abandoned();		await this.waitHydrated(run);		if (signal.aborted) throw abandoned();		await this.initCells(run, signal);		if (signal.aborted) throw abandoned();		this.host.checkRegistrations(this, run.decls);		this.decls = run.decls;⋯ lược bớt dòng 1114–1130		for (const derived of run.created.derives) derived.attach();		this.refreshSchemas();		// Served only by a running reload's pending code version: new code must not run (listeners, tasks,		// registrations) before that reload commits.		this.deferredRuntime = !this.implementations.has(this.host.currentCodeVersion());		if (!this.deferredRuntime) {			if (run.decls.registrations.size > 0) this.host.registriesChanged(this);			this.startRuntime();		}		// A listener starting up may have released the last reference synchronously: no resurrection.		if (signal.aborted) throw abandoned();		this.status = "live";

async root<T>(fn: (ctx: Ctx) => T | Promise<T>, options: RootOptions = {}): Promise<T> {	// The incarnation keeps root ids unique across restarts of a host with the same name.	const rootId = `${this.name}#${this.#incarnation}:${++this.#rootSeq}`;	this.#local.add(rootId);	return this.joinRoot(rootId, options.codeVersion ?? this.current.id, fn, options);}⋯ lược bớt dòng 618–624async joinRoot<T>(	rootId: string,	codeVersionId: number,	fn: (ctx: Ctx) => T | Promise<T>,	options: { readonly principal?: unknown; readonly signal?: AbortSignal; readonly from?: Connection } = {},): Promise<T> {	const codeVersion = this.codeVersions.get(codeVersionId);	if (codeVersion === undefined) {		this.#local.delete(rootId);		throw new ChordError("superseded", `code version ${codeVersionId} is not retained by ${this.name}`);	}⋯ lược bớt dòng 636–643	codeVersion.roots.set(rootId, (codeVersion.roots.get(rootId) ?? 0) + 1);	const ctx = new CtxImpl(this, {		// Forced retirement aborts what still runs on a code version before closing what only it used.		signals:			options.signal === undefined ? [codeVersion.abort.signal] : [options.signal, codeVersion.abort.signal],		rootId,		codeVersion: codeVersion.id,		principal: options.principal ?? this.principal,	});	try {		return await fn(ctx);	} finally {		ctx.close();		const count = (codeVersion.roots.get(rootId) ?? 1) - 1;		if (count > 0) codeVersion.roots.set(rootId, count);		else {			codeVersion.roots.delete(rootId);			if (!this.#foreign.has(rootId)) this.endRoot(rootId);			this.maybeRetire();			this.#retirementChanged();		}	}}

/** Error codes callers see (guide table). A refused or aborted reload keeps the code of its cause. */export const CHORD_ERROR_CODES = [	"aborted",	"contract_mismatch",	"forbidden",	"gone",	"invalid",	"not_in_variant",	"not_ready",	"released",	"superseded",	"unavailable",	"upgrade_conflict",	"upgrade_required",] as const;export type ChordErrorCode = (typeof CHORD_ERROR_CODES)[number];⋯ lược bớt dòng 17–23const RETRYABLE: ReadonlySet<ChordErrorCode> = new Set(["superseded", "upgrade_conflict"]);export class ChordError extends Error {	readonly code: ChordErrorCode;	readonly reason: ChordErrorReason | undefined;	readonly retryable: boolean;	readonly details: unknown;	constructor(		code: ChordErrorCode,		message: string,		options?: { cause?: unknown; details?: unknown; reason?: ChordErrorReason },	) {		super(`${code}${options?.reason ? `/${options.reason}` : ""}: ${message}`, {			cause: options?.cause,		});		this.name = "ChordError";		this.code = code;		this.reason = options?.reason;		this.retryable = RETRYABLE.has(code);		this.details = options?.details;	}}

// Node facet loader: `createHost({ loader: nodeLoader })`. Kept out of index.ts so the core stays browser-safe.//// TODO(loader): replace with an isolating, evicting loader. Every reload imports fresh module copies under a new// query and the old copies are never evicted (they leak until the process exits). TUI plugins need isolation, which// this loader does not provide.⋯ lược bớt dòng 6–43export const nodeLoader: FacetLoader = async (modules) => {	hook();	const version = String(++generation);	const facets: Facet[] = [];	const urls = modules.map((module) => {		const url = toUrl(module);		url.search = "";		return url;	});	// Every root first: a facet importing another facet's directory gets this load's copy, whatever the order.	for (const url of urls) roots.add(new URL(".", url).href);	for (const url of urls) {		const name = url.href;		url.searchParams.set(PARAM, version);		// Owner-approved exception to the no-dynamic-import rule; see TODO(loader) above.		const loaded = (await import(url.href)) as { default?: FacetDefinition };		const definition = loaded.default;		if (definition === undefined || typeof definition.setup !== "function") {			throw new TypeError(`${name}: default export must be a facet (defineFacet)`);		}		facets.push({ name, setup: definition.setup, uses: definition.uses });	}	return facets;};

/** Everything up to the commit. On failure the host is rolled back and the underlying error is thrown. */async prepare(): Promise<void> {	try {		await this.waitForRetirement();		this.load();		this.checkProviders();		await this.prepareAll();		// From here new objects of changed types are rejected (upgrade_conflict, retryable) until commit or abort,		// so the candidate set is final once the activations already in flight have been prepared.		this.host.gating = this.changed;		await this.withDeadline(this.prepareAll(), "activations in flight did not finish");		this.checkRefs();		this.tables = this.host.buildTables(this.entries());		await this.gate();		this.#alive();		this.#stage();	} catch (error) {		await this.rollback();		throw reloadError(error);	}}

// Each cell carries its own version: a raised one gates and migrates, a schema change without one is refused.for (const [name, cell] of run.decls.cells) {	if (object.decls.cells.get(name) !== cell) continue;	const from = object.decls.cellVersions.get(name) ?? 1;	const to = run.decls.cellVersions.get(name) ?? 1;	if (to < from)		throw invalid("state", `${object.address}: version of state ${name} lowered from ${from} to ${to}`);	if (to > from) candidate.reasons.push(`state ${name} version ${from} -> ${to}`);	else if (!deepEqual(cell.schema, run.decls.cellSchemas.get(name)))		throw invalid("state", `${object.address}: schema of state ${name} changed without a version bump`);}

async gate(): Promise<void> {	this.gated = [...this.candidates.values()].filter(		(candidate) => candidate.reasons.length > 0 && candidate.object.status === "live",	);	const drained = this.gated.map(({ object }) => {		const grandfathered = new Set(object.leases.keys());		return new Promise<void>((resolve) => {			object.gate = { grandfathered, queue: [], drained: resolve };			object.stopTasks();			if (grandfathered.size === 0) resolve();		});	});⋯ lược bớt dòng 469–477	if (timedOut) {		const blocking = this.gated.flatMap(({ object }) => [...(object.gate?.grandfathered ?? [])]);		throw new ChordError(			"aborted",			`gate deadline (${deadline} ms) missed; blocking roots: ${blocking.join(", ")}`,			{				reason: "gate_deadline",				details: { blocking },			},		);	}

/** * Publishes the new code version in one synchronous step. Everything fallible ran in prepare (factories, opens, * migration and derivation checks against the prospective schemas, listener subscriptions), so this only installs. * Observers (state, shapes, registries, listener roots) are notified after the whole step. If an install still * throws, the code version is half-published: the host fails terminally and admits no more work. */commit(beforeStart?: () => void): ReloadReport {	try {		return holdPublications(() => {			const report = this.#publish();			// Installed, not yet started: the coordinator queues its commit frames here, so nothing this code version			// starts (tasks, host object activations, then the held listener/registry/state notifications) can reach a child			// first.			beforeStart?.();			for (const start of this.#starts.splice(0)) start();			return report;		});	} catch (error) {		this.host.fail(error);		throw reloadError(error);	}}

/** An ordered (FIFO), bidirectional frame channel between two hosts. */export interface Link {	readonly closed: boolean;	/**	 * Returns the size of the encoded frame as sent (JSON text length, or bytes of a packed message) when the transport	 * knows it. The connection's flow control counts these sizes; frames without one count as zero, so a link that never	 * reports sizes is not windowed.	 */	send(frame: Frame): number | undefined;	/** Sets the single receiver. Frames arriving earlier are queued. */	receive(handler: (frame: Frame) => void): void;	onClose(handler: (reason: string) => void): void;	/** Closes both ends: the peer gets a `bye`. */	close(reason?: string): void;}

// JSON frames between two hosts (or a host and a client). Ids are chosen by the side that creates the hold, call or// subscription; every frame type has one direction, so ids never collide://   consumer → provider: hold, release, call, cancel, sub, unsub, resync//   provider → consumer: shape, gone, result, state// State is addressed by (hold id, member): a hold implies the state of its variant, sent as units (a shape followed by// a snapshot of each member the consumer lacks; see Connection).//   both: hello, ping, pong, bye, root_end, ack; parent → child: invalidate, prepare, commit, abort, retire;//   child → parent: vote//// Each frame is declared once, as the schema every received frame is checked against (`isFrame`); the types are derived// from it. Payloads (hold inputs, call args, results, snapshots) stay `unknown`: they are validated where they are// used, against the contract. State ops are checked op by op by the delta decoder./** * The link protocol version, announced in `hello`. Peers of different versions do not talk: the side that notices * closes the link with reason `protocol-version <its version>`, so the other side learns it too * (`LinkOptions.onProtocolMismatch`). */export const PROTOCOL_VERSION = 4;

Type.Object({	t: Type.Literal("state"),	hold: Id,	m: Name,	/** `snap`/`reset` without `value`: the value is `undefined`. */	as: Type.Union([Type.Literal("snap"), Type.Literal("reset"), Type.Literal("ops"), Type.Literal("error")]),	value: Type.Optional(Type.Unknown()),	/** Encoded with the stream's delta encoder; the decoder validates each op. */	ops: Type.Optional(List(List(Type.Unknown()))),	error: Type.Optional(WireErrorSchema),}),

/** * Principal stamping: a downstream link acts as the principal it was accepted with, whatever the frame says. Only * the parent may forward a principal (relayed requests); its own requests act as the principal given to `connect`. */#principalOf(frame: { readonly principal?: unknown }): unknown {	if (this.role === "downstream") return this.principal;	return frame.principal ?? this.principal;}/** * The client connection a hold comes from: this link's own id for a client link (whatever the frame says), the id * the parent forwarded for a relayed hold, none for a host's own holds. */#originOf(frame: { readonly origin?: string }): string | undefined {	return this.role === "downstream" ? this.originId : frame.origin;}