zettelkasten.syllabus

zettelkasten.syllabus

Reading-syllabus scoring: a two-axis, debiased notion of paper importance.

This module is the deterministic scoring engine behind the Workshop / Papers view. Like :mod:zettelkasten.framing, it is a pure, cacheable, unit-testable layer parametrized by a get_graph callable so the MCP server and the dashboard backend (separate graph caches) can share it. There is no LLM here — every score is a reproducible function of the corpus.

Importance is defined along two independent axes that are exposed separately so the Papers view can sort/color by either alone:

• influence — cohort-normalized standing of cited_by_count. It is the percentile of a work's raw count within its cohort (a publication-year window × theme), computed over the corpus. The raw count is never used directly: this debiases for paper age, field size, and survey citation inflation, so a heavily-cited paper in a citation-rich cohort does not automatically outrank a modestly-cited paper that leads a sparse cohort.

• salience — how load-bearing a work is inside our own graph: a 0–1 blend of (1) the weighted in-degree of incoming supports/replicates/ extends/cites edges from distinct sources, (2) graph centrality (PageRank over the note/citation graph), and (3) claim-coreness (how many key claims a work is the core paper for). The claim layer does not exist yet, so claim-coreness defaults to 0 and is dropped from the blend until it appears — a clean seam that composes later without re-tuning today's scores.

Derived from those:

• importance — a configurable blend of influence + salience (default 50/50).
• anchor_score — a separate facet (is_review / out-degree / co-citation): good entry points into a literature, distinct from importance.
• read_priority — importance × unread(coverage.human) × claim-coverage: what to read next. Drops to 0 once a work is fully read by a human.

The pure primitives operate on a list of :class:Work records, so they are trivial to unit-test against a synthetic corpus. :func:build_corpus assembles those records from the live graph (citations + sources + note edges) and is the seam the later M0 waves (interestingness, trend layer, situate()) hook into.

ImportanceWeights dataclass

Blend of the two top-level axes into a single importance score.

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class ImportanceWeights:
    """Blend of the two top-level axes into a single importance score."""

    influence: float = 0.5
    salience: float = 0.5

SalienceWeights dataclass

Blend of salience's sub-signals. Signals absent across the whole corpus are dropped and the remaining weights renormalized (see :func:salience).

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class SalienceWeights:
    """Blend of salience's sub-signals. Signals absent across the whole corpus
    are dropped and the remaining weights renormalized (see :func:`salience`)."""

    in_degree: float = 0.5
    centrality: float = 0.3
    claim_coreness: float = 0.2

AnchorWeights dataclass

Blend of the anchor facet's sub-signals (entry-point quality).

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class AnchorWeights:
    """Blend of the anchor facet's sub-signals (entry-point quality)."""

    is_review: float = 0.5
    out_degree: float = 0.3
    co_citation: float = 0.2

Work dataclass

A single scorable paper-like entity (a citation or a promoted source).

Graph-derived signals (weighted_in_degree, centrality, out_degree, co_citation) and the claim hooks default to 0 / neutral so a bare Work is well-defined and the scorers compose even before :func:build_corpus has populated them.

Source code in zettelkasten/syllabus.py

@dataclass
class Work:
    """A single scorable paper-like entity (a citation or a promoted source).

    Graph-derived signals (``weighted_in_degree``, ``centrality``,
    ``out_degree``, ``co_citation``) and the claim hooks default to 0 / neutral
    so a bare ``Work`` is well-defined and the scorers compose even before
    :func:`build_corpus` has populated them.
    """

    id: str
    title: str = ""
    year: int | None = None
    theme: str = ""
    doi: str = ""
    cited_by_count: int = 0
    # Whether ``cited_by_count`` is a KNOWN value vs an unenriched placeholder.
    # Defaults True so a directly-constructed ``Work`` (tests, ad-hoc callers)
    # behaves as before — only :func:`build_corpus` downgrades it to False when a
    # work's stored metadata carries no ``cited_by_count`` key at all. A DOI-less
    # arXiv paper that was never resolved then keeps ``cited_by_count == 0`` but
    # is *unknown*, not a genuinely-uncited work — :func:`influence` excludes
    # unknown works from the cohort ranking so they neither earn a misleading
    # percentile nor drag down the enriched works they sit beside.
    cited_by_count_known: bool = True
    counts_by_year: list[dict] = field(default_factory=list)
    coverage_human: str = "none"  # none | skimmed | full
    authors: list[str] = field(default_factory=list)

    # Non-None means this work is backed by a writable LOCAL source folder (a
    # "promoted" source) whose name is held here — coverage writes and
    # click-to-open target this folder. Citation-only works keep it None.
    source_graph: str | None = None

    # Salience signals (filled by build_corpus).
    weighted_in_degree: float = 0.0  # weighted edges from *distinct* sources
    centrality: float = 0.0  # PageRank over the note/citation graph
    core_claim_count: int = 0  # claims this work is the *core* paper for (claim hook)

    # Anchor-facet signals.
    is_review: bool = False
    out_degree: int = 0
    co_citation: int = 0

    # Read-priority claim-coverage hook. ``None`` until the claim layer exists,
    # which makes the claim-coverage factor neutral (see _claim_coverage_factor).
    unlocks_uncovered_claim: bool | None = None

TrendMetrics dataclass

Per-paper temporal signals, all reconstructed as of reference_year.

Frozen and self-contained so later waves can pass it around as a clean value. velocity/acceleration are in raw citations-per-year (and per-year²); citation_recency/publication_recency are 0–1 shares; trajectory is one of :data:TRAJECTORY_CLASSES.

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class TrendMetrics:
    """Per-paper temporal signals, all reconstructed as of ``reference_year``.

    Frozen and self-contained so later waves can pass it around as a clean value.
    ``velocity``/``acceleration`` are in raw citations-per-year (and per-year²);
    ``citation_recency``/``publication_recency`` are 0–1 shares; ``trajectory`` is
    one of :data:`TRAJECTORY_CLASSES`.
    """

    work_id: str
    reference_year: int
    velocity: float            # OLS slope of citations/year over the recent window
    acceleration: float        # OLS slope of the year-over-year deltas (2nd derivative)
    citation_recency: float    # share of total citations received in the last window
    publication_recency: float # how new the work itself is, from its publication year
    trajectory: str            # one of TRAJECTORY_CLASSES
    recent_citations: int      # raw citations within the window (frontier helper)
    total_citations: int       # raw citations across counts_by_year truncated to <= T

InterestingnessWeights dataclass

Blend weights for the interestingness sub-signals (used by the blend term; the max term is weight-free). Absent signals contribute 0 and are simply out-weighed rather than dropped, since each is already a 0–1 normalized quantity.

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class InterestingnessWeights:
    """Blend weights for the interestingness sub-signals (used by the blend term;
    the max term is weight-free). Absent signals contribute 0 and are simply
    out-weighed rather than dropped, since each is already a 0–1 normalized
    quantity."""

    surprise: float = 0.25
    brokerage: float = 0.20
    dissent: float = 0.20
    citation_anomaly: float = 0.20
    novelty: float = 0.15

Interestingness dataclass

Per-paper interestingness, with the dominant firing sub-signal recorded.

score is 0–1. dominant_signal is the name of the strongest firing sub-signal (one of the five fields below), or None when nothing fired (or the work was floored out by the relevance guard) — which the rationale layer reads to decide between an explanation and a neutral return.

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class Interestingness:
    """Per-paper interestingness, with the dominant firing sub-signal recorded.

    ``score`` is 0–1. ``dominant_signal`` is the name of the strongest firing
    sub-signal (one of the five fields below), or ``None`` when nothing fired
    (or the work was floored out by the relevance guard) — which the rationale
    layer reads to decide between an explanation and a neutral return.
    """

    work_id: str
    score: float
    dominant_signal: str | None
    surprise: float
    brokerage: float
    dissent: float
    citation_anomaly: float
    novelty: float

Situation dataclass

Per-paper context labels. Any label may be None when nothing applies.

temporal ∈ {seminal, recent, origin-of-claim, None}; position ∈ {foundational, extension, superseded, replication, review, bridge, frontier-rising, None}; influence is the wave-1 cohort percentile (0–1) with influence_label its tier and sleeper flagging a core-but-undercited work.

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class Situation:
    """Per-paper context labels. Any label may be ``None`` when nothing applies.

    ``temporal`` ∈ {seminal, recent, origin-of-claim, None}; ``position`` ∈
    {foundational, extension, superseded, replication, review, bridge,
    frontier-rising, None}; ``influence`` is the wave-1 cohort percentile (0–1)
    with ``influence_label`` its tier and ``sleeper`` flagging a core-but-undercited
    work.
    """

    work_id: str
    temporal: str | None
    position: str | None
    influence: float | None
    influence_label: str | None
    sleeper: bool

Theme dataclass

A single theme: a labeled group of member works plus its weight.

work_ids are the in-corpus members (sorted for stable output). weight is len(work_ids) × Σ member cited_by_count — size times external citation mass. source records which priority tier produced it (landscape / cluster / cold-start).

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class Theme:
    """A single theme: a labeled group of member works plus its weight.

    ``work_ids`` are the in-corpus members (sorted for stable output). ``weight``
    is ``len(work_ids) × Σ member cited_by_count`` — size times external citation
    mass. ``source`` records which priority tier produced it (``landscape`` /
    ``cluster`` / ``cold-start``).
    """

    theme_id: str
    label: str
    work_ids: tuple[str, ...]
    weight: float
    source: str

ThemeModel dataclass

The corpus's theme partition.

themes is ordered by descending weight then theme_id (a stable, deterministic order). assignment maps each member work to its primary theme — the highest-weight theme it belongs to — for display; the per-theme work_ids remain the authoritative (possibly overlapping) membership that :func:set_cover consumes. source is the priority tier that produced the partition.

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class ThemeModel:
    """The corpus's theme partition.

    ``themes`` is ordered by descending weight then ``theme_id`` (a stable,
    deterministic order). ``assignment`` maps each member work to its *primary*
    theme — the highest-weight theme it belongs to — for display; the per-theme
    ``work_ids`` remain the authoritative (possibly overlapping) membership that
    :func:`set_cover` consumes. ``source`` is the priority tier that produced the
    partition.
    """

    themes: tuple[Theme, ...]
    assignment: dict[str, str]
    source: str

CoverSelection dataclass

One paper chosen for the reading list, with the units it covers.

theme_ids are the themes it covers (theme-coverage mode); claim_ids are the key-claim ids it is the CORE paper for (claim-coverage mode). Exactly one is populated depending on the cover universe (the other stays an empty tuple), so existing theme-mode consumers reading theme_ids are unaffected.

acquisition marks a coverer whose work is UNOWNED (source_graph is None): it covers its unit only by being ACQUIRED, not by being read. It is set at the :func:set_cover boundary so an owned-only read-hook consumer can filter it out deterministically (an unowned work must never appear in owned-only reading recommendations) without re-deriving ownership from the work list — the invariant lives in set_cover, not its callers.

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class CoverSelection:
    """One paper chosen for the reading list, with the units it covers.

    ``theme_ids`` are the themes it covers (theme-coverage mode); ``claim_ids``
    are the key-claim ids it is the CORE paper for (claim-coverage mode). Exactly
    one is populated depending on the cover universe (the other stays an empty
    tuple), so existing theme-mode consumers reading ``theme_ids`` are unaffected.

    ``acquisition`` marks a coverer whose work is UNOWNED (``source_graph is
    None``): it covers its unit only by being ACQUIRED, not by being read. It is
    set at the :func:`set_cover` boundary so an owned-only read-hook consumer can
    filter it out deterministically (an unowned work must never appear in
    owned-only reading recommendations) without re-deriving ownership from the
    work list — the invariant lives in ``set_cover``, not its callers.
    """

    work_id: str
    theme_ids: tuple[str, ...]
    value: float
    claim_ids: tuple[str, ...] = ()
    acquisition: bool = False

