Upload folder using huggingface_hub

app/config.py

@@ -47,14 +47,36 @@ class Settings(BaseSettings):
     europepmc_url: str = "https://www.ebi.ac.uk/europepmc/webservices/rest"
     fulltext_timeout: float = 20.0
     fulltext_max_lines: int = Field(default=160, ge=20, le=600)
-    # Gap map: a deterministic random sample of embedded studies, clustered once.
+    # Gap map: every embedded study, scanned in pages and clustered once.
     # The region count is a resolution, and what counts as an empty band depends on
     # it: measured on 6000 documents of the live index, 24 regions leave the corpus
     # continuous (no band under the occupancy threshold) while 80 keep a median of
     # 64 primary attempts per region and still expose bands.
-    gapmap_sample: int = Field(default=6000, ge=100, le=10000)
     gapmap_regions: int = Field(default=80, ge=2, le=200)
     gapmap_seed: int = 0
+    # A cap on how much of the embedded corpus one build reads; 0 is no cap, and
+    # the coverage line and the warning always report what was actually read.
+    # Measured on the live index (2,078,515 embedded studies, 384 dimensions):
+    # the scan runs at ~11k documents/s and the vectors weigh ~3.0 GiB, so an
+    # uncapped build is ~3 minutes and needs a host with several GB to spare.
+    gapmap_scan_limit: int = Field(default=0, ge=0)
+    # Page size and concurrent point-in-time slices for that scan. Sixteen slices
+    # measured ~11k docs/s against ~3.2k for a single unsliced walk.
+    gapmap_batch: int = Field(default=2000, ge=100, le=10000)
+    gapmap_slices: int = Field(default=16, ge=1, le=64)
+    # How many of the clustered studies the canvas draws. Every study shapes the
+    # regions; a browser cannot paint two million marks, so the drawn subset is
+    # thinned by a hash of the document id.
+    gapmap_points: int = Field(default=6000, ge=100, le=50000)
+    # Rows the 2-D projection is fitted on. The basis is a drawing choice, and a
+    # full-corpus SVD would stall the build for minutes to move points by pixels.
+    gapmap_projection_sample: int = Field(default=50000, ge=1000, le=500000)
+    # Bound on the full-corpus k-means. Measured at 200k documents and 80
+    # regions: 10 iterations ~3.2 s, 40 ~7.7 s; the assignment usually settles
+    # first and the loop stops when it does.
+    gapmap_iterations: int = Field(default=25, ge=1, le=200)
+    # Minimum seconds between two streamed frames while the corpus is scanning.
+    gapmap_tick_seconds: float = Field(default=0.4, ge=0.0, le=10.0)
     max_concurrent_searches: int = 4
     frontend_dist: str = str(Path(__file__).resolve().parents[2] / "dist")
     cors_origins: list[str] = ["http://localhost:5173"]

app/gapmap.py

@@ -33,6 +33,9 @@ MAP_VERSION = "gapmap-v1"
 # A region needs this many primary studies before its verdict mix means anything;
 # below it the honest label is "thin", not a finding about the science.
 MIN_REGION_ATTEMPTS = 5
+# Rows per chunk when multiplying the corpus against the centroids. Bounds the
+# intermediate to a few tens of megabytes whatever the corpus size.
+ASSIGN_CHUNK = 50_000
 # Share of attempts that must be null (or unreported) before the region is called
 # saturated (or dark). Both are deliberately majority-ish rather than "any".
 NULL_SATURATED_SHARE = 0.5
@@ -75,13 +78,8 @@ def normalize(vectors: np.ndarray) -> np.ndarray:
     return vectors / np.clip(norms, 1e-8, None)
 
 
-def kmeans(vectors: np.ndarray, k: int, seed: int = 0, iterations: int = 40) -> np.ndarray:
-    """Spherical k-means with k-means++ seeding, returning a label per row.
-
-    Deterministic for a given seed so a map can be rebuilt and compared. Written
-    out here rather than pulled from scikit-learn because the ingest extra that
-    provides it is optional and the serving path must not depend on it.
-    """
+def seed_centroids(vectors: np.ndarray, k: int, seed: int = 0) -> np.ndarray:
+    """k-means++ seeding over a cosine space, deterministic for a given seed."""
     rng = np.random.default_rng(seed)
     count = len(vectors)
     k = max(1, min(k, count))
@@ -93,20 +91,123 @@ def kmeans(vectors: np.ndarray, k: int, seed: int = 0, iterations: int = 40) ->
         total = distance.sum()
         pick = rng.integers(count) if total <= 0 else rng.choice(count, p=distance / total)
         centroids[index] = vectors[pick]
-    labels = np.full(count, -1, dtype=np.int64)
-    for _ in range(iterations):
-        assigned = (vectors @ centroids.T).argmax(axis=1)
-        if np.array_equal(assigned, labels):
-            break
+    return centroids
+
+
+def assign(vectors: np.ndarray, centroids: np.ndarray, chunk: int = ASSIGN_CHUNK) -> np.ndarray:
+    """Nearest centroid per row, in chunks so a corpus-sized matrix is never built.
+
+    ``vectors @ centroids.T`` over two million rows and eighty centroids is a
+    640 MB intermediate; the chunked loop is the same arithmetic in 16 MB pieces.
+    """
+    if not len(vectors):
+        return np.zeros(0, dtype=np.int64)
+    labels = np.empty(len(vectors), dtype=np.int64)
+    for start in range(0, len(vectors), chunk):
+        stop = start + chunk
+        labels[start:stop] = (vectors[start:stop] @ centroids.T).argmax(axis=1)
+    return labels
+
+
+def _recentre(vectors: np.ndarray, labels: np.ndarray, centroids: np.ndarray) -> np.ndarray:
+    """Move each centroid onto the mean direction of the rows assigned to it.
+
+    Empty clusters keep their previous position rather than being reseeded, so a
+    warm-started run cannot make the map jump between successive iterations.
+    """
+    moved = centroids.copy()
+    order = np.argsort(labels, kind="stable")
+    bounds = np.searchsorted(labels[order], np.arange(len(centroids) + 1))
+    for index in range(len(centroids)):
+        members = order[bounds[index] : bounds[index + 1]]
+        if len(members):
+            moved[index] = vectors[members].mean(axis=0)
+    return normalize(moved)
+
+
+def kmeans_steps(
+    vectors: np.ndarray,
+    k: int,
+    seed: int = 0,
+    iterations: int = 40,
+    centroids: np.ndarray | None = None,
+):
+    """Spherical k-means, yielding ``(centroids, labels)`` after every iteration.
+
+    Yielding the intermediate states is what lets a caller show the map settling
+    instead of a spinner: each yielded pair is a real assignment of the corpus,
+    not an interpolation. Deterministic for a given seed and starting position,
+    and stops early once the assignment stops changing.
+    """
+    if not len(vectors):
+        return
+    if centroids is None:
+        centroids = seed_centroids(vectors, k, seed)
+    centroids = normalize(np.asarray(centroids, dtype=np.float32))
+    labels = np.full(len(vectors), -1, dtype=np.int64)
+    for _ in range(max(1, iterations)):
+        assigned = assign(vectors, centroids)
+        settled = np.array_equal(assigned, labels)
         labels = assigned
