Blame for app/gapmap.py

"""A map of what the literature is missing, built from the indexed corpus alone.

Retrieval answers "what exists". This module answers the complementary question:
for a region of the embedding space, *why* is nothing there. Three kinds of absence
are distinguishable from the fields the index already stores and are not the same
thing at all:

- **null-saturated** — the work was done and came back null or inconclusive. An idea
  landing here is not novel, it is refuted, and the refutations are usually
  uncited, which is exactly why a reader (or a language model summarising the top
  ten hits) concludes the question is open.
- **dark** — trials completed, results never reported. Absence of a paper, not
  absence of evidence.
- **gap** — no work between two populated regions. A combination the neighbouring
  literature implies but nobody has run.

Regions are clusters of real documents, so they can only describe where work exists.
Gaps therefore come from *interpolation*: the midpoint between two nearby region
centroids is a point the literature implies but may never have occupied, and its
cosine to the nearest real paper measures how unoccupied it is. Nothing here calls
an LLM or Elasticsearch, and no score is a probability: every number is a count, a
share of counts, or a cosine.
"""

from __future__ import annotations

from collections import Counter

import numpy as np

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
DARK_SHARE = 0.34
# Both directions must be well represented for a region to count as contested.
CONTESTED_SHARE = 0.25
# Two regions are neighbours, and so worth interpolating between, when their
# centroid cosine is in the top decile of all region pairs; below that the
# midpoint is empty for the boring reason that the two literatures are unrelated.
# A quantile rather than a fixed cosine because what counts as "close" is a
# property of the embedding model and of how narrow the indexed corpus is: on
# MiniLM, unrelated clinical topics sit around 0.1-0.3, so any constant either
# admits everything in a single-disease index or nothing in a broad one. The
# floor stops a corpus of entirely unrelated clusters from producing neighbours
# by ranking alone.
NEIGHBOUR_QUANTILE = 0.9
NEIGHBOUR_FLOOR = 0.35
# The band between two regions is open when it holds at most this share of the
# papers the thinner of the two parent cores holds. A ratio, because an absolute
# cosine cannot express "between": every midpoint is close to its own parents.
GAP_OCCUPANCY = 0.1
# Below this share of readable outcomes a region says nothing about the science,
# only about what the index could read.
READABLE_SHARE = 0.1

REGION_LABELS = ("active", "contested", "null_saturated", "dark", "unread", "thin")
NULL_BUCKETS = ("credible_null", "reported_null")


def _vector(document: dict) -> np.ndarray | None:
    embedding = document.get("embedding")
    if not isinstance(embedding, (list, tuple)) or not embedding:
        return None
    vector = np.asarray(embedding, dtype=np.float32)
    return vector if np.isfinite(vector).all() and vector.any() else None


