/**
 * Runtime local Collab host registry.
 *
 * Every connected Collab host publishes a private, per-process IPC endpoint
 * (Unix domain socket on POSIX, named pipe on Windows) so that a separate local
 * process can discover live hosts and, on explicit request, retrieve a
 * shareable URL. The registry exposes two contracts over that endpoint:
 *
 * - `snapshot`: non-capability metadata (identity, session, cwd, model,
 *   participants, relay/attention/activity state) suitable for listing and
 *   polling;
 * - `link`: one browser URL for one exact host generation and access level,
 *   rejected when the generation moved or the access level is not published.
 *
 * Full-control and view-only URLs, room keys, and write tokens stay in host
 * memory; disk carries only ephemeral discovery metadata (protocol version,
 * instance ID, PID, endpoint, creation time, random bearer token) with
 * owner-only permissions. Endpoints die with the host process, so a crash
 * leaves at most stale metadata that the next list operation prunes best-effort.
 *
 * Transport follows the launch daemon broker conventions: the native
 * `IpcServer`/`IpcConnection` line framing, newline-delimited JSON envelopes,
 * per-request bearer authentication, and bounded frames. Guests never publish;
 * this registry is host-only.
 */
import * as crypto from "node:crypto";
import * as fs from "node:fs";
import * as path from "node:path";
import { type } from "@oh-my-pi/omptype";
import { IpcConnection, IpcServer, ipcEndpoint } from "@oh-my-pi/pi-natives";
import { getBaseConfigRoot, isEnoent, isFsError } from "@oh-my-pi/pi-utils";

/** Discovery metadata / IPC protocol version. Mixed omp versions fail safely. */
export const COLLAB_REGISTRY_VERSION = 1;

/** Reject request lines beyond this size; a valid request is <300 bytes. */
const MAX_REQUEST_BYTES = 4 * 1024;
/** Reject responses beyond this size. */
const MAX_RESPONSE_BYTES = 64 * 1024;
/**
 * Longest string any snapshot field is sent with. Session names, imported
 * session ids, and working directories have no length limit of their own;
 * bounding them here keeps every valid snapshot response (five such fields,
 * worst case fully escaped) well inside {@link MAX_RESPONSE_BYTES}, so an
 * unusual title never makes an otherwise healthy host invisible.
 */
const MAX_SNAPSHOT_FIELD_CHARS = 1024;
/** Per-entry connect+response deadline during listing. */
const DEFAULT_QUERY_TIMEOUT_MS = 1_500;
/** Concurrency bound for querying discovery entries. */
const LIST_CONCURRENCY = 8;

/** Access a link grants: `view` (bare room key) or `control` (room key + write token). */
export type CollabAccess = "view" | "control";

/**
 * Non-capability host state, computed by the host process at query time.
 * Free-form strings (session id and name, cwd, model) are bounded to
 * {@link MAX_SNAPSHOT_FIELD_CHARS} characters on the wire.
 */
export interface CollabHostSnapshot {
	/** Random per-process identity; stable across the host's room generations. */
	instanceId: string;
	/** Increments every time this process starts a new room (session switch, restart). */
	generation: number;
	/** Host process ID. */
	pid: number;
	/** Session ID of the hosted conversation. */
	sessionId: string;
	/** Human-readable session name, when one is set. */
	sessionName: string | null;
	/** Host working directory. */
	cwd: string;
	/** Model the host session is currently using, when one is selected. */
	model: { provider: string; id: string } | null;
	/** Epoch milliseconds when the host first connected to the relay. */
	startedAt: number;
	/** Current participant count, including the host. */
	participants: number;
	/** Whether the host currently holds an open relay connection. */
	relayConnected: boolean;
	/** Whether a host-side question is waiting for an answer a writable guest could give. */
	inputRequired: boolean;
	/**
	 * Whether the session is running a turn: streaming a response or executing
	 * tools. A poller watching this fall from `true` to `false` sees the host
	 * stop working while it is still published, which no other field reports —
	 * disappearing from discovery means the process died or became unreachable,
	 * not that the agent finished.
	 *
	 * `null` when the host does not report it (an omp older than this field).
	 * Unknown is not idle: a consumer must not read the absence as a session
	 * that stopped.
	 */
	busy: boolean | null;
	/** Highest access the registry will hand out for this host. */
	access: CollabAccess;
}