-        for index in range(k):
-            members = vectors[labels == index]
-            if len(members):
-                centroids[index] = members.mean(axis=0)
-        centroids = normalize(centroids)
+        centroids = _recentre(vectors, labels, centroids)
+        yield centroids, labels
+        if settled:
+            return
+
+
+def kmeans(vectors: np.ndarray, k: int, seed: int = 0, iterations: int = 40) -> np.ndarray:
+    """Spherical k-means with k-means++ seeding, returning a label per row.
+
+    Deterministic for a given seed so a map can be rebuilt and compared. Written
+    out here rather than pulled from scikit-learn because the ingest extra that
+    provides it is optional and the serving path must not depend on it.
+    """
+    labels = np.zeros(len(vectors), dtype=np.int64)
+    for _, labels in kmeans_steps(vectors, k, seed=seed, iterations=iterations):
+        pass
     return labels
 
 
+class RunningClusters:
+    """Mini-batch spherical k-means: centroids that exist before the scan ends.
+
+    The full corpus cannot be clustered until it has all arrived, but a map that
+    only appears at the end cannot be watched. Each batch is assigned to the
+    current centroids and each centroid is moved towards the mean of what it
+    just received, weighted by how much it has seen — the standard mini-batch
+    update. These centroids are provisional and are replaced by full-corpus
+    k-means once the scan finishes; nothing is reported as final until then.
+    """
+
+    def __init__(self, k: int, seed: int = 0):
+        self.k = max(1, k)
+        self.seed = seed
+        self.centroids: np.ndarray | None = None
+        self.counts: np.ndarray | None = None
+
+    def update(self, vectors: np.ndarray) -> np.ndarray | None:
+        if not len(vectors):
+            return self.centroids
+        if self.centroids is None:
+            if len(vectors) < self.k:
+                return None  # too few rows to seed k distinct centroids honestly
+            self.centroids = normalize(seed_centroids(vectors, self.k, self.seed))
+            self.counts = np.zeros(len(self.centroids), dtype=np.float64)
+        labels = assign(vectors, self.centroids)
+        sums = np.zeros_like(self.centroids, dtype=np.float32)
+        np.add.at(sums, labels, vectors)
+        counts = np.bincount(labels, minlength=len(self.centroids)).astype(np.float64)
+        assert self.counts is not None
+        seen = self.counts + counts
+        rate = np.divide(counts, seen, out=np.zeros_like(seen), where=seen > 0)[:, None]
+        self.centroids = normalize(
+            self.centroids * (1 - rate)
+            + np.divide(sums, np.clip(counts, 1, None)[:, None]) * rate
+        )
+        self.counts = seen
+        return self.centroids
+
+
 def fit_projection(vectors: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
     """A deterministic 2-D basis for drawing the map: PCA over centered vectors.
 
@@ -207,51 +308,88 @@ def _exemplars(documents: list[dict], limit: int = 3) -> list[dict]:
     ]
 
 
-def build_regions(documents: list[dict], regions: int = 24, seed: int = 0) -> dict:
-    """Cluster embedded documents and describe each cluster by what happened in it.
+def group(labels: np.ndarray, clusters: int) -> list[np.ndarray]:
+    """Row indices per cluster. One sort rather than a scan per cluster: the
+    obvious ``labels == index`` loop is O(rows × clusters), which is hours of
+    Python over a corpus of millions."""
+    order = np.argsort(labels, kind="stable")
+    bounds = np.searchsorted(labels[order], np.arange(clusters + 1))
+    return [order[bounds[index] : bounds[index + 1]] for index in range(clusters)]
 
-    Documents without a stored vector are counted in ``skipped`` rather than
-    dropped silently: a region map over a partly unembedded corpus describes the
-    embedded part only, and saying so is the difference between a map and a claim.
+
+def describe_regions(
+    documents: list[dict], vectors: np.ndarray, labels: np.ndarray, clusters: int
+) -> list[dict]:
+    """Describe each occupied cluster by what happened in the documents it holds.
+
+    Cluster indices are the region ids, and an empty cluster yields no region, so
+    a region keeps its identity between two runs that share centroids — which is
+    what lets a partially built map be redrawn rather than replaced.
     """
-    embedded = [(doc, vector) for doc in documents if (vector := _vector(doc)) is not None]
-    skipped = len(documents) - len(embedded)
-    if not embedded:
-        return {"version": MAP_VERSION, "regions": [], "documents": 0, "skipped": skipped}
-    corpus = [doc for doc, _ in embedded]
-    vectors = normalize(np.vstack([vector for _, vector in embedded]))
-    labels = kmeans(vectors, regions, seed=seed)
     described = []
-    for index in sorted(set(labels.tolist())):
-        members = [doc for doc, label in zip(corpus, labels) if label == index]
-        centroid = normalize(vectors[labels == index].mean(axis=0, keepdims=True))[0]
-        primary = [doc for doc in members if _is_primary(doc)]
+    for index, members in enumerate(group(labels, clusters)):
+        if not len(members):
+            continue
+        rows = vectors[members]
+        centroid = normalize(rows.mean(axis=0, keepdims=True))[0]
+        held = [documents[row] for row in members.tolist()]
+        primary = [doc for doc in held if _is_primary(doc)]
         counts = Counter(_bucket(doc) for doc in primary)
-        years = [doc["year"] for doc in members if isinstance(doc.get("year"), int)]
+        years = [doc["year"] for doc in held if isinstance(doc.get("year"), int)]
         citations = [doc.get("cited_by_count") or 0 for doc in primary]
         described.append(
             {
                 "id": int(index),
-                "size": len(members),
+                "size": len(held),
                 "attempts": len(primary),
                 "centroid": [float(value) for value in centroid],
                 "label": label_region(counts, len(primary)),
                 "bucketCounts": dict(counts),
                 "medianYear": int(np.median(years)) if years else None,
                 "medianCitations": float(np.median(citations)) if citations else None,
-                "coherence": float((vectors[labels == index] @ centroid).mean()),
-                "exemplars": _exemplars(members),
+                "coherence": float((rows @ centroid).mean()),
+                "exemplars": _exemplars(held),
             }
         )
+    return described
+
+
+def build_regions(
+    documents: list[dict],
+    regions: int = 24,
+    seed: int = 0,
+    vectors: np.ndarray | None = None,
+) -> dict:
+    """Cluster embedded documents and describe each cluster by what happened in it.
+
+    Documents without a stored vector are counted in ``skipped`` rather than
+    dropped silently: a region map over a partly unembedded corpus describes the
+    embedded part only, and saying so is the difference between a map and a claim.
+
+    ``vectors`` lets a caller that already holds the corpus as one normalized
+    matrix — the streaming build does — skip rebuilding it from the documents,
+    which would cost a second copy of several gigabytes.
+    """
+    corpus, matrix = _corpus(documents, vectors)
+    skipped = len(documents) - len(corpus)
+    if not len(matrix):
+        return {"version": MAP_VERSION, "regions": [], "documents": 0, "skipped": skipped}
+    clusters = max(1, min(regions, len(matrix)))
+    labels = kmeans(matrix, clusters, seed=seed)
     return {
         "version": MAP_VERSION,
-        "regions": described,
-        "documents": len(embedded),
+        "regions": describe_regions(corpus, matrix, labels, clusters),
+        "documents": len(corpus),
         "skipped": skipped,
     }
 
 
-def _corpus(documents: list[dict]) -> tuple[list[dict], np.ndarray]:
+def _corpus(
+    documents: list[dict], vectors: np.ndarray | None = None
+) -> tuple[list[dict], np.ndarray]:
+    """Documents paired with their unit vectors, from the dicts or from a matrix."""
+    if vectors is not None:
+        return documents, vectors
     embedded = [(doc, vector) for doc in documents if (vector := _vector(doc)) is not None]
     if not embedded:
         return [], np.zeros((0, 0), dtype=np.float32)
@@ -270,7 +408,11 @@ def nearest_neighbors(
     real papers near a point is the legible form of "how occupied is this spot".
     """
     similarity = vectors @ vector
-    order = np.argsort(-similarity)[: max(1, limit)]
+    limit = max(1, min(limit, len(similarity)))
+    # argpartition first: a full sort of a corpus-sized array to read five rows
+    # off the top is seconds of wasted work on the live index.
+    top = np.argpartition(-similarity, limit - 1)[:limit]
+    order = top[np.argsort(-similarity[top])]
     return [
         {
             "id": documents[index].get("id", ""),
@@ -302,6 +444,7 @@ def find_gaps(
     limit: int = 10,
     neighbour_cosine: float | None = None,
     occupancy: float = GAP_OCCUPANCY,
+    vectors: np.ndarray | None = None,
 ) -> list[dict]:
     """Bands between neighbouring regions that the indexed corpus barely occupies.
 
@@ -322,13 +465,13 @@ def find_gaps(
     The parent labels carry the warning that matters — an open band between two
     null-saturated regions is unexplored *because the neighbours failed*.
     """
-    corpus, vectors = _corpus(documents)
+    corpus, vectors = _corpus(documents, vectors)
     if not corpus or len(regions) < 2:
         return []
     centroids = normalize(np.vstack([region["centroid"] for region in regions]).astype(np.float32))
     if neighbour_cosine is None:
         neighbour_cosine = neighbour_threshold(centroids)
-    assigned = (vectors @ centroids.T).argmax(axis=1)
+    assigned = assign(vectors, centroids)
     gaps = []
     for left in range(len(regions)):
         for right in range(left + 1, len(regions)):
@@ -340,14 +483,10 @@ def find_gaps(
             others = np.delete(to_midpoint, [left, right])
             if len(others) and others.max() >= min(to_midpoint[left], to_midpoint[right]):
                 continue  # a third literature already sits in the band
-            members = (assigned == left) | (assigned == right)
-            owner = np.vstack(
-                [
-                    vectors[members] @ centroids[left],
-                    vectors[members] @ centroids[right],
-                    vectors[members] @ midpoint,
-                ]
-            ).argmax(axis=0)
+            members = vectors[(assigned == left) | (assigned == right)]
+            owner = (
+                members @ np.vstack([centroids[left], centroids[right], midpoint]).T
+            ).argmax(axis=1)
             support = min(int((owner == 0).sum()), int((owner == 1).sum()))
             if not support:
                 continue
@@ -380,6 +519,7 @@ def place(
     regions: list[dict],
     gaps: list[dict] | None = None,
     nearest_limit: int = 5,
+    vectors: np.ndarray | None = None,
 ) -> dict:
     """Locate an idea on the map: how redundant it is and what surrounds it.
 
@@ -388,7 +528,7 @@ def place(
     published, and the nearest papers are shown so the claim can be checked by
     reading them.
     """
-    corpus, vectors = _corpus(documents)
+    corpus, vectors = _corpus(documents, vectors)
     query = normalize(np.asarray(vector, dtype=np.float32)[None, :])[0]
     if not corpus:
         return {
@@ -442,6 +582,7 @@ def calibrate(
     cutoff_year: int,
     regions: int = 24,
     seed: int = 0,
+    vectors: np.ndarray | None = None,
 ) -> dict:
     """Rebuild the map as it would have looked at ``cutoff_year`` and score it since.
 
@@ -453,11 +594,21 @@ def calibrate(
     study: a region can be mislabelled and still look right if later authors
     simply copied the earlier design.
     """
-    past = [doc for doc in documents if isinstance(doc.get("year"), int) and doc["year"] < cutoff_year]
-    future = [
-        doc for doc in documents if isinstance(doc.get("year"), int) and doc["year"] >= cutoff_year
+    corpus, matrix = _corpus(documents, vectors)
+    dated = [
+        index for index, doc in enumerate(corpus) if isinstance(doc.get("year"), int)
     ]
-    built = build_regions(past, regions=regions, seed=seed)
+    before = np.array(
+        [index for index in dated if corpus[index]["year"] < cutoff_year], dtype=np.int64
+    )
+    after = np.array(
+        [index for index in dated if corpus[index]["year"] >= cutoff_year], dtype=np.int64
+    )
+    past = [corpus[index] for index in before.tolist()]
+    future = [corpus[index] for index in after.tolist()]
+    built = build_regions(
+        past, regions=regions, seed=seed, vectors=matrix[before] if len(before) else None
+    )
     if not built["regions"] or not future:
         return {
             "version": MAP_VERSION,
@@ -470,7 +621,7 @@ def calibrate(
         np.vstack([region["centroid"] for region in built["regions"]]).astype(np.float32)
     )
     later: dict[int, list[dict]] = {region["id"]: [] for region in built["regions"]}
-    corpus, vectors = _corpus(future)
+    corpus, vectors = _corpus(future, matrix[after] if len(after) else None)
     for document, vector in zip(corpus, vectors):
         later[built["regions"][int((centroids @ vector).argmax())]["id"]].append(document)
     rows: dict[str, dict] = {}

app/gapmap_service.py

@@ -1,71 +1,130 @@
-"""Serves the gap map: sample the index once, describe it, place ideas on it.
+"""Serves the gap map: cluster the whole embedded index, describe it, place ideas on it.
 
-The map is a property of the corpus, not of a query, so it is computed from a
-deterministic sample and cached until asked to refresh. Placing an idea then
-costs one query embedding and two matrix products, which is what makes the
-"where does this land" answer instant while still being computed over thousands
-of papers rather than the ten a chat model can read.
+The map is a property of the corpus, not of a query, so it is computed once and
+cached. Placing an idea then costs one query embedding and two matrix products,
+which is what makes the "where does this land" answer instant while still being
+computed over every embedded paper rather than the ten a chat model can read.
+
+Covering the whole index means the build is long enough to be worth watching, so
+it is written as a stream: documents are scanned in pages, provisional centroids
+move as pages arrive, and full-corpus k-means then runs to convergence, emitting
+the real intermediate state after every iteration. Nothing emitted is
+interpolated or replayed — a partial map is labelled partial, and only the final
+event describes the whole corpus. Measured on the live index (2,096,736 embedded
+studies): 264 s end to end — roughly 130 s to scan at sixteen slices and 25
+bounded k-means iterations over the whole corpus in the rest, peaking at 6.9 GB.
 """
 
 from __future__ import annotations
 
 import asyncio
+import contextlib
+import logging
+import zlib
+from collections.abc import Awaitable, Callable
 
 import numpy as np
 
 from app.config import Settings, settings
 from app.embeddings import QueryEmbedder
 from app.gapmap import (
+    MAP_VERSION,
+    RunningClusters,
     _bucket,
-    _corpus,
-    build_regions,
+    assign,
     calibrate,
     combine,
+    describe_regions,
     find_gaps,
     fit_projection,
+    kmeans_steps,
     normalize,
     place,
     project_with,
 )
 from app.models import MapArithmetic
 
+logger = logging.getLogger(__name__)
+
+Progress = Callable[[dict], Awaitable[None]]
+
 
 def _public(region: dict) -> dict:
     """Region without its centroid: a 384-float vector is not UI payload."""
     return {key: value for key, value in region.items() if key != "centroid"}
 
 
-def _layout(documents: list[dict], regions: list[dict]) -> dict | None:
-    """2-D positions for the sampled documents and each region centroid.
+def _drawn(identifier: str, stride: int) -> bool:
+    """Whether a document is one of the ones drawn on the canvas.
 
-    Membership is recomputed rather than returned by build_regions — k-means
-    assignment is argmax over centroid cosines, so this is exact. The fitted
-    basis is kept with the map so a placed idea lands in the same plane.
+    A browser cannot render two million marks, so the canvas shows a subset —
+    chosen by a hash of the document id, which is deterministic, independent of
+    the order pages happen to arrive in, and unrelated to anything the map
+    labels. Every document is still clustered; only the drawing is thinned.
     """
-    corpus, vectors = _corpus(documents)
-    if not len(vectors) or not regions:
-        return None
-    mean, basis = fit_projection(vectors)
-    coords = project_with(mean, basis, vectors)
-    centroids = np.vstack([region["centroid"] for region in regions]).astype(np.float32)
-    assigned = (vectors @ centroids.T).argmax(axis=1)
-    for region, (x, y) in zip(regions, project_with(mean, basis, centroids)):
-        region["x"], region["y"] = float(x), float(y)
-    points = []
-    for index, doc in enumerate(corpus):
-        title = str(doc.get("title") or "")
-        points.append(
-            {
-                "id": str(doc.get("id", "")),
-                "x": float(coords[index, 0]),
-                "y": float(coords[index, 1]),
-                "region": regions[int(assigned[index])]["id"],
-                "bucket": _bucket(doc),
-                "title": title if len(title) <= 120 else f"{title[:119]}…",
-                "year": doc.get("year") if isinstance(doc.get("year"), int) else None,
-            }
-        )
-    return {"points": points, "mean": mean, "basis": basis}
+    return stride <= 1 or zlib.crc32(identifier.encode("utf-8")) % stride == 0
+
+
+class MapBuild:
+    """The corpus as it accumulates: vectors in one matrix, documents beside it.
+
+    The matrix is grown by doubling and handed out as a view, so clustering a
+    corpus of millions never holds two copies of it and adding a page costs the
+    page, not the corpus. The embedding is dropped from each document once it
+    has been copied into the matrix: keeping both is three gigabytes of Python
+    floats for nothing.
+    """
+
+    def __init__(self, corpus: int, capacity: int = 0):
+        self.corpus = corpus
+        self.documents: list[dict] = []
+        self.dimensions: int | None = None
+        self._capacity = max(1024, capacity)
+        self._buffer: np.ndarray | None = None
+        self._rows = 0
+
+    def _reserve(self, rows: int) -> None:
+        assert self.dimensions is not None
+        if self._buffer is None:
+            self._buffer = np.empty((self._capacity, self.dimensions), dtype=np.float32)
+        while self._rows + rows > len(self._buffer):
+            grown = np.empty((len(self._buffer) * 2, self.dimensions), dtype=np.float32)
+            grown[: self._rows] = self._buffer[: self._rows]
+            self._buffer = grown
+
+    def add(self, documents: list[dict]) -> int:
+        """Append a page's vectors to the matrix; returns how many were usable."""
+        rows, kept = [], []
+        for document in documents:
+            embedding = document.pop("embedding", None)
+            if not isinstance(embedding, (list, tuple)) or not embedding:
+                continue
+            vector = np.asarray(embedding, dtype=np.float32)
+            if not np.isfinite(vector).all() or not vector.any():
+                continue
+            if self.dimensions is None:
+                self.dimensions = len(vector)
+            if len(vector) != self.dimensions:
+                continue
+            rows.append(vector)
+            kept.append(document)
+        if not rows:
+            return 0
+        self._reserve(len(rows))
+        assert self._buffer is not None
+        self._buffer[self._rows : self._rows + len(rows)] = normalize(np.vstack(rows))
+        self._rows += len(rows)
+        self.documents.extend(kept)
+        return len(kept)
+
+    @property
+    def vectors(self) -> np.ndarray:
+        if self._buffer is None:
+            return np.zeros((0, self.dimensions or 1), dtype=np.float32)
+        return self._buffer[: self._rows]
+
+    def __len__(self) -> int:
+        return self._rows
 
 
 class GapMapService:
@@ -75,40 +134,272 @@ class GapMapService:
         self.embedder = QueryEmbedder(config)
         self._built: dict | None = None
         self._lock = asyncio.Lock()
+        self._subscribers: set[asyncio.Queue] = set()
+        self._task: asyncio.Task | None = None
 
     async def embed(self, text: str) -> list[float] | None:
         return await self.embedder.embed(text)
 
+    # ---------------------------------------------------------------- building
+
+    async def _publish(self, payload: dict) -> None:
+        for queue in list(self._subscribers):
+            if queue.full():  # a slow reader must not stall the build
+                continue
+            queue.put_nowait(payload)
+
     async def build(self, refresh: bool = False) -> dict:
+        """The cached map, building it (once, shared) if it is not there yet.
+
+        The build is a task of its own rather than a coroutine of whichever
+        request arrived first: a viewer who closes the tab after twenty seconds
+        must not throw away a three-minute clustering run that other viewers,
+        and the cache, are waiting for.
+        """
+        if self._built is not None and not refresh:
+            return self._built
         async with self._lock:
             if self._built is not None and not refresh:
                 return self._built
-            sample = await self.repo.sample_embedded(
-                limit=self.config.gapmap_sample, seed=self.config.gapmap_seed
+            if self._task is None or self._task.done():
+                self._task = asyncio.create_task(self._build())
+            task = self._task
+        self._built = await asyncio.shield(task)
+        return self._built
+
+    async def _build(self) -> dict:
+        """Scan the embedded corpus, cluster it, and describe the result."""
+        corpus = await self.repo.count_embedded()
+        limit = max(0, self.config.gapmap_scan_limit)
+        target = min(corpus, limit) if limit else corpus
+        stride = max(1, target // max(1, self.config.gapmap_points))
+        build = MapBuild(corpus, capacity=target)
+        clusters = RunningClusters(self.config.gapmap_regions, self.config.gapmap_seed)
+        drawn: list[int] = []  # row indices of the documents the canvas shows
+        basis: tuple[np.ndarray, np.ndarray] | None = None
+        emitted = 0
+        ticked = 0.0
+        loop = asyncio.get_running_loop()
+
+        async for batch in self.repo.scan_embedded(
+            batch_size=self.config.gapmap_batch,
+            slices=self.config.gapmap_slices,
+            limit=limit,
+        ):
+            start = len(build)
+            build.add(batch)
+            drawn.extend(
+                row
+                for row in range(start, len(build))
+                if _drawn(str(build.documents[row].get("id", row)), stride)
             )
-            documents = sample["documents"]
-            built = await asyncio.to_thread(
-                build_regions,
-                documents,
+            await asyncio.to_thread(clusters.update, build.vectors[start:])
+            if basis is None and len(build) >= self.config.gapmap_projection_sample:
+                basis = await asyncio.to_thread(fit_projection, build.vectors)
+            # One frame per interval, not one per page: at eleven thousand
+            # documents a second the pages are far faster than anything a canvas
+            # can show, and every frame resends where each drawn point now sits.
+            if (
+                basis is not None
+                and clusters.centroids is not None
+                and loop.time() - ticked >= self.config.gapmap_tick_seconds
+            ):
+                ticked = loop.time()
+                emitted = await self._emit_partial(
+                    build, clusters.centroids, basis, drawn, emitted, target
+                )
+
+        if basis is None and len(build):
+            basis = await asyncio.to_thread(fit_projection, build.vectors)
+
+        centroids = clusters.centroids
+        labels = np.zeros(len(build), dtype=np.int64)
+        iteration = 0
+        if len(build):
+            # Stepped from a thread one iteration at a time: each step is a pass
+            # over the whole corpus, so running the generator inline would block
+            # the event loop for exactly as long as the frames it is producing.
+            steps = kmeans_steps(
+                build.vectors,
                 self.config.gapmap_regions,
-                self.config.gapmap_seed,
+                seed=self.config.gapmap_seed,
+                iterations=self.config.gapmap_iterations,
+                centroids=centroids,
+            )
+            while True:
+                step = await asyncio.to_thread(next, steps, None)
+                if step is None:
+                    break
+                centroids, labels = step
+                iteration += 1
+                if basis is not None:
+                    emitted = await self._emit_partial(
+                        build,
+                        centroids,
+                        basis,
+                        drawn,
+                        emitted,
+                        target,
+                        iteration=iteration,
+                        labels=labels,
+                    )
+
+        clusters_used = len(centroids) if centroids is not None else 0
+        regions = await asyncio.to_thread(
+            describe_regions, build.documents, build.vectors, labels, clusters_used
+        )
+        gaps = await asyncio.to_thread(
+            find_gaps, build.documents, regions, vectors=build.vectors
+        )
+        points = self._points(build, regions, labels, basis, drawn)
+        if basis is not None and regions:
+            centre, plane = basis
+            positions = project_with(
+                centre,
+                plane,
+                np.vstack([region["centroid"] for region in regions]).astype(np.float32),
             )
-            gaps = await asyncio.to_thread(find_gaps, documents, built["regions"])
-            layout = await asyncio.to_thread(_layout, documents, built["regions"])
-            self._built = {
-                "documents": documents,
-                "regions": built["regions"],
-                "gaps": gaps,
-                "points": layout["points"] if layout else [],
-                "basis": (layout["mean"], layout["basis"]) if layout else None,
+            for region, (x, y) in zip(regions, positions):
+                region["x"], region["y"] = float(x), float(y)
+        built = {
+            "documents": build.documents,
+            "vectors": build.vectors,
+            "regions": regions,
+            "gaps": gaps,
+            "points": points,
+            "basis": basis,
+            "coverage": {
+                "clustered": len(build),
+                "corpus": corpus,
+                "regions": len(regions),
+                "drawn": len(points),
+                "complete": True,
+            },
+            "version": MAP_VERSION,
+        }
+        logger.info(
+            "Gap map built over %s of %s embedded studies in %s regions after %s iterations",
+            len(build),
+            corpus,
+            len(regions),
+            iteration,
+        )
+        await self._publish({"event": "built"})
+        return built
+
+    async def _emit_partial(
+        self,
+        build: MapBuild,
+        centroids: np.ndarray,
+        basis: tuple[np.ndarray, np.ndarray],
+        drawn: list[int],
+        emitted: int,
+        target: int,
+        iteration: int | None = None,
+        labels: np.ndarray | None = None,
+    ) -> int:
+        """Publish the map as it stands: new points, where the regions are now.
+
+        Only the points the canvas has not seen are sent in full; the ones it
+        already holds are recoloured through ``pointRegions``, because a region
+        a point belongs to changes on every k-means iteration while its position
+        does not.
+        """
+        if not self._subscribers:
+            return emitted
+        centre, plane = basis
+        rows = np.array(drawn, dtype=np.int64)
+        if not len(rows):
+            return emitted
+        if labels is None:
+            member = assign(build.vectors[rows], centroids)
+        else:
+            member = labels[rows]
+        fresh = []
+        if len(rows) > emitted:
+            new_rows = rows[emitted:]
+            coords = project_with(centre, plane, build.vectors[new_rows])
+            for offset, row in enumerate(new_rows.tolist()):
+                document = build.documents[row]
+                title = str(document.get("title") or "")
+                fresh.append(
+                    {
+                        "id": str(document.get("id", "")),
+                        "x": float(coords[offset, 0]),
+                        "y": float(coords[offset, 1]),
+                        "region": int(member[emitted + offset]),
+                        "bucket": _bucket(document),
+                        "title": title if len(title) <= 120 else f"{title[:119]}…",
+                        "year": document.get("year") if isinstance(document.get("year"), int) else None,
+                    }
+                )
+        positions = project_with(centre, plane, centroids)
+        counts = np.bincount(member, minlength=len(centroids))
+        await self._publish(
+            {
+                "event": "progress",
+                "stage": "clustering" if iteration else "scanning",
+                "iteration": iteration,
                 "coverage": {
-                    "sampled": built["documents"],
-                    "corpus": sample["corpus"],
-                    "regions": len(built["regions"]),
+                    "clustered": len(build),
+                    "corpus": build.corpus,
+                    "regions": int((counts > 0).sum()),
+                    "drawn": len(rows),
+                    "target": target,
+                    "complete": False,
                 },
-                "version": built["version"],
+                "points": fresh,
+                "pointRegions": [int(value) for value in member.tolist()],
+                "regions": [
+                    {
+                        "id": index,
+                        "size": int(counts[index]),
+                        "x": float(positions[index, 0]),
+                        "y": float(positions[index, 1]),
+                    }
+                    for index in range(len(centroids))
+                    if counts[index]
+                ],
             }
-            return self._built
+        )
+        return len(rows)
+
+    def _points(
+        self,
+        build: MapBuild,
+        regions: list[dict],
+        labels: np.ndarray,
+        basis: tuple[np.ndarray, np.ndarray] | None,
+        drawn: list[int],
+    ) -> list[dict]:
+        """2-D positions for the drawn subset, each carrying its region."""
+        if basis is None or not regions or not drawn:
+            return []
+        centre, plane = basis
+        rows = np.array(drawn, dtype=np.int64)
+        coords = project_with(centre, plane, build.vectors[rows])
+        known = {region["id"] for region in regions}
+        points = []
+        for offset, row in enumerate(rows.tolist()):
+            region = int(labels[row])
+            if region not in known:
+                continue
+            document = build.documents[row]
+            title = str(document.get("title") or "")
+            points.append(
+                {
+                    "id": str(document.get("id", "")),
+                    "x": float(coords[offset, 0]),
+                    "y": float(coords[offset, 1]),
+                    "region": region,
+                    "bucket": _bucket(document),
+                    "title": title if len(title) <= 120 else f"{title[:119]}…",
+                    "year": document.get("year") if isinstance(document.get("year"), int) else None,
+                }
+            )
+        return points
+
+    # ---------------------------------------------------------------- serving
 
     def _decorate(
         self, placement: dict, vector: np.ndarray | list[float], built: dict
@@ -147,7 +438,12 @@ class GapMapService:
             )
             return None
         placement = await asyncio.to_thread(
-            place, combined, built["documents"], built["regions"], built["gaps"]
+            place,
+            combined,
+            built["documents"],
+            built["regions"],
+            built["gaps"],
+            vectors=built["vectors"],
         )
         placement["expression"] = {
             "start": expression.start,
@@ -166,10 +462,15 @@ class GapMapService:
         built = await self.build(refresh=refresh)
         warnings: list[str] = []
         coverage = built["coverage"]
-        if coverage["corpus"] > coverage["sampled"]:
+        if coverage["corpus"] > coverage["clustered"]:
+            warnings.append(
+                f"The map clusters {coverage['clustered']} of {coverage['corpus']} "
+                "embedded studies, not the whole index."
+            )
+        if coverage["drawn"] < coverage["clustered"]:
             warnings.append(
-                f"The map describes a random sample of {coverage['sampled']} of "
-                f"{coverage['corpus']} embedded studies, not the whole index."
+                f"All {coverage['clustered']} clustered studies shape the regions; the canvas "
+                f"draws {coverage['drawn']} of them."
             )
         placement = None
         if arithmetic is not None:
@@ -180,7 +481,12 @@ class GapMapService:
                 warnings.append("Embeddings are disabled, so the idea could not be placed.")
             else:
                 placement = await asyncio.to_thread(
-                    place, vector, built["documents"], built["regions"], built["gaps"]
+                    place,
+                    vector,
+                    built["documents"],
+                    built["regions"],
+                    built["gaps"],
+                    vectors=built["vectors"],
                 )
                 placement = self._decorate(placement, vector, built)
         calibration = None
@@ -191,6 +497,7 @@ class GapMapService:
                 cutoff_year,
                 self.config.gapmap_regions,
                 self.config.gapmap_seed,
+                vectors=built["vectors"],
             )
         return {
             "version": built["version"],
@@ -202,3 +509,38 @@ class GapMapService:
             "calibration": calibration,
             "warnings": warnings,
         }
+
+    async def stream(
+        self,
+        idea: str | None = None,
+        arithmetic: MapArithmetic | None = None,
+        cutoff_year: int | None = None,
+        refresh: bool = False,
+        progress: Progress | None = None,
+    ) -> dict:
+        """``assess`` with the build published as it happens.
+
+        A cached map produces no progress events: replaying a finished build as
+        though it were happening would be theatre, and the caller can tell the
+        difference because the first thing it receives is the finished result.
+        """
+        if progress is None:
+            return await self.assess(idea, arithmetic, cutoff_year, refresh)
+        queue: asyncio.Queue = asyncio.Queue(maxsize=64)
+        self._subscribers.add(queue)
+        pump = asyncio.create_task(self._pump(queue, progress))
+        try:
+            return await self.assess(idea, arithmetic, cutoff_year, refresh)
+        finally:
+            self._subscribers.discard(queue)
+            queue.put_nowait(None)
+            with contextlib.suppress(asyncio.CancelledError, Exception):
+                await pump
+
+    @staticmethod
+    async def _pump(queue: asyncio.Queue, progress: Progress) -> None:
+        while True:
+            payload = await queue.get()
+            if payload is None or payload.get("event") == "built":
+                return
+            await progress({key: value for key, value in payload.items() if key != "event"})

app/main.py

@@ -172,6 +172,70 @@ async def gap_map(body: MapRequest, request: Request):
         ) from None
 
 
+@app.post("/map/stream")
+@app.post("/api/map/stream", include_in_schema=False)
+async def gap_map_stream(body: MapRequest, request: Request):
+    """The map as it is built: the same payload as ``/map``, plus the states on
+    the way to it. A cached map arrives as one `result` event with no progress
+    before it, which is the honest thing for work that did not happen."""
+
+    async def stream():
+        queue: asyncio.Queue = asyncio.Queue(maxsize=32)
+
+        async def progress(payload):
+            await queue.put(("progress", payload))
+
+        async def run():
+            try:
+                result = await request.app.state.gapmap.stream(
+                    idea=body.idea,
+                    arithmetic=body.arithmetic,
+                    cutoff_year=body.cutoffYear,
+                    refresh=body.refresh,
+                    progress=progress,
+                )
+                await queue.put(("result", result))
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                logger.error("Streamed gap map failed: %s", type(exc).__name__)
+                await queue.put(
+                    (
+                        "error",
+                        {
+                            "message": "The evidence map could not be built. "
+                            "Please retry or inspect the backend logs."
+                        },
+                    )
+                )
+
+        task = asyncio.create_task(run())
+        try:
+            while True:
+                if await request.is_disconnected():
+                    break
+                try:
+                    event, payload = await asyncio.wait_for(queue.get(), timeout=10)
+                except TimeoutError:
+                    yield ": keepalive\n\n"
+                    continue
+                yield f"event: {event}\ndata: {json.dumps(payload, allow_nan=False)}\n\n"
+                if event in ("result", "error"):
+                    break
+        finally:
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+
+    return StreamingResponse(
+        stream(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )
+
+
 @app.post("/concepts/search")
 @app.post("/api/concepts/search", include_in_schema=False)
 async def concept_search(body: ConceptSearchRequest, request: Request):

app/repository.py

@@ -2,9 +2,11 @@
 
 import asyncio
 import calendar
+import contextlib
 import math
 import re
 import unicodedata
+from collections.abc import AsyncIterator
 from datetime import UTC, date, datetime
 from typing import Any
 
@@ -57,6 +59,34 @@ _PAPER_FIELDS = set(
 )
 
 
+# Embedded studies, the corpus the gap map is drawn from. Reviews stay in: they
+# occupy the space too, and `gapmap` excludes them from a region's attempts.
+_EMBEDDED_QUERY = {
+    "bool": {"filter": [{"term": {"record_kind": "study"}}, {"exists": {"field": "embedding"}}]}
+}
+# Everything the map needs per document and nothing else: abstracts would
+# multiply the bytes on the wire without changing a single position.
+_MAP_FIELDS = [
+    "embedding",
+    "title",
+    "year",
+    "bucket",
+    "record_kind",
+    "is_review",
+    "cited_by_count",
+    "source",
+]
+_PIT_KEEP_ALIVE = "10m"
+
+
+def _split_budget(limit: int, slices: int) -> list[int | None]:
+    """Share a document budget between concurrent slices; no limit means no cap."""
+    if limit <= 0:
+        return [None] * slices
+    base, extra = divmod(limit, slices)
+    return [base + (1 if index < extra else 0) for index in range(slices)]
+
+
 def index_mapping(dimensions: int = 384) -> dict:
     properties: dict[str, Any] = {}
     for name in (
@@ -849,11 +879,7 @@ class ElasticRepository:
         sample so a caller can say which fraction of the corpus it describes.
         """
         await self.ensure_index()
-        query = {
-            "bool": {
-                "filter": [{"term": {"record_kind": "study"}}, {"exists": {"field": "embedding"}}],
-            }
-        }
+        query = _EMBEDDED_QUERY
         total = await self.client.count(index=self.index, query=query)
         body = {
             "index": self.index,
@@ -870,20 +896,89 @@ class ElasticRepository:
                     "boost_mode": "replace",
                 }
             },
-            "source_includes": [
-                "embedding",
-                "title",
-                "year",
-                "bucket",
-                "record_kind",
-                "is_review",
-                "cited_by_count",
-                "source",
-            ],
+            "source_includes": _MAP_FIELDS,
         }
         result = await self._search_with_vectors(body)
         return {"documents": self.hits(result), "corpus": total["count"]}
 
+    async def count_embedded(self) -> int:
+        """How many embedded studies exist, so coverage can be stated as a fraction."""
+        await self.ensure_index()
+        result = await self.client.count(index=self.index, query=_EMBEDDED_QUERY)
+        return int(result["count"])
+
+    async def scan_embedded(
+        self,
+        batch_size: int = 2000,
+        slices: int = 8,
+        limit: int = 0,
+    ) -> AsyncIterator[list[dict]]:
+        """Every embedded study, in batches, over one point in time.
+
+        Paginated rather than fetched at once: the embedded corpus is millions of
+        384-float vectors, so a single oversized search would neither fit in one
+        response nor let a caller show anything before the last document arrived.
+        Slices run concurrently because a single `search_after` walk is limited by
+        round-trip latency, not by Elasticsearch — measured against the live index,
+        one walk moved ~3.2k documents/s and sixteen slices ~11k/s.
+
+        A point in time keeps the walk consistent, so the same document is never
+        yielded twice and none is missed while the index is written to.
+        """
+        await self.ensure_index()
+        slices = max(1, slices)
+        pit = (await self.client.open_point_in_time(index=self.index, keep_alive=_PIT_KEEP_ALIVE))
+        pit_id = pit["id"]
+        queue: asyncio.Queue = asyncio.Queue(maxsize=slices * 2)
+        budgets = _split_budget(limit, slices)
+
+        async def walk(slice_id: int, budget: int | None) -> None:
+            after: list | None = None
+            taken = 0
+            while budget is None or taken < budget:
+                size = batch_size if budget is None else min(batch_size, budget - taken)
+                # No index: a point in time already names one, and Elasticsearch
+                # rejects a search that specifies both.
+                body: dict[str, Any] = {
+                    "size": size,
+                    "query": _EMBEDDED_QUERY,
+                    "pit": {"id": pit_id, "keep_alive": _PIT_KEEP_ALIVE},
+                    "sort": [{"_shard_doc": "asc"}],
+                    "source_includes": _MAP_FIELDS,
+                }
+                if slices > 1:
+                    body["slice"] = {"id": slice_id, "max": slices}
+                if after is not None:
+                    body["search_after"] = after
+                response = await self._search_with_vectors(body)
+                hits = response["hits"]["hits"]
+                if not hits:
+                    return
+                after = hits[-1]["sort"]
+                taken += len(hits)
+                await queue.put(self.hits(response))
+
+        async def produce() -> None:
+            try:
+                await asyncio.gather(*(walk(index, budgets[index]) for index in range(slices)))
+            finally:
+                await queue.put(None)
+
+        producer = asyncio.create_task(produce())
+        try:
+            while True:
+                batch = await queue.get()
+                if batch is None:
+                    break
+                yield batch
+            await producer
+        finally:
+            producer.cancel()
+            with contextlib.suppress(asyncio.CancelledError, Exception):
+                await producer
+            with contextlib.suppress(Exception):  # the walk is over either way
+                await self.client.close_point_in_time(id=pit_id)
+
     async def _search_with_vectors(self, body: dict) -> Any:
         """Search keeping the stored vectors in `_source`, on 9.1 and on 9.2+."""
         if not self._vector_source_filter: