aggregate.js

/**
 * Cluster aggregation: fold N nodes' admin payloads into one cluster answer, at read time.
 *
 * WHY THE CONSOLE DOES THIS AND NOT THE PLUGIN. Almost everything interesting in a prerender
 * deployment is node-local by construction — `hdb_analytics` rows are written per node, the
 * backlog snapshot and the queue keeper cover only the residency-pinned keys THAT node owns, the
 * lease table is a node-local shared buffer, and the unrouted tally is per worker. A plugin-side "cluster"
 * endpoint would have to fan out to its peers, which means every prerender node becomes a
 * client of every other prerender node on a path that also serves crawler traffic. The console
 * is already the one component that holds every node's session and talks to all of them, and it
 * runs off the serve path. So the fan-out lives here, and the plugin stays a set of honest
 * per-node endpoints.
 *
 * THREE MERGE CLASSES, and picking the wrong one is how a dashboard lies:
 *
 *   MERGED  — node-local numbers that genuinely add up (analytics, the backlog snapshot, the
 *             unrouted tally). Summed here. This is the only class that produces a number no
 *             single node can report: the true cluster backlog, the true serve rate.
 *   SHARED  — replicated tables every node can answer for identically (pages, sitemaps,
 *             invalidations, the crawl sketches, the static metric catalog). Fanning these out
 *             would multiply load to produce four copies of one answer, so ONE node answers and
 *             the payload says which.
 *   COMPARED — config. Identical by intent, so the interesting content is any DISAGREEMENT: a
 *             component deploy that silently skipped a node shows up here and nowhere else.
 *
 * PARTIAL ANSWERS ARE LABELLED, NEVER SILENTLY SHORT. A sum missing a node is not a smaller
 * number, it is a WRONG number, and it looks exactly like a real decline. Every merged payload
 * carries `sources` — who answered, who didn't, and why — and the UI refuses to present an
 * incomplete sum without saying so.
 *
 * Everything here is pure: `(results) => payload`. No I/O, no Harper globals, so the merge math
 * is unit-tested directly.
 */

/** The scope sentinel the UI sends in place of a node origin. Never a hostname. */
export const CLUSTER = 'cluster';

/**
 * Routes whose numbers are node-local and additive. Everything else in PROXIED_GET reads a
 * replicated table (or is static), so it is answered by ONE node — see SHARED_NOTE.
 */
export const MERGED_GET = Object.freeze([
	'overview',
	'analytics',
	'unrouted',
	'config',
	// Neither of these ADDS the way analytics does — a probe pass and a purge pass are run one node
	// at a time over the keys that node owns, so what the merge produces is every node's own slice
	// side by side plus the running tally. They are here rather than in SHARED_NOTE because the
	// alternative is worse in the one way this classification exists to prevent: one node's
	// "deleted 240,118" answered under a cluster label reads as done when three quarters of the
	// keyspace has not been touched.
	'change-probe',
	'discovery-purge',
	// Each node's queue keeper holds only the rows that node owns, so the cluster's queue is the sum —
	// and a node that cannot vouch for its numbers answers 503 with the reason. See mergeQueueState.
	'queue-state',
]);

/**
 * Why a route is answered by a single node under cluster scope, keyed by route. Shown in the
 * UI so "cluster" never implies a fan-out that didn't happen.
 */
export const SHARED_NOTE = Object.freeze({
	'pages': 'page_cache replicates — any node holds the whole corpus',
	'page-content': 'page_cache replicates — any node holds the whole corpus',
	'sitemaps': 'the sitemaps database replicates',
	'invalidations': 'the invalidation database replicates',
	'crawl-breadth': 'crawl sketches replicate; a read already merges every node’s shard',
	'metrics': 'the metric catalog is static, compiled into the plugin',
});

/**
 * POST routes that act on ONE node's own state and therefore cannot be sent to an arbitrary
 * node. Under cluster scope these are refused with an instruction to pick a node, rather than
 * quietly landing on whichever node happened to be first — "Run repair sweep" hitting one of
 * four nodes while the console says "all nodes" is the kind of thing an operator discovers
 * three incidents later.
 *
 * `queue` is absent on purpose: a control write is replicated intent, and its own `scope` field
 * already says which node(s) it applies to.
 */
export const NODE_LOCAL_POST = Object.freeze({
	'reconcile': 'The repair sweep runs over the keys ONE node owns. Pick a node to sweep it.',
	// Node-scoped for a second reason on top of residency: the in-flight check reads THIS node's
	// lease buffer, so asked of a non-owner it cannot tell an orphan from a key mid-render — and
	// this sweep deletes.
	'sweep-orphans': 'The orphan sweep deletes among the keys ONE node owns. Pick a node to sweep it.',
	'backlog': 'A backlog snapshot covers the keys ONE node owns. Pick a node to recompute it.',
	// A probe pass walks the registry rows THIS node owns and writes their baselines, and the
	// canary probes the cohort built from that same slice. Fanned out it would be N passes at N
	// times the agreed origin rate, which is the one thing changeProbe.ratePerSecond exists to
	// stop — the rate is a promise made to whoever runs the origin, per node.
	'change-probe':
		'A probe pass covers the URLs ONE node owns, at that node’s agreed probe rate. Pick a node to run it.',
	// Deletes, owner-scoped, and paced per node — the sweep-orphans argument exactly.
	'discovery-purge':
		'A purge deletes among the keys ONE node owns. Pick a node to run it, and run every node to cover the keyspace.',
	// `schedule` reads RenderSchedule node-locally (replicateFrom: false, because the table is
	// residency-pinned and a cross-node read has no timeout). Asked of a node that does not own
	// the key it truthfully answers "no row" — which reads as "this URL will never render". The
	// UI never calls it (it is the peer endpoint `explain` uses, and `explain` already proxies
	// to the owner itself); a caller reaching it directly has to name the node they mean.
	'schedule':
		'A schedule row is only readable on the node that owns the key. Pick a node — or use explain, which finds the owner for you.',
});

/**
 * POST routes whose cluster-scoped answer carries a provenance envelope, and the reason one node
 * is the whole write. The read side's SHARED_NOTE, pointed the other way.
 *
 * This is NOT the inverse of NODE_LOCAL_POST. Every route absent from that map already lands on a
 * single node (see `writeNode`); membership here only adds the envelope naming which node took it
 * and why that is cluster-wide.
 *
 * `config-override` is here and an invalidation is not, because of what reads the answer next. A
 * config write is followed, seconds later, by the one panel that treats nodes disagreeing as a
 * DEPLOY FAILURE — and until the write replicates the nodes do disagree. The response has to have
 * already said "one node took this, the rows replicate, the others converge", or the console's own
 * write path reads as the alarm it exists to raise. Nothing reads an invalidation's answer that
 * way, so those keep their byte-exact pass-through.
 */
export const REPLICATED_POST_NOTE = Object.freeze({
	'config-override': 'stored config overrides replicate — one node takes the write and every node converges',
});

// ---------------------------------------------------------------- small helpers

const num = (v) => (Number.isFinite(v) ? v : null);
const finite = (v) => (typeof v === 'number' && Number.isFinite(v) ? v : NaN);

/** Sum of a field over results, or null when no result carried a finite value. */
function sumOf(items, read) {
	let total = 0;
	let seen = false;
	for (const item of items) {
		const value = finite(read(item));
		if (Number.isFinite(value)) {
			total += value;
			seen = true;
		}
	}
	return seen ? total : null;
}

const maxOf = (items, read) => {
	let best = null;
	for (const item of items) {
		const value = finite(read(item));
		if (Number.isFinite(value) && (best === null || value > best)) best = value;
	}
	return best;
};

const minOf = (items, read) => {
	let best = null;
	for (const item of items) {
		const value = finite(read(item));
		if (Number.isFinite(value) && (best === null || value < best)) best = value;
	}
	return best;
};

/** Element-wise sum of equal-length numeric arrays; shorter arrays contribute what they have. */
function sumArrays(arrays, length) {
	const out = new Array(length).fill(0);
	for (const arr of arrays) {
		if (!Array.isArray(arr)) continue;
		for (let i = 0; i < length && i < arr.length; i++) {
			const value = finite(arr[i]);
			if (Number.isFinite(value)) out[i] += value;
		}
	}
	return out;
}

/**
 * The result whose `read` timestamp is newest. Used wherever every node holds a replicated copy
 * of the same row: they should agree, and when they don't the freshest copy is the one to show.
 */
function freshest(items, read) {
	let best = null;
	let bestAt = -Infinity;
	for (const item of items) {
		const at = finite(read(item));
		if (Number.isFinite(at) && at > bestAt) {
			bestAt = at;
			best = item;
		}
	}
	return best ?? items[0] ?? null;
}

const msOf = (value) => {
	if (value === null || value === undefined) return NaN;
	const n = typeof value === 'number' ? value : Date.parse(value);
	return Number.isFinite(n) ? n : NaN;
};

const isoOrNull = (ms) => (Number.isFinite(ms) ? new Date(ms).toISOString() : null);

// ---------------------------------------------------------------- the envelope

/**
 * The provenance block every cluster-scoped response carries. This is the honesty contract:
 * a merged number is only as trustworthy as `complete`, and `nodes[]` says exactly which node
 * failed and how, so a dip in a chart can always be told apart from a node that stopped
 * answering.
 */
export function sourcesOf(results, { mode, scope = 'cluster', extra = null } = {}) {
	const nodes = results.map((r) => ({
		origin: r.origin,
		hostname: r.hostname,
		ok: !!r.ok,
		status: r.status ?? 0,
		error: r.error ?? null,
		ms: Number.isFinite(r.ms) ? r.ms : null,
		// How long ago the console's read cache fetched this node's answer; null when it was read now.
		ageMs: Number.isFinite(r.ageMs) ? r.ageMs : null,
	}));
	const answered = nodes.filter((n) => n.ok).length;
	return {
		mode,
		scope,
		answered,
		configured: nodes.length,
		complete: answered === nodes.length && answered > 0,
		nodes,
		...(extra ?? {}),
	};
}

/** Everything that answered with a usable body. */
const okBodies = (results) => results.filter((r) => r.ok && r.body && typeof r.body === 'object');

/**
 * The response when nothing answered. A 502 rather than an empty aggregate: zero serves across
 * a cluster that is actually up is the single most dangerous number this console could print.
 */