/** Live host state served over the IPC endpoint. */
export interface CollabHostRegistrySource {
	/** Current metadata; throws when the host can no longer vouch for its session. */
	snapshot(): CollabHostSnapshot;
	/** Browser URL for `access`, or `null` when that access is not published. */
	link(access: CollabAccess): string | null;
}

/** One resolved capability returned by {@link resolveCollabHostLink}. */
export interface CollabResolvedLink {
	instanceId: string;
	generation: number;
	access: CollabAccess;
	url: string;
}

/** Handle returned by {@link publishCollabHost}; closing withdraws the host. */
export interface CollabHostPublication {
	/** Endpoint the host listens on (test/diagnostic use; not secret). */
	readonly endpoint: string;
	/** Stop serving requests and remove the discovery metadata. Idempotent. */
	close(): Promise<void>;
}

export interface CollabRegistryOptions {
	/** Override the discovery metadata directory (tests). */
	dir?: string;
}

export interface CollabPublishOptions extends CollabRegistryOptions {
	/**
	 * Identity recorded in the metadata and matched by `omp collab link <id>`.
	 * Defaults to a fresh random ID; a host that rotates rooms passes its
	 * process-lifetime instance ID so the replacement room keeps the same id.
	 * The metadata file and endpoint are always named per publication, so a
	 * stale entry and its successor never share artifacts.
	 */
	instanceId?: string;
}

export interface CollabListOptions extends CollabRegistryOptions {
	/** Per-entry query deadline in milliseconds. */
	timeoutMs?: number;
}

/** Stable failure codes for {@link resolveCollabHostLink}; never carry URLs. */
export type CollabLinkErrorCode = "not_found" | "ambiguous" | "stale_generation" | "access_unavailable" | "unreachable";

export class CollabLinkError extends Error {
	readonly code: CollabLinkErrorCode;
	constructor(code: CollabLinkErrorCode, message: string) {
		super(message);
		this.name = "CollabLinkError";
		this.code = code;
	}
}

/**
 * Discovery metadata directory. Deliberately under the profile-independent
 * config root (`~/.omp/run/collab-hosts`) — unlike the launch broker's
 * profile-scoped runtime dir — so hosts started under any profile are
 * discoverable from any other (issue #6099 user story 18).
 */
export function collabHostsRuntimeDir(): string {
	return path.join(getBaseConfigRoot(), "run", "collab-hosts");
}

interface DiscoveryMetadata {
	version: number;
	instanceId: string;
	pid: number;
	endpoint: string;
	createdAt: number;
	token: string;
}

const INSTANCE_ID_PATTERN = /^[a-z0-9-]{8,64}$/;
const InstanceId = type(INSTANCE_ID_PATTERN);
const Access = type("'view' | 'control'");

const DiscoveryMetadataShape = type({
	"+": "delete",
	version: "number",
	instanceId: InstanceId,
	pid: "number.integer > 0",
	endpoint: "string > 0",
	createdAt: "number",
	token: "string > 0",
});

function parseDiscoveryMetadata(text: string): DiscoveryMetadata | null {
	let raw: unknown;
	try {
		raw = JSON.parse(text);
	} catch {
		return null;
	}
	const meta = DiscoveryMetadataShape(raw);
	return meta instanceof type.errors ? null : meta;
}

function tokenMatches(expected: string, presented: unknown): boolean {
	if (typeof presented !== "string") return false;
	const a = Buffer.from(expected, "utf8");
	const b = Buffer.from(presented, "utf8");
	if (a.length !== b.length) return false;
	return crypto.timingSafeEqual(a, b);
}

