zettelkasten.papers

zettelkasten.papers

Papers-view payload assembly: the EXPOSURE layer over the syllabus engine.

:mod:zettelkasten.syllabus is the deterministic, LLM-free scoring/trend/theme engine. This module is the thin seam that turns its primitives into the single JSON payload the Papers view consumes — exactly once, in one place, so the MCP build_syllabus tool and the dashboard's /papers endpoints can never drift. Like :mod:zettelkasten.framing it is parametrized by a get_graph callable (the MCP server and the dashboard keep separate graph caches) and is pure given its inputs: every score is a reproducible function of the corpus.

The engine is scoped only by project (a source list) or globally. A single graph is a one-source scope the public engine does not express directly, so :func:_graph_scoped_corpus realises it by driving :func:build_corpus with a get_graph restricted to the target graph (every other source raises and is skipped, so endorsement edges, PageRank, and the referenced-citation pool are all confined to that one graph) and then keeping only the works the graph actually engages with. That yields the same single-source semantics a one-source project would, without touching the locked engine.

Semantic inputs the engine cannot derive itself — note embeddings and cluster membership — are accepted as explicit work_id → … mappings (the same convention the engine's :func:interestingness/:func:situate use). The dashboard supplies them (projected from its note-level embedding/cluster machinery); the MCP server omits them and the corresponding fields degrade to null rather than failing. relation_edges and landscape hubs ARE derivable from the graph's own links, so this module derives them for both callers.

assemble_papers

assemble_papers(get_graph: GetGraph, *, project: str = '', graph: str = '', budget: int = DEFAULT_COVER_BUDGET, suggested_only: bool = False, as_of: 'str | int | None' = '', sort: str = '', reverse: bool = False, scope_name: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None, embeddings: 'dict[str, Sequence[float]] | None' = None, clusters: 'dict[str, str] | None' = None, now_year: int | None = None) -> dict[str, Any]

Assemble the full Papers-view payload from the syllabus engine.

Scoped to a single graph (one source) or a project (its source set), or globally when both are empty. embeddings/clusters are optional work_id → … mappings; when omitted (the MCP server) the embedding-driven signals degrade to 0 and relation_edges/trends still populate interestingness/situate. relation_edges and landscape hubs are derived from the graph's own links for both callers.

Source code in zettelkasten/papers.py

def assemble_papers(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    budget: int = DEFAULT_COVER_BUDGET,
    suggested_only: bool = False,
    as_of: "str | int | None" = "",
    sort: str = "",
    reverse: bool = False,
    scope_name: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
    embeddings: "dict[str, Sequence[float]] | None" = None,
    clusters: "dict[str, str] | None" = None,
    now_year: int | None = None,
) -> dict[str, Any]:
    """Assemble the full Papers-view payload from the syllabus engine.

    Scoped to a single ``graph`` (one source) or a ``project`` (its source set),
    or globally when both are empty. ``embeddings``/``clusters`` are optional
    ``work_id → …`` mappings; when omitted (the MCP server) the embedding-driven
    signals degrade to 0 and ``relation_edges``/trends still populate
    interestingness/situate. ``relation_edges`` and landscape hubs are derived
    from the graph's own links for both callers.
    """
    base = graphs_dir or GRAPHS_DIR
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    as_of_int = _parse_as_of(as_of)

    if graph:
        works, scoped = _graph_scoped_corpus(get_graph, graph, graphs_dir=base, namespace=ns)
    else:
        works, scoped = build_corpus(get_graph, project=project, graphs_dir=base, namespace=ns)

    work_ids = {w.id for w in works}
    relation_edges = _derive_relation_edges(get_graph, scoped, loc)
    landscape_hubs = _derive_landscape_hubs(get_graph, scoped, loc, work_ids)
    citations = load_citations(graphs_dir=base)

    # ── Claim read-layer enrichment (the natural orchestration point) ──────────
    # Build the scoped claim index over the SAME scope as the corpus, rank its key
    # claims, resolve each one's CORE paper using BASE (pre-claim) importance — so
    # the core-paper tie-break does not feed back into the salience it is about to
    # change — and populate the two Work claim hooks (core_claim_count,
    # unlocks_uncovered_claim) IN PLACE, before any score is computed. A
    # claim-free scope is a no-op: the hooks stay 0/None and every downstream
    # score is byte-identical to the pre-claim behavior. ``key_claims`` with no
    # limit/banding returns every resolved claim ranked, so "key claims" here is
    # the full claim set — core_claim_count counts over all of them.
    cindex = build_claim_index(
        get_graph, project=project, graph=graph, graphs_dir=base,
        namespace=ns, localize=loc,
    )
    # BASE (pre-claim) paper importance, keyed by BOTH work id AND writable
    # source-graph folder (``_paper_importance``, the same id+folder map the
    # standalone claim engine uses). The claim layer (key_claims supporter-quality,
    # core_paper_atom selection) keys on source-graph folders, so a plain
    # id-keyed ``importance(works)`` would miss canonical_citation_key-merged
    # sources whose work id != folder. Computed before enrichment so the
    # core-paper tie-break does not feed back into the salience it is about to
    # change (core_claim_count is still 0 here).
    base_importance = _paper_importance(works)
    claim_centrality_map = claim_centrality(cindex)
    ranked_claims = key_claims(
        cindex, importance_map=base_importance, centrality=claim_centrality_map
    )
    core_map, core_work_id = enrich_works_with_claims(
        works, cindex,
        key_claim_keys=[kc.key for kc in ranked_claims],
        citations=citations,
        importance_map=base_importance,
    )
    has_claims = bool(ranked_claims)

    # Wave-1 scores (the same composition score_corpus performs), computed once
    # and threaded into the trend/interestingness/situate/cover layers so every
    # facet of a work is scored against one consistent corpus. The claim hooks
    # above are now folded into salience (claim-coreness) and read_priority
    # (claim-coverage).
    inf = influence(works)
    sal = salience(works)
    imp = importance(works, influence_scores=inf, salience_scores=sal)
    anc = anchor_score(works)
    rp = read_priority(works, importance_scores=imp)

    trends = corpus_trends(works, as_of=as_of_int, now_year=now_year)
    situations = situate(
        works,
        relation_edges=relation_edges,
        clusters=clusters,
        trends=trends,
        influence_scores=inf,
        salience_scores=sal,
        as_of=as_of_int,
        now_year=now_year,
    )
    interest = interestingness(
        works,
        embeddings=embeddings,
        clusters=clusters,
        relation_edges=relation_edges,
        trends=trends,
        influence_scores=inf,
        salience_scores=sal,
        read_state_aware=True,
    )

    tm = theme_model(
        works,
        clusters=clusters,
        relation_edges=relation_edges,
        landscape_hubs=landscape_hubs or None,
        anchor_scores=anc,
    )
    # Composite cover when key claims exist: the cover covers BOTH the theme units
    # AND the key-claim units (each claim covered by its OWNED core work). Every
    # key claim is a unit — including one whose core does NOT resolve to an owned
    # corpus Work (``core_work_id`` None, e.g. a _cross synthesis claim with no
    # citation/cross-source supporter, OR a claim whose core is an UNOWNED
    # citation). Such a claim has no readable/owned coverer here, so it stays a
    # PERMANENT ACQUISITION GAP in the reading list (it is NOT dropped from
    # key_claims, and the claim-space invariant key_claims == covered ∪ claim_gaps
    # ∪ pre_covered holds); acquiring it is surfaced by the dedicated
    # missing-papers ranker (claims.rank_missing_papers), not silently folded into
    # the reading list. The claim-free path keeps claim_cover None → byte-identical
    # theme cover.
    claim_cover: dict[str, str | None] | None = None
    if has_claims:
        claim_cover = {
            claim_uid(kc.key): core_work_id.get(kc.key) for kc in ranked_claims
        }
        claim_cover = claim_cover or None

    cover = set_cover(
        works,
        tm,
        budget=budget,
        importance_scores=imp,
        read_priority_scores=rp,
        relation_edges=relation_edges,
        embeddings=embeddings,
        claim_cover=claim_cover,
    )
    suggested_ids = {cs.work_id for cs in cover.selected}

    # Per-work claims_covered: the key claims a work is the CORE paper for, in the
    # global key-claim ranking order (deterministic). Built by inverting the
    # core_work_id map computed during enrichment.
    claims_by_work: dict[str, list[dict[str, Any]]] = {}
    for kc in ranked_claims:
        wid = core_work_id.get(kc.key)
        if wid is None:
            continue
        claims_by_work.setdefault(wid, []).append(
            {"id": kc.id, "graph": kc.graph, "uid": claim_uid(kc.key), "title": kc.title}
        )

    rows: list[dict[str, Any]] = []
    for w in works:
        venue, wtype, abstract = _meta_lookup(w.id, citations, base)
        sit = situations.get(w.id)
        it = interest.get(w.id)
        tmet = trends.get(w.id)
        row = {
            "id": w.id,
            "title": w.title,
            "year": w.year,
            "venue": venue,
            "type": wtype,
            "doi": w.doi,
            "abstract": abstract,
            # ``None`` (→ "—" in the UI) when the citation count was never
            # enriched: the work is absent from ``inf`` rather than scored a
            # misleading neutral 0.5. See syllabus.influence.
            "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": round(anc.get(w.id, 0.0), 4),
            "read_priority": round(rp.get(w.id, 0.0), 4),
            "coverage_human": w.coverage_human,
            "authors": list(w.authors),
            # Namespace the writable backing graph at the exposure boundary so a
            # federated source routes to its repo, exactly as
            # ``_namespace_project_graph`` does for the graph/reading-list
            # payloads. ``ns`` is identity for the local/non-federated store, so
            # a local source_graph stays the bare folder name that
            # ``/graphs/{name}/coverage`` resolves.
            "source_graph": ns(w.source_graph) if w.source_graph is not None else None,
            "promoted": w.source_graph is not None,
            "interestingness": round(it.score, 4) if it is not None else None,
            "dominant_signal": it.dominant_signal if it is not None else None,
            "temporal": sit.temporal if sit is not None else None,
            "position": sit.position if sit is not None else None,
            "influence_label": sit.influence_label if sit is not None else None,
            "sleeper": sit.sleeper if sit is not None else None,
            "velocity": round(tmet.velocity, 4) if tmet is not None else None,
            "acceleration": round(tmet.acceleration, 4) if tmet is not None else None,
            "trajectory": tmet.trajectory if tmet is not None else None,
            "recency": round(tmet.citation_recency, 4) if tmet is not None else None,
            "publication_recency": round(tmet.publication_recency, 4) if tmet is not None else None,
            "why_read": why_read(
                w,
                salience=sal.get(w.id, 0.0),
                importance=imp.get(w.id, 0.0),
                anchor=anc.get(w.id, 0.0),
                situation=sit,
                contested=(it is not None and it.dissent > 0.0),
            ),
            "why_interesting": why_interesting(it),
            "in_suggested_set": w.id in suggested_ids,
            "claims_covered": claims_by_work.get(w.id, []),
        }
        rows.append(row)

    _sort_rows(rows, sort, reverse)
    if suggested_only:
        rows = [r for r in rows if r["in_suggested_set"]]

    name = scope_name or graph or project or "all"
    return {
        "scope": {"type": "graph" if graph else "project", "name": name},
        "work_count": len(works),
        "as_of": as_of_int,
        "theme_source": tm.source,
        "themes": [
            {
                "theme_id": t.theme_id,
                "label": t.label,
                "work_ids": list(t.work_ids),
                "weight": round(t.weight, 4),
                "source": t.source,
            }
            for t in tm.themes
        ],
        "works": rows,
        "key_claims": [
            {
                "id": kc.id,
                "graph": kc.graph,
                "uid": claim_uid(kc.key),
                "title": kc.title,
                "type": kc.type,
                "strength": kc.strength,
                "status": kc.status,
                "support_span": kc.support_span,
                "oppose_span": kc.oppose_span,
                "core_paper": _key_claim_core_payload(core_map.get(kc.key)),
            }
            for kc in ranked_claims
        ],
        "suggested_set": [
            {
                "id": cs.work_id,
                "theme_ids": list(cs.theme_ids),
                "claim_ids": list(cs.claim_ids),
                "value": round(cs.value, 4),
                "acquisition": cs.acquisition,
            }
            for cs in cover.selected
        ],
        "cover_universe": cover.universe,
        "residual_gaps": list(cover.gaps),
        "pre_covered": list(cover.pre_covered),
        "budget": cover.budget,
        "field_direction": field_direction(
            works, grouping=tm.assignment or None, as_of=as_of_int, now_year=now_year
        ),
    }

apply_papers_view

apply_papers_view(payload: dict[str, Any], *, sort: str = '', reverse: bool = False, suggested_only: bool = False) -> dict[str, Any]

Apply the per-request sort/reverse/suggested_only view over a base payload.

:func:assemble_papers already scores every work and (when called with the default sort="", reverse=False, suggested_only=False) returns the rows in the engine's DEFAULT order (read_priority then importance, both descending). This helper turns that one scored base payload into the view a specific request wants — WITHOUT rescoring — so a sort/filter toggle is a cheap O(n) reorder over a cached payload instead of a full syllabus recompute.

Byte-identical to calling assemble_papers directly with the same sort/reverse/suggested_only: the default order's tie-break (read_priority, importance) is a subset of every other sort's tie-break, so re-sorting the default-ordered rows preserves the exact build-corpus tie order a from-scratch sort would produce. Non-works fields (themes, key_claims, suggested_set, counts, …) are independent of the view and are carried through unchanged. The input payload is never mutated — the top-level dict and the works list are copied — so a shared cache entry stays reusable across requests.

The DEFAULT view (no sort, no reverse, no filter) is returned AS-IS — the same object — because the base payload is already assembled in that exact order. This keeps the cache-hit contract "an unchanged corpus returns the memoized object" intact for the common poll path (a repeat default request is object- identical to the cached payload, not merely equal).

Source code in zettelkasten/papers.py

def apply_papers_view(
    payload: dict[str, Any],
    *,
    sort: str = "",
    reverse: bool = False,
    suggested_only: bool = False,
) -> dict[str, Any]:
    """Apply the per-request sort/reverse/suggested_only view over a base payload.

    :func:`assemble_papers` already scores every work and (when called with the
    default ``sort=""``, ``reverse=False``, ``suggested_only=False``) returns the
    rows in the engine's DEFAULT order (``read_priority`` then ``importance``,
    both descending). This helper turns that one scored base payload into the
    view a specific request wants — WITHOUT rescoring — so a sort/filter toggle
    is a cheap O(n) reorder over a cached payload instead of a full syllabus
    recompute.

    Byte-identical to calling ``assemble_papers`` directly with the same
    ``sort``/``reverse``/``suggested_only``: the default order's tie-break
    (``read_priority``, ``importance``) is a subset of every other sort's
    tie-break, so re-sorting the default-ordered rows preserves the exact
    build-corpus tie order a from-scratch sort would produce.     Non-``works`` fields
    (themes, key_claims, suggested_set, counts, …) are independent of the view and
    are carried through unchanged. The input ``payload`` is never mutated — the
    top-level dict and the ``works`` list are copied — so a shared cache entry
    stays reusable across requests.

    The DEFAULT view (no sort, no reverse, no filter) is returned AS-IS — the same
    object — because the base payload is already assembled in that exact order.
    This keeps the cache-hit contract "an unchanged corpus returns the memoized
    object" intact for the common poll path (a repeat default request is object-
    identical to the cached payload, not merely equal).
    """
    if not sort and not reverse and not suggested_only:
        return payload
    result = dict(payload)
    rows = list(payload.get("works", []))
    _sort_rows(rows, sort, reverse)
    if suggested_only:
        rows = [r for r in rows if r.get("in_suggested_set")]
    result["works"] = rows
    return result