def normalize(vectors: np.ndarray) -> np.ndarray:
    norms = np.linalg.norm(vectors, axis=-1, keepdims=True)
    return vectors / np.clip(norms, 1e-8, None)


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))
    centroids = np.empty((k, vectors.shape[1]), dtype=np.float32)
    centroids[0] = vectors[rng.integers(count)]
    for index in range(1, k):
        distance = 1.0 - (vectors @ centroids[:index].T).max(axis=1)
        distance = np.clip(distance, 0.0, None) ** 2
        total = distance.sum()
        pick = rng.integers(count) if total <= 0 else rng.choice(count, p=distance / total)
        centroids[index] = vectors[pick]
    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
        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.

    Returns ``(mean, basis)`` so centroids and idea vectors can be projected
    through the same fit via :func:`project_with`. SVD is deterministic for a
    given input, but the sign of each component is arbitrary, so every axis is
    flipped until its largest-magnitude weight is positive — without that the
    same map could render mirrored between builds.
    """
    mean = vectors.mean(axis=0, keepdims=True).astype(np.float32)
    _, _, vt = np.linalg.svd(vectors - mean, full_matrices=False)
    basis = np.zeros((2, vectors.shape[1]), dtype=np.float32)
    rows = min(2, len(vt))  # a single document yields only one axis
    basis[:rows] = vt[:rows]
    for component in basis:
        if component[np.argmax(np.abs(component))] < 0:
            component *= -1
    return mean, basis


def project_with(mean: np.ndarray, basis: np.ndarray, vectors: np.ndarray) -> np.ndarray:
    """Project ``vectors`` through a basis fitted by :func:`fit_projection`."""
    return (np.atleast_2d(np.asarray(vectors, dtype=np.float32)) - mean) @ basis.T


def neighbor_edges(
    vectors: np.ndarray,
    start: int = 0,
    neighbors: int = 3,
    floor: float = 0.0,
    chunk: int = 1024,
) -> list[tuple[int, int, float]]:
    """Cosine-nearest-neighbor edges among unit ``vectors``.

    Each row from ``start`` on is joined to its ``neighbors`` most similar rows
    anywhere in the matrix, provided the cosine reaches ``floor``; rows before
    ``start`` already had their turn on an earlier call, so a stream can call
    this once per frame with only the new rows and never repeats an edge. Pairs
    are returned once each as ``(i, j, cosine)`` with ``i < j``, indices being
    positions in ``vectors``. The degree cap is what keeps a dense literature
    from becoming one hub; the floor is what keeps an isolated paper isolated.
    """
    total = len(vectors)
    if total < 2 or start >= total or neighbors < 1:
        return []
    matrix = np.asarray(vectors, dtype=np.float32)
    edges: dict[tuple[int, int], float] = {}
    for begin in range(start, total, chunk):
        end = min(begin + chunk, total)
        scores = matrix[begin:end] @ matrix.T
        scores[np.arange(end - begin), np.arange(begin, end)] = -np.inf
        top = min(neighbors, total - 1)
        candidates = np.argpartition(-scores, top - 1, axis=1)[:, :top]
        for offset, row in enumerate(range(begin, end)):
            for column in candidates[offset].tolist():
                cosine = float(scores[offset, column])
                if cosine < floor:
                    continue
                key = (row, column) if row < column else (column, row)
                edges[key] = cosine
    return [(i, j, cosine) for (i, j), cosine in edges.items()]


def project(vectors: np.ndarray) -> np.ndarray:
    """2-D coordinates for ``vectors`` under a basis fitted on themselves."""
    if not len(vectors):
        return np.zeros((0, 2), dtype=np.float32)
    mean, basis = fit_projection(vectors)
    return project_with(mean, basis, vectors)


def combine(positive: list[np.ndarray], negative: list[np.ndarray]) -> np.ndarray | None:
    """Embedding arithmetic: a starting point plus added contexts, minus removed ones.

    ``start - context + context`` is the document-level analogue of word2vec
    analogy: the result points at the neighbourhood the combination implies,
    and the corpus decides whether anything actually sits there. The sum is
    renormalized because only direction carries meaning in a cosine space.
    ``None`` when the terms cancel out — a zero-length vector has no direction,
    and placing it would land somewhere meaningless.
    """
    total = np.zeros_like(np.asarray(positive[0], dtype=np.float64))
    for vector in positive:
        total += np.asarray(vector, dtype=np.float64)
    for vector in negative:
        total -= np.asarray(vector, dtype=np.float64)
    norm = np.linalg.norm(total)
    if norm < 1e-6:
        return None
    return (total / norm).astype(np.float32)


def _bucket(document: dict) -> str:
    return str(document.get("query_bucket") or document.get("bucket") or "inconclusive")


def _is_primary(document: dict) -> bool:
    """Reviews restate other people's results, so they cannot evidence a region."""
    return document.get("is_review") is not True and document.get("record_kind", "study") == "study"