const SnapshotShape = type({
	"+": "delete",
	instanceId: InstanceId,
	generation: "number.integer >= 1",
	pid: "number.integer",
	sessionId: "string",
	sessionName: "string | null",
	cwd: "string",
	model: type({ "+": "delete", provider: "string", id: "string" }).or("null"),
	startedAt: "number",
	participants: "number",
	relayConnected: "boolean",
	inputRequired: "boolean",
	// `busy` was added after the protocol version shipped, and the version is a
	// hard gate on both sides: bumping it would make every host invisible to a
	// differently versioned lister on the same machine. So an older host's
	// snapshot simply omits the field, and a missing one reads as unknown
	// (`null`) rather than rejecting an otherwise healthy host.
	"busy?": "boolean | null",
	access: Access,
});

function parseSnapshot(raw: unknown): CollabHostSnapshot | null {
	const snapshot = SnapshotShape(raw);
	if (snapshot instanceof type.errors) return null;
	return { ...snapshot, busy: snapshot.busy ?? null };
}

function boundField(value: string): string {
	return value.length > MAX_SNAPSHOT_FIELD_CHARS ? value.slice(0, MAX_SNAPSHOT_FIELD_CHARS) : value;
}

/** The snapshot as sent on the wire: every free-form string bounded to {@link MAX_SNAPSHOT_FIELD_CHARS}. */
function boundSnapshot(snapshot: CollabHostSnapshot): CollabHostSnapshot {
	return {
		...snapshot,
		sessionId: boundField(snapshot.sessionId),
		sessionName: snapshot.sessionName === null ? null : boundField(snapshot.sessionName),
		cwd: boundField(snapshot.cwd),
		model: snapshot.model
			? { provider: boundField(snapshot.model.provider), id: boundField(snapshot.model.id) }
			: null,
	};
}

/** One request per connection: authenticate, dispatch the op, respond, close. */
function handleConnection(
	connection: IpcConnection,
	token: string,
	source: CollabHostRegistrySource,
	onClose: () => void,
): void {
	let handled = false;
	const respond = (payload: object): void => {
		handled = true;
		connection.send(JSON.stringify(payload));
		connection.end();
	};
	const fail = (error: string): void => respond({ ok: false, v: COLLAB_REGISTRY_VERSION, error });
	// Oversized requests never arrive: the `maxFrameBytes` cap closes the connection natively.
	connection.start(line => {
		if (handled) return;
		let request: unknown;
		try {
			request = JSON.parse(line.trim());
		} catch {
			fail("malformed_request");
			return;
		}
		if (typeof request !== "object" || request === null) {
			fail("malformed_request");
			return;
		}
		const { v, token: presented, op, access, generation } = request as Record<string, unknown>;
		if (v !== COLLAB_REGISTRY_VERSION) {
			fail("unsupported_protocol");
			return;
		}
		if (!tokenMatches(token, presented)) {
			fail("authentication_failed");
			return;
		}
		let snapshot: CollabHostSnapshot;
		try {
			snapshot = source.snapshot();
		} catch {
			// Never let source errors (or URLs) leak into the wire error.
			fail("snapshot_unavailable");
			return;
		}
		if (op === "snapshot") {
			respond({ ok: true, v: COLLAB_REGISTRY_VERSION, snapshot: boundSnapshot(snapshot) });
			return;
		}
		if (op !== "link") {
			fail("invalid_operation");
			return;
		}
		if (!Access.allows(access)) {
			fail("invalid_access");
			return;
		}
		// A capability is bound to the exact generation the caller listed: a room
		// that rotated underneath a stale card must not hand out its successor.
		if (generation !== snapshot.generation) {
			fail("stale_generation");
			return;
		}
		if (access === "control" && snapshot.access !== "control") {
			fail("access_unavailable");
			return;
		}
		let url: string | null;
		try {
			url = source.link(access);
		} catch {
			fail("snapshot_unavailable");
			return;
		}
		if (!url) {
			fail("access_unavailable");
			return;
		}
		respond({ ok: true, v: COLLAB_REGISTRY_VERSION, url });
	}, onClose);
}

/**
 * The registry must be a real directory; POSIX also verifies its owner.
 * Both publication and listing check this: listing prunes malformed
 * entries, so following a symlink into an unrelated directory would let a
 * planted link turn `omp collab list` into a deletion tool.
 */
async function assertPrivateDir(dir: string): Promise<fs.Stats | null> {
	const stat = await fs.promises.lstat(dir);
	if (stat.isSymbolicLink()) throw new Error(`collab registry directory is a symlink: ${dir}`);
	if (!stat.isDirectory()) throw new Error(`collab registry path is not a directory: ${dir}`);
	if (process.platform === "win32") return null;
	const uid = process.getuid?.();
	if (uid !== undefined && stat.uid !== uid) {
		throw new Error(`collab registry directory is not owned by the current user: ${dir}`);
	}
	return stat;
}

/**
 * Create the directory with owner-only POSIX permissions. Windows retains
 * the config root's ACL. `mkdir` with a mode leaves an
 * existing directory's permissions alone, so an already-present directory is
 * tightened explicitly; a symlink or a directory owned by another user is
 * refused rather than published into.
 */
async function ensurePrivateDir(dir: string): Promise<void> {
	await fs.promises.mkdir(dir, { recursive: true, mode: 0o700 });
	const stat = await assertPrivateDir(dir);
	if (stat && (stat.mode & 0o077) !== 0) await fs.promises.chmod(dir, 0o700);
}

/**
 * Unix files an endpoint leaves on disk: the socket and the native listener's
 * `<socket>.bind.lock`. Endpoint names are unique per publication, so no other
 * binder can contend for that lock and unlinking it is race-free.
 */
function endpointFiles(endpoint: string): string[] {
	return process.platform === "win32" ? [] : [endpoint, `${endpoint}.bind.lock`];
}

/** Best-effort synchronous unlink; a survivor is pruned by the next list. */
function removeFilesSync(files: string[]): void {
	for (const file of files) {
		try {
			fs.rmSync(file, { force: true });
		} catch {
			// Best-effort.
		}
	}
}

/**
 * Publish a live Collab host to the local registry.
 *
 * Creates the owner-only runtime dir, starts a private IPC endpoint backed by
 * `source`, and writes discovery metadata (never URLs or room secrets).
 * Call {@link CollabHostPublication.close} on every teardown path; a process
 * exit hook removes the on-disk state for normal shutdown, and the OS closing
 * the endpoint covers crashes.
 */
export async function publishCollabHost(
	source: CollabHostRegistrySource,
	options?: CollabPublishOptions,
): Promise<CollabHostPublication> {
	const dir = options?.dir ?? collabHostsRuntimeDir();
	await ensurePrivateDir(dir);

	const instanceId = options?.instanceId ?? crypto.randomBytes(8).toString("hex");
	if (!INSTANCE_ID_PATTERN.test(instanceId)) throw new Error("invalid collab registry instance id");
	// Unpredictable per-publication entry ID names the endpoint and the metadata
	// file: PID reuse cannot attach stale metadata to an unrelated process, and a
	// room that rotates never shares artifact names with its predecessor, so a
	// lister pruning the stale entry can never remove the live successor's.
	const entryId = crypto.randomBytes(8).toString("hex");
	const token = crypto.randomBytes(32).toString("hex");
	// `<dir>/<entryId>.sock` on Unix (natively relocated when it would overflow
	// `sun_path`); listers never guess it, the metadata records the endpoint.
	const endpoint = ipcEndpoint(dir, entryId);
	const metaPath = path.join(dir, `${entryId}.json`);

	const liveConnections = new Set<IpcConnection>();
	const server = await IpcServer.listen(
		endpoint,
		{ framing: "lines", maxFrameBytes: MAX_REQUEST_BYTES },
		connection => {
			liveConnections.add(connection);
			handleConnection(connection, token, source, () => liveConnections.delete(connection));
		},
	);
	try {
		const meta: DiscoveryMetadata = {
			version: COLLAB_REGISTRY_VERSION,
			instanceId,
			pid: process.pid,
			endpoint,
			createdAt: Date.now(),
			token,
		};
		// Write-then-rename so a concurrent list never observes a partial file
		// (it would classify the entry as malformed and prune it, leaving this
		// host published but undiscoverable). The temp suffix keeps it outside
		// the `*.json` listing filter; the entry ID makes the name unique. Any
		// failure after the exclusive create removes the temp file again.
		const tmpPath = `${metaPath}.tmp`;
		const handle = await fs.promises.open(tmpPath, "wx", 0o600);
		try {
			try {
				await handle.writeFile(JSON.stringify(meta), "utf8");
			} finally {
				await handle.close();
			}
			await fs.promises.rename(tmpPath, metaPath);
		} catch (err) {
			fs.rmSync(tmpPath, { force: true });
			throw err;
		}
	} catch (err) {
		await server.close();
		removeFilesSync(endpointFiles(endpoint));
		throw err;
	}

	// Normal process shutdown without an explicit stop still withdraws the host.
	// The exit hook is synchronous, so it unlinks the socket itself instead of
	// awaiting `server.close()`.
	const removeArtifactsOnExit = (): void => removeFilesSync([metaPath, ...endpointFiles(endpoint)]);
	process.once("exit", removeArtifactsOnExit);

	let closed = false;
	return {
		endpoint,
		async close(): Promise<void> {
			if (closed) return;
			closed = true;
			process.off("exit", removeArtifactsOnExit);
			// Sever any lingering clients so no request is answered after withdrawal.
			for (const connection of liveConnections) connection.destroy();
			removeFilesSync([metaPath]);
			// Stops accepting and unlinks the socket file; the bind lock is ours to remove.
			await server.close();
			removeFilesSync(endpointFiles(endpoint));
		},
	};
}

type QueryResult<T> = { status: "ok"; value: T } | { status: "dead" } | { status: "skip"; error?: string };

/** Query one endpoint: connect, authenticate, send one request, read one bounded response line. */
function query(meta: DiscoveryMetadata, request: object, timeoutMs: number): Promise<QueryResult<unknown>> {
	const { promise, resolve } = Promise.withResolvers<QueryResult<unknown>>();
	let connection: IpcConnection | undefined;
	let settled = false;
	const finish = (result: QueryResult<unknown>): void => {
		if (settled) return;
		settled = true;
		clearTimeout(timer);
		connection?.destroy();
		resolve(result);
	};
	const timer = setTimeout(() => finish({ status: "skip" }), timeoutMs);
	const onLine = (line: string): void => {
		let response: unknown;
		try {
			response = JSON.parse(line);
		} catch {
			finish({ status: "skip" });
			return;
		}
		if (typeof response !== "object" || response === null) {
			finish({ status: "skip" });
			return;
		}
		const { ok, error } = response as Record<string, unknown>;
		if (ok !== true) {
			// Authentication failure or structured error: an unrelated endpoint
			// cannot satisfy stale metadata without the matching token.
			finish({ status: "skip", error: typeof error === "string" ? error : undefined });
			return;
		}
		finish({ status: "ok", value: response });
	};
	// An oversized response (past `maxFrameBytes`) closes the connection without a frame: skip.
	IpcConnection.connect(meta.endpoint, { framing: "lines", timeoutMs, maxFrameBytes: MAX_RESPONSE_BYTES }).then(
		conn => {
			if (settled) {
				conn.destroy();
				return;
			}
			connection = conn;
			conn.start(onLine, () => finish({ status: "skip" }));
			conn.send(JSON.stringify({ v: COLLAB_REGISTRY_VERSION, token: meta.token, ...request }));
		},
		(err: unknown) => {
			// Endpoints die with their host process: a refused or missing socket
			// means the host is gone. Any other error (EMFILE, EACCES, ETIMEDOUT, …)
			// says nothing about liveness and must not prune a live host.
			const code = isFsError(err) ? err.code : undefined;
			finish({ status: code === "ENOENT" || code === "ECONNREFUSED" ? "dead" : "skip" });
		},
	);
	return promise;
}