SetCover dataclass

A suggested reading list.

selected is the flat list of chosen papers in selection order; by_theme groups the selected work ids by the themes they cover; gaps are units left with unmet residual need (budget exhausted, or no selectable coverer at all — every member out of budget, already fully read, or an unresolved claim core); pre_covered are units that needed nothing because a member was already fully read.

by_theme/by_claim group the picks by the THEME and CLAIM key spaces respectively; claim_ids is the claim-space subset of the covered units. gaps UNIONS both key spaces (theme ids ∪ "<graph>::<id>" claim uids) — consult universe ("theme" / "claim" / "composite") to know which spaces are populated. The claim fields default empty so theme-only callers and persisted theme covers are byte-for-byte unchanged.

Source code in zettelkasten/syllabus.py

@dataclass(frozen=True)
class SetCover:
    """A suggested reading list.

    ``selected`` is the flat list of chosen papers in selection order;
    ``by_theme`` groups the selected work ids by the themes they cover; ``gaps``
    are units left with unmet residual need (budget exhausted, or no selectable
    coverer at all — every member out of budget, already fully read, or an
    unresolved claim core); ``pre_covered`` are units that needed nothing because
    a member was already fully read.

    ``by_theme``/``by_claim`` group the picks by the THEME and CLAIM key spaces
    respectively; ``claim_ids`` is the claim-space subset of the covered units.
    ``gaps`` UNIONS both key spaces (theme ids ∪ ``"<graph>::<id>"`` claim uids) —
    consult ``universe`` (``"theme"`` / ``"claim"`` / ``"composite"``) to know
    which spaces are populated. The claim fields default empty so theme-only
    callers and persisted theme covers are byte-for-byte unchanged.
    """

    selected: tuple[CoverSelection, ...]
    by_theme: dict[str, tuple[str, ...]]
    gaps: tuple[str, ...]
    pre_covered: tuple[str, ...]
    budget: int
    by_claim: dict[str, tuple[str, ...]] = field(default_factory=dict)
    claim_ids: tuple[str, ...] = ()
    universe: str = "theme"

influence

influence(works: list[Work], *, cohort_window: int = DEFAULT_COHORT_WINDOW) -> dict[str, float]

Cohort-normalized influence in 0–1: the percentile of cited_by_count within each work's (year-window × theme) cohort.