def label_region(counts: Counter, attempts: int) -> str:
    """Name the kind of absence a region represents, most specific rule first.

    ``unread`` is the label for a region whose documents carry no readable
    outcome at all — registrations, retractions and unreadable reports. It exists
    because the fallback must not be ``active``: measured against the live index,
    every region held only ``inconclusive``/``failed``/``unreported`` documents,
    and calling those "active" would assert a literature that was never read.
    """
    if attempts < MIN_REGION_ATTEMPTS:
        return "thin"
    nulls = sum(counts[bucket] for bucket in NULL_BUCKETS)
    effects, dark = counts["effect"], counts["unreported"]
    if dark / attempts >= DARK_SHARE:
        return "dark"
    if nulls / attempts >= NULL_SATURATED_SHARE:
        return "null_saturated"
    if min(nulls, effects) / attempts >= CONTESTED_SHARE:
        return "contested"
    if (nulls + effects) / attempts < READABLE_SHARE:
        return "unread"
    return "active"


def _exemplars(documents: list[dict], limit: int = 3) -> list[dict]:
    ranked = sorted(documents, key=lambda doc: doc.get("cited_by_count") or 0, reverse=True)
    return [
        {
            "id": doc.get("id", ""),
            "title": doc.get("title", ""),
            "year": doc.get("year"),
            "bucket": _bucket(doc),
        }
        for doc in ranked[:limit]
    ]


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)]


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.
    """
    described = []
    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 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(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((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": describe_regions(corpus, matrix, labels, clusters),
        "documents": len(corpus),
        "skipped": skipped,
    }


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)
    return (
        [doc for doc, _ in embedded],
        normalize(np.vstack([vector for _, vector in embedded])),
    )


def nearest_neighbors(
    vector: np.ndarray, documents: list[dict], vectors: np.ndarray, limit: int = 1
) -> list[dict]:
    """The ``limit`` closest sampled papers to a point, closest first.

    Callers show these so a bare cosine never has to stand alone: the list of
    real papers near a point is the legible form of "how occupied is this spot".
    """
    similarity = vectors @ vector
    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", ""),
            "title": documents[index].get("title", ""),
            "year": documents[index].get("year"),
            "bucket": _bucket(documents[index]),
            "cosine": float(similarity[index]),
        }
        for index in order
    ]


def _nearest(vector: np.ndarray, documents: list[dict], vectors: np.ndarray) -> dict:
    return nearest_neighbors(vector, documents, vectors, limit=1)[0]


def neighbour_threshold(centroids: np.ndarray) -> float:
    """Centroid cosine above which two regions count as adjacent literatures."""
    pairs = centroids @ centroids.T
    upper = pairs[np.triu_indices(len(centroids), k=1)]
    if not len(upper):
        return NEIGHBOUR_FLOOR
    return max(NEIGHBOUR_FLOOR, float(np.quantile(upper, NEIGHBOUR_QUANTILE)))


def find_gaps(
    documents: list[dict],
    regions: list[dict],
    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.

    Occupancy is counted, not measured by distance: the two regions' own members
    are given to whichever of the two centroids and their midpoint they are
    closest to, and the band is open when the midpoint keeps almost nothing while
    both parents keep plenty. Pairs whose midpoint is better described by some
    third region are skipped — that band is occupied, by a whole literature.

    Distance alone cannot express any of this: a midpoint is always near its own
    parents, so an absolute cosine threshold either rejects every gap or accepts
    every one, and a fixed-radius ball around each point holds nothing at all in
    a real high-dimensional corpus (measured against the live index: every such
    ball was empty, so no pair could ever have been a gap).

    ``support`` is the thinner parent core: a band between two substantial
    literatures is a combination many people were positioned to try and did not.
    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, 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 = assign(vectors, centroids)
    gaps = []
    for left in range(len(regions)):
        for right in range(left + 1, len(regions)):
            separation = float(np.clip(centroids[left] @ centroids[right], -1.0, 1.0))
            if separation < neighbour_cosine:
                continue
            midpoint = normalize(((centroids[left] + centroids[right]) / 2)[None, :])[0]
            to_midpoint = centroids @ midpoint
            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 = 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
            band = int((owner == 2).sum())
            if band > occupancy * support:
                continue
            parents = [regions[left], regions[right]]
            gaps.append(
                {
                    "regions": [region["id"] for region in parents],
                    "parentLabels": [region["label"] for region in parents],
                    "openness": 1.0 - band / support,
                    "support": support,
                    "band": band,
                    "separation": separation,
                    "nearest": _nearest(midpoint, corpus, vectors),
                    "exemplars": [region["exemplars"][:1] for region in parents],
                    "discouraged": any(
                        region["label"] in {"null_saturated", "dark"} for region in parents
                    ),
                }
            )
    gaps.sort(key=lambda gap: (gap["openness"], gap["support"]), reverse=True)
    return gaps[:limit]


def place(
    vector: list[float] | np.ndarray,
    documents: list[dict],
    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.

    ``redundancy`` is the cosine to the single closest indexed paper. It is the
    honest headline: most proposals are a near-duplicate of something already
    published, and the nearest papers are shown so the claim can be checked by
    reading them.
    """
    corpus, vectors = _corpus(documents, vectors)
    query = normalize(np.asarray(vector, dtype=np.float32)[None, :])[0]
    if not corpus:
        return {
            "redundancy": None,
            "nearest": None,
            "neighbors": [],
            "region": None,
            "nearestGap": None,
        }
    neighbors = nearest_neighbors(query, corpus, vectors, limit=nearest_limit)
    region = None
    if regions:
        centroids = normalize(
            np.vstack([entry["centroid"] for entry in regions]).astype(np.float32)
        )
        similarity = centroids @ query
        position = int(similarity.argmax())
        region = {**regions[position], "cosine": float(similarity[position])}
        region.pop("centroid", None)
    return {
        "redundancy": neighbors[0]["cosine"],
        "nearest": neighbors[0],
        "neighbors": neighbors[1:],
        "region": region,
        "nearestGap": _nearest_gap(query, regions, gaps or []),
    }