async function querySnapshot(meta: DiscoveryMetadata, timeoutMs: number): Promise<QueryResult<CollabHostSnapshot>> {
	const result = await query(meta, { op: "snapshot" }, timeoutMs);
	if (result.status !== "ok") return result;
	const snapshot = parseSnapshot((result.value as Record<string, unknown>).snapshot);
	return snapshot ? { status: "ok", value: snapshot } : { status: "skip" };
}

function pidAlive(pid: number): boolean {
	try {
		process.kill(pid, 0);
		return true;
	} catch {
		return false;
	}
}

/**
 * Remove one stale entry. Artifact names are unique per publication, so the
 * metadata and endpoint observed dead can only belong to that publication;
 * a room that rotated meanwhile lives under different names and is untouched.
 */
async function pruneEntry(dir: string, name: string, meta: DiscoveryMetadata | null): Promise<void> {
	try {
		await fs.promises.rm(path.join(dir, name), { force: true });
		// Only unlink sockets this registry could have created: beside the
		// metadata, or at the exact relocated endpoint this entry's publication
		// derives (under the shared `/tmp/omp-<euid>/`, never another socket there).
		const ownsEndpoint =
			meta !== null &&
			process.platform !== "win32" &&
			(meta.endpoint.startsWith(dir + path.sep) || meta.endpoint === ipcEndpoint(dir, path.basename(name, ".json")));
		if (ownsEndpoint) {
			for (const file of endpointFiles(meta.endpoint)) await fs.promises.rm(file, { force: true });
		}
	} catch {
		// Best-effort cleanup only (a missing file means someone else already pruned it).
	}
}

interface LiveEntry {
	meta: DiscoveryMetadata;
	snapshot: CollabHostSnapshot;
}

async function listEntry(dir: string, name: string, timeoutMs: number): Promise<LiveEntry | null> {
	let text: string;
	try {
		text = await Bun.file(path.join(dir, name)).text();
	} catch {
		return null;
	}
	const meta = parseDiscoveryMetadata(text);
	if (!meta) {
		// Malformed metadata can never become listable; remove it.
		await pruneEntry(dir, name, null);
		return null;
	}
	if (meta.version !== COLLAB_REGISTRY_VERSION) {
		// A different omp version owns this entry. Never show it; prune only
		// once the owning process is gone so newer versions keep their state.
		if (!pidAlive(meta.pid)) await pruneEntry(dir, name, meta);
		return null;
	}
	const result = await querySnapshot(meta, timeoutMs);
	if (result.status === "ok") return { meta, snapshot: result.value };
	if (result.status === "dead") await pruneEntry(dir, name, meta);
	return null;
}

async function listLiveEntries(options?: CollabListOptions): Promise<LiveEntry[]> {
	const dir = options?.dir ?? collabHostsRuntimeDir();
	const timeoutMs = options?.timeoutMs ?? DEFAULT_QUERY_TIMEOUT_MS;
	let names: string[];
	try {
		await assertPrivateDir(dir);
		names = await fs.promises.readdir(dir);
	} catch (err) {
		if (isEnoent(err)) return [];
		throw err;
	}
	const entries = names.filter(name => name.endsWith(".json")).sort();
	const live: LiveEntry[] = [];
	// Bounded worker pool: LIST_CONCURRENCY entries in flight at once.
	let next = 0;
	const worker = async (): Promise<void> => {
		while (next < entries.length) {
			const name = entries[next++];
			const entry = await listEntry(dir, name, timeoutMs);
			if (entry) live.push(entry);
		}
	};
	await Promise.all(Array.from({ length: Math.min(LIST_CONCURRENCY, entries.length) }, worker));
	live.sort(
		(a, b) =>
			a.snapshot.startedAt - b.snapshot.startedAt ||
			a.snapshot.pid - b.snapshot.pid ||
			a.snapshot.instanceId.localeCompare(b.snapshot.instanceId),
	);
	return live;
}