const allFailed = (results, mode) => ({
	status: 502,
	body: {
		error: 'No prerender node answered.',
		sources: sourcesOf(results, { mode }),
	},
});

// ---------------------------------------------------------------- analytics

const comboKey = (s) => `${s.metric}${s.path ?? ''}${s.method ?? ''}${s.type ?? ''}`;

/**
 * Merge per-node analytics windows into one cluster window.
 *
 * BUCKETS ALIGN BY INDEX, NOT BY TIMESTAMP. Every node derives `bucketMs` deterministically
 * from the same requested range, so bucket *i* means the same elapsed offset everywhere; the
 * nodes' window starts differ only by the few milliseconds between the fan-out's requests
 * (reported as `skewMs`, and always far under a ≥60s bucket). A node that clamped the range
 * differently — a peer running an older plugin, or a smaller `management.analytics.maxRange` —
 * produces a DIFFERENT bucket width, and index alignment would then silently smear two
 * timebases together. Those nodes are dropped from the merge and reported as failed sources
 * with the reason, which is the one outcome an operator can act on.
 *
 * DISTRIBUTIONS ARE COUNT-WEIGHTED AND APPROXIMATE, twice over: once inside each node (merging
 * Harper's per-combo aggregates) and again here. The weight available in the payload is the
 * bucket's `count`, which slightly over-weights buckets containing rows that carried no stats;
 * for value metrics — where a row without a mean is vanishing — the error is negligible. The UI
 * writes "≈" on every merged percentile for exactly this reason. A merged p95 is a trend line,
 * never an SLO.
 */
/**
 * Which series get PER-NODE BUCKETS in the merged analytics payload: bot serves (by source and
 * verdict) and render outcomes. That is what the traffic-by-instance and renders-by-node charts
 * draw, and nothing else needs a per-node time axis.
 *
 * Collapsed over the remaining slot (the bot name on `bot_serve`) before shipping, which keeps it to
 * a dozen or so arrays per node instead of one per (bot × verdict × source) — the per-instance view
 * compares nodes, and a bot filter cannot narrow it (the tag on its card says so).
 */
const PER_NODE_BUCKETED = (s) => s.metric === 'bot_serve' || (s.metric === 'render' && s.path === 'outcome');

function perNodeBuckets(series, bucketCount) {
	const out = new Map();
	for (const s of series ?? []) {
		if (!PER_NODE_BUCKETED(s)) continue;
		const key = `${s.metric}\u0000${s.path ?? ''}\u0000${s.method ?? ''}`;
		let acc = out.get(key);
		if (!acc) {
			acc = {
				metric: s.metric,
				path: s.path ?? null,
				method: s.method ?? null,
				count: 0,
				counts: new Array(bucketCount).fill(0),
			};
			out.set(key, acc);
		}
		acc.count += finite(s.count) || 0;
		const counts = Array.isArray(s.counts) ? s.counts : [];
		for (let i = 0; i < bucketCount && i < counts.length; i++) {
			const c = finite(counts[i]);
			if (Number.isFinite(c)) acc.counts[i] += c;
		}
	}
	return [...out.values()];
}

export function mergeAnalytics(results) {
	const usable = okBodies(results).filter((r) => r.body.available !== false);
	if (!usable.length) {
		// Every node answered "analytics unavailable" (or none answered at all). Pass the first
		// node's explanation through rather than inventing an empty window.
		const explained = okBodies(results)[0];
		if (explained) {
			return {
				status: 200,
				body: { ...explained.body, sources: sourcesOf(results, { mode: 'merged' }) },
			};
		}
		return allFailed(results, 'merged');
	}

	// DOUBLE-COUNT GUARD. When a deployment sets `analytics_replicate: true`, every node's
	// hdb_analytics already holds every node's rows — the plugin says so with `scope: 'cluster'`.
	// Summing would then multiply the whole cluster by N. One node answers instead.
	const replicated = usable.find((r) => r.body.scope === 'cluster');
	if (replicated) {
		return {
			status: 200,
			body: {
				...replicated.body,
				sources: sourcesOf(results, {
					mode: 'shared',
					extra: {
						servedBy: replicated.hostname,
						note: 'analytics_replicate is on — each node’s table already holds the whole cluster, so summing would multiply it.',
					},
				}),
			},
		};
	}

	// Group by bucket width; the largest group wins and the rest are reported as mismatched.
	const byWidth = new Map();
	for (const r of usable) {
		const width = finite(r.body.bucketMs);
		const key = Number.isFinite(width) ? width : 'invalid';
		byWidth.set(key, [...(byWidth.get(key) ?? []), r]);
	}
	let merged = usable;
	let mismatched = [];
	if (byWidth.size > 1) {
		const groups = [...byWidth.entries()].sort((a, b) => b[1].length - a[1].length);
		merged = groups[0][1];
		mismatched = groups.slice(1).flatMap(([, group]) => group);
	}

	const canonical = merged[0].body;
	const bucketMs = canonical.bucketMs;
	const bucketCount = Math.max(...merged.map((r) => finite(r.body.bucketCount) || 0), 1);

	// One accumulator per (metric, path, method, type) combo.
	const combos = new Map();
	for (const r of merged) {
		for (const s of r.body.series ?? []) {
			const key = comboKey(s);
			let acc = combos.get(key);
			if (!acc) {
				acc = {
					metric: s.metric,
					path: s.path ?? null,
					method: s.method ?? null,
					type: s.type ?? null,
					count: 0,
					total: 0,
					counts: new Array(bucketCount).fill(0),
					value: false,
					meanSum: 0,
					medianSum: 0,
					p95Sum: 0,
					statWeight: 0,
					meanSums: new Array(bucketCount).fill(0),
					medianSums: new Array(bucketCount).fill(0),
					p95Sums: new Array(bucketCount).fill(0),
					statWeights: new Array(bucketCount).fill(0),
				};
				combos.set(key, acc);
			}

			const count = finite(s.count) || 0;
			acc.count += count;
			acc.total += finite(s.total) || 0;
			const counts = Array.isArray(s.counts) ? s.counts : [];
			for (let i = 0; i < bucketCount && i < counts.length; i++) {
				const c = finite(counts[i]);
				if (Number.isFinite(c)) acc.counts[i] += c;
			}

			// A series is a distribution if the node sent one. Weight by the node's own count.
			const mean = finite(s.mean);
			if (Number.isFinite(mean) && count > 0) {
				acc.value = true;
				acc.meanSum += mean * count;
				acc.statWeight += count;
				const median = finite(s.median);
				if (Number.isFinite(median)) acc.medianSum += median * count;
				const p95 = finite(s.p95);
				if (Number.isFinite(p95)) acc.p95Sum += p95 * count;
			}
			if (Array.isArray(s.means)) {
				acc.value = true;
				for (let i = 0; i < bucketCount && i < s.means.length; i++) {
					const weight = finite(counts[i]);
					const m = finite(s.means[i]);
					if (Number.isFinite(m) && Number.isFinite(weight) && weight > 0) {
						acc.meanSums[i] += m * weight;
						acc.statWeights[i] += weight;
						// Carried through the cluster merge as well, or the typical-case line exists on one
						// node and vanishes the moment an operator switches to cluster scope — which is the
						// default view, so it would effectively not exist at all.
						const med = finite(s.medians?.[i]);
						if (Number.isFinite(med)) acc.medianSums[i] += med * weight;
						const p = finite(s.p95s?.[i]);
						if (Number.isFinite(p)) acc.p95Sums[i] += p * weight;
					}
				}
			}
		}
	}

	const series = [...combos.values()].map((acc) => {
		const out = {
			metric: acc.metric,
			path: acc.path,
			method: acc.method,
			type: acc.type,
			count: acc.count,
			total: acc.total,
			counts: acc.counts,
		};
		if (acc.value && acc.statWeight > 0) {
			out.mean = acc.meanSum / acc.statWeight;
			out.median = acc.medianSum / acc.statWeight;
			out.p95 = acc.p95Sum / acc.statWeight;
			out.means = acc.statWeights.map((w, i) => (w > 0 ? acc.meanSums[i] / w : null));
			out.medians = acc.statWeights.map((w, i) => (w > 0 ? acc.medianSums[i] / w : null));
			out.p95s = acc.statWeights.map((w, i) => (w > 0 ? acc.p95Sums[i] / w : null));
		}
		return out;
	});
	series.sort((a, b) => b.count - a.count || a.metric.localeCompare(b.metric));

	// Per-node totals WITHOUT buckets: enough for the node table's throughput column (which
	// otherwise has to print "—" for every node but one) at a few hundred bytes per node,
	// instead of shipping four full bucketed payloads.
	const byNode = merged.map((r) => ({
		origin: r.origin,
		hostname: r.hostname,
		// The node's OWN `server.hostname`, which is the key QueueStatus rows use. `hostname`
		// above is the configured origin's host and carries the port, so it does not join
		// against the node table — this is the field that does.
		node: r.body.node ?? null,
		rangeMs: r.body.rangeMs ?? null,
		truncated: !!r.body.truncated,
		scan: r.body.scan ?? null,
		totals: (r.body.series ?? []).map((s) => ({
			metric: s.metric,
			path: s.path ?? null,
			method: s.method ?? null,
			type: s.type ?? null,
			count: finite(s.count) || 0,
		})),
		buckets: perNodeBuckets(r.body.series, bucketCount),
		// Harper's own per-minute resource rows for that node (plugin v0.92.0+): CPU, memory, event loop,
		// disk. Passed through per node, never summed — see `hosts` on the overview merge.
		system: r.body.system ?? null,
	}));

	// Config that should be identical across nodes; a disagreement changes what the reference
	// bands on the charts mean, so it is flagged rather than quietly taken from node one.
	const intervalsJson = merged.map((r) => JSON.stringify(r.body.intervals ?? null));
	const intervalsDiverge = new Set(intervalsJson).size > 1;

	const truncated = merged.some((r) => r.body.truncated);
	const startMs = minOf(merged, (r) => r.body.startMs);
	const skewMs = (maxOf(merged, (r) => r.body.startMs) ?? 0) - (startMs ?? 0);

	const failedForMerge = [
		...results.filter((r) => !r.ok),
		...mismatched.map((r) => ({
			...r,
			ok: false,
			error: `bucket width ${r.body.bucketMs}ms ≠ ${bucketMs}ms (different analytics range clamp) — excluded from the merge`,
		})),
	];
	const sourceRows = [...merged.map((r) => ({ ...r, ok: true })), ...failedForMerge];

	return {
		status: 200,
		body: {
			available: true,
			scope: 'cluster',
			node: null,
			workerIndex: null,
			rangeMs: canonical.rangeMs,
			startMs,
			endMs: maxOf(merged, (r) => r.body.endMs),
			bucketMs,
			bucketCount,
			// The COVERED window is the intersection, not the union: a number summed across
			// nodes is only honest over the span every contributing node actually reached.
			coveredFromMs: truncated ? maxOf(merged, (r) => r.body.coveredFromMs) : startMs,
			coveredToMs: truncated ? minOf(merged, (r) => r.body.coveredToMs) : maxOf(merged, (r) => r.body.endMs),
			truncated,
			skewMs,
			intervals: canonical.intervals ?? null,
			intervalsDiverge,
			scan: {
				scans: merged.length,
				ms: maxOf(merged, (r) => r.body.scan?.ms) ?? 0,
				scanned: sumOf(merged, (r) => r.body.scan?.scanned) ?? 0,
				kept: sumOf(merged, (r) => r.body.scan?.kept) ?? 0,
				cap: sumOf(merged, (r) => r.body.scan?.cap) ?? 0,
			},
			// The stalest cache in the merge bounds how fresh the whole window is.
			cacheAgeMs: maxOf(merged, (r) => r.body.cacheAgeMs),
			series,
			byNode,
			sources: sourcesOf(sourceRows, { mode: 'merged' }),
		},
	};
}

