feat(page-cluster)!: streaming API for large corpora

cspell.json

@@ -17,6 +17,8 @@
 		"hrefs",
 		"Murtagh",
 		"unioned",
+		"retags",
+		"Vitter",
 
 		//
 		"gaxios",

packages/@d-zero/page-cluster/src/merge-cross-block-clusters.spec.ts

@@ -4,6 +4,20 @@ import type { CrossBlockUnit } from './merge-cross-block-clusters.js';
 import { describe, expect, test } from 'vitest';
 
 import { mergeCrossBlockClusters } from './merge-cross-block-clusters.js';
+import { computePerPageLandmarkInstances } from './per-page-landmark-signatures.js';
+
+/**
+ * Converts an old-style per-member `ExtractLandmarksResult[]` fixture into
+ * the pre-tokenized `PerPageLandmarkInstance[][]` shape that `CrossBlockUnit`
+ * now stores. Every test in this file historically constructed landmark
+ * fixtures as full `ExtractLandmarksResult` objects; this shim funnels them
+ * through the same tokenization Stage B used to do at read time, keeping
+ * every existing test's semantics intact under the new field name.
+ * @param landmarks
+ */
+function toInstances(landmarks: readonly ExtractLandmarksResult[]) {
+	return computePerPageLandmarkInstances(landmarks);
+}
 
 const noLandmarks: ExtractLandmarksResult = {
 	header: [],
@@ -28,7 +42,7 @@ describe('mergeCrossBlockClusters', () => {
 		const unit: CrossBlockUnit = {
 			key: 'k1',
 			memberTokenSets: [new Set(['body>main>.card'])],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const result = mergeCrossBlockClusters([unit], {});
 		expect(result.get('k1')).toBe('k1');
@@ -39,12 +53,12 @@ describe('mergeCrossBlockClusters', () => {
 		const unit1: CrossBlockUnit = {
 			key: 'k1',
 			memberTokenSets: [tokens],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const unit2: CrossBlockUnit = {
 			key: 'k2',
 			memberTokenSets: [tokens],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const result = mergeCrossBlockClusters([unit1, unit2], {});
 		expect(result.get('k1')).toBe(result.get('k2'));
@@ -56,14 +70,14 @@ describe('mergeCrossBlockClusters', () => {
 			memberTokenSets: [
 				new Set(['body>main>article', 'body>main>article>h1', 'body>main>article>p']),
 			],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const unit2: CrossBlockUnit = {
 			key: 'k2',
 			memberTokenSets: [
 				new Set(['body>aside>section', 'body>aside>section>ul', 'body>footer>nav']),
 			],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const result = mergeCrossBlockClusters([unit1, unit2], {});
 		expect(result.get('k1')).toBe('k1');
@@ -85,7 +99,7 @@ describe('mergeCrossBlockClusters', () => {
 					'body>main>section.c-reports>ul.c-reports__list',
 				]),
 			],
-			memberLandmarks: [noLandmarks, noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks, noLandmarks]),
 		};
 		const unit2: CrossBlockUnit = {
 			key: 'projects',
@@ -99,7 +113,7 @@ describe('mergeCrossBlockClusters', () => {
 					'body>main>section.c-projects>ul.c-projects__list',
 				]),
 			],
-			memberLandmarks: [noLandmarks, noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks, noLandmarks]),
 		};
 		const result = mergeCrossBlockClusters([unit1, unit2], {});
 		expect(result.get('reports')).toBe(result.get('projects'));
@@ -113,14 +127,14 @@ describe('mergeCrossBlockClusters', () => {
 			memberTokenSets: [
 				new Set(['body>main>section.type-a', 'body>main>section.type-a>p']),
 			],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const unit2: CrossBlockUnit = {
 			key: 'solo-b',
 			memberTokenSets: [
 				new Set(['body>main>section.type-b', 'body>main>section.type-b>p']),
 			],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const result = mergeCrossBlockClusters([unit1, unit2], {});
 		expect(result.get('solo-a')).toBe('solo-a');
@@ -131,17 +145,17 @@ describe('mergeCrossBlockClusters', () => {
 		const unit1: CrossBlockUnit = {
 			key: 'a',
 			memberTokenSets: [new Set(['x'])],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const unit2: CrossBlockUnit = {
 			key: 'b',
 			memberTokenSets: [new Set(['y'])],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const unit3: CrossBlockUnit = {
 			key: 'c',
 			memberTokenSets: [new Set(['z'])],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const result = mergeCrossBlockClusters([unit1, unit2, unit3], {});
 		expect(result.has('a')).toBe(true);
@@ -154,12 +168,12 @@ describe('mergeCrossBlockClusters', () => {
 		const unit1: CrossBlockUnit = {
 			key: 'k1',
 			memberTokenSets: [tokens],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const unit2: CrossBlockUnit = {
 			key: 'k2',
 			memberTokenSets: [tokens],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const r1 = mergeCrossBlockClusters([unit1, unit2], {});
 		const r2 = mergeCrossBlockClusters([unit1, unit2], {});
@@ -172,12 +186,12 @@ describe('mergeCrossBlockClusters', () => {
 		const unit1: CrossBlockUnit = {
 			key: 'block-a',
 			memberTokenSets: [tokens],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const unit2: CrossBlockUnit = {
 			key: 'block-b',
 			memberTokenSets: [tokens],
-			memberLandmarks: [noLandmarks],
+			memberLandmarkInstances: toInstances([noLandmarks]),
 		};
 		const result = mergeCrossBlockClusters([unit1, unit2], {});
 		const rootKey = result.get('block-a')!;
@@ -193,12 +207,12 @@ describe('mergeCrossBlockClusters', () => {
 		const unit1: CrossBlockUnit = {
 			key: 'l1',
 			memberTokenSets: [tokens],
-			memberLandmarks: [landmarks],
+			memberLandmarkInstances: toInstances([landmarks]),
 		};
 		const unit2: CrossBlockUnit = {
 			key: 'l2',
 			memberTokenSets: [tokens],
-			memberLandmarks: [landmarks],
+			memberLandmarkInstances: toInstances([landmarks]),
 		};
 		const result = mergeCrossBlockClusters([unit1, unit2], {});
 		expect(result.get('l1')).toBe(result.get('l2'));
@@ -216,7 +230,7 @@ describe('mergeCrossBlockClusters shellQuorum (via mergeCrossBlockClusters exerc
 		const unit: CrossBlockUnit = {
 			key: 'k',
 			memberTokenSets: Array.from({ length: 5 }, () => tokens),
-			memberLandmarks: Array.from({ length: 5 }, () => landmarksAll),
+			memberLandmarkInstances: toInstances(Array.from({ length: 5 }, () => landmarksAll)),
 		};
 		// The clustering itself just needs to run without crashing to prove
 		// the histogram path works with a well-populated signature.
@@ -244,12 +258,12 @@ describe('mergeCrossBlockClusters shellQuorum (via mergeCrossBlockClusters exerc
 		const unitA: CrossBlockUnit = {
 			key: 'A',
 			memberTokenSets: Array.from({ length: 5 }, () => tokens),
-			memberLandmarks: landmarksList,
+			memberLandmarkInstances: toInstances(landmarksList),
 		};
 		const unitB: CrossBlockUnit = {
 			key: 'B',
 			memberTokenSets: Array.from({ length: 5 }, () => tokens),
-			memberLandmarks: landmarksList,
+			memberLandmarkInstances: toInstances(landmarksList),
 		};
 		const result = mergeCrossBlockClusters([unitA, unitB], {});
 		// Same tokens sets merge via fine stage (Jaccard = 1.0), independent
@@ -275,12 +289,12 @@ describe('mergeCrossBlockClusters shellQuorum (via mergeCrossBlockClusters exerc
 		const unitA: CrossBlockUnit = {
 			key: 'A',
 			memberTokenSets: Array.from({ length: 5 }, () => new Set(['body>main>.article'])),
-			memberLandmarks: landmarksList,
+			memberLandmarkInstances: toInstances(landmarksList),
 		};
 		const unitB: CrossBlockUnit = {
 			key: 'B',
 			memberTokenSets: Array.from({ length: 5 }, () => new Set(['body>aside>.widget'])),
-			memberLandmarks: landmarksList,
+			memberLandmarkInstances: toInstances(landmarksList),
 		};
 		const result = mergeCrossBlockClusters([unitA, unitB], {});
 		// Cores are disjoint and shells don't corroborate → A and B stay

packages/@d-zero/page-cluster/src/merge-cross-block-clusters.ts

@@ -1,4 +1,4 @@
-import type { ExtractLandmarksResult } from './extract-landmarks.js';
+import type { PerPageLandmarkInstance } from './per-page-landmark-signatures.js';
 
 import { assignContainedClusters } from './assign-contained-clusters.js';
 import { autoCutThreshold } from './auto-cut-threshold.js';
@@ -9,17 +9,28 @@ import {
 } from './complete-linkage-dendrogram.js';
 import { computeDocumentFrequency } from './compute-document-frequency.js';
 import { jaccardSimilarity } from './jaccard-similarity.js';
-import { computePerPageLandmarkInstances } from './per-page-landmark-signatures.js';
+import { reservoirSample } from './reservoir-sample.js';
 import { shapeToken } from './shape-token.js';
 import { splitTokensByFrequency } from './split-tokens-by-frequency.js';
 
 /**
  * One cluster (post-Stage-A) entering cross-block comparison.
+ *
+ * ## Why `memberLandmarkInstances` rather than raw `ExtractLandmarksResult[]`
+ *
+ * Stage B never reads the full `ExtractLandmarksResult` — it only needs each
+ * member page's landmark instances (per-type, per-signature, per-token-set)
+ * for `shellQuorum`'s per-token page-frequency histogram. Callers pre-run
+ * {@link ./per-page-landmark-signatures.js | computePerPageLandmarkInstances}
+ * once at unit creation and hand Stage B the compact instance list instead
+ * of the ~10× larger raw landmark HTML strings. This is the single largest
+ * memory reduction the streaming path relies on: the difference between a
+ * 176k-page corpus fitting in an 8 GB heap and exhausting a 12 GB heap.
  */
 export type CrossBlockUnit = {
 	readonly key: string;
 	readonly memberTokenSets: readonly ReadonlySet<string>[];
-	readonly memberLandmarks: readonly ExtractLandmarksResult[];
+	readonly memberLandmarkInstances: readonly (readonly PerPageLandmarkInstance[])[];
 };
 
 /**
@@ -247,15 +258,14 @@ function l2Contained(xSig: Map<string, number>, ySig: Map<string, number>): bool
  * shell-corroboration jaccard between two landmark-less pages is 0 rather
  * than 1 (which it would be if we handed back a `<body></body>`-derived
  * `{body}` fallback set to both sides).
- * @param memberLandmarks
+ * @param perPageInstances
  */
 function shellQuorum(
-	memberLandmarks: readonly ExtractLandmarksResult[],
+	perPageInstances: readonly (readonly PerPageLandmarkInstance[])[],
 ): ReadonlySet<string> {
-	const pageCount = memberLandmarks.length;
+	const pageCount = perPageInstances.length;
 	if (pageCount === 0) return new Set();
 
-	const perPageInstances = computePerPageLandmarkInstances(memberLandmarks);
 	// Union all instance token sets per page (dedupe within page: a token
 	// present on both header and footer of the same page still counts once
 	// for that page's contribution).
@@ -316,28 +326,42 @@ function shellQuorum(
  * @param units Post-Stage-A clusters.
  * @param options Forwarded `similarityThreshold` (defaults to 0.8).
  * @param options.similarityThreshold
+ * @param options.capMembers
  */
 export function mergeCrossBlockClusters(
 	units: readonly CrossBlockUnit[],
-	options?: { similarityThreshold?: number },
+	options?: {
+		similarityThreshold?: number;
+		/**
+		 * Opt-in post-merge cap on a group's retained member count. When
+		 * set, each merged group is reservoir-sampled back down to this
+		 * cap after every merge so a chain of merges cannot balloon past
+		 * the per-unit cap Stage A applied at creation. Callers on the
+		 * streaming path pass the same value as `capMembers` used in
+		 * Stage A; the in-memory path omits it so validated corpora keep
+		 * full-membership merge behavior unchanged.
+		 */
+		capMembers?: number;
+	},
 ): Map<string, string> {
 	if (units.length <= 1) {
 		return new Map(units.map((u) => [u.key, u.key]));
 	}
 
 	const threshold = options?.similarityThreshold ?? CROSS_BLOCK_THRESHOLD;
+	const capMembers = options?.capMembers;
 
 	// Mutable group state: rootKey → combined member arrays
 	type GroupMembers = {
 		tokenSets: ReadonlySet<string>[];
-		landmarks: ExtractLandmarksResult[];
+		landmarkInstances: (readonly PerPageLandmarkInstance[])[];
 	};
 
 	const groups = new Map<string, GroupMembers>();
 	for (const unit of units) {
 		groups.set(unit.key, {
 			tokenSets: [...unit.memberTokenSets],
-			landmarks: [...unit.memberLandmarks],
+			landmarkInstances: [...unit.memberLandmarkInstances],
 		});
 	}
 
@@ -355,8 +379,23 @@ export function mergeCrossBlockClusters(
 			const rootG = groups.get(root);
 			if (!absorbedG || !rootG) continue;
 
-			rootG.tokenSets = [...rootG.tokenSets, ...absorbedG.tokenSets];
-			rootG.landmarks = [...rootG.landmarks, ...absorbedG.landmarks];
+			const mergedTokenSets = [...rootG.tokenSets, ...absorbedG.tokenSets];
+			const mergedLandmarkInstances = [
+				...rootG.landmarkInstances,
+				...absorbedG.landmarkInstances,
+			];
+			// Only down-sample when the caller explicitly opts in (streaming
+			// path). Same-index sampling keeps memberTokenSets[i] and
+			// landmarkInstances[i] parallel.
+			if (capMembers !== undefined && mergedTokenSets.length > capMembers) {
+				const indices = mergedTokenSets.map((_, i) => i);
+				const kept = reservoirSample(indices, capMembers, root);
+				rootG.tokenSets = kept.map((i) => mergedTokenSets[i]!);
+				rootG.landmarkInstances = kept.map((i) => mergedLandmarkInstances[i]!);
+			} else {
+				rootG.tokenSets = mergedTokenSets;
+				rootG.landmarkInstances = mergedLandmarkInstances;
+			}
 			groups.delete(absorbed);
 
 			for (const [origKey, cur] of keyToRoot) {
@@ -517,7 +556,7 @@ export function mergeCrossBlockClusters(
 
 		const getShell = (key: string): ReadonlySet<string> => {
 			if (!shellCache.has(key)) {
-				shellCache.set(key, shellQuorum(groups.get(key)?.landmarks ?? []));
+				shellCache.set(key, shellQuorum(groups.get(key)?.landmarkInstances ?? []));
 			}
 			return shellCache.get(key) ?? new Set();
 		};