/**
 * List live Collab hosts under this config root.
 *
 * Reads every discovery entry, queries the live hosts concurrently (bounded,
 * short independent deadlines), prunes stale or malformed entries
 * best-effort, and returns healthy hosts sorted by start time, PID, then
 * instance ID. Unreachable, unauthenticated, malformed, or version-mismatched
 * entries are omitted without failing the listing. The result carries no URLs.
 */
export async function listCollabHosts(options?: CollabListOptions): Promise<CollabHostSnapshot[]> {
	return (await listLiveEntries(options)).map(entry => entry.snapshot);
}

/**
 * Resolve one browser URL for the host selected by `selector` — an exact
 * instance ID, or a PID when no instance matches. The link request carries the
 * generation observed while listing, so a host that rotated rooms in between
 * answers `stale_generation` instead of leaking its successor's capability.
 */
export async function resolveCollabHostLink(
	selector: string,
	access: CollabAccess,
	options?: CollabListOptions,
): Promise<CollabResolvedLink> {
	const wanted = selector.trim();
	const live = await listLiveEntries(options);
	let matches = live.filter(entry => entry.snapshot.instanceId === wanted);
	if (matches.length === 0 && /^[1-9][0-9]*$/.test(wanted)) {
		const pid = Number(wanted);
		matches = live.filter(entry => entry.snapshot.pid === pid);
	}
	if (matches.length === 0) {
		throw new CollabLinkError("not_found", `no active Collab host matches ${wanted}`);
	}
	if (matches.length > 1) {
		const ids = matches.map(entry => entry.snapshot.instanceId).join(", ");
		throw new CollabLinkError("ambiguous", `${wanted} matches more than one Collab host; use an instance id: ${ids}`);
	}
	const [{ meta, snapshot }] = matches;
	// No local access precheck: the host decides, and it checks the generation
	// before the access level, so a room that rotated underneath the listing
	// reports `stale_generation` rather than a verdict about its predecessor.
	const timeoutMs = options?.timeoutMs ?? DEFAULT_QUERY_TIMEOUT_MS;
	const result = await query(meta, { op: "link", access, generation: snapshot.generation }, timeoutMs);
	if (result.status === "ok") {
		const { url } = result.value as Record<string, unknown>;
		if (typeof url === "string" && url.length > 0) {
			return { instanceId: snapshot.instanceId, generation: snapshot.generation, access, url };
		}
		throw new CollabLinkError("unreachable", `host ${snapshot.instanceId} returned an invalid link response`);
	}
	if (result.status === "skip" && result.error === "stale_generation") {
		throw new CollabLinkError(
			"stale_generation",
			`host ${snapshot.instanceId} started a new room since it was listed; list again and retry`,
		);
	}
	if (result.status === "skip" && result.error === "access_unavailable") {
		throw new CollabLinkError("access_unavailable", `host ${snapshot.instanceId} does not publish ${access} access`);
	}
	if (result.status === "dead") {
		// Every room generation has its own endpoint, so a host that rotated
		// since the listing is simply gone from this one rather than answering
		// `stale_generation` itself. Look the instance up again before giving up.
		const rotated = (await listLiveEntries(options)).some(
			entry => entry.snapshot.instanceId === snapshot.instanceId && entry.snapshot.generation > snapshot.generation,
		);
		if (rotated) {
			throw new CollabLinkError(
				"stale_generation",
				`host ${snapshot.instanceId} started a new room since it was listed; list again and retry`,
			);
		}
	}
	throw new CollabLinkError("unreachable", `host ${snapshot.instanceId} did not answer the link request`);
}