Uses the mean/Hazen percentile rank (#below + 0.5·#equal) / cohort_size, which gives two works that hold the same standing in their respective cohorts the same score regardless of their raw counts, and never uses the raw count directly. A singleton cohort scores 0.5 (neutral).

Only works with a KNOWN citation count (cited_by_count_known) participate: cohorts are formed from them and the returned dict has an entry only for them. A work whose citation count was never enriched (a DOI-less paper stuck at 0) is intentionally absent — callers treat a missing id as "influence unknown" rather than a real low percentile, so it can't depress the cohort it happens to share a year/theme with, nor read as a confidently low score.

Source code in zettelkasten/syllabus.py

def influence(works: list[Work], *, cohort_window: int = DEFAULT_COHORT_WINDOW) -> dict[str, float]:
    """Cohort-normalized influence in 0–1: the percentile of ``cited_by_count``
    within each work's (year-window × theme) cohort.

    Uses the mean/Hazen percentile rank ``(#below + 0.5·#equal) / cohort_size``,
    which gives two works that hold the *same standing in their respective
    cohorts* the same score regardless of their raw counts, and never uses the
    raw count directly. A singleton cohort scores 0.5 (neutral).

    Only works with a KNOWN citation count (``cited_by_count_known``) participate:
    cohorts are formed from them and the returned dict has an entry *only* for
    them. A work whose citation count was never enriched (a DOI-less paper stuck
    at 0) is intentionally absent — callers treat a missing id as "influence
    unknown" rather than a real low percentile, so it can't depress the cohort it
    happens to share a year/theme with, nor read as a confidently low score.
    """
    known = [w for w in works if w.cited_by_count_known]
    cohorts: dict[tuple[Any, str], list[Work]] = {}
    for w in known:
        cohorts.setdefault(_cohort_key(w, cohort_window), []).append(w)

    scores: dict[str, float] = {}
    for w in known:
        cohort = cohorts[_cohort_key(w, cohort_window)]
        c = w.cited_by_count or 0
        below = sum(1 for o in cohort if (o.cited_by_count or 0) < c)
        equal = sum(1 for o in cohort if (o.cited_by_count or 0) == c)
        scores[w.id] = (below + 0.5 * equal) / len(cohort)
    return scores

salience

salience(works: list[Work], *, weights: SalienceWeights | None = None) -> dict[str, float]

In-graph salience in 0–1: blend of distinct-source weighted in-degree, PageRank centrality, and claim-coreness.

Each sub-signal is max-normalized so it spans 0–1, then blended by :func:_blend (which drops absent signals). The result is monotonic in weighted_in_degree while only that signal is present, which is the intended "load-bearing-ness" behavior.

Source code in zettelkasten/syllabus.py

def salience(works: list[Work], *, weights: SalienceWeights | None = None) -> dict[str, float]:
    """In-graph salience in 0–1: blend of distinct-source weighted in-degree,
    PageRank centrality, and claim-coreness.

    Each sub-signal is max-normalized so it spans 0–1, then blended by
    :func:`_blend` (which drops absent signals). The result is monotonic in
    ``weighted_in_degree`` while only that signal is present, which is the
    intended "load-bearing-ness" behavior.
    """
    weights = weights or SalienceWeights()
    work_ids = [w.id for w in works]
    in_degree = _max_normalize({w.id: max(0.0, w.weighted_in_degree) for w in works})
    centrality = _max_normalize({w.id: max(0.0, w.centrality) for w in works})
    claim = _max_normalize({w.id: max(0.0, float(w.core_claim_count)) for w in works})
    return _blend(
        work_ids,
        [
            (in_degree, weights.in_degree),
            (centrality, weights.centrality),
            (claim, weights.claim_coreness),
        ],
    )

importance

importance(works: list[Work], *, weights: ImportanceWeights | None = None, influence_scores: dict[str, float] | None = None, salience_scores: dict[str, float] | None = None, cohort_window: int = DEFAULT_COHORT_WINDOW) -> dict[str, float]

Configurable blend of influence + salience in 0–1 (default 50/50).

Blended through :func:_blend, so an axis that is entirely absent across the corpus (e.g. salience on a graph with no note edges) is dropped and the surviving weight renormalized — matching salience/anchor_score. This preserves the seam invariant: a no-edge corpus yields importance == influence (not a halved blend), so today's scores compose without re-tuning when the missing axis lights up later.

Precomputed axis scores can be passed in to avoid recomputation when a caller already has them (the Papers view needs all axes at once).

Source code in zettelkasten/syllabus.py

def importance(
    works: list[Work],
    *,
    weights: ImportanceWeights | None = None,
    influence_scores: dict[str, float] | None = None,
    salience_scores: dict[str, float] | None = None,
    cohort_window: int = DEFAULT_COHORT_WINDOW,
) -> dict[str, float]:
    """Configurable blend of influence + salience in 0–1 (default 50/50).

    Blended through :func:`_blend`, so an axis that is entirely absent across the
    corpus (e.g. salience on a graph with no note edges) is dropped and the
    surviving weight renormalized — matching ``salience``/``anchor_score``. This
    preserves the seam invariant: a no-edge corpus yields ``importance ==
    influence`` (not a halved blend), so today's scores compose without re-tuning
    when the missing axis lights up later.

    Precomputed axis scores can be passed in to avoid recomputation when a caller
    already has them (the Papers view needs all axes at once).
    """
    weights = weights or ImportanceWeights()
    inf = influence_scores if influence_scores is not None else influence(works, cohort_window=cohort_window)
    sal = salience_scores if salience_scores is not None else salience(works)
    # Works with a KNOWN influence blend influence + salience as before. Works
    # whose influence is unknown (no citation data — absent from ``inf``) fall
    # back to salience alone rather than being blended against a fictitious 0,
    # which would otherwise halve their importance for a missing signal we simply
    # don't have. This mirrors ``_blend``'s "drop the absent axis" contract, but
    # applied per-work instead of corpus-wide.
    known_ids = [w.id for w in works if w.id in inf]
    blended = _blend(known_ids, [(inf, weights.influence), (sal, weights.salience)])
    return {
        w.id: (blended.get(w.id, 0.0) if w.id in inf else sal.get(w.id, 0.0))
        for w in works
    }

anchor_score

anchor_score(works: list[Work], *, weights: AnchorWeights | None = None) -> dict[str, float]

Anchor facet in 0–1: how good a entry point into the literature a work is.

A separate facet from importance — reviews, well-connected hubs, and frequently co-cited works are good places to start reading even when they are not the most important results. Blended (and absent-signal-dropped) the same way as salience.

Source code in zettelkasten/syllabus.py

def anchor_score(works: list[Work], *, weights: AnchorWeights | None = None) -> dict[str, float]:
    """Anchor facet in 0–1: how good a *entry point* into the literature a work is.

    A separate facet from importance — reviews, well-connected hubs, and
    frequently co-cited works are good places to start reading even when they are
    not the most important results. Blended (and absent-signal-dropped) the same
    way as salience.
    """
    weights = weights or AnchorWeights()
    work_ids = [w.id for w in works]
    review = {w.id: (1.0 if w.is_review else 0.0) for w in works}
    out_degree = _max_normalize({w.id: float(max(0, w.out_degree)) for w in works})
    co_citation = _max_normalize({w.id: float(max(0, w.co_citation)) for w in works})
    return _blend(
        work_ids,
        [
            (review, weights.is_review),
            (out_degree, weights.out_degree),
            (co_citation, weights.co_citation),
        ],
    )

read_priority

read_priority(works: list[Work], *, importance_scores: dict[str, float] | None = None, cohort_window: int = DEFAULT_COHORT_WINDOW) -> dict[str, float]

What to read next, in 0–1: importance × unread(coverage.human) × claim-coverage.

A fully human-read work (coverage.human == "full") has an unread factor of 0, so its read_priority is 0 regardless of importance.

Source code in zettelkasten/syllabus.py

def read_priority(
    works: list[Work],
    *,
    importance_scores: dict[str, float] | None = None,
    cohort_window: int = DEFAULT_COHORT_WINDOW,
) -> dict[str, float]:
    """What to read next, in 0–1: ``importance × unread(coverage.human) ×
    claim-coverage``.

    A fully human-read work (``coverage.human == "full"``) has an unread factor of
    0, so its read_priority is 0 regardless of importance.
    """
    imp = importance_scores if importance_scores is not None else importance(works, cohort_window=cohort_window)
    scores: dict[str, float] = {}
    for w in works:
        unread = _UNREAD_FACTOR.get((w.coverage_human or "none").lower(), 1.0)
        scores[w.id] = imp.get(w.id, 0.0) * unread * _claim_coverage_factor(w)
    return scores

score_corpus

score_corpus(works: list[Work], *, importance_weights: ImportanceWeights | None = None, salience_weights: SalienceWeights | None = None, cohort_window: int = DEFAULT_COHORT_WINDOW) -> list[dict[str, Any]]

Score every work along all axes and return rows sorted by read_priority.

This is the composable output the Papers view consumes — each axis is present independently (influence and salience separately, plus the derived importance, anchor_score and read_priority) so the UI can sort or color by any of them.

Source code in zettelkasten/syllabus.py

def score_corpus(
    works: list[Work],
    *,
    importance_weights: ImportanceWeights | None = None,
    salience_weights: SalienceWeights | None = None,
    cohort_window: int = DEFAULT_COHORT_WINDOW,
) -> list[dict[str, Any]]:
    """Score every work along all axes and return rows sorted by read_priority.

    This is the composable output the Papers view consumes — each axis is present
    independently (``influence`` and ``salience`` separately, plus the derived
    ``importance``, ``anchor_score`` and ``read_priority``) so the UI can sort or
    color by any of them.
    """
    inf = influence(works, cohort_window=cohort_window)
    sal = salience(works, weights=salience_weights)
    imp = importance(
        works,
        weights=importance_weights,
        influence_scores=inf,
        salience_scores=sal,
    )
    anc = anchor_score(works)
    rp = read_priority(works, importance_scores=imp)

    rows = [
        {
            "id": w.id,
            "title": w.title,
            "year": w.year,
            "theme": w.theme,
            "cited_by_count": w.cited_by_count,
            "coverage_human": w.coverage_human,
            # ``None`` when the citation count is unknown (never enriched) — the
            # work is absent from ``inf`` and reads as "—" rather than a fake 0.
            "influence": (round(inf[w.id], 4) if w.id in inf else None),
            "salience": round(sal.get(w.id, 0.0), 4),
            "importance": round(imp.get(w.id, 0.0), 4),
            "anchor_score": round(anc.get(w.id, 0.0), 4),
            "read_priority": round(rp.get(w.id, 0.0), 4),
        }
        for w in works
    ]
    rows.sort(key=lambda r: (r["read_priority"], r["importance"]), reverse=True)
    return rows

build_corpus

build_corpus(get_graph: GetGraph, *, project: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, citation_scope: 'set[str] | None' = None) -> tuple[list[Work], list[str]]

Assemble the scorable corpus from the live graph.

Returns (works, search_sources). Works are built from the citation store and from the note-graph sources (deduplicated against citations by canonical key, with the source's coverage merged onto the matching citation). Graph signals are derived here:

• weighted_in_degree — for each work, the summed :data:SALIENCE_RELATIONS weight of incoming edges, counting each source graph at most once (its strongest edge), i.e. distinct-source in-degree.
• centrality — PageRank over the source→target edge graph.
• out_degree — a source's count of outgoing salience edges.

Claim signals and inter-citation co-citation are left at their defaults (later M0 waves). Pure: no LLM, no network.

Dedup & merge. Records that resolve to the same work by :func:canonical_citation_key are collapsed onto a single :class:Work, whether the duplicates are citation-vs-citation (two files sharing a DOI) or source-vs-citation (a promoted source that is also a citation). On merge the graph signals are folded together — distinct-source weighted_in_degree (max per source across both id forms, then summed), centrality (summed, as each id-node accrued its own PageRank mass), cited_by_count (max) — and counts_by_year/coverage are filled from whichever record carries them, so no endorsement edge or PageRank mass is orphaned by the dedup. A degenerate canonical key (the constant "meta:" for metadata-poor works, or a placeholder DOI like doi:n/a) does not discriminate works, so such records are never deduped — each keeps its storage id as identity (see :func:_is_dedupable_key), keeping distinct works distinct.

Project scope. Sources and edges are scoped to project via :func:frame_search_sources. Citations are scoped too: when project is given, only citations referenced by the in-project sources enter the corpus, so cohort percentiles are computed over the project's own pool. A citation is "referenced" if any in-project note links to it — by any graph (_citations, the default graph, _cross), derived from the very same link walk that collects salience, so a citation whose in-degree is counted is never then dropped by the scope filter. When project is empty the full global citation pool is used (there is no per-citation project tag — the association is derived from note links).

Explicit citation scope. citation_scope (when not None) restricts the citation pool to exactly that id set, overriding the project-derived scoping. The single-source graph scope (:func:papers._graph_scoped_corpus) uses it to avoid materializing the entire global citation pool: it passes a dedup-group-closed set of the citations that one graph engages with, so the surviving Works — and every canonical-key merge — are identical to the full-pool corpus, only without building the unreferenced citations.

Source code in zettelkasten/syllabus.py

def build_corpus(
    get_graph: GetGraph,
    *,
    project: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    citation_scope: "set[str] | None" = None,
) -> tuple[list[Work], list[str]]:
    """Assemble the scorable corpus from the live graph.

    Returns ``(works, search_sources)``. Works are built from the citation store
    and from the note-graph sources (deduplicated against citations by canonical
    key, with the source's coverage merged onto the matching citation). Graph
    signals are derived here:

    * ``weighted_in_degree`` — for each work, the summed
      :data:`SALIENCE_RELATIONS` weight of incoming edges, counting each source
      graph at most once (its strongest edge), i.e. *distinct-source* in-degree.
    * ``centrality`` — PageRank over the source→target edge graph.
    * ``out_degree`` — a source's count of outgoing salience edges.

    Claim signals and inter-citation co-citation are left at their defaults
    (later M0 waves). Pure: no LLM, no network.

    **Dedup & merge.** Records that resolve to the same work by
    :func:`canonical_citation_key` are collapsed onto a single :class:`Work`,
    whether the duplicates are citation-vs-citation (two files sharing a DOI) or
    source-vs-citation (a promoted source that is also a citation). On merge the
    graph signals are folded together — distinct-source ``weighted_in_degree``
    (max per source across both id forms, then summed), ``centrality`` (summed,
    as each id-node accrued its own PageRank mass), ``cited_by_count`` (max) —
    and ``counts_by_year``/coverage are filled from whichever record carries
    them, so no endorsement edge or PageRank mass is orphaned by the dedup. A
    *degenerate* canonical key (the constant ``"meta:"`` for metadata-poor works,
    or a placeholder DOI like ``doi:n/a``) does not discriminate works, so such
    records are never deduped — each keeps its storage id as identity (see
    :func:`_is_dedupable_key`), keeping distinct works distinct.

    **Project scope.** Sources and edges are scoped to ``project`` via
    :func:`frame_search_sources`. Citations are scoped too: when ``project`` is
    given, only citations *referenced by the in-project sources* enter the
    corpus, so cohort percentiles are computed over the project's own pool. A
    citation is "referenced" if any in-project note links to it — by any graph
    (``_citations``, the default graph, ``_cross``), derived from the very same
    link walk that collects salience, so a citation whose in-degree is counted is
    never then dropped by the scope filter. When ``project`` is empty the full
    global citation pool is used (there is no per-citation project tag — the
    association is derived from note links).

    **Explicit citation scope.** ``citation_scope`` (when not ``None``) restricts
    the citation pool to exactly that id set, overriding the project-derived
    scoping. The single-source graph scope (:func:`papers._graph_scoped_corpus`)
    uses it to avoid materializing the entire global citation pool: it passes a
    dedup-group-closed set of the citations that one graph engages with, so the
    surviving Works — and every canonical-key merge — are identical to the
    full-pool corpus, only without building the unreferenced citations.
    """
    ns = namespace or (lambda s: s)
    base = graphs_dir or GRAPHS_DIR
    search_sources = frame_search_sources(project, graphs_dir=base, namespace=ns)

    citations = load_citations(graphs_dir=base)
    citation_ids = set(citations.keys())

    # Walk every source's notes once: collect salience edges (for centrality), the
    # set of distinct sources that point at each target via a salience edge, and
    # every citation referenced from an in-project source (for project scoping).
    edges: list[tuple[str, str]] = []
    in_sources: dict[str, dict[str, float]] = {}  # target id -> {source: max edge weight}
    out_degree: dict[str, int] = {}
    referenced_citation_ids: set[str] = set()
    for sname in search_sources:
        try:
            zg = get_graph(sname)
        except Exception:
            continue
        for note in zg.notes.values():
            for link in note.links:
                # A citation is "in project" if any in-project note links to it,
                # regardless of the link's graph — derived from this same pass so
                # scope stays consistent with the in-degree it contributes below.
                if link.target in citation_ids:
                    referenced_citation_ids.add(link.target)
                weight = SALIENCE_RELATIONS.get(link.relation)
                if weight is None:
                    continue
                target = link.target
                edges.append((sname, target))
                out_degree[sname] = out_degree.get(sname, 0) + 1
                per_source = in_sources.setdefault(target, {})
                # Distinct-source: a source contributes only its strongest edge.
                per_source[sname] = max(per_source.get(sname, 0.0), weight)
    centrality = _pagerank(edges)

    works: list[Work] = []
    fp_to_work: dict[str, Work] = {}
    fp_to_sources: dict[str, dict[str, float]] = {}  # fp -> merged {source: weight}

    if citation_scope is not None:
        # Explicit caller-provided citation scope (the graph-scope corpus path).
        # ``papers._graph_scoped_corpus`` computes the dedup-group-closed set of
        # citations the single graph actually engages with and passes it here, so
        # we materialize ONLY those Works instead of the entire global citation
        # pool. Because the set is closed over canonical-key merge groups, every
        # citation-vs-citation and source-vs-citation merge resolves to the SAME
        # kept Work it would over the full pool — the works the caller then keeps
        # are byte-identical to filtering the full-pool corpus post-hoc, we just
        # never build the unreferenced Works. Iteration order is preserved by the
        # comprehension, so merge "first wins" identity is unchanged.
        citations = {
            cid: data for cid, data in citations.items() if cid in citation_scope
        }
    elif project:
        # Scope citations to those the project's sources actually reference; a
        # global pool would skew every cohort percentile (see docstring).
        citations = {
            cid: data for cid, data in citations.items() if cid in referenced_citation_ids
        }
    for cid, data in citations.items():
        fp = canonical_citation_key(data)
        # A degenerate/placeholder key collides unrelated works; fall back to the
        # unique storage id so distinct works are never fused into one.
        key = fp if _is_dedupable_key(fp) else f"_nodedup:cit:{cid}"
        per_source = in_sources.get(cid, {})
        existing = fp_to_work.get(key)
        if existing is not None:
            # Citation-vs-citation duplicate (same DOI): fold onto the kept work
            # rather than inflating the cohort with a phantom second member.
            # Backfill authors from the duplicate only when the kept work lacks
            # them, mirroring the source-merge guard — never clobber non-empty.
            if not existing.authors:
                existing.authors = list(data.get("authors") or [])
            _merge_signals(
                existing,
                fp_to_sources[key],
                per_source,
                centrality.get(cid, 0.0),
                int(data.get("cited_by_count") or 0),
                data.get("counts_by_year"),
                add_cited_by_count_known=data.get("cited_by_count") is not None,
                add_doi=str(data.get("doi") or ""),
            )
            continue
        work = Work(
            id=cid,
            title=str(data.get("title") or cid),
            year=_coerce_year(data.get("year")),
            theme=str(data.get("theme") or ""),
            doi=str(data.get("doi") or ""),
            cited_by_count=int(data.get("cited_by_count") or 0),
            cited_by_count_known=data.get("cited_by_count") is not None,
            counts_by_year=list(data.get("counts_by_year") or []),
            coverage_human="none",
            authors=list(data.get("authors") or []),
            weighted_in_degree=sum(per_source.values()),
            centrality=centrality.get(cid, 0.0),
            is_review=str(data.get("doc_type") or "").lower() == "review",
        )
        works.append(work)
        fp_to_work[key] = work
        fp_to_sources[key] = dict(per_source)

    # Note-graph sources become works too. A source that is the same work as an
    # existing citation (same canonical key) is merged onto it — carrying human
    # coverage, out-degree, AND its own graph signals (endorsement edges that
    # target the source-name id) — rather than added as a duplicate.
    for sname in note_graph_names(graphs_dir=base):
        if search_sources and ns(sname) not in search_sources:
            continue
        # Infra folders (notably ``_cross``, the synthesis/claims store) are
        # walked above for their salience edges but are NOT real papers — never
        # materialize them as a Work, or they surface as a phantom theme and a
        # Papers-view row. ``note_graph_names`` returns only real sources plus
        # ``_cross``, so the underscore guard is exactly the infra exclusion.
        if sname.startswith("_"):
            continue
        meta = load_source_meta(sname, graphs_dir=base)
        coverage = meta.get("coverage") if isinstance(meta.get("coverage"), dict) else {}
        human = str(coverage.get("human", "none") or "none")
        fp = canonical_citation_key({**meta, "id": sname})
        # Same degenerate-key guard as the citation loop: a metadata-poor source
        # must not fuse onto an unrelated work that merely shares ``"meta:"``.
        key = fp if _is_dedupable_key(fp) else f"_nodedup:src:{sname}"
        per_source = in_sources.get(sname, {})
        existing = fp_to_work.get(key)
        if existing is not None:
            existing.out_degree = max(existing.out_degree, out_degree.get(sname, 0))
            # The source is the writable LOCAL backing for this work; record its
            # folder so coverage writes / click-to-open target it (not work.id),
            # AND take its human read-state from the SAME folder so the displayed
            # coverage matches the writable backing graph. Only claim them when
            # unset: a citation has source_graph None so it gets set, but when two
            # LOCAL source folders share a canonical key the FIRST one (whose id
            # the kept work already holds) retains both the writable identity and
            # the read-state rather than an alphabetically-later folder clobbering
            # them and de-syncing the two.
            if existing.source_graph is None:
                existing.source_graph = sname
                existing.coverage_human = human
            # Backfill authors from the source _meta only when the citation has
            # none — never clobber a citation's own authors.
            if not existing.authors:
                existing.authors = list(meta.get("authors") or [])
            _merge_signals(
                existing,
                fp_to_sources[key],
                per_source,
                centrality.get(sname, 0.0),
                int(meta.get("cited_by_count") or 0),
                meta.get("counts_by_year"),
                add_cited_by_count_known=meta.get("cited_by_count") is not None,
                add_doi=str(meta.get("doi") or ""),
            )
            continue
        work = Work(
            id=sname,
            title=str(meta.get("title") or meta.get("name") or sname),
            year=_coerce_year(meta.get("year")),
            theme=str(meta.get("theme") or ""),
            doi=str(meta.get("doi") or ""),
            cited_by_count=int(meta.get("cited_by_count") or 0),
            cited_by_count_known=meta.get("cited_by_count") is not None,
            counts_by_year=list(meta.get("counts_by_year") or []),
            coverage_human=human,
            authors=list(meta.get("authors") or []),
            source_graph=sname,
            weighted_in_degree=sum(per_source.values()),
            centrality=centrality.get(sname, 0.0),
            out_degree=out_degree.get(sname, 0),
            is_review=str(meta.get("doc_type") or "").lower() == "review",
        )
        works.append(work)
        fp_to_work[key] = work
        fp_to_sources[key] = dict(per_source)

    return works, search_sources

score_syllabus

score_syllabus(get_graph: GetGraph, *, project: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, importance_weights: ImportanceWeights | None = None, salience_weights: SalienceWeights | None = None, cohort_window: int = DEFAULT_COHORT_WINDOW) -> dict[str, Any]

End-to-end: build the corpus from the graph and score every work.

The top-level seam the dashboard/MCP call. Returns a result dict (callers serialize as needed) with per-work rows and a small summary.

Source code in zettelkasten/syllabus.py

def score_syllabus(
    get_graph: GetGraph,
    *,
    project: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    importance_weights: ImportanceWeights | None = None,
    salience_weights: SalienceWeights | None = None,
    cohort_window: int = DEFAULT_COHORT_WINDOW,
) -> dict[str, Any]:
    """End-to-end: build the corpus from the graph and score every work.

    The top-level seam the dashboard/MCP call. Returns a result dict (callers
    serialize as needed) with per-work rows and a small summary.
    """
    works, search_sources = build_corpus(
        get_graph, project=project, graphs_dir=graphs_dir, namespace=namespace
    )
    rows = score_corpus(
        works,
        importance_weights=importance_weights,
        salience_weights=salience_weights,
        cohort_window=cohort_window,
    )
    return {
        "project": project or "all",
        "source_count": len(search_sources),
        "work_count": len(rows),
        "works": rows,
    }

trend_metrics

trend_metrics(work: Work, reference_year: int, *, window: int = TREND_WINDOW) -> TrendMetrics

Per-paper trend metrics for work as of reference_year.

The clean, importable accessor for velocity / acceleration / citation-recency / trajectory that wave 3 consumes. counts_by_year is truncated to <= reference_year first, so this is the literal point-in-time view.

Source code in zettelkasten/syllabus.py

def trend_metrics(work: Work, reference_year: int, *, window: int = TREND_WINDOW) -> TrendMetrics:
    """Per-paper trend metrics for ``work`` as of ``reference_year``.

    The clean, importable accessor for velocity / acceleration / citation-recency
    / trajectory that wave 3 consumes. ``counts_by_year`` is truncated to
    ``<= reference_year`` first, so this is the literal point-in-time view.
    """
    items = _year_count_items(work, reference_year)
    pub_year = _coerce_year(work.year)
    (
        velocity,
        acceleration,
        citation_recency,
        publication_recency,
        recent_citations,
        total_citations,
        mean_recent,
    ) = _trend_core(items, pub_year, reference_year, window)
    trajectory = _classify_trajectory(
        velocity=velocity,
        citation_recency=citation_recency,
        recent_citations=recent_citations,
        mean_recent=mean_recent,
        pub_year=pub_year,
        reference_year=reference_year,
        window=window,
    )
    return TrendMetrics(
        work_id=work.id,
        reference_year=reference_year,
        velocity=velocity,
        acceleration=acceleration,
        citation_recency=citation_recency,
        publication_recency=publication_recency,
        trajectory=trajectory,
        recent_citations=recent_citations,
        total_citations=total_citations,
    )

corpus_reference_year

corpus_reference_year(works: list[Work], as_of: int | None = None, *, now_year: int | None = None) -> int

Resolve the reference year T for a corpus.

T is resolved one of two ways, kept strictly separate so the point-in-time replay stays pure while the live default is the only path that ever reads the wall clock:

• Point-in-time replay (as_of given): T is int(as_of) exactly — a pure, deterministic function of the argument, with no wall-clock access. This is what makes an as_of=T result reproducible regardless of when it is computed.
• Live default (as_of is None): T is today, a wall-clock value. A caller (notably a test) injects a deterministic "now" via now_year; absent that, T falls back to datetime.now().year.

T is deliberately NOT inferred from the corpus's observed years. Deriving it from data — by bare max, range-clamp, or even multi-work corroboration — let a single corrupt future-dated record (a 2040 typo, a 9999 sentinel) drag T into the future, all-zeroing every real paper's recent window and flipping a thriving field to "all dormant"; and corroboration conversely mislabeled a legitimately-fresh sole-holder-of-the-newest-year paper as dormant on clean data. Pinning T to the wall clock removes the whole class of failures: an outlier no longer contaminates its neighbors because T is not a function of it — the outlier merely falls beyond T and is handled per-paper by the existing future-publication guard (publication_recency 0, never "emerging") and the counts_by_year truncation to <= T.

works is accepted for call-site symmetry with the rest of the trends layer (and forward compatibility) but is intentionally unused: nothing about T is read from the corpus anymore.

Source code in zettelkasten/syllabus.py

def corpus_reference_year(
    works: list[Work],
    as_of: int | None = None,
    *,
    now_year: int | None = None,
) -> int:
    """Resolve the reference year T for a corpus.

    T is resolved one of two ways, kept strictly separate so the point-in-time
    replay stays pure while the live default is the *only* path that ever reads
    the wall clock:

    * **Point-in-time replay** (``as_of`` given): T is ``int(as_of)`` exactly —
      a pure, deterministic function of the argument, with no wall-clock access.
      This is what makes an ``as_of=T`` result reproducible regardless of when it
      is computed.
    * **Live default** (``as_of is None``): T is *today*, a wall-clock value. A
      caller (notably a test) injects a deterministic "now" via ``now_year``;
      absent that, T falls back to ``datetime.now().year``.

    T is deliberately NOT inferred from the corpus's observed years. Deriving it
    from data — by bare max, range-clamp, or even multi-work corroboration — let
    a single corrupt future-dated record (a ``2040`` typo, a ``9999`` sentinel)
    drag T into the future, all-zeroing every real paper's recent window and
    flipping a thriving field to "all dormant"; and corroboration conversely
    mislabeled a legitimately-fresh sole-holder-of-the-newest-year paper as
    dormant on clean data. Pinning T to the wall clock removes the whole class of
    failures: an outlier no longer contaminates its neighbors because T is not a
    function of it — the outlier merely falls beyond T and is handled per-paper
    by the existing future-publication guard (``publication_recency`` 0, never
    "emerging") and the ``counts_by_year`` truncation to ``<= T``.

    ``works`` is accepted for call-site symmetry with the rest of the trends
    layer (and forward compatibility) but is intentionally unused: nothing about
    T is read from the corpus anymore.
    """
    if as_of is not None:
        return int(as_of)
    if now_year is not None:
        return int(now_year)
    return datetime.now().year

corpus_trends

corpus_trends(works: list[Work], *, as_of: int | None = None, window: int = TREND_WINDOW, now_year: int | None = None) -> dict[str, TrendMetrics]

Per-work :class:TrendMetrics across the corpus, keyed by work id.

Honors the point-in-time contract: works published after as_of are dropped and each surviving work's history is truncated to <= as_of. With as_of omitted the live-default reference year is used (now_year if given, else the wall-clock year — see :func:corpus_reference_year).

Source code in zettelkasten/syllabus.py

def corpus_trends(
    works: list[Work],
    *,
    as_of: int | None = None,
    window: int = TREND_WINDOW,
    now_year: int | None = None,
) -> dict[str, TrendMetrics]:
    """Per-work :class:`TrendMetrics` across the corpus, keyed by work id.

    Honors the point-in-time contract: works published after ``as_of`` are
    dropped and each surviving work's history is truncated to ``<= as_of``. With
    ``as_of`` omitted the live-default reference year is used (``now_year`` if
    given, else the wall-clock year — see :func:`corpus_reference_year`).
    """
    reference_year = corpus_reference_year(works, as_of, now_year=now_year)
    return {
        w.id: trend_metrics(w, reference_year, window=window)
        for w in works
        if _within_as_of(w, as_of)
    }

field_direction

field_direction(works: list[Work], grouping: dict[str, str] | None = None, *, as_of: int | None = None, window: int = TREND_WINDOW, frontier_limit: int = 10, emerging_limit: int = 5, now_year: int | None = None) -> dict[str, Any]

Aggregate per-paper trajectories into a "field direction" roll-up.

Groups works (default: by :attr:Work.theme, the membership the corpus already carries — pass an explicit grouping of work_id → label to roll up by claims/concepts/clusters instead, so this composes with later layers) and summarizes, per group, whether it is accelerating, cooling, or stable. The top-level summary surfaces the accelerating and cooling themes, the emerging groups (fastest-growing recent citation share), and the current frontier (recent, high-velocity works).

Point-in-time aware: with as_of=T set, works published after T are excluded and each history truncated to <= T (see module notes on the honest limitations of this replay). With as_of omitted the live-default reference year is used (now_year if given, else the wall-clock year).

Source code in zettelkasten/syllabus.py

def field_direction(
    works: list[Work],
    grouping: dict[str, str] | None = None,
    *,
    as_of: int | None = None,
    window: int = TREND_WINDOW,
    frontier_limit: int = 10,
    emerging_limit: int = 5,
    now_year: int | None = None,
) -> dict[str, Any]:
    """Aggregate per-paper trajectories into a "field direction" roll-up.

    Groups works (default: by :attr:`Work.theme`, the membership the corpus
    already carries — pass an explicit ``grouping`` of ``work_id → label`` to roll
    up by claims/concepts/clusters instead, so this composes with later layers)
    and summarizes, per group, whether it is **accelerating**, **cooling**, or
    **stable**. The top-level summary surfaces the accelerating and cooling
    themes, the emerging groups (fastest-growing recent citation share), and the
    current **frontier** (recent, high-velocity works).

    Point-in-time aware: with ``as_of=T`` set, works published after T are
    excluded and each history truncated to ``<= T`` (see module notes on the
    honest limitations of this replay). With ``as_of`` omitted the live-default
    reference year is used (``now_year`` if given, else the wall-clock year).
    """
    present = [w for w in works if _within_as_of(w, as_of)]
    reference_year = corpus_reference_year(works, as_of, now_year=now_year)
    metrics = {w.id: trend_metrics(w, reference_year, window=window) for w in present}
    work_by_id = {w.id: w for w in present}

    # Default grouping is the theme each Work already carries; undated/untyped
    # works fall into an explicit bucket rather than being silently dropped.
    if grouping is None:
        grouping = {w.id: (w.theme or "(untyped)") for w in present}

    grouped: dict[str, list[TrendMetrics]] = {}
    for wid, tm in metrics.items():
        label = grouping.get(wid)
        if label is None:
            continue
        grouped.setdefault(label, []).append(tm)

    groups: list[dict[str, Any]] = []
    for label, members in grouped.items():
        n = len(members)
        traj_counts = {cls: 0 for cls in TRAJECTORY_CLASSES}
        for tm in members:
            traj_counts[tm.trajectory] += 1
        mean_velocity = sum(tm.velocity for tm in members) / n
        mean_acceleration = sum(tm.acceleration for tm in members) / n
        mean_recency = sum(tm.citation_recency for tm in members) / n
        recent_citations = sum(tm.recent_citations for tm in members)
        rising_like = traj_counts["rising"] + traj_counts["emerging"]
        cooling_like = traj_counts["cooling"] + traj_counts["dormant"]
        groups.append({
            "group": label,
            "work_count": n,
            "direction": _group_direction(mean_velocity, rising_like, cooling_like),
            "trajectory_counts": traj_counts,
            "mean_velocity": round(mean_velocity, 4),
            "mean_acceleration": round(mean_acceleration, 4),
            "mean_citation_recency": round(mean_recency, 4),
            "recent_citations": recent_citations,
        })

    groups.sort(key=lambda g: g["mean_velocity"], reverse=True)

    accelerating = [g["group"] for g in groups if g["direction"] == "accelerating"]
    cooling = [g["group"] for g in sorted(groups, key=lambda g: g["mean_velocity"])
               if g["direction"] == "cooling"]
    # Emerging concepts = groups whose citations are most concentrated in the
    # recent window (fastest-growing recent share); all-dormant groups (0 share)
    # are not "emerging".
    emerging_concepts = [
        g["group"]
        for g in sorted(groups, key=lambda g: g["mean_citation_recency"], reverse=True)
        if g["mean_citation_recency"] > 0.0
    ][:emerging_limit]

    # Frontier = the recent, still-growing work itself: positive velocity and a
    # citation history dominated by the recent window.
    frontier_candidates = [
        tm for tm in metrics.values()
        if tm.recent_citations > 0
        and tm.velocity > 0.0
        and tm.citation_recency >= _FRONTIER_MIN_RECENCY
    ]
    frontier_candidates.sort(key=lambda tm: (tm.velocity, tm.citation_recency), reverse=True)
    frontier = [
        {
            "id": tm.work_id,
            "title": work_by_id[tm.work_id].title,
            "year": work_by_id[tm.work_id].year,
            "trajectory": tm.trajectory,
            "velocity": round(tm.velocity, 4),
            "citation_recency": round(tm.citation_recency, 4),
        }
        for tm in frontier_candidates[:frontier_limit]
    ]

    return {
        "reference_year": reference_year,
        "window": window,
        "as_of": as_of,
        "truncated": as_of is not None,
        "group_count": len(groups),
        "groups": groups,
        "accelerating": accelerating,
        "cooling": cooling,
        "emerging_concepts": emerging_concepts,
        "frontier": frontier,
    }

trend_summary

trend_summary(get_graph: GetGraph, *, project: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, as_of: int | None = None, window: int = TREND_WINDOW, now_year: int | None = None) -> dict[str, Any]

End-to-end trends seam: build the corpus from the graph and roll it up.

Mirrors :func:score_syllabus (and reuses :func:build_corpus, never duplicating corpus assembly) so the dashboard/MCP can request the temporal view the same way they request scores. Pass as_of to replay the field as of an earlier year T; pass now_year to pin the live-default reference year deterministically (otherwise the wall-clock year is used).

Source code in zettelkasten/syllabus.py

def trend_summary(
    get_graph: GetGraph,
    *,
    project: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    as_of: int | None = None,
    window: int = TREND_WINDOW,
    now_year: int | None = None,
) -> dict[str, Any]:
    """End-to-end trends seam: build the corpus from the graph and roll it up.

    Mirrors :func:`score_syllabus` (and reuses :func:`build_corpus`, never
    duplicating corpus assembly) so the dashboard/MCP can request the temporal
    view the same way they request scores. Pass ``as_of`` to replay the field as
    of an earlier year T; pass ``now_year`` to pin the live-default reference
    year deterministically (otherwise the wall-clock year is used).
    """
    works, search_sources = build_corpus(
        get_graph, project=project, graphs_dir=graphs_dir, namespace=namespace
    )
    direction = field_direction(works, as_of=as_of, window=window, now_year=now_year)
    return {
        "project": project or "all",
        "source_count": len(search_sources),
        "work_count": sum(g["work_count"] for g in direction["groups"]),
        **direction,
    }

interestingness

interestingness(works: list[Work], *, embeddings: dict[str, Sequence[float]] | None = None, clusters: dict[str, str] | None = None, relation_edges: Iterable[tuple[str, str, str]] | None = None, trends: dict[str, TrendMetrics] | None = None, influence_scores: dict[str, float] | None = None, salience_scores: dict[str, float] | None = None, weights: InterestingnessWeights | None = None, read_state_aware: bool = False, cohort_window: int = DEFAULT_COHORT_WINDOW) -> dict[str, Interestingness]

Composite interestingness per work, decorrelated from salience.

The five sub-signals each measure something salience cannot:

• surprise — embedding distance to the work's cluster centroid.
• brokerage — a gated indicator that the work bridges ≥2 otherwise-separate clusters. Gated to {0,1} rather than a graded cluster count so it cannot rank-track citation volume (its neighbor set includes cites edges, which also drive salience), keeping it decorrelated from salience.
• dissent — how contested a claim is: the count/strength of incoming contradicts/qualifies edges challenging it (and, in read_state_aware mode, boosted against claims already fully read). Independent of the challenged claim's citation weight, so it is not a salience proxy.
• citation_anomaly — a gated "sleeper" indicator (a structurally-core and cohort-under-cited work in a real cohort, NOT a monotone function of raw salience) OR a positive citation acceleration surge from the wave-2 :class:TrendMetrics, whichever is larger.
• novelty — embedding distance to the GLOBAL corpus centroid.

Each sub-signal is max-normalized to 0–1, then combined as MAX_CONTRIB·max(signal) + (1-MAX_CONTRIB)·weighted_blend and gated by a relevance floor (:func:_is_grounded) so disconnected/off-topic notes cannot score high. The dominant firing sub-signal is recorded on each result.

Pure: embeddings/clusters/edges/trends are all supplied by the caller (the dashboard owns the embedding + clustering machinery); absent inputs make the corresponding signals 0 rather than raising. Wave-1 influence/salience and wave-2 TrendMetrics are reused, never re-derived.

Source code in zettelkasten/syllabus.py

def interestingness(
    works: list[Work],
    *,
    embeddings: dict[str, Sequence[float]] | None = None,
    clusters: dict[str, str] | None = None,
    relation_edges: Iterable[tuple[str, str, str]] | None = None,
    trends: dict[str, TrendMetrics] | None = None,
    influence_scores: dict[str, float] | None = None,
    salience_scores: dict[str, float] | None = None,
    weights: InterestingnessWeights | None = None,
    read_state_aware: bool = False,
    cohort_window: int = DEFAULT_COHORT_WINDOW,
) -> dict[str, Interestingness]:
    """Composite interestingness per work, decorrelated from salience.

    The five sub-signals each measure something salience cannot:

    * **surprise** — embedding distance to the work's cluster centroid.
    * **brokerage** — a gated indicator that the work bridges ≥2 otherwise-separate
      clusters. Gated to {0,1} rather than a graded cluster count so it cannot
      rank-track citation volume (its neighbor set includes ``cites`` edges, which
      also drive salience), keeping it decorrelated from salience.
    * **dissent** — how contested a claim is: the count/strength of incoming
      contradicts/qualifies edges challenging it (and, in ``read_state_aware``
      mode, boosted against claims already fully read). Independent of the
      challenged claim's citation weight, so it is not a salience proxy.
    * **citation_anomaly** — a gated "sleeper" indicator (a structurally-core
      *and* cohort-under-cited work in a real cohort, NOT a monotone function of
      raw salience) OR a positive citation acceleration surge from the wave-2
      :class:`TrendMetrics`, whichever is larger.
    * **novelty** — embedding distance to the GLOBAL corpus centroid.

    Each sub-signal is max-normalized to 0–1, then combined as
    ``MAX_CONTRIB·max(signal) + (1-MAX_CONTRIB)·weighted_blend`` and gated by a
    relevance floor (:func:`_is_grounded`) so disconnected/off-topic notes cannot
    score high. The dominant firing sub-signal is recorded on each result.

    Pure: embeddings/clusters/edges/trends are all supplied by the caller (the
    dashboard owns the embedding + clustering machinery); absent inputs make the
    corresponding signals 0 rather than raising. Wave-1 ``influence``/``salience``
    and wave-2 ``TrendMetrics`` are reused, never re-derived.
    """
    weights = weights or InterestingnessWeights()
    ids = [w.id for w in works]
    cl = _resolve_clusters(works, clusters)
    inf = influence_scores if influence_scores is not None else influence(works, cohort_window=cohort_window)
    sal = salience_scores if salience_scores is not None else salience(works)
    edges = _iter_relation_edges(relation_edges)
    # Only salience-bearing / dissent edges ground a work for the relevance floor;
    # a stray ``related`` association does not (see :data:`_GROUNDING_RELATIONS`).
    participants = {
        e for src, rel, tgt in edges if rel in _GROUNDING_RELATIONS for e in (src, tgt)
    }

    surprise = _max_normalize(_surprise_raw(works, embeddings, cl))
    novelty = _max_normalize(_novelty_raw(works, embeddings))
    brokerage = _max_normalize(_brokerage_raw(works, edges, cl))
    dissent = _max_normalize(_dissent_raw(works, edges, read_state_aware=read_state_aware))
    surge = _max_normalize({
        w.id: max(0.0, trends[w.id].acceleration) if (trends and w.id in trends) else 0.0
        for w in works
    })
    # citation anomaly = a genuine "sleeper" gate OR a recent surge. The sleeper
    # term is a gated INDICATOR, not ``salience·(1-influence)``: that product is
    # monotone in salience for the whole low-influence population, so it merely
    # re-expresses salience. Instead a work is a sleeper only when it is BOTH
    # structurally core (high salience) AND demonstrably under-cited (low cohort
    # influence) within a real cohort (size ≥ 2, so a neutral singleton can't
    # qualify) — matching :func:`situate`'s ``sleeper`` flag. The result is
    # decorrelated from raw salience at every influence level.
    cohort_size = _cohort_sizes(works, cohort_window)
    anomaly = {
        wid: max(
            1.0 if (
                wid in inf  # a sleeper must be *demonstrably* under-cited; an
                            # unknown-influence work can't qualify (see influence)
                and sal.get(wid, 0.0) >= _SLEEPER_MIN_SALIENCE
                and inf[wid] <= _SLEEPER_MAX_INFLUENCE
                and cohort_size.get(wid, 0) >= 2
            ) else 0.0,
            surge.get(wid, 0.0),
        )
        for wid in ids
    }

    wmap = {
        "surprise": weights.surprise,
        "brokerage": weights.brokerage,
        "dissent": weights.dissent,
        "citation_anomaly": weights.citation_anomaly,
        "novelty": weights.novelty,
    }
    total_w = sum(wmap.values())

    result: dict[str, Interestingness] = {}
    for w in works:
        sigs = {
            "surprise": surprise.get(w.id, 0.0),
            "brokerage": brokerage.get(w.id, 0.0),
            "dissent": dissent.get(w.id, 0.0),
            "citation_anomaly": anomaly.get(w.id, 0.0),
            "novelty": novelty.get(w.id, 0.0),
        }
        max_signal = max(sigs.values(), default=0.0)
        blend = (
            sum(sigs[k] * wmap[k] for k in sigs) / total_w if total_w > 0 else 0.0
        )
        composite = _INTEREST_MAX_CONTRIB * max_signal + (1.0 - _INTEREST_MAX_CONTRIB) * blend
        score = composite if _is_grounded(w, participants) else 0.0
        dominant = _dominant_signal(sigs) if score > _EPS else None
        result[w.id] = Interestingness(
            work_id=w.id,
            score=score,
            dominant_signal=dominant,
            surprise=sigs["surprise"],
            brokerage=sigs["brokerage"],
            dissent=sigs["dissent"],
            citation_anomaly=sigs["citation_anomaly"],
            novelty=sigs["novelty"],
        )
    return result

situate

situate(works: list[Work], *, relation_edges: Iterable[tuple[str, str, str]] | None = None, clusters: dict[str, str] | None = None, trends: dict[str, TrendMetrics] | None = None, influence_scores: dict[str, float] | None = None, salience_scores: dict[str, float] | None = None, as_of: int | None = None, window: int = TREND_WINDOW, now_year: int | None = None, cohort_window: int = DEFAULT_COHORT_WINDOW) -> dict[str, Situation]

Per-paper context labels (temporal / literature position / influence).

A shared deterministic helper the rationale layer and the Papers view both read. Reuses wave-1 :func:influence/:func:salience and wave-2 :func:corpus_trends (for trajectory/velocity) rather than re-deriving them; pass precomputed scores/trends to avoid recomputation. relation_edges ((source, relation, target) triples) supply citation topology, and clusters the bridge detection — both optional and defaulted the same way as :func:interestingness.

Label derivation (each axis independent; within an axis the first matching rule in priority order wins, so labels are mutually exclusive):

• temporal — origin-of-claim (claim-core, the claim-layer hook) > recent (published within the trend window) > seminal (older and high cohort influence).
• position — review > superseded (target of a supersedes edge) > frontier-rising (rising/emerging trajectory) > foundational (high influence with others building on it) > replication > extension > bridge.
• influence — the wave-1 cohort percentile, tiered into high/medium/low, with sleeper set when a work is core (high salience) yet under-cited (low influence).

Source code in zettelkasten/syllabus.py

def situate(
    works: list[Work],
    *,
    relation_edges: Iterable[tuple[str, str, str]] | None = None,
    clusters: dict[str, str] | None = None,
    trends: dict[str, TrendMetrics] | None = None,
    influence_scores: dict[str, float] | None = None,
    salience_scores: dict[str, float] | None = None,
    as_of: int | None = None,
    window: int = TREND_WINDOW,
    now_year: int | None = None,
    cohort_window: int = DEFAULT_COHORT_WINDOW,
) -> dict[str, Situation]:
    """Per-paper context labels (temporal / literature position / influence).

    A shared deterministic helper the rationale layer and the Papers view both
    read. Reuses wave-1 :func:`influence`/:func:`salience` and wave-2
    :func:`corpus_trends` (for trajectory/velocity) rather than re-deriving them;
    pass precomputed scores/trends to avoid recomputation. ``relation_edges``
    (``(source, relation, target)`` triples) supply citation topology, and
    ``clusters`` the bridge detection — both optional and defaulted the same way
    as :func:`interestingness`.

    Label derivation (each axis independent; within an axis the first matching
    rule in priority order wins, so labels are mutually exclusive):

    * **temporal** — origin-of-claim (claim-core, the claim-layer hook) >
      recent (published within the trend window) > seminal (older and high
      cohort influence).
    * **position** — review > superseded (target of a ``supersedes`` edge) >
      frontier-rising (rising/emerging trajectory) > foundational (high influence
      with others building on it) > replication > extension > bridge.
    * **influence** — the wave-1 cohort percentile, tiered into high/medium/low,
      with ``sleeper`` set when a work is core (high salience) yet under-cited
      (low influence).
    """
    inf = influence_scores if influence_scores is not None else influence(works, cohort_window=cohort_window)
    sal = salience_scores if salience_scores is not None else salience(works)
    if trends is None:
        trends = corpus_trends(works, as_of=as_of, window=window, now_year=now_year)
    reference_year = corpus_reference_year(works, as_of, now_year=now_year)
    cl = _resolve_clusters(works, clusters)
    edges = _iter_relation_edges(relation_edges)

    incoming_superseded: set[str] = set()
    incoming_building: dict[str, int] = {}
    replicates_involved: set[str] = set()
    outgoing_extends: set[str] = set()
    for src, rel, tgt in edges:
        if rel == "supersedes":
            incoming_superseded.add(tgt)
        if rel == "replicates":
            replicates_involved.add(src)
            replicates_involved.add(tgt)
        if rel == "extends":
            outgoing_extends.add(src)
        if rel in _BUILDING_RELATIONS:
            incoming_building[tgt] = incoming_building.get(tgt, 0) + 1
    brokerage_raw = _brokerage_raw(works, edges, cl)

    result: dict[str, Situation] = {}
    for w in works:
        tm = trends.get(w.id)
        pub_year = _coerce_year(w.year)
        # ``influence_v`` is None when the work's citation count is unknown
        # (never enriched) — influence-gated labels below must not fire off a
        # value we don't actually have, so each gate guards for None explicitly.
        influence_v = inf.get(w.id)
        salience_v = sal.get(w.id, 0.0)

        # ── temporal ──
        recent = pub_year is not None and 0 <= reference_year - pub_year < window
        old = pub_year is not None and reference_year - pub_year >= window
        if (w.core_claim_count or 0) > 0:
            temporal: str | None = "origin-of-claim"
        elif recent:
            temporal = "recent"
        elif old and influence_v is not None and influence_v >= _SEMINAL_MIN_INFLUENCE:
            temporal = "seminal"
        else:
            temporal = None

        # ── literature position ──
        if w.is_review:
            position: str | None = "review"
        elif w.id in incoming_superseded:
            position = "superseded"
        elif tm is not None and tm.trajectory in ("rising", "emerging"):
            position = "frontier-rising"
        elif influence_v is not None and influence_v >= _FOUNDATIONAL_MIN_INFLUENCE and incoming_building.get(w.id, 0) > 0:
            position = "foundational"
        elif w.id in replicates_involved:
            position = "replication"
        elif w.id in outgoing_extends:
            position = "extension"
        elif brokerage_raw.get(w.id, 0.0) > 0:
            position = "bridge"
        else:
            position = None

        # ── influence tier + sleeper ──
        # Unknown influence → no tier ("unknown") and never a sleeper, since a
        # sleeper is defined as core-but-*demonstrably*-under-cited.
        influence_label: str | None
        if influence_v is None:
            influence_label = "unknown"
        elif influence_v >= _INFLUENCE_HIGH:
            influence_label = "high"
        elif influence_v <= _INFLUENCE_LOW:
            influence_label = "low"
        else:
            influence_label = "medium"
        sleeper = (
            influence_v is not None
            and salience_v >= _SLEEPER_MIN_SALIENCE
            and influence_v <= _SLEEPER_MAX_INFLUENCE
        )

        result[w.id] = Situation(
            work_id=w.id,
            temporal=temporal,
            position=position,
            influence=influence_v,
            influence_label=influence_label,
            sleeper=sleeper,
        )
    return result

why_read

why_read(work: Work, *, salience: float = 0.0, importance: float = 0.0, anchor: float = 0.0, situation: Situation | None = None, contested: bool = False, threshold: float = _READ_REASON_THRESHOLD) -> str | None

Render a one-sentence "why read this" rationale from selection signals.

Picks the 1–2 STRONGEST firing reasons among core-paper / multi-claim / anchor / salience / contested / foundational / importance. The continuous signals (anchor, salience, importance) fire only above threshold; the categorical ones (core/multi-claim from Work.core_claim_count, contested, foundational from the work's :class:Situation) fire on their own. Returns None when nothing fires, so a caller can omit the rationale entirely.

Source code in zettelkasten/syllabus.py

def why_read(
    work: Work,
    *,
    salience: float = 0.0,
    importance: float = 0.0,
    anchor: float = 0.0,
    situation: Situation | None = None,
    contested: bool = False,
    threshold: float = _READ_REASON_THRESHOLD,
) -> str | None:
    """Render a one-sentence "why read this" rationale from selection signals.

    Picks the 1–2 STRONGEST firing reasons among core-paper / multi-claim /
    anchor / salience / contested / foundational / importance. The continuous
    signals (anchor, salience, importance) fire only above ``threshold``; the
    categorical ones (core/multi-claim from ``Work.core_claim_count``, contested,
    foundational from the work's :class:`Situation`) fire on their own. Returns
    ``None`` when nothing fires, so a caller can omit the rationale entirely.
    """
    candidates: list[tuple[float, str]] = []
    claims = work.core_claim_count or 0
    if claims >= 2:
        candidates.append((1.0, "multi-claim"))
    elif claims >= 1:
        candidates.append((0.95, "core-paper"))
    if situation is not None and situation.position == "foundational":
        candidates.append((0.90, "foundational"))
    if contested:
        candidates.append((0.85, "contested"))
    if importance >= threshold:
        candidates.append((importance, "importance"))
    if anchor >= threshold:
        candidates.append((anchor, "anchor"))
    if salience >= threshold:
        candidates.append((salience, "salience"))
    if not candidates:
        return None
    candidates.sort(key=lambda c: (-c[0], _READ_REASON_PRIORITY.index(c[1])))
    phrases = [_READ_REASON_PHRASES[reason] for _strength, reason in candidates[:2]]
    return f"Worth reading: it {_join_phrases(phrases)}."

why_interesting

why_interesting(interest: Interestingness | None) -> str | None

Render a one-sentence rationale from the dominant interestingness signal.

Returns None (neutral — nothing remarkable fired) when the score is ~0 or no sub-signal dominated, otherwise a sentence keyed off the dominant signal.

Source code in zettelkasten/syllabus.py

def why_interesting(interest: Interestingness | None) -> str | None:
    """Render a one-sentence rationale from the dominant interestingness signal.

    Returns ``None`` (neutral — nothing remarkable fired) when the score is ~0 or
    no sub-signal dominated, otherwise a sentence keyed off the dominant signal.
    """
    if interest is None or interest.dominant_signal is None or interest.score <= _EPS:
        return None
    phrase = _INTEREST_PHRASES.get(interest.dominant_signal)
    if phrase is None:
        return None
    return f"Interesting because it {phrase}."

theme_model

theme_model(works: list[Work], *, clusters: dict[str, str] | None = None, relation_edges: Iterable[tuple[str, str, str]] | None = None, landscape_hubs: dict[str, Any] | None = None, anchor_scores: dict[str, float] | None = None, cold_start_limit: int = DEFAULT_COLD_START_LIMIT, merge_remainder: bool = False) -> ThemeModel

Group works into weighted themes by a strict source priority.

Theme source (first non-empty tier wins):

• (a) landscape hubs — when landscape_hubs (concept hubs tagged landscape in the graph) yield at least one in-corpus member. Each hub becomes a theme; membership is the hub's explicit work_ids and/or the works it surveys/cites in relation_edges.
• (b) embedding clusters — else, when clusters (work_id → label, the convention :func:interestingness/:func:situate use) is supplied. Works are grouped by :func:_resolve_clusters, so works absent from the mapping fall back to their :attr:Work.theme (or (untyped)).
• (c) cold start — else, the top works by anchor_score + most-cited become singleton seed themes (up to cold_start_limit).

When merge_remainder is set AND landscape hubs win tier (a), the works that no hub covers are NOT abandoned: they are themed by the cluster/cold-start tier (b/c) and those themes are appended alongside the landscape themes. The resulting source is "hybrid" when both tiers contribute. This keeps a freshly-minted landscape hub (e.g. a promoted review section) from collapsing the whole corpus into landscape-only — without it, the first hub would drop every cluster/cold-start theme and orphan their claims. Default False preserves the strict single-tier partition for non-review callers.

Each theme's weight is size × external citation mass = member count times the summed OpenAlex cited_by_count of its in-corpus members (the pull the theme exerts on the outside literature). Themes are returned sorted by descending weight then theme_id; ties everywhere break on id, so the result is fully deterministic. Members not present in works are dropped, and a theme with no in-corpus members is omitted (it could never be covered).

Pure: no graph access beyond the supplied relation_edges, no LLM, no network, no wall clock.

Source code in zettelkasten/syllabus.py

def theme_model(
    works: list[Work],
    *,
    clusters: dict[str, str] | None = None,
    relation_edges: Iterable[tuple[str, str, str]] | None = None,
    landscape_hubs: dict[str, Any] | None = None,
    anchor_scores: dict[str, float] | None = None,
    cold_start_limit: int = DEFAULT_COLD_START_LIMIT,
    merge_remainder: bool = False,
) -> ThemeModel:
    """Group ``works`` into weighted themes by a strict source priority.

    Theme source (first non-empty tier wins):

    * **(a) landscape hubs** — when ``landscape_hubs`` (concept hubs tagged
      ``landscape`` in the graph) yield at least one in-corpus member. Each hub
      becomes a theme; membership is the hub's explicit ``work_ids`` and/or the
      works it ``surveys``/``cites`` in ``relation_edges``.
    * **(b) embedding clusters** — else, when ``clusters`` (``work_id → label``,
      the convention :func:`interestingness`/:func:`situate` use) is supplied.
      Works are grouped by :func:`_resolve_clusters`, so works absent from the
      mapping fall back to their :attr:`Work.theme` (or ``(untyped)``).
    * **(c) cold start** — else, the top works by ``anchor_score`` + most-cited
      become singleton seed themes (up to ``cold_start_limit``).

    When ``merge_remainder`` is set AND landscape hubs win tier (a), the works
    that no hub covers are NOT abandoned: they are themed by the cluster/cold-start
    tier (b/c) and those themes are appended alongside the landscape themes. The
    resulting ``source`` is ``"hybrid"`` when both tiers contribute. This keeps a
    freshly-minted landscape hub (e.g. a promoted review section) from collapsing
    the whole corpus into landscape-only — without it, the first hub would drop
    every cluster/cold-start theme and orphan their claims. Default ``False``
    preserves the strict single-tier partition for non-review callers.

    Each theme's weight is ``size × external citation mass`` = member count times
    the summed OpenAlex ``cited_by_count`` of its in-corpus members (the pull the
    theme exerts on the outside literature). Themes are returned sorted by
    descending weight then ``theme_id``; ties everywhere break on id, so the
    result is fully deterministic. Members not present in ``works`` are dropped,
    and a theme with no in-corpus members is omitted (it could never be covered).

    Pure: no graph access beyond the supplied ``relation_edges``, no LLM, no
    network, no wall clock.
    """
    work_by_id = {w.id: w for w in works}
    edges = _iter_relation_edges(relation_edges)

    def _present_mass(member_ids: Iterable[str]) -> tuple[list[str], int, int]:
        present = sorted(m for m in set(member_ids) if m in work_by_id)
        return present, len(present), sum(work_by_id[m].cited_by_count for m in present)

    hubs = _normalize_landscape_hubs(landscape_hubs, edges)
    landscape_themes: list[Theme] = []
    for hub_id, (label, members) in hubs.items():
        present, size, mass = _present_mass(members)
        if present:
            landscape_themes.append(
                Theme(hub_id, label, tuple(present), float(size * mass), "landscape")
            )

    def _cluster_or_cold_start(subset: list[Work]) -> tuple[list[Theme], str]:
        """Tier (b) cluster — else tier (c) cold-start — themes over ``subset``.

        Shared by the strict single-tier path (``subset`` = the whole corpus) and
        the ``merge_remainder`` hybrid path (``subset`` = the hub-uncovered works),
        so both produce byte-identical cluster/cold-start themes.
        """
        if clusters:
            labels = _resolve_clusters(subset, clusters)
            by_label: dict[str, list[str]] = {}
            for w in subset:
                by_label.setdefault(labels[w.id], []).append(w.id)
            out: list[Theme] = []
            for label, member_ids in by_label.items():
                present, size, mass = _present_mass(member_ids)
                if present:
                    out.append(Theme(label, label, tuple(present), float(size * mass), "cluster"))
            return out, "cluster"
        anc = anchor_scores if anchor_scores is not None else anchor_score(subset)
        cited_norm = _max_normalize({w.id: float(max(0, w.cited_by_count)) for w in subset})
        seed_score = {w.id: anc.get(w.id, 0.0) + cited_norm.get(w.id, 0.0) for w in subset}
        # Order every work by descending seed strength then id and keep the top
        # ``cold_start_limit``. Zero-score works are NOT filtered out: the
        # cold-start tier exists to bootstrap a bare corpus, so a freshly-ingested
        # corpus (all anchor_score/cited_by_count == 0) must still seed at least
        # one theme. Only a truly empty corpus yields no themes.
        seeds = sorted(
            subset,
            key=lambda w: (-seed_score[w.id], w.id),
        )[: max(0, cold_start_limit)]
        out = []
        for w in seeds:
            present, size, mass = _present_mass((w.id,))
            out.append(Theme(w.id, w.title or w.id, tuple(present), float(size * mass), "cold-start"))
        return out, "cold-start"

    if landscape_themes:
        source = "landscape"
        themes = list(landscape_themes)
        if merge_remainder:
            # The hub tier won, but don't let it swallow the corpus: theme the
            # works no hub covers with the cluster/cold-start tier and append them.
            covered = {wid for t in landscape_themes for wid in t.work_ids}
            remaining = [w for w in works if w.id not in covered]
            if remaining:
                remainder_themes, _ = _cluster_or_cold_start(remaining)
                if remainder_themes:
                    themes.extend(remainder_themes)
                    source = "hybrid"
    else:
        themes, source = _cluster_or_cold_start(works)

    themes.sort(key=lambda t: (-t.weight, t.theme_id))
    themes_tuple = tuple(themes)

    # Primary theme per work = the first (highest-weight) theme it appears in.
    assignment: dict[str, str] = {}
    for t in themes_tuple:
        for wid in t.work_ids:
            assignment.setdefault(wid, t.theme_id)

    return ThemeModel(themes=themes_tuple, assignment=assignment, source=source)

set_cover

set_cover(works: list[Work], themes: 'ThemeModel | Sequence[Theme]', *, budget: int = DEFAULT_COVER_BUDGET, importance_scores: dict[str, float] | None = None, read_priority_scores: dict[str, float] | None = None, relation_edges: Iterable[tuple[str, str, str]] | None = None, embeddings: dict[str, Sequence[float]] | None = None, redundancy_threshold: float = DEFAULT_REDUNDANCY_THRESHOLD, cohort_window: int = DEFAULT_COHORT_WINDOW, claim_cover: dict[str, str | None] | None = None) -> SetCover

Greedy, redundancy-penalized set cover over themes AND/OR key claims.

Builds a minimal high-value suggested reading list. The coverage universe is COMPOSITE: theme units and key-claim units compete in ONE greedy loop within ONE budget.

• themes — each theme is a unit covered by any of its member works.
• key claims — when claim_cover (claim_uid → core work id) is supplied, each KEY CLAIM is a unit covered by its single CORE paper. The core work id may be an OWNED source OR an UNOWNED citation: an unowned core is an acquisition candidate — acquiring it would cover the claim — and it is admitted as a coverer just like an owned one. Only a None core work id (a claim with no candidate paper at all) has no selectable coverer and is a PERMANENT ACQUISITION GAP. Owned coverers are preferred over unowned ones on an exact tie (an unowned candidate never DISPLACES an equivalent owned coverer).

universe reports which spaces are live: "theme" when there are no claim units (then EVERY output is byte-identical to the original theme-only cover), "composite" when both theme and claim units exist, or "claim" when only claim units exist. Theme units are NEVER abandoned when claims are present — both are covered together.

The greedy machinery is identical for both universes — only the unit→coverer mapping and the result labelling differ:

1. Pre-cover. A unit with a member the user has already fully read (coverage_human == "full") needs nothing — its residual need drops to 0 and it is reported in pre_covered (never a gap, never selected for). For a key claim, the only member is its core paper, so a claim is pre-covered iff its core paper is fully read.
2. Greedy marginal gain. While under budget and units remain uncovered, pick the unread candidate that covers the most still-uncovered units, breaking ties by value then id. Value is read_priority_scores if supplied, else importance_scores, else a freshly computed :func:importance (what to read next / how important).
3. Redundancy (coverage-first penalty). A candidate that is a near-duplicate of an already-selected paper (see :func:_is_redundant) is deferred: a non-redundant coverer is always preferred when one exists. But redundancy is a penalty, NOT an absolute veto — if the only remaining coverers of an uncovered unit are near-duplicates, the best such coverer is still selected (coverage wins) rather than abandoning the unit. A contradicts/qualifies edge exempts a pair from redundancy entirely, so a claim and its rebuttal are both directly selectable.
4. Stop at the budget or when no candidate (redundant or not) yields positive marginal gain — i.e. every remaining uncovered unit has no selectable coverer left.

redundancy_threshold is an embedding cosine distance in [0, 2] (0 = identical, 2 = opposite; smaller = stricter deduplication). It is clamped into that range, so a similarity-like or out-of-range value cannot collapse the corpus.

Returns the flat selection (in order), the selection grouped by_theme (or by_claim in claim mode), the uncovered gaps, and the pre_covered units. Pure and deterministic (stable tie-breaks; no RNG, no wall clock).

Source code in zettelkasten/syllabus.py

def set_cover(
    works: list[Work],
    themes: "ThemeModel | Sequence[Theme]",
    *,
    budget: int = DEFAULT_COVER_BUDGET,
    importance_scores: dict[str, float] | None = None,
    read_priority_scores: dict[str, float] | None = None,
    relation_edges: Iterable[tuple[str, str, str]] | None = None,
    embeddings: dict[str, Sequence[float]] | None = None,
    redundancy_threshold: float = DEFAULT_REDUNDANCY_THRESHOLD,
    cohort_window: int = DEFAULT_COHORT_WINDOW,
    claim_cover: dict[str, str | None] | None = None,
) -> SetCover:
    """Greedy, redundancy-penalized set cover over themes AND/OR key claims.

    Builds a minimal high-value suggested reading list. The **coverage universe**
    is COMPOSITE: theme units and key-claim units compete in ONE greedy loop
    within ONE budget.

    * **themes** — each theme is a unit covered by *any* of its member works.
    * **key claims** — when ``claim_cover`` (``claim_uid → core work id``) is
      supplied, each KEY CLAIM is a unit covered by its single CORE paper. The
      core work id may be an OWNED source OR an UNOWNED citation: an unowned
      core is an *acquisition candidate* — acquiring it would cover the claim —
      and it is admitted as a coverer just like an owned one. Only a ``None``
      core work id (a claim with no candidate paper at all) has no selectable
      coverer and is a PERMANENT ACQUISITION GAP. Owned coverers are preferred
      over unowned ones on an exact tie (an unowned candidate never DISPLACES an
      equivalent owned coverer).

    ``universe`` reports which spaces are live: ``"theme"`` when there are no
    claim units (then EVERY output is byte-identical to the original theme-only
    cover), ``"composite"`` when both theme and claim units exist, or ``"claim"``
    when only claim units exist. Theme units are NEVER abandoned when claims are
    present — both are covered together.

    The greedy machinery is identical for both universes — only the unit→coverer
    mapping and the result labelling differ:

    1. **Pre-cover.** A unit with a member the user has already *fully* read
       (``coverage_human == "full"``) needs nothing — its residual need drops to
       0 and it is reported in ``pre_covered`` (never a gap, never selected for).
       For a key claim, the only member is its core paper, so a claim is
       pre-covered iff its core paper is fully read.
    2. **Greedy marginal gain.** While under ``budget`` and units remain
       uncovered, pick the unread candidate that covers the most still-uncovered
       units, breaking ties by value then id. Value is ``read_priority_scores``
       if supplied, else ``importance_scores``, else a freshly computed
       :func:`importance` (what to read next / how important).
    3. **Redundancy (coverage-first penalty).** A candidate that is a
       near-duplicate of an already-selected paper (see :func:`_is_redundant`) is
       *deferred*: a non-redundant coverer is always preferred when one exists. But
       redundancy is a penalty, NOT an absolute veto — if the only remaining
       coverers of an uncovered unit are near-duplicates, the best such coverer is
       still selected (coverage wins) rather than abandoning the unit. A
       contradicts/qualifies edge exempts a pair from redundancy entirely, so a
       claim and its rebuttal are both directly selectable.
    4. **Stop** at the budget or when no candidate (redundant or not) yields
       positive marginal gain — i.e. every remaining uncovered unit has no
       selectable coverer left.

    ``redundancy_threshold`` is an embedding cosine *distance* in ``[0, 2]``
    (``0`` = identical, ``2`` = opposite; smaller = stricter deduplication). It is
    clamped into that range, so a similarity-like or out-of-range value cannot
    collapse the corpus.

    Returns the flat selection (in order), the selection grouped ``by_theme`` (or
    ``by_claim`` in claim mode), the uncovered ``gaps``, and the ``pre_covered``
    units. Pure and deterministic (stable tie-breaks; no RNG, no wall clock).
    """
    theme_list = themes.themes if isinstance(themes, ThemeModel) else tuple(themes)
    work_by_id = {w.id: w for w in works}
    edges = _iter_relation_edges(relation_edges)
    contradiction_pairs = _contradiction_pairs(edges)
    # ``redundancy_threshold`` is a cosine DISTANCE in [0, 2]; clamp it so a caller
    # passing a similarity-like (e.g. 0.9) or out-of-range (>=2.0) value cannot
    # mark everything redundant against the first pick.
    redundancy_threshold = max(0.0, min(2.0, redundancy_threshold))

    if read_priority_scores is not None:
        value = read_priority_scores
    elif importance_scores is not None:
        value = importance_scores
    else:
        value = importance(works, cohort_window=cohort_window)

    # COMPOSITE coverage universe: THEME units (each covered by any of its member
    # works) and KEY-CLAIM units (each covered by its single CORE paper) compete
    # in ONE greedy loop within ONE budget. The two live in DIFFERENT key spaces —
    # theme ids vs ``"<graph>::<id>"`` claim uids — so ``theme_unit_ids`` and
    # ``claim_unit_ids`` partition the units, and the rest of the algorithm is
    # universe-agnostic over the merged ``unit_members``.
    #
    #   * no claim units  → ``universe == "theme"`` and EVERY output (selection,
    #     by_theme, gaps, pre_covered, claim fields) is byte-identical to the
    #     original theme-only cover — a hard regression guard.
    #   * claim + theme    → ``universe == "composite"`` (both covered together).
    #   * claim, no theme  → ``universe == "claim"`` (claim-only corpus).
    theme_members = {t.theme_id: list(t.work_ids) for t in theme_list}
    # A claim's core work id is its single coverer — OWNED source or UNOWNED
    # citation (an acquisition candidate) alike; ``source_graph`` (consulted via
    # ``owned_ids`` below) decides the tie-break, not membership. Only a falsy
    # core (None — no candidate paper at all) has NO selectable coverer: it is a
    # PERMANENT ACQUISITION GAP, kept as a unit so it surfaces in the gaps rather
    # than being silently dropped.
    claim_members = (
        {uid: ([wid] if wid else []) for uid, wid in claim_cover.items()}
        if claim_cover else {}
    )
    has_claims = bool(claim_members)
    has_themes = bool(theme_members)
    if not has_claims:
        universe = "theme"
    elif has_themes:
        universe = "composite"
    else:
        universe = "claim"

    theme_unit_ids = set(theme_members)
    claim_unit_ids = set(claim_members)
    unit_members: dict[str, list[str]] = {**theme_members, **claim_members}

    unit_of_work: dict[str, set[str]] = {}
    for uid, members in unit_members.items():
        for wid in members:
            # P2: a member work id that is not in ``works`` cannot be selected (the
            # greedy only iterates ``works``), so it is effectively dropped from the
            # selectable set while its unit's residual need stays unmet → the unit
            # remains a gap. No crash, no spam: a phantom id here is simply never
            # reached by the selection loop below.
            unit_of_work.setdefault(wid, set()).add(uid)

    residual: dict[str, int] = {}
    pre_covered: list[str] = []
    for uid, members in unit_members.items():
        has_read = any(
            wid in work_by_id and (work_by_id[wid].coverage_human or "none").lower() == "full"
            for wid in members
        )
        residual[uid] = 0 if has_read else 1
        if has_read:
            pre_covered.append(uid)
    uncovered = {uid for uid, need in residual.items() if need > 0}

    selected: list[CoverSelection] = []
    selected_order: list[str] = []
    selected_set: set[str] = set()
    by_unit: dict[str, list[str]] = {}

    # A work backed by a writable local source folder is OWNED; a citation-only
    # work (``source_graph is None``) is UNOWNED — an acquisition candidate that
    # must not displace an equivalent owned coverer on a tie.
    owned_ids = {w.id for w in works if w.source_graph is not None}

    def _outranks(cand: tuple[int, float, str, set[str]], cur: tuple[int, float, str, set[str]] | None) -> bool:
        if cur is None:
            return True
        gain, v, wid, _ = cand
        cgain, cv, cwid = cur[0], cur[1], cur[2]
        if gain != cgain:
            return gain > cgain
        if v != cv:
            return v > cv
        # Exact (gain, value) tie: prefer an OWNED coverer over an UNOWNED
        # (citation) acquisition candidate, then fall back to the stable id order.
        cand_owned = wid in owned_ids
        cur_owned = cwid in owned_ids
        if cand_owned != cur_owned:
            return cand_owned
        return wid < cwid

    while len(selected) < budget and uncovered:
        best: tuple[int, float, str, set[str]] | None = None
        best_redundant: tuple[int, float, str, set[str]] | None = None
        for w in works:
            if w.id in selected_set or (w.coverage_human or "none").lower() == "full":
                continue
            covers = uncovered & unit_of_work.get(w.id, set())
            if not covers:
                continue
            gain = len(covers)
            v = value.get(w.id, 0.0)
            cand = (gain, v, w.id, covers)
            if _is_redundant(w.id, selected_order, embeddings, redundancy_threshold, contradiction_pairs):
                # Defer near-duplicates: keep the best one as a coverage-first
                # fallback in case no non-redundant coverer exists this round.
                if _outranks(cand, best_redundant):
                    best_redundant = cand
                continue
            if _outranks(cand, best):
                best = cand
        # Prefer a non-redundant coverer; otherwise fall back to the best
        # near-duplicate so a unit whose only coverers are redundant is still
        # covered while budget remains (redundancy is a penalty, not a veto).
        chosen = best if best is not None else best_redundant
        if chosen is None:
            break
        _gain, v, wid, covers = chosen
        # A single pick can cover BOTH theme and claim units; route each covered
        # unit into its own key space so theme_ids and claim_ids stay disjoint.
        theme_ids = tuple(sorted(covers & theme_unit_ids))
        claim_ids = tuple(sorted(covers & claim_unit_ids))
        # Mark an UNOWNED coverer at the boundary: it covers its unit only by being
        # acquired, so owned-only read-hook consumers exclude it deterministically
        # rather than relying on the caller to remap unowned cores (see
        # claims.enrich_works_with_claims). ``owned_ids`` holds every work backed
        # by a writable source folder, so anything outside it is an acquisition.
        acquisition = wid not in owned_ids
        selected.append(
            CoverSelection(
                work_id=wid, theme_ids=theme_ids, value=v,
                claim_ids=claim_ids, acquisition=acquisition,
            )
        )
        selected_order.append(wid)
        selected_set.add(wid)
        for uid in covers:
            by_unit.setdefault(uid, []).append(wid)
            residual[uid] -= 1
            if residual[uid] <= 0:
                uncovered.discard(uid)

    # Partition the selection / gaps back into their two key spaces. ``residual_gaps``
    # on the SetCover (``gaps``) UNIONS both spaces (theme ids ∪ claim uids) for the
    # exposure layer; ``by_theme``/``by_claim`` keep them separate, and ``claim_ids``
    # is the claim-space subset of the covered units. Which key space each output
    # field carries depends on ``universe`` (theme / claim / composite).
    by_theme = {uid: tuple(wids) for uid, wids in by_unit.items() if uid in theme_unit_ids}
    by_claim = {uid: tuple(wids) for uid, wids in by_unit.items() if uid in claim_unit_ids}
    return SetCover(
        selected=tuple(selected),
        by_theme=by_theme,
        gaps=tuple(sorted(uid for uid, need in residual.items() if need > 0)),
        pre_covered=tuple(sorted(pre_covered)),
        budget=budget,
        by_claim=by_claim,
        claim_ids=tuple(sorted(uid for uid in by_unit if uid in claim_unit_ids)),
        universe=universe,
    )