// ---------------------------------------------------------------- overview

/**
 * Merge per-node overviews.
 *
 * The headline is `backlog`: each node's snapshot covers only the RenderSchedule keys it owns
 * (the table is residency-pinned, and a node cannot read another node's rows without an
 * unbounded replication fetch), so summing the per-node snapshots is the ONLY way to see the
 * cluster's real render backlog. No single node can report it, and until now the console
 * couldn't either — an operator had to open four consoles and add up the tiles.
 *
 * The table COUNTS are the opposite case and must not be summed: `Target`, `PrerenderedPage`
 * and `Sitemap` all replicate, so every node counts the same corpus. They are taken from the
 * freshest snapshot and checked for disagreement — which is itself a finding, since a
 * persistent spread between nodes means replication has a gap.
 */
export function mergeOverview(results) {
	const usable = okBodies(results);
	if (!usable.length) return allFailed(results, 'merged');

	const bodies = usable.map((r) => ({ ...r, b: r.body }));

	// ---- nodes: QueueStatus replicates, so every node holds a copy of every node's row. That
	// makes this the one place a fan-out sees the SAME row through four different eyes, and the
	// two things worth knowing here both come out of that, without any node writing anything.
	//
	// LIVENESS IS NOT IN THE ROW. `statusChangedTime` is when the status last moved, and a busy
	// node holding `queued` legitimately never moves — so no age threshold over it can separate
	// "healthy and steady" from "dead". But the console just asked every node for this payload,
	// so it knows exactly which ones answered, as of now. That is the honest liveness signal and
	// it costs nothing. (Under node scope there is no fan-out and therefore no claim: the field
	// is left null rather than guessed.)
	//
	// DISAGREEMENT IS THE OTHER SIGNAL, and it is the one nothing else can see. All four copies
	// of a row should be identical once replication has caught up, so a persistent spread means
	// replication is not delivering that node's writes. Measured in the field: one node's own
	// copy of its row was 3 hours newer than every peer's, because its outbound replication for
	// that database had been dead since morning while every other database on the same link
	// flowed. Taking the freshest copy — which is what this merge used to do, silently — hides
	// precisely that.
	// Both keys, lowercased and port-stripped: a QueueStatus row is keyed by the node's OWN
	// `server.hostname`, while a failed result only ever carries the configured origin's host —
	// so neither alone joins every case. Same two-sided join, for the same reason, as the
	// per-node throughput column.
	const asked = new Map(); // hostname -> did it answer
	const hostKey = (value) => String(value).toLowerCase().split(':')[0];
	for (const r of results) {
		for (const key of [r.body?.node, r.hostname]) {
			if (!key) continue;
			asked.set(hostKey(key), (asked.get(hostKey(key)) ?? false) || !!r.ok);
		}
	}

	const copies = new Map(); // hostname -> [{ reporter, at, status }]
	for (const { hostname: reporter, b } of bodies) {
		for (const node of b.nodes ?? []) {
			if (!copies.has(node.hostname)) copies.set(node.hostname, []);
			copies.get(node.hostname).push({ reporter, at: msOf(node.statusChangedTime), row: node });
		}
	}

	const nodes = [...copies.entries()]
		.map(([hostname, seen]) => {
			// Newest copy for the displayed status: it is the one that has actually heard from
			// the node most recently. The spread beside it is what says whether to trust that.
			const newest = seen.reduce((best, c) => (!best || !(best.at >= c.at) ? c : best), null);
			const times = seen.map((c) => c.at).filter(Number.isFinite);
			const spreadMs = times.length > 1 ? Math.max(...times) - Math.min(...times) : 0;
			const answered = asked.get(hostKey(hostname));
			return {
				...newest.row,
				// `isThisNode` is meaningless in a cluster view — there is no "this node".
				isThisNode: false,
				// null = not one of the configured nodes, so it was never asked and no claim is made.
				responding: answered === undefined ? null : answered,
				copies: seen.length,
				spreadMs,
				// Which reporters hold a copy older than the newest, and by how far. This is the
				// replication-gap detail — named reporters, so it points at a link, not a node.
				behind: seen
					.filter((c) => Number.isFinite(c.at) && Number.isFinite(newest.at) && newest.at - c.at > 0)
					.map((c) => ({ reporter: c.reporter, byMs: newest.at - c.at }))
					.sort((a, b) => b.byMs - a.byMs),
			};
		})
		.sort((a, b) => String(a.hostname).localeCompare(String(b.hostname)));

	// ---- table counts: replicated, so compare rather than add.
	const freshestCounts = freshest(bodies, (r) => msOf(r.b.countsAsOf));
	const counts = mergeCounts(bodies, freshestCounts?.b?.counts ?? null);

	// ---- backlog: node-local slices of one cluster-wide queue. SUM.
	const runs = bodies.map((r) => r.b.backlog?.lastRun).filter(Boolean);
	const bucketLength = Math.max(0, ...runs.map((run) => (Array.isArray(run.buckets) ? run.buckets.length : 0)));
	const summedBuckets = sumArrays(
		runs.map((run) => (run.buckets ?? []).map((bucket) => bucket.count)),
		bucketLength
	).map((count, hour) => ({ hour, count }));

	// Since plugin v0.93.0 a snapshot's queue half comes from the node's queue keeper, and a node
	// whose keeper was not live at snapshot time reports `queueUnavailable` with a null `overdue`.
	// `sumOf` skips a null, so that node would silently drop out of "due now" — named instead, the
	// same way as a node with no snapshot at all.
	const unavailable = bodies
		.filter((r) => r.b.backlog?.lastRun && !r.b.backlog.lastRun.error && !Number.isFinite(r.b.backlog.lastRun.overdue))
		.map((r) => r.hostname);
	const backlogRun = runs.length
		? {
				overdue: sumOf(runs, (run) => run.overdue),
				inFlight: sumOf(runs, (run) => run.inFlight),
				buckets: summedBuckets,
				scanned: sumOf(runs, (run) => run.scanned),
				truncated: runs.some((run) => run.truncated),
				horizonMs: runs[0]?.horizonMs ?? null,
				source: runs.find((run) => run.source)?.source ?? null,
				// A sum is only as fresh as its STALEST input — and only as complete as its
				// narrowest, so a node with no snapshot yet is named rather than counted as zero.
				finishedAt: minOf(runs, (run) => msOf(run.finishedAt)),
				startedAt: minOf(runs, (run) => msOf(run.startedAt)),
				counts,
				error: runs.find((run) => run.error)?.error ?? null,
				nodes: runs.length,
				missing: bodies.filter((r) => !r.b.backlog?.lastRun).map((r) => r.hostname),
				unavailable,
			}
		: null;

	// ---- leases: live, per node. In flight ADDS across nodes — it is the cluster's render
	// concurrency — but slot pressure is per node (each node has its own lease table, and a full one
	// refuses grants there however empty the others are), so the worst node is kept beside the sum.
	const leaseRows = bodies
		.filter((r) => r.b.leases && typeof r.b.leases === 'object')
		.map((r) => ({
			hostname: r.hostname,
			occupancy: num(finite(r.b.leases.occupancy)),
			maxLeases: num(finite(r.b.leases.maxLeases)),
			oldestLeaseAgeMs: num(finite(r.b.leases.oldestLeaseAgeMs)),
			oldestLeaseExpiresAt: num(finite(r.b.leases.oldestLeaseExpiresAt)),
		}));
	const slotShare = (row) =>
		row.maxLeases > 0 && Number.isFinite(row.occupancy) ? row.occupancy / row.maxLeases : NaN;
	let fullest = null;
	for (const row of leaseRows) {
		const share = slotShare(row);
		if (Number.isFinite(share) && (fullest === null || share > slotShare(fullest))) fullest = row;
	}
	const leases = leaseRows.length
		? {
				occupancy: sumOf(leaseRows, (row) => row.occupancy),
				maxLeases: sumOf(leaseRows, (row) => row.maxLeases),
				oldestLeaseAgeMs: maxOf(leaseRows, (row) => row.oldestLeaseAgeMs),
				oldestLeaseExpiresAt: minOf(leaseRows, (row) => row.oldestLeaseExpiresAt),
				fullestNode: fullest?.hostname ?? null,
				fullestShare: fullest ? slotShare(fullest) : null,
				// A node that answered without the block (an older plugin) is named, not counted as idle.
				missing: bodies.filter((r) => !(r.b.leases && typeof r.b.leases === 'object')).map((r) => r.hostname),
				byNode: leaseRows,
			}
		: null;

	// ---- reconcile: per-node sweeps over each node's own keys. Sum the work, and name any
	// node where the sweep is OFF — one node with reconcile disabled is a silent hole in the
	// corpus, and a cluster view that reported "enabled" because three nodes were would hide it.
	const reconciles = bodies.map((r) => ({ hostname: r.hostname, ...(r.b.reconcile ?? {}) }));
	const sweeps = reconciles.map((r) => r.lastRun).filter(Boolean);
	const reconcile = {
		enabled: reconciles.every((r) => r.enabled !== false),
		disabledOn: reconciles.filter((r) => r.enabled === false).map((r) => r.hostname),
		interval: reconciles.find((r) => Number.isFinite(r.interval))?.interval ?? null,
		running: reconciles.some((r) => r.running),
		lastRun: sweeps.length
			? {
					examined: sumOf(sweeps, (s) => s.examined),
					owned: sumOf(sweeps, (s) => s.owned),
					restored: sumOf(sweeps, (s) => s.restored),
					truncated: sweeps.some((s) => s.truncated),
					finishedAt: minOf(sweeps, (s) => msOf(s.finishedAt)),
					error: sweeps.find((s) => s.error)?.error ?? null,
					nodes: sweeps.length,
				}
			: null,
	};

	// ---- orphan sweep: same node scope as reconcile, so the same sum — but MANUAL, which
	// changes what the merge has to say. There is no cadence to report and no "disabled on"
	// hole to name; the interesting shortfall is the opposite one, a node nobody has swept.
	// `sweptNodes` vs the answering node count is what says so, since a node that has never run
	// it contributes zero to every total here and that is indistinguishable from a clean node.
	const orphanRuns = bodies.map((r) => ({ hostname: r.hostname, ...(r.b.orphanSweep ?? {}) }));
	const orphanSweeps = orphanRuns.map((r) => r.lastRun).filter(Boolean);
	const orphanSweep = {
		maxDeletes: orphanRuns.find((r) => Number.isFinite(r.maxDeletes))?.maxDeletes ?? null,
		dryRunDefault: orphanRuns.some((r) => r.dryRunDefault),
		running: orphanRuns.some((r) => r.running),
		sweptNodes: orphanSweeps.length,
		unsweptNodes: orphanRuns.filter((r) => !r.lastRun).map((r) => r.hostname),
		lastRun: orphanSweeps.length
			? {
					examined: sumOf(orphanSweeps, (s) => s.examined),
					owned: sumOf(orphanSweeps, (s) => s.owned),
					orphaned: sumOf(orphanSweeps, (s) => s.orphaned),
					deleted: sumOf(orphanSweeps, (s) => s.deleted),
					leaseSkipped: sumOf(orphanSweeps, (s) => s.leaseSkipped),
					truncated: orphanSweeps.some((s) => s.truncated),
					// A cluster figure counts as a dry run only if EVERY node's was: one node that
					// actually deleted makes "nothing was deleted" false.
					dryRun: orphanSweeps.every((s) => s.dryRun),
					finishedAt: minOf(orphanSweeps, (s) => msOf(s.finishedAt)),
					error: orphanSweeps.find((s) => s.error)?.error ?? null,
					nodes: orphanSweeps.length,
				}
			: null,
	};

	const controlRows = bodies.map((r) => r.b.control?.cluster).filter(Boolean);
	const intervalsJson = bodies.map((r) => JSON.stringify(r.b.intervals ?? null));

	return {
		status: 200,
		body: {
			// The aggregate is as of its oldest constituent, not its newest.
			generatedAt: minOf(bodies, (r) => r.b.generatedAt),
			node: null,
			workerIndex: null,
			scope: 'cluster',
			localQueueStatus: null,
			queueStatusByNode: Object.fromEntries(bodies.map((r) => [r.hostname, r.b.localQueueStatus ?? null])),
			// Point-in-time host vitals (plugin v0.92.0+), one entry per node that answered and NEVER merged:
			// a CPU, memory or uptime figure means something only for the node that reported it, and an
			// average of four would hide exactly the node that is in trouble. `host` is null for a plugin
			// that predates it.
			hosts: bodies.map((r) => ({
				origin: r.origin,
				hostname: r.hostname,
				node: r.b.node ?? null,
				host: r.b.host ?? null,
			})),
			control: {
				cluster: freshest(controlRows, (row) => msOf(row.updatedTime)) ?? null,
				knownScopes: [...new Set(bodies.flatMap((r) => r.b.control?.knownScopes ?? []))],
			},
			nodes,
			counts,
			countsAsOf: freshestCounts?.b?.countsAsOf ?? null,
			backlog: {
				enabled: bodies.some((r) => r.b.backlog?.enabled),
				disabledOn: bodies.filter((r) => r.b.backlog?.enabled === false).map((r) => r.hostname),
				interval: bodies.find((r) => Number.isFinite(r.b.backlog?.interval))?.b.backlog.interval ?? null,
				running: bodies.some((r) => r.b.backlog?.running),
				lastRun: backlogRun,
			},
			intervals: bodies[0].b.intervals ?? null,
			intervalsDiverge: new Set(intervalsJson).size > 1,
			leases,
			reconcile,
			orphanSweep,
			sources: sourcesOf(results, { mode: 'merged' }),
		},
	};
}

/**
 * Table counts across nodes. Every counted table replicates, so the nodes should agree;
 * `divergent` marks the ones that don't. A count is an estimate on a large table, so the
 * threshold is generous — this is looking for a replication GAP, not for jitter.
 */
const DIVERGENCE_THRESHOLD = 0.02;

function mergeCounts(bodies, representative) {
	if (!representative) return null;
	const out = {};
	for (const [table, value] of Object.entries(representative)) {
		const observed = bodies
			.map((r) => ({ hostname: r.hostname, count: finite(r.b.counts?.[table]?.recordCount) }))
			.filter((o) => Number.isFinite(o.count));
		const high = Math.max(...observed.map((o) => o.count), 0);
		const low = observed.length ? Math.min(...observed.map((o) => o.count)) : 0;
		const divergent = observed.length > 1 && high > 0 && (high - low) / high > DIVERGENCE_THRESHOLD;
		out[table] = {
			...value,
			divergent,
			...(divergent ? { spread: { low, high, byNode: observed } } : {}),
		};
	}
	return out;
}

// ---------------------------------------------------------------- unrouted

/**
 * The unrouted tally, summed across nodes.
 *
 * It stays labelled `perWorker` because it still is: each node answers from ONE worker's
 * in-process counters, so the cluster sum is a sum of N single-worker slices, not the cluster's
 * traffic. That understatement is why the panel exists at all (it answers "is anything hitting
 * a route we don't classify", not "how much"), and the payload keeps saying so.
 */
export function mergeUnrouted(results) {
	const usable = okBodies(results);
	if (!usable.length) return allFailed(results, 'merged');

	const classes = new Set();
	for (const r of usable) {
		for (const key of Object.keys(r.body.report ?? {})) {
			if (Array.isArray(r.body.report[key])) classes.add(key);
		}
	}

	const report = { overflowed: usable.some((r) => r.body.report?.overflowed) };
	for (const routeClass of classes) {
		const merged = new Map();
		for (const r of usable) {
			for (const row of r.body.report?.[routeClass] ?? []) {
				const existing = merged.get(row.bucket);
				if (existing) existing.count += finite(row.count) || 0;
				else merged.set(row.bucket, { bucket: row.bucket, count: finite(row.count) || 0, samplePath: row.samplePath });
			}
		}
		report[routeClass] = [...merged.values()].sort((a, b) => b.count - a.count);
	}

	return {
		status: 200,
		body: {
			node: null,
			workerIndex: null,
			scope: 'cluster',
			perWorker: true,
			workers: usable.length,
			interval: usable[0].body.interval ?? null,
			report,
			sources: sourcesOf(results, { mode: 'merged' }),
		},
	};
}

// ---------------------------------------------------------------- config

/** Bound on the reported divergence list — a truly divergent pair would otherwise be huge. */
const MAX_DIVERGENCES = 60;

/**
 * Flatten an object to dotted leaf paths. Arrays are leaves (order matters and an element-wise
 * diff of, say, a route list is noise, not a finding).
 */
function flatten(value, prefix = '', out = new Map()) {
	if (value && typeof value === 'object' && !Array.isArray(value)) {
		for (const [key, child] of Object.entries(value)) flatten(child, prefix ? `${prefix}.${key}` : key, out);
	} else {
		out.set(prefix, JSON.stringify(value ?? null));
	}
	return out;
}

/**
 * Every option path some node reports a stored override for.
 *
 * TWO SOURCES, and both are load-bearing, because they disagree in exactly the window this
 * function exists to explain. `overrides.rows` is the replicated TABLE as that node last read it;
 * `layers[].overridden` is the layer that node has actually APPLIED. A set still in flight is a
 * row nobody has applied yet (table only); a CLEAR still in flight is a node running an override
 * no row backs any more (applied layer only). Reading either alone misfiles one of the two as a
 * deploy failure — see the classification below.
 */
function overriddenPaths(usable) {
	const paths = new Set();
	for (const r of usable) {
		for (const row of r.body.overrides?.rows ?? []) {
			if (typeof row?.path === 'string' && row.path) paths.add(row.path);
		}
		for (const layer of r.body.layers ?? []) {
			if (layer?.overridden && typeof layer.path === 'string' && layer.path) paths.add(layer.path);
		}
	}
	return paths;
}

/**
 * The override layer, cluster-wide.
 *
 * The ROWS replicate, so they are the union and `missingOn` names any node a row has not reached —
 * the direct readout of "the edit is still landing" that keeps an operator from reading the
 * divergence list as a broken deploy.
 *
 * The WATCH STATE does not replicate and is per node and per worker: whether that node's
 * subscription is live, when its backstop last re-read, and whether either is erroring. It is
 * therefore reported per node and the scalar `watch` is deliberately ABSENT from the cluster
 * answer — a single watch object here would be node one's, and a node whose override watch has
 * died is precisely the node that silently stops honouring every edit made from this console.
 * `degraded` rolls up as ANY, for the same reason.
 *
 * Returns null for a plugin that predates the override layer, so the reference body passes
 * through untouched rather than growing an empty section.
 */
function mergeOverrides(usable) {
	if (!usable.some((r) => r.body.overrides && typeof r.body.overrides === 'object')) return null;

	const hostnames = usable.map((r) => r.hostname);
	const rows = new Map();
	for (const r of usable) {
		for (const row of r.body.overrides?.rows ?? []) {
			if (typeof row?.path !== 'string' || !row.path) continue;
			let entry = rows.get(row.path);
			if (!entry) rows.set(row.path, (entry = { row, presentOn: [] }));
			entry.presentOn.push(r.hostname);
		}
	}

	return {
		// True only if EVERY node honours the layer: the kill switch is a file option, so a node
		// where it is off is running the deployed config while the console lists rows it obeys.
		// (That difference is also a real config divergence, and shows up as one.)
		enabled: usable.every((r) => r.body.overrides?.enabled !== false),
		rows: [...rows.values()].map(({ row, presentOn }) => ({
			...row,
			missingOn: hostnames.filter((hostname) => !presentOn.includes(hostname)),
		})),
		degraded: usable.some((r) => r.body.overrides?.degraded === true),
		truncated: usable.some((r) => r.body.overrides?.truncated === true),
		error: usable.map((r) => r.body.overrides?.error).find((error) => error) ?? null,
		nodes: usable.map((r) => ({
			hostname: r.hostname,
			enabled: r.body.overrides?.enabled ?? null,
			degraded: r.body.overrides?.degraded ?? null,
			truncated: r.body.overrides?.truncated ?? null,
			error: r.body.overrides?.error ?? null,
			rowCount: (r.body.overrides?.rows ?? []).length,
			watch: r.body.overrides?.watch ?? null,
		})),
	};
}

/**
 * Compare each node's effective config against the first node's.
 *
 * THIS IS THE ONE PLACE A HALF-APPLIED DEPLOY IS VISIBLE. A component deploy that restarts
 * three nodes and silently skips the fourth leaves a cluster running two versions of its own
 * configuration, and every other panel keeps looking healthy — the skipped node serves traffic,
 * reports queue status, and answers this API. The only symptom is that its options differ.
 *
 * SINCE THE OVERRIDE LAYER, DISAGREEMENT HAS A SECOND CAUSE, and it is a benign one. An override
 * written from this console commits on one node and reaches the rest by replication, so between
 * the write and every node re-reading the table the cluster genuinely disagrees about that path —
 * a state that ends by itself in about a second, or by `management.overrides.syncInterval` if a
 * node's subscription never established. Left unclassified, every config edit an operator makes
 * here raises the alarm that means "a deploy skipped a node", and an alarm that fires on its own
 * console's normal operation stops being read at all. So each divergence carries `overridden`:
 * true means some node has an override in play at that path and this is the layer converging (or,
 * if it persists, an override one node is refusing — the `override-rejected` state, visible in
 * that node's `layers`); false is the original finding, unchanged and still a deploy failure.
 */
export function mergeConfig(results) {
	const usable = okBodies(results);
	if (!usable.length) return allFailed(results, 'compared');

	const reference = usable[0];
	const referenceFlat = flatten(reference.body.config ?? {});
	const overridden = overriddenPaths(usable);

	const divergences = [];
	let truncated = false;
	for (const r of usable.slice(1)) {
		const flat = flatten(r.body.config ?? {});
		const paths = new Set([...referenceFlat.keys(), ...flat.keys()]);
		for (const path of paths) {
			if (referenceFlat.get(path) === flat.get(path)) continue;
			if (divergences.length >= MAX_DIVERGENCES) {
				truncated = true;
				break;
			}
			let entry = divergences.find((d) => d.path === path);
			if (!entry) {
				entry = {
					path,
					overridden: overridden.has(path),
					values: [{ hostname: reference.hostname, value: referenceFlat.get(path) ?? null }],
				};
				divergences.push(entry);
			}
			entry.values.push({ hostname: r.hostname, value: flat.get(path) ?? null });
		}
	}

	const tagged = (key) => usable.flatMap((r) => (r.body[key] ?? []).map((item) => ({ ...item, hostname: r.hostname })));
	const overrides = mergeOverrides(usable);

	return {
		status: 200,
		body: {
			...reference.body,
			scope: 'cluster',
			node: null,
			workerIndex: null,
			// The reference node's own config, named — the pre view is one node's truth, and
			// pretending it is "the cluster's config" is exactly what divergence disproves. `layers`
			// rides along from the same node and is covered by the same caveat: it is that node's
			// per-option provenance, including its own `override-rejected` verdicts. Publishing all N
			// nodes' layers would be N×~130 rows to say what the divergence list already says in the
			// only terms that matter — the values that actually differ.
			configFrom: reference.hostname,
			divergences,
			divergencesTruncated: truncated,
			warnings: tagged('warnings'),
			pendingRestart: tagged('pendingRestart'),
			...(overrides ? { overrides } : {}),
			sources: sourcesOf(results, { mode: 'compared' }),
		},
	};
}

// ---------------------------------------------------------------- change probe

/**
 * Pass counters that ADD across nodes: each node probes a disjoint slice of the keyspace.
 *
 * THIS LIST IS NOT A PARTITION and summing across it means nothing. `probed` is the total the
 * outcome counters divide up; `throttled` sits INSIDE `failed` (a pushback response is one kind of
 * failed probe, tagged with who caused it); `fresh` sits outside `probed` entirely, because a
 * skipped URL was never attempted. Each entry sums across nodes on its own, and only on its own.
 *
 * `unreadable` is deliberately NOT here: it is a fault of the registry WALK, not an outcome of a
 * pass, and only the sweep produces it. The canary's per-rule merge below iterates this list, and
 * a walk fault has no per-rule meaning to iterate.
 */
const PROBE_COUNTERS = [
	'examined',
	'owned',
	'matched',
	'probed',
	'seeded',
	'unchanged',
	'changed',
	// Changes acted on — page hard-expired, render filed ahead of rotation (plugin v0.94.0; before it,
	// re-renders filed). Plugin v0.94.0 removed `deferred` and `queued` with the trigger queue.
	'triggered',
	'failed',
	'fresh',
	'throttled',
	// OVERLAYS the outcome buckets rather than joining them: a row whose cached page disagreed
	// with the origin (pageCheck, plugin v0.58.0) is ALSO inside `changed` or `unchanged`, so this
	// sums on its own and belongs in no subtraction — same footing as `throttled` inside `failed`.
	'pageMismatch',
	// Actions that threw: page left as it was, baseline stale, so the next probe acts again.
	'errors',
	// Added after the list was first written, and each one was DROPPED by this merge until console
	// v0.16.0 — the cluster view simply did not have them, while every node reported them.
	// `rebaselined` is the one that mattered most: the compared denominator subtracts it, so without
	// it a cluster pass right after a rule edit read as "nothing changed" over rows never compared.
	'rebaselined',
	'extended', // plugin v0.86.0 — overlays unchanged/changed, like pageMismatch
	'caughtUp', // v0.88.0 — overlays changed
	'ignored', // v0.88.0 — overlays unchanged
	'outOfScope', // v0.89.0 — outside matched entirely
	'behindBatches',
];

/**
 * Per-pass maps of counts, `{ [rule]: { [key]: n } }`: origin changes per extract slot and page
 * mismatches per mapped field (plugin v0.88.0). Each node counts its own slice, so the leaves add.
 */
const PROBE_COUNT_MAPS = ['slotChanges', 'fieldMismatch'];

const sumCountMaps = (runs, key) => {
	const out = {};
	let seen = false;
	for (const run of runs) {
		const map = run[key];
		if (!map || typeof map !== 'object') continue;
		seen = true;
		for (const [rule, counts] of Object.entries(map)) {
			for (const [leaf, n] of Object.entries(counts ?? {})) {
				const value = finite(n);
				if (!Number.isFinite(value)) continue;
				out[rule] ??= {};
				out[rule][leaf] = (out[rule][leaf] ?? 0) + value;
			}
		}
	}
	return seen ? out : null;
};

/**
 * The mapping-defect guard, across nodes. The guard is per PROCESS — each node judges its own
 * witnessed comparisons — so the counts add but the verdict does not: a field disarmed on one node
 * is still triggering on the others, and the merge names the node rather than folding it.
 */
const mergeFieldGuards = (runs) => {
	const out = {};
	let seen = false;
	for (const run of runs) {
		const guard = run.fieldGuard;
		if (!guard || typeof guard !== 'object') continue;
		seen = true;
		for (const [rule, fields] of Object.entries(guard)) {
			for (const [field, entry] of Object.entries(fields ?? {})) {
				out[rule] ??= {};
				const merged = (out[rule][field] ??= { witnessed: 0, disagreed: 0, armed: true, disarmedOn: [] });
				merged.witnessed += finite(entry?.witnessed) || 0;
				merged.disagreed += finite(entry?.disagreed) || 0;
				if (entry?.armed === false) {
					merged.armed = false;
					merged.disarmedOn.push(run.hostname);
				}
			}
		}
	}
	return seen ? out : null;
};

/**
 * What makes two nodes' rule lists the same, as a comparable string per node.
 *
 * NOT the raw objects: plugin v0.91.0 added `fingerprint`, `extract` and `endpoint` to each rule, so
 * a JSON comparison would flag every mixed-version cluster — every rolling deploy — as "the nodes
 * disagree on the rules". The fields every version reports are compared always; the fingerprint
 * (which sees an edited extract path or header, where the rest cannot) only when every node has one.
 */
const ruleIdentity = (bodies) => {
	const withFingerprints = bodies.every((r) => (r.b.rules ?? []).every((rule) => typeof rule.fingerprint === 'string'));
	return bodies.map((r) =>
		JSON.stringify(
			(r.b.rules ?? []).map((rule) => [
				rule.label ?? null,
				rule.pathPattern ?? null,
				rule.source ?? null,
				rule.invalidateScope ?? null,
				rule.pageFields ?? null,
				rule.ignoreChanges ?? null,
				withFingerprints ? rule.fingerprint : null,
			])
		)
	);
};

/** Bound on the pooled failure samples — they exist to name a shape change, not to be a log. */
const MAX_FAILURE_SAMPLES = 6;

const sumCounters = (runs, keys) => Object.fromEntries(keys.map((key) => [key, sumOf(runs, (r) => r[key])]));

/**
 * Which nodes' pass records lack a counter — a node on a plugin older than the counter. Their slice
 * is missing from that one sum, which reads as a smaller number rather than a partial one unless
 * the payload says so. Only counters some node DOES report are listed.
 */
const unreportedCounters = (runs, keys) => {
	const out = {};
	for (const key of keys) {
		const missing = runs.filter((run) => !Number.isFinite(finite(run[key]))).map((run) => run.hostname);
		if (missing.length && missing.length < runs.length) out[key] = missing;
	}
	return out;
};

/** The one value every node that reports it agrees on, else null (the caller reports the split). */
const agreed = (values) => {
	const present = values.filter((value) => value !== null && value !== undefined);
	if (!present.length) return null;
	const first = JSON.stringify(present[0]);
	return present.every((value) => JSON.stringify(value) === first) ? present[0] : null;
};

const diverges = (values) =>
	new Set(values.filter((value) => value !== null && value !== undefined).map((value) => JSON.stringify(value))).size >
	1;

/**
 * Merge the change-probe status of every node into one cluster answer.
 *
 * THE COUNTERS ADD, THE VERDICTS DO NOT. A probe pass is owner-scoped — each node walks the
 * registry rows it owns — so `probed`, `changed` and the rest partition the corpus and summing
 * them is the only way to get a cluster change rate at all. The canary's TRIP is the opposite: it
 * is a threshold crossed against ONE node's cohort, and a cluster-wide "tripped: true" folded out
 * of four independent verdicts would name no node and match no log line. So trips are reported as
 * the nodes that tripped, and the pooled fraction rides alongside as the wider sample it is.
 *
 * TIME IS TAKEN AS THE OLDEST, never the newest — the same rule the orphan sweep merge follows.
 * Nodes stagger their first sweep on purpose (`startJitter`), so their last passes finished at
 * different times, and a summary stamped with the freshest of them would present a figure that is
 * hours stale on three nodes out of four as current.
 *
 * A NODE THAT HAS NOT SWEPT CONTRIBUTES ZERO TO EVERY SUM, which is indistinguishable from a node
 * that swept and found nothing. `unsweptNodes` is what separates them, and the UI says so rather
 * than letting the shortfall read as a smaller change rate.
 */
export function mergeChangeProbe(results) {
	const usable = okBodies(results);
	if (!usable.length) return allFailed(results, 'merged');

	const bodies = usable.map((r) => ({ hostname: r.hostname, b: r.body }));

	const sweeps = bodies
		.map((r) => ({ hostname: r.hostname, ...(r.b.sweep?.lastRun ?? {}) }))
		.filter((r) => r.startedAt);
	const canaries = bodies
		.map((r) => ({ hostname: r.hostname, ...(r.b.canary?.lastRun ?? {}) }))
		.filter((r) => r.startedAt);

	// Rules are configuration, so they are identical by intent and one node's copy is the answer.
	// A disagreement is a deploy that skipped a node, which is the Config view's finding and not
	// this one's — flagged here, diagnosed there.
	const ruleJson = ruleIdentity(bodies);
	const rules = bodies.find((r) => (r.b.rules ?? []).length)?.b.rules ?? [];
	const modes = bodies.map((r) => r.b.mode ?? null);
	const armedSweeps = bodies.map((r) => r.b.sweep?.armedInterval ?? null);

	// Per RULE, across nodes: the counters add (disjoint cohorts), the verdict is named per node.
	const perRule = new Map();
	for (const run of canaries) {
		for (const entry of run.perRule ?? []) {
			const merged = perRule.get(entry.rule) ?? {
				rule: entry.rule,
				cohort: 0,
				...Object.fromEntries(PROBE_COUNTERS.map((key) => [key, 0])),
				compared: 0,
				trippedOn: [],
				actions: [],
				emptyCohortOn: [],
			};
			merged.cohort += finite(entry.cohort) || 0;
			for (const key of PROBE_COUNTERS) merged[key] += finite(entry[key]) || 0;
			merged.compared += finite(entry.compared) || 0;
			if (entry.tripped) merged.trippedOn.push(run.hostname);
			if (entry.skipped) merged.emptyCohortOn.push(run.hostname);
			if (entry.action) merged.actions.push({ hostname: run.hostname, ...entry.action });
			perRule.set(entry.rule, merged);
		}
	}
	// Pooled over the union of the nodes' cohorts: a wider sample of the same population, and the
	// number to read when one node's cohort is too small to decide anything on its own.
	for (const entry of perRule.values()) {
		entry.fraction = entry.compared > 0 ? entry.changed / entry.compared : null;
	}

	const failureSamples = sweeps
		.flatMap((run) => (run.failureSamples ?? []).map((sample) => ({ hostname: run.hostname, ...sample })))
		.slice(0, MAX_FAILURE_SAMPLES);

	return {
		status: 200,
		body: {
			node: null,
			workerIndex: null,
			scope: 'cluster',
			// ONE node with the probe off is the finding, not "the cluster has it on": that node's
			// owned slice is never probed and every panel here keeps looking healthy.
			enabled: bodies.some((r) => r.b.enabled),
			disabledOn: bodies.filter((r) => r.b.enabled === false).map((r) => r.hostname),
			// Dry run only if EVERY node is: one node acting makes "nothing is being triggered" false.
			dryRun: bodies.every((r) => r.b.dryRun !== false),
			liveOn: bodies.filter((r) => r.b.dryRun === false).map((r) => r.hostname),
			ownerScopeNote: bodies[0].b.ownerScopeNote ?? null,
			rules,
			rulesDiverge: new Set(ruleJson).size > 1,
			// The sweep's scheduling mode. `null` with `modeDiverge` is the finding: the nodes are not
			// running the same schedule, which is a deploy or override that did not land everywhere.
			mode: agreed(modes),
			modeDiverge: diverges(modes),
			settingsDiverge: diverges(bodies.map((r) => r.b.settings ?? null)),
			// 1 is every plugin before v0.91.0, which reports no running-pass counters, no next run and
			// no settings. The OLDEST node decides what the cluster view can promise.
			statusVersion: minOf(bodies, (r) => r.b.statusVersion ?? 1),
			stateUpdatedAt: minOf(bodies, (r) => msOf(r.b.stateUpdatedAt)),
			// Plugin v0.62.0 publishes the probe's state to a node-local row every worker can read,
			// and says when that row could NOT be read: `stateAvailable: false` is "this answer is not
			// trustworthy", which the old per-worker shape could not express (it reported `running:
			// false, lastRun: null` — indistinguishable from the probe being off). Named per node,
			// because a node whose row is unreadable contributes zeros to every sum above exactly the
			// way an unswept node does, and the UI has to say which it is. An older plugin that has no
			// such field is not flagged: absent is "not reported", not "unavailable".
			stateUnavailableOn: bodies.filter((r) => r.b.stateAvailable === false).map((r) => r.hostname),
			sweep: {
				running: bodies.some((r) => r.b.sweep?.running),
				runningOn: bodies.filter((r) => r.b.sweep?.running).map((r) => r.hostname),
				// Live progress of the passes in flight, per node — a sweep runs for hours, and "running"
				// alone cannot tell a pass that is moving from one whose worker died (the plugin's own
				// heartbeat staleness handles the latter; this is the number an operator watches move).
				progress: bodies
					.filter((r) => r.b.sweep?.running && r.b.sweep?.progress)
					.map((r) => ({ hostname: r.hostname, ...r.b.sweep.progress })),
				// ANY armed value, not only a numeric one. The plugin tags the mode into this field
				// ('continuous', or 'anchored:<time>|<zone>'), and a `Number.isFinite` filter here turned
				// every continuous or anchored cluster into "not armed" on the cluster view — the default
				// view — while every node was armed and sweeping.
				armedInterval: agreed(armedSweeps) ?? armedSweeps.find((value) => value !== null) ?? null,
				armedIntervalDiverge: diverges(armedSweeps),
				cycleTarget: agreed(bodies.map((r) => r.b.sweep?.cycleTarget ?? null)),
				sliceSize: sumOf(bodies, (r) => r.b.sweep?.sliceSize),
				// The EARLIEST next run across nodes: the soonest anything will start.
				nextAnchoredRunAt: isoOrNull(minOf(bodies, (r) => msOf(r.b.sweep?.nextAnchoredRunAt))),
				nextRunAt: minOf(bodies, (r) => msOf(r.b.sweep?.nextRunAt)),
				sweptNodes: sweeps.length,
				unsweptNodes: bodies.filter((r) => !r.b.sweep?.lastRun?.startedAt).map((r) => r.hostname),
				lastRun: sweeps.length
					? {
							...sumCounters(sweeps, PROBE_COUNTERS),
							...Object.fromEntries(PROBE_COUNT_MAPS.map((key) => [key, sumCountMaps(sweeps, key)])),
							fieldGuard: mergeFieldGuards(sweeps),
							unreported: unreportedCounters(sweeps, [...PROBE_COUNTERS, 'unreadable']),
							// A walk fault, summed like a counter but reported apart from the pass outcomes.
							unreadable: sumOf(sweeps, (run) => run.unreadable),
							// The most actions any node had in flight at once (plugin v0.94.0) — a high-water mark, not a sum.
							maxActionsInFlight: maxOf(sweeps, (run) => run.maxActionsInFlight),
							// The longest any node's walk was blocked on a full action pipeline — the worst node, since
							// each wait is a share of that node's own pass.
							actionWaitMs: maxOf(sweeps, (run) => run.actionWaitMs),
							// Nodes whose last pass finished one a restart cut short (plugin v0.94.0): their counts
							// cover only the rows the interrupted pass had not reached.
							resumedOn: sweeps.filter((run) => run.startedBy === 'resume').map((run) => run.hostname),
							dryRun: sweeps.every((run) => run.dryRun !== false),
							aborted: sweeps.some((run) => run.aborted),
							// WHY `some` AND NOT A COUNT. One node that gave up on a refusing origin covered
							// none of its slice, and every total above is short by that slice — which reads
							// as a smaller corpus, not as a missing one. The same rule as `unsweptNodes`.
							abortedOnDistress: sweeps.some((run) => run.abortedOnDistress),
							distressedOn: sweeps.filter((run) => run.abortedOnDistress).map((run) => run.hostname),
							// The WORST window any node ended on, never a mean: the question this answers is
							// "was the probe being held back", and averaging one throttled node against three
							// healthy ones answers it with a number that was true nowhere.
							throttleLevel: maxOf(sweeps, (run) => run.throttleLevel),
							throttledOn: sweeps.filter((run) => run.throttled > 0).map((run) => run.hostname),
							startedAt: minOf(sweeps, (run) => msOf(run.startedAt)),
							finishedAt: minOf(sweeps, (run) => msOf(run.finishedAt)),
							error: sweeps.find((run) => run.error)?.error ?? null,
							nodes: sweeps.length,
							failureSamples,
						}
					: null,
			},
			canary: {
				running: bodies.some((r) => r.b.canary?.running),
				armedInterval: bodies.find((r) => Number.isFinite(r.b.canary?.armedInterval))?.b.canary.armedInterval ?? null,
				nextRunAt: minOf(bodies, (r) => msOf(r.b.canary?.nextRunAt)),
				cohortSizes: bodies.reduce((acc, r) => {
					for (const [label, size] of Object.entries(r.b.canary?.cohortSizes ?? {})) {
						acc[label] = (acc[label] ?? 0) + (finite(size) || 0);
					}
					return acc;
				}, {}),
				ranNodes: canaries.length,
				unrunNodes: bodies.filter((r) => !r.b.canary?.lastRun?.startedAt).map((r) => r.hostname),
				lastRun: canaries.length
					? {
							perRule: [...perRule.values()],
							dryRun: canaries.every((run) => run.dryRun !== false),
							startedAt: minOf(canaries, (run) => msOf(run.startedAt)),
							finishedAt: minOf(canaries, (run) => msOf(run.finishedAt)),
							error: canaries.find((run) => run.error)?.error ?? null,
							nodes: canaries.length,
						}
					: null,
			},
			// EVERY NODE'S OWN PAYLOAD, UNMERGED. The merge above is a summary, and it can only carry the
			// fields it knows; every field a plugin added after it was written used to vanish here —
			// `mode`, `nextAnchoredRunAt`, `stateUpdatedAt`, the page-record counters — while every node
			// reported them. The view reads each node's state from its own payload, so a field this
			// merge has never heard of still reaches the screen.
			perNode: bodies.map((r) => ({ ...r.b, hostname: r.hostname })),
			// Every node's own slice, unsummed — the purge/sweep actions are run per node, so this
			// is the table an operator works down.
			byNode: bodies.map((r) => ({
				hostname: r.hostname,
				enabled: r.b.enabled ?? null,
				dryRun: r.b.dryRun ?? null,
				// null on a plugin that predates the shared state row; false is the finding.
				stateAvailable: typeof r.b.stateAvailable === 'boolean' ? r.b.stateAvailable : null,
				sweepRunning: !!r.b.sweep?.running,
				sweepProgress: r.b.sweep?.running ? (r.b.sweep?.progress ?? null) : null,
				canaryRunning: !!r.b.canary?.running,
				sweepFinishedAt: msOf(r.b.sweep?.lastRun?.finishedAt) || null,
				canaryFinishedAt: msOf(r.b.canary?.lastRun?.finishedAt) || null,
				probed: num(r.b.sweep?.lastRun?.probed),
				changed: num(r.b.sweep?.lastRun?.changed),
				failed: num(r.b.sweep?.lastRun?.failed),
				// Per node BECAUSE THE RATE LIMIT IS PER NODE: a probe rate is agreed with whoever runs
				// the origin, each node holds its own, and origin pushback is therefore a fact about one
				// node's pass. Summed into a cluster figure it names nobody to go and look at.
				throttled: num(r.b.sweep?.lastRun?.throttled),
				abortedOnDistress: !!r.b.sweep?.lastRun?.abortedOnDistress,
				error: r.b.sweep?.lastRun?.error ?? r.b.canary?.lastRun?.error ?? null,
			})),
			sources: sourcesOf(results, { mode: 'merged' }),
		},
	};
}

// ---------------------------------------------------------------- discovery purge

/**
 * Purge counters that ADD across nodes: each node purges the keys it owns.
 *
 * `leaseSkipped`, `visitedSkipped` and `errors` earn their place beside `deleted` because the UI
 * subtracts all four from `discovered` to say what a pass never reached. A term missing from that
 * subtraction does not read as a missing term — it reads as rows the pass left behind for no
 * stated reason, which is the shape of an incomplete pass rather than a deliberate one.
 *
 * `unreadable` is summed here too but is NOT part of that subtraction: the walk skips those rows
 * before anything reads their `sitemapUrl`, so they never entered `discovered` in the first place.
 */
const PURGE_COUNTERS = [
	'examined',
	'owned',
	'discovered',
	'leaseSkipped',
	'visitedSkipped',
	'unreadable',
	'errors',
	'deleted',
];

/**
 * Merge every node's discovered-target purge state.
 *
 * A PURGE IS RUN PER NODE, ONE AT A TIME, and the keyspace is only covered once every node has
 * run — so `byNode` is the substance here and the totals are the running tally, not a result. The
 * merge exists to stop the alternative: a single node's "deleted 240,118" answered under a
 * "cluster" label, which reads as done when three quarters of the corpus has not been touched.
 *
 * DIFFERENT NODES MAY HAVE RUN DIFFERENT PREFIXES, and a total summed across two prefixes answers
 * a question nobody asked. `urlPrefix` is therefore only reported when every node that ran used
 * the same one; otherwise the per-node rows are the whole answer and the UI says so.
 */
export function mergeDiscoveryPurge(results) {
	const usable = okBodies(results);
	if (!usable.length) return allFailed(results, 'merged');

	const bodies = usable.map((r) => ({ hostname: r.hostname, b: r.body }));
	const runs = bodies.map((r) => ({ hostname: r.hostname, ...r.b })).filter((run) => run.startedAt);
	const prefixes = [...new Set(runs.map((run) => run.urlPrefix).filter(Boolean))];

	return {
		status: 200,
		body: {
			node: null,
			scope: 'cluster',
			running: bodies.some((r) => r.b.running),
			runningOn: bodies.filter((r) => r.b.running).map((r) => r.hostname),
			ranNodes: runs.length,
			unrunNodes: bodies.filter((r) => !r.b.startedAt).map((r) => r.hostname),
			urlPrefix: prefixes.length === 1 ? prefixes[0] : null,
			urlPrefixes: prefixes,
			// A cluster figure counts as a dry run only if every node's was — one node that actually
			// deleted makes "nothing was deleted" false.
			dryRun: runs.length ? runs.every((run) => run.dryRun !== false) : true,
			canceled: runs.some((run) => run.canceled),
			// Same rule as `dryRun` above: "targets the ladder promoted were spared" is only true of
			// the cluster if EVERY node that ran spared them. One node that ran without the flag
			// deleted visited rows in its slice, and the totals cannot say which.
			skipVisited: runs.length ? runs.every((run) => run.skipVisited === true) : false,
			skipVisitedOn: runs.filter((run) => run.skipVisited === true).map((run) => run.hostname),
			// A pass that gave up mid-prefix covered part of its slice; the prefix is not done on
			// that node whatever the totals look like.
			abortedOnErrors: runs.some((run) => run.abortedOnErrors),
			errorSamples: runs
				.flatMap((run) => (run.errorSamples ?? []).map((sample) => ({ hostname: run.hostname, ...sample })))
				.slice(0, MAX_FAILURE_SAMPLES),
			startedAt: minOf(runs, (run) => msOf(run.startedAt)),
			// Oldest, and null while any node is still running: a finish time for a pass three nodes
			// are still in the middle of is the wrong shape of answer.
			finishedAt: runs.some((run) => run.running) ? null : minOf(runs, (run) => msOf(run.finishedAt)),
			error: runs.find((run) => run.error)?.error ?? null,
			totals: runs.length ? sumCounters(runs, PURGE_COUNTERS) : null,
			ownerScopeNote: bodies[0].b.ownerScopeNote ?? null,
			byNode: bodies.map((r) => ({
				hostname: r.hostname,
				running: !!r.b.running,
				urlPrefix: r.b.urlPrefix ?? null,
				dryRun: r.b.dryRun ?? null,
				ratePerSecond: num(r.b.ratePerSecond),
				startedAt: msOf(r.b.startedAt) || null,
				finishedAt: msOf(r.b.finishedAt) || null,
				canceled: !!r.b.canceled,
				skipVisited: !!r.b.skipVisited,
				abortedOnErrors: !!r.b.abortedOnErrors,
				error: r.b.error ?? null,
				...Object.fromEntries(PURGE_COUNTERS.map((key) => [key, num(r.b[key])])),
			})),
			sources: sourcesOf(results, { mode: 'merged' }),
		},
	};
}

// ---------------------------------------------------------------- queue state

/**
 * One node's `GET /prerender_admin/queue-state` answer, as the merge and the Queue view read it.
 *
 * A 503 IS AN ANSWER. The plugin answers 503 whenever its queue keeper cannot vouch for its numbers
 * (starting, waiting for peers, loading, failed, or a state document gone stale), and that body
 * carries the reason, the keeper's own stats and the live `now` fields (in flight, pause, status).
 * An operator needs every one of those, so it is kept — `answered` — and only its COUNTS are absent.
 * A transport failure or a 401 carries no `trust` block and is not answered.
 *
 * A NODE WITHOUT A KEEPER is its own case: a plugin before v0.93.0 answers 404 "Unknown route" (a
 * canary rollout runs mixed versions). It is up and claiming from its own index — `noKeeper`, judged as
 * a watch by the view at either scope. A 404 "Management API is disabled" is not that; it is no answer.
 *
 * `fanOut` passes a non-200's parsed body as `errorBody`; `body` stays reserved for a usable 200.
 */
export function queueStateRow(result) {
	const body = result.ok ? result.body : result.errorBody;
	const answered = !!(body && typeof body === 'object' && body.trust && typeof body.trust === 'object');
	const live = answered && !!result.ok && body.trust.live !== false;
	const noKeeper =
		!answered && result.status === 404 && /^Unknown route/.test(result.errorBody?.error ?? result.error ?? '');
	return {
		origin: result.origin ?? null,
		hostname: result.hostname ?? body?.node ?? null,
		node: answered ? (body.node ?? null) : null,
		httpStatus: result.status ?? 0,
		answered,
		noKeeper,
		live,
		error: live ? null : answered ? (body.error ?? `answered ${result.status}`) : (result.error ?? null),
		now: answered ? (body.now ?? null) : null,
		coming: live ? (body.coming ?? null) : null,
		// Per node: the bins and the oldest due row. The route and class lists are cluster-merged below.
		lateness: live
			? {
					edges: body.lateness?.edges ?? null,
					sitemap: body.lateness?.sitemap ?? null,
					discovered: body.lateness?.discovered ?? null,
					oldestDueAt: body.lateness?.oldestDueAt ?? null,
				}
			: null,
		trust: answered ? body.trust : null,
	};
}

/**
 * Every node's `now.changedByDemand` (plugin v0.95.0) as one list: due changed rows by the demand period
 * they carry — `periodMs` between bot visits, `null` when the tracker could not say. Periods are whole-slot
 * fractions of the tracker's window, so the same period on two nodes is the same value and merges exactly:
 * `due` sums, `oldestDueAt` is the earliest. Ordered as the plugin orders it, most-asked-for first and
 * unknown last.
 */
function mergeChangedByDemand(rows) {
	const byPeriod = new Map();
	for (const row of rows) {
		for (const entry of row.now.changedByDemand) {
			const period = finite(entry?.periodMs);
			const periodMs = Number.isFinite(period) && period > 0 ? period : null;
			const acc = byPeriod.get(periodMs) ?? { periodMs, due: 0, oldestDueAt: null };
			acc.due += finite(entry?.due) || 0;
			const at = finite(entry?.oldestDueAt);
			if (Number.isFinite(at) && (acc.oldestDueAt === null || at < acc.oldestDueAt)) acc.oldestDueAt = at;
			byPeriod.set(periodMs, acc);
		}
	}
	return [...byPeriod.values()].sort((a, b) => (a.periodMs ?? Infinity) - (b.periodMs ?? Infinity));
}

/** How many classes the merged list keeps: the panel shows the worst few, and each node sends ≤200. */
const MAX_CLASSES = 25;

/** Element-wise sum of lateness bins, or null when the nodes disagree on the edges (a version skew). */
function sumBins(rows, key, edgesJson) {
	if (edgesJson === null) return null;
	const arrays = rows.map((row) => row.lateness?.[key]).filter(Array.isArray);
	const length = Math.max(0, ...arrays.map((a) => a.length));
	return length ? sumArrays(arrays, length) : null;
}

/**
 * Per-node queue state, merged.
 *
 * THE CLUSTER TOTAL IS WITHHELD, NOT FLOORED, until every configured node is live. Each keeper holds
 * only the rows its node owns, so the sum over three of four nodes is not a smaller queue — it is a
 * quarter of the queue missing, and it reads exactly like the backlog draining. `cluster` is null
 * then, `withheld` names who could not vouch and why, and `nodes` still carries every node's own
 * answer (the live nodes' counts included), which is what an operator acts on during a restart.
 *
 * Because a 503 is an answer, `sources` counts it as one: the view names those nodes itself, with the
 * keeper's reason, where the generic "N of M nodes answered" banner would call a loading node down.
 */
export function mergeQueueState(results) {
	const rows = results.map(queueStateRow);
	// A cluster entirely on a plugin before v0.93.0 is still an answer: every node without a keeper.
	if (!rows.some((row) => row.answered || row.noKeeper)) return allFailed(results, 'merged');

	const live = rows.filter((row) => row.live);
	const complete = live.length === rows.length;
	const edgesSeen = new Set(live.map((row) => JSON.stringify(row.lateness?.edges ?? null)));
	const edgesJson = edgesSeen.size === 1 ? [...edgesSeen][0] : null;

	let cluster = null;
	if (complete) {
		const sum = (read) => sumOf(live, read);

		// Classes (route × cadence × sitemap flag × changed flag) add, keeping the worst head and naming its
		// node the way every other per-node field here does: by the fan-out's host. The changed flag (plugin
		// v0.94.0: rows the change probe filed, ranked ahead) keeps those rows' class apart from the routine
		// one it would otherwise fold into — a 0.93 node sends no flag, and its classes read as routine. A 0.95
		// node also splits a changed class by demand period; those fold together here, because this list ranks
		// how late classes are and the demand split has its own list (`now.changedByDemand`).
		// (The per-route list is not merged — nothing reads it, and its bins would need the same edge check.)
		const classes = new Map();
		for (const result of results) {
			for (const klass of result.body?.lateness?.classes ?? []) {
				const id = `${klass.route}\u0000${klass.cadenceMs}\u0000${!!klass.fromSitemap}\u0000${!!klass.changed}`;
				const acc = classes.get(id) ?? {
					route: klass.route,
					cadenceMs: klass.cadenceMs,
					fromSitemap: !!klass.fromSitemap,
					changed: !!klass.changed,
					rows: 0,
					due: 0,
					oldestDueAt: null,
					oldestLatenessCadences: null,
					worstNode: null,
				};
				acc.rows += finite(klass.rows) || 0;
				acc.due += finite(klass.due) || 0;
				const at = finite(klass.oldestDueAt);
				if (Number.isFinite(at) && (acc.oldestDueAt === null || at < acc.oldestDueAt)) acc.oldestDueAt = at;
				const late = finite(klass.oldestLatenessCadences);
				if (Number.isFinite(late) && (acc.oldestLatenessCadences === null || late > acc.oldestLatenessCadences)) {
					acc.oldestLatenessCadences = late;
					acc.worstNode = result.hostname ?? null;
				}
				classes.set(id, acc);
			}
		}

		// Flow is per minute on every node's clock; the minutes are wall-clock, so they align.
		const flow = new Map();
		for (const result of results) {
			for (const slot of result.body?.flow ?? []) {
				const minute = finite(slot.minute);
				if (!Number.isFinite(minute)) continue;
				const acc = flow.get(minute) ?? { minute, cameDue: 0, added: 0, triggered: 0, rescheduled: 0, removed: 0 };
				for (const key of ['cameDue', 'added', 'triggered', 'rescheduled', 'removed'])
					acc[key] += finite(slot[key]) || 0;
				flow.set(minute, acc);
			}
		}

		cluster = {
			now: {
				due: sum((row) => row.now?.due),
				dueSitemap: sum((row) => row.now?.dueSitemap),
				dueDiscovered: sum((row) => row.now?.dueDiscovered),
				// Due rows the change probe filed (plugin v0.94.0): pages known changed, expired and served from
				// the origin until they re-render. Summed only when EVERY node reports it: a node on an older
				// plugin does not, and a sum without its slice would read as the whole cluster's.
				dueChanged: live.every((row) => Number.isFinite(row.now?.dueChanged))
					? sum((row) => row.now?.dueChanged)
					: null,
				// The longest-waiting changed page's due minute: the EARLIEST across nodes, and only when every
				// node reports the field. `null` is an answer (no changed page is due), so an unreported cluster
				// value is left OUT rather than set to null — exactly as a node that does not report it sends it.
				...(live.every((row) => row.now && 'oldestChangedAt' in row.now)
					? { oldestChangedAt: minOf(live, (row) => row.now.oldestChangedAt) }
					: {}),
				// The same rows by the demand estimate they carry (plugin v0.95.0): merged by period, and left
				// out unless every node sends the list — the same rule as the two fields above.
				...(live.every((row) => Array.isArray(row.now?.changedByDemand))
					? { changedByDemand: mergeChangedByDemand(live) }
					: {}),
				inFlight: sum((row) => row.now?.inFlight),
				// Each node's own `due − inFlight` (clamped at zero there), summed: a node with more leases
				// than due rows must not cancel another node's waiting rows.
				unclaimed: sum((row) => row.now?.unclaimed),
			},
			coming: {
				next15m: sum((row) => row.coming?.next15m),
				next60m: sum((row) => row.coming?.next60m),
				next24h: sum((row) => row.coming?.next24h),
				byHour: sumArrays(
					live.map((row) => row.coming?.byHour),
					Math.max(0, ...live.map((row) => (Array.isArray(row.coming?.byHour) ? row.coming.byHour.length : 0)))
				),
			},
			lateness: {
				edges: edgesJson === null ? null : JSON.parse(edgesJson),
				sitemap: sumBins(live, 'sitemap', edgesJson),
				discovered: sumBins(live, 'discovered', edgesJson),
				edgesDiverge: edgesJson === null,
				oldestDueAt: minOf(live, (row) => row.lateness?.oldestDueAt),
				// Plugin v0.93.1+: a node cut its lists at 200. Absent (0.93.0) reads as not cut.
				listsTruncated: results.some((result) => result.body?.lateness?.listsTruncated === true),
				classes: [...classes.values()]
					.sort((a, b) => (b.oldestLatenessCadences ?? -1) - (a.oldestLatenessCadences ?? -1))
					.slice(0, MAX_CLASSES),
			},
			flow: [...flow.values()].sort((a, b) => a.minute - b.minute),
			// The plugin's own flag, every node's. The view refines it (a membership change's gained rows
			// turn `exact` false without making a count short) from the per-node rows.
			exact: live.every((row) => row.trust?.exact === true),
		};
	}

	return {
		status: 200,
		body: {
			scope: 'cluster',
			generatedAt: minOf(live, (row) => row.trust?.stateAt),
			live: live.length,
			configured: rows.length,
			cluster,
			withheld: rows.filter((row) => !row.live).map((row) => ({ hostname: row.hostname, reason: row.error })),
			nodes: rows,
			// A 503 and a node without a keeper are answers the view names itself; only a node that did not
			// answer at all banners the page as incomplete.
			sources: sourcesOf(
				results.map((r, i) => (rows[i].answered || rows[i].noKeeper ? { ...r, ok: true, error: null } : r)),
				{ mode: 'merged' }
			),
		},
	};
}

// ---------------------------------------------------------------- dispatch

const MERGERS = {
	'overview': mergeOverview,
	'analytics': mergeAnalytics,
	'unrouted': mergeUnrouted,
	'config': mergeConfig,
	'change-probe': mergeChangeProbe,
	'discovery-purge': mergeDiscoveryPurge,
	'queue-state': mergeQueueState,
};

export const mergerFor = (route) => MERGERS[route] ?? null;