def _nearest_gap(query: np.ndarray, regions: list[dict], gaps: list[dict]) -> dict | None:
    if not gaps or not regions:
        return None
    by_id = {region["id"]: region for region in regions}
    best, best_cosine = None, -1.0
    for gap in gaps:
        parents = [by_id[identifier] for identifier in gap["regions"] if identifier in by_id]
        if len(parents) != 2:
            continue
        midpoint = normalize(
            np.vstack([parent["centroid"] for parent in parents]).astype(np.float32).mean(
                axis=0, keepdims=True
            )
        )[0]
        cosine = float(midpoint @ query)
        if cosine > best_cosine:
            best, best_cosine = gap, cosine
    return {**best, "cosine": best_cosine} if best else None


def calibrate(
    documents: list[dict],
    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.

    The labels claim something checkable: work published after the cutoff should
    behave differently depending on the label its region carried before it. A
    null-saturated region that keeps producing nulls, and an active region that
    keeps producing effects, is the map earning its labels. This is a calibration
    record over one corpus and one cutoff, not a forecast and not an accuracy
    study: a region can be mislabelled and still look right if later authors
    simply copied the earlier design.
    """
    corpus, matrix = _corpus(documents, vectors)
    dated = [
        index for index, doc in enumerate(corpus) if isinstance(doc.get("year"), int)
    ]
    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,
            "cutoffYear": cutoff_year,
            "past": len(past),
            "future": 0,
            "labels": [],
        }
    centroids = normalize(
        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, 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] = {}
    for region in built["regions"]:
        arrivals = [doc for doc in later[region["id"]] if _is_primary(doc)]
        row = rows.setdefault(
            region["label"],
            {"label": region["label"], "regions": 0, "priorAttempts": 0, "laterPapers": 0,
             "laterNull": 0, "laterEffect": 0},
        )
        row["regions"] += 1
        row["priorAttempts"] += region["attempts"]
        row["laterPapers"] += len(arrivals)
        row["laterNull"] += sum(_bucket(doc) in NULL_BUCKETS for doc in arrivals)
        row["laterEffect"] += sum(_bucket(doc) == "effect" for doc in arrivals)
    for row in rows.values():
        row["laterNullShare"] = row["laterNull"] / row["laterPapers"] if row["laterPapers"] else None
    return {
        "version": MAP_VERSION,
        "cutoffYear": cutoff_year,
        "past": len(past),
        "future": len(corpus),
        "labels": sorted(rows.values(), key=lambda row: row["label"]),
    }