zettelkasten.claims

zettelkasten.claims

Claim read layer: the deterministic claim-level engine over the Zettelkasten.

Where :mod:zettelkasten.syllabus scores papers (Works) and :mod:zettelkasten.papers exposes them, this module is the pure, LLM-free read layer over claims and findings — the notes that carry assertions, the relation edges that endorse, replicate, contest, or qualify them, and the quote notes that supply verbatim evidence. Like :mod:zettelkasten.framing and :mod:zettelkasten.papers 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.

There is no file write and no new note type here. The engine only reads the existing claim graph; the WS3 discovery layer (:func:discover_contradictions, :func:debate_map, :func:find_supersessions) is likewise PROPOSE-ONLY — it emits candidate edges but writes nothing, and its only optional LLM touch is an INJECTABLE, write-free stance classify_fn (defaulted off under tests, exactly like :func:claim_producer.test_claim). The engine extracts active claim/finding notes across the scoped graphs, builds a single one-pass reverse index of every relation edge (the only tractable way to find supporters/attackers that live in other source graphs, since :meth:ZettelGraph.get_backlinks only scans one graph), and from that assembles per-claim anatomy — evidence, corroboration, counterclaims, caveats, a bounded :func:claim_strength, a derived status, a :func:core_paper_atom, and a :func:claim_path.

Conventions it relies on (verified against the live data model):

• Evidence is a quote note linking --supports--> claim (the support edge is outgoing from the quote), so a claim's verbatim evidence is its supports backlinks whose source note is a quote.
• Cross-source corroboration is a supports/replicates backlink from a note living in a different source graph.
• Contestation is contradicts / responds-to / qualifies (incoming = attacks on this claim; outgoing = this claim attacks others), with qualifies doubling as a caveat.
• A claim's home paper is the source-graph folder it lives in; secondary papers are cross-graph supporters' home folders and outgoing _citations links.

Scoring reuses the locked syllabus primitives (_pagerank, SALIENCE_RELATIONS, the Work/importance machinery) rather than re-deriving them, so claim scores and paper scores stay on one consistent basis.

Edge dataclass

A single resolved relation edge between two notes.

src_* is the note that authored the link; dst_* is its (namespace- resolved) target. dst_graph resolves Link.graph to a concrete namespace — the source's own graph when Link.graph is empty, the literal "_citations" for citation targets, otherwise the named foreign graph — so the edge can be indexed by where its target actually lives even though :meth:ZettelGraph.get_backlinks cannot see across graphs.

Source code in zettelkasten/claims.py

@dataclass(frozen=True)
class Edge:
    """A single resolved relation edge between two notes.

    ``src_*`` is the note that authored the link; ``dst_*`` is its (namespace-
    resolved) target. ``dst_graph`` resolves ``Link.graph`` to a concrete
    namespace — the source's own graph when ``Link.graph`` is empty, the literal
    ``"_citations"`` for citation targets, otherwise the named foreign graph — so
    the edge can be indexed by where its target actually lives even though
    :meth:`ZettelGraph.get_backlinks` cannot see across graphs.
    """

    src_graph: str
    src_id: str
    src_type: str
    relation: str
    dst_graph: str
    dst_id: str

    @property
    def src_key(self) -> Key:
        return (self.src_graph, self.src_id)

    @property
    def dst_key(self) -> Key:
        return (self.dst_graph, self.dst_id)

ResolvedClaim dataclass

A claim/finding note with its incident edges already partitioned.

Built once per claim by :func:build_claim_index. Holds the raw edge partitions (incoming support, both-direction counters/structural, …) plus the materialized supporting quotes, so the scorers and the anatomy assembler are pure functions of this record and never re-walk the graph.

Source code in zettelkasten/claims.py

@dataclass
class ResolvedClaim:
    """A claim/finding note with its incident edges already partitioned.

    Built once per claim by :func:`build_claim_index`. Holds the raw edge
    partitions (incoming support, both-direction counters/structural, …) plus
    the materialized supporting quotes, so the scorers and the anatomy assembler
    are pure functions of this record and never re-walk the graph.
    """

    key: Key
    note: Note
    home_graph: str
    support_edges: list[Edge] = field(default_factory=list)        # incoming supports/replicates
    support_sources: list[str] = field(default_factory=list)       # distinct supporting source graphs
    supporting_quotes: list[dict[str, Any]] = field(default_factory=list)
    corroborations: list[Edge] = field(default_factory=list)       # support from OTHER sources
    replication_count: int = 0
    counter_in: list[Edge] = field(default_factory=list)           # attacks on this claim
    counter_out: list[Edge] = field(default_factory=list)          # this claim attacking others
    caveats: list[Edge] = field(default_factory=list)              # incoming qualifies
    superseded_by: list[Edge] = field(default_factory=list)        # incoming supersedes
    structural_out: list[Edge] = field(default_factory=list)
    structural_in: list[Edge] = field(default_factory=list)

ClaimIndex dataclass

The scoped claim graph: notes, claims, edges, and per-claim resolutions.

sources are the (local) graph names walked; real_sources drops the non-paper folders (_cross) so home-paper identity can be checked. notes_by_key indexes every note (not just claims) so edge endpoints — quotes, methods, opposing claims — resolve. edges is the flat one-pass edge list; by_source/by_target are its forward/reverse views.

Source code in zettelkasten/claims.py

@dataclass
class ClaimIndex:
    """The scoped claim graph: notes, claims, edges, and per-claim resolutions.

    ``sources`` are the (local) graph names walked; ``real_sources`` drops the
    non-paper folders (``_cross``) so home-paper identity can be checked.
    ``notes_by_key`` indexes *every* note (not just claims) so edge endpoints —
    quotes, methods, opposing claims — resolve. ``edges`` is the flat one-pass
    edge list; ``by_source``/``by_target`` are its forward/reverse views.
    """

    sources: list[str]
    real_sources: set[str]
    notes_by_key: dict[Key, Note]
    claims: dict[Key, Note]
    resolved: dict[Key, ResolvedClaim]
    edges: list[Edge]
    by_source: dict[Key, list[Edge]]
    by_target: dict[Key, list[Edge]]

KeyClaim dataclass

A ranked claim: its salience plus the scored facets the view needs.

Source code in zettelkasten/claims.py

@dataclass(frozen=True)
class KeyClaim:
    """A ranked claim: its salience plus the scored facets the view needs."""

    key: Key
    graph: str
    id: str
    title: str
    type: str
    salience: float
    strength: float
    status: str
    support_span: int
    oppose_span: int
    contested: float
    theme: str = ""

CorePaper dataclass

The single best paper to acquire/read for a claim.

unowned flags a paper that is only a citation (no writable local source folder) — i.e. an acquisition recommendation. distributed flags a claim with no single clear owner (degraded to a survey or top supporters). reason is a short human string explaining the pick.

Source code in zettelkasten/claims.py

@dataclass(frozen=True)
class CorePaper:
    """The single best paper to acquire/read for a claim.

    ``unowned`` flags a paper that is only a citation (no writable local source
    folder) — i.e. an *acquisition* recommendation. ``distributed`` flags a
    claim with no single clear owner (degraded to a survey or top supporters).
    ``reason`` is a short human string explaining the pick.
    """

    paper_id: str
    title: str
    year: int | None
    source_graph: str | None
    unowned: bool
    distributed: bool
    reason: str
    importance: float

ClaimPath dataclass

An ordered reading path to a claim: prerequisites first, then the claim.

steps are claim refs ({graph,id,title,type}) in dependency order, ending with the target claim. core_paper is the paper to start from.

Source code in zettelkasten/claims.py

@dataclass(frozen=True)
class ClaimPath:
    """An ordered reading path to a claim: prerequisites first, then the claim.

    ``steps`` are claim refs (``{graph,id,title,type}``) in dependency order,
    ending with the target claim. ``core_paper`` is the paper to start from.
    """

    claim: Key
    steps: tuple[dict[str, Any], ...]
    core_paper: CorePaper | None

claim_search_sources

claim_search_sources(project: str, graph: str, graphs_dir: 'Path | None', namespace: Callable[[str], str]) -> list[str]

Resolve which graphs the claim index should walk.

A single graph scope confines the walk to that one source (its claims, its evidence) — the claim analogue of papers' graph scope. Otherwise the project's source set (or the global set) is used, including _cross so cross-source synthesis claims are seen.

Source code in zettelkasten/claims.py

def claim_search_sources(
    project: str,
    graph: str,
    graphs_dir: "Path | None",
    namespace: Callable[[str], str],
) -> list[str]:
    """Resolve which graphs the claim index should walk.

    A single ``graph`` scope confines the walk to that one source (its claims,
    its evidence) — the claim analogue of papers' graph scope. Otherwise the
    project's source set (or the global set) is used, including ``_cross`` so
    cross-source synthesis claims are seen.
    """
    if graph:
        return [namespace(graph)]
    return frame_search_sources(project, graphs_dir=graphs_dir, namespace=namespace)

build_claim_index

build_claim_index(get_graph: GetGraph, *, project: str = '', graph: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None) -> ClaimIndex

Walk the scoped graphs once and build the claim index.

One pass over every scoped graph's notes collects (a) a note lookup keyed by (graph, id), (b) the flat edge list with each Link.graph resolved to a concrete target namespace, and (c) the set of active claim/finding notes. A second cheap pass partitions each claim's incident edges into the support / corroboration / counter / caveat / structural buckets. draft notes are excluded (an in-progress assertion is not a scorable claim).

Pure: the only I/O is the get_graph reads (and the graph helpers those already use). Tolerant of missing graphs, empty graphs, and dangling edges.

Source code in zettelkasten/claims.py

def build_claim_index(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
) -> ClaimIndex:
    """Walk the scoped graphs once and build the claim index.

    One pass over every scoped graph's notes collects (a) a note lookup keyed by
    ``(graph, id)``, (b) the flat edge list with each ``Link.graph`` resolved to
    a concrete target namespace, and (c) the set of active claim/finding notes.
    A second cheap pass partitions each claim's incident edges into the support /
    corroboration / counter / caveat / structural buckets. ``draft`` notes are
    excluded (an in-progress assertion is not a scorable claim).

    Pure: the only I/O is the ``get_graph`` reads (and the graph helpers those
    already use). Tolerant of missing graphs, empty graphs, and dangling edges.
    """
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    sources = claim_search_sources(project, graph, graphs_dir, ns)

    notes_by_key: dict[Key, Note] = {}
    edges: list[Edge] = []
    by_source: dict[Key, list[Edge]] = {}
    by_target: dict[Key, list[Edge]] = {}
    claims: dict[Key, Note] = {}
    real_sources: set[str] = set()

    for display in sources:
        local = loc(display)
        if local not in _NON_PAPER_GRAPHS:
            real_sources.add(local)
        try:
            zg = get_graph(display)
        except Exception:
            continue
        for note in zg.notes.values():
            key = (local, note.id)
            notes_by_key[key] = note
            if note.type in CLAIM_TYPES and (note.status or "complete") in ACTIVE_STATUSES:
                claims[key] = note
            for link in note.links:
                # A tombstoned edge is a reversible soft-delete (the claim
                # producer sets it on stance edges incident to a discarded claim).
                # Skip it so it never inflates another claim's support/counter
                # buckets or surfaces stale evidence in the read layer.
                if link.tombstoned:
                    continue
                dst_graph = _resolve_dst_graph(link.graph, local)
                edge = Edge(
                    src_graph=local,
                    src_id=note.id,
                    src_type=note.type,
                    relation=link.relation,
                    dst_graph=dst_graph,
                    dst_id=link.target,
                )
                edges.append(edge)
                by_source.setdefault(edge.src_key, []).append(edge)
                by_target.setdefault(edge.dst_key, []).append(edge)

    # Per-project ``_cross`` membership guard. The synthesis ``_cross`` folder is
    # a single GLOBAL store, but ``frame_search_sources`` appends it to EVERY
    # project's source set — so without scoping, a synthesis claim grounded in
    # project Q's papers would leak into project P's projection (it resolves from
    # the shared ``_cross`` walk even though none of P's own sources touch it),
    # surfacing in P's review as a stray "unplaced" claim. Project membership of a
    # ``_cross`` note is EXPLICIT — the project manifest's ``cross:`` list, the
    # same authority ``project_graph`` honors — so under a project scope we keep
    # only the ``_cross`` claims that project actually claims. Global scope (no
    # project) is unchanged, and graph scope never walks ``_cross`` at all.
    if project:
        base = graphs_dir or GRAPHS_DIR
        try:
            cross_ids = set(project_cross_ids(load_project(project, graphs_dir=base)))
        except Exception:
            cross_ids = set()
        for key in [k for k in claims if k[0] == "_cross" and k[1] not in cross_ids]:
            claims.pop(key, None)

    resolved: dict[Key, ResolvedClaim] = {}
    for key, note in claims.items():
        resolved[key] = _resolve_claim(key, note, by_source, by_target, notes_by_key)

    return ClaimIndex(
        sources=list(sources),
        real_sources=real_sources,
        notes_by_key=notes_by_key,
        claims=claims,
        resolved=resolved,
        edges=edges,
        by_source=by_source,
        by_target=by_target,
    )

claim_strength

claim_strength(rc: ResolvedClaim, *, importance_map: dict[str, float] | None = None) -> float

A bounded, deterministic claim-strength score in [0,1].

Composed of: support breadth (distinct supporting sources, not raw quote count, saturated) scaled by supporter quality (mean paper importance of those sources, degrading to neutral 0.5 when no importance map is supplied); a replication bonus (replicates edges); and evidence richness (verified/paged quote notes). The sum is then discounted by the counterclaim strength. A claim with no evidence scores a defined 0.0.

Source code in zettelkasten/claims.py

def claim_strength(
    rc: ResolvedClaim,
    *,
    importance_map: dict[str, float] | None = None,
) -> float:
    """A bounded, deterministic claim-strength score in [0,1].

    Composed of: **support breadth** (distinct supporting *sources*, not raw
    quote count, saturated) scaled by **supporter quality** (mean paper
    importance of those sources, degrading to neutral 0.5 when no importance map
    is supplied); a **replication bonus** (``replicates`` edges); and **evidence
    richness** (verified/paged quote notes). The sum is then discounted by the
    **counterclaim strength**. A claim with no evidence scores a defined 0.0.
    """
    n_sources = len(rc.support_sources)
    breadth = _saturate(n_sources, 2.0)

    if importance_map and rc.support_sources:
        quals = [importance_map.get(s) for s in rc.support_sources]
        quals = [q for q in quals if q is not None]
        quality = sum(quals) / len(quals) if quals else 0.5
    else:
        quality = 0.5
    # Quality modulates support but never zeroes a genuinely broadly-supported
    # claim: it scales the breadth term between 0.5× and 1.0×.
    support_component = breadth * (0.5 + 0.5 * quality)

    replication_bonus = 0.15 * _saturate(rc.replication_count, 1.0)

    if rc.supporting_quotes:
        verified = sum(1 for q in rc.supporting_quotes if q["grounded"])
        richness = 0.15 * _saturate(verified, 1.0) if verified else 0.05
    else:
        richness = 0.0

    raw = min(1.0, support_component + replication_bonus + richness)
    strength = raw * (1.0 - _COUNTER_DISCOUNT * _counter_strength(rc))
    return max(0.0, min(1.0, strength))

claim_status

claim_status(rc: ResolvedClaim) -> str

Derive a claim's status: superseded | contested | established.

Source code in zettelkasten/claims.py

def claim_status(rc: ResolvedClaim) -> str:
    """Derive a claim's status: ``superseded`` | ``contested`` | ``established``."""
    if rc.superseded_by:
        return "superseded"
    if _counter_strength(rc) >= CONTESTED_THRESHOLD:
        return "contested"
    return "established"

claim_centrality

claim_centrality(index: ClaimIndex) -> dict[Key, float]

Max-normalized PageRank centrality over the claim-to-claim edge graph.

The graph is the structural + endorsement + contestation edges whose both endpoints are claims in the index, so centrality measures how load-bearing a claim is among other claims. Isolated claims score 0. Reuses the locked :func:syllabus._pagerank.

Source code in zettelkasten/claims.py

def claim_centrality(index: ClaimIndex) -> dict[Key, float]:
    """Max-normalized PageRank centrality over the claim-to-claim edge graph.

    The graph is the structural + endorsement + contestation edges whose *both*
    endpoints are claims in the index, so centrality measures how load-bearing a
    claim is among other claims. Isolated claims score 0. Reuses the locked
    :func:`syllabus._pagerank`.
    """
    rel_set = set(STRUCTURAL_RELATIONS) | set(SUPPORT_RELATIONS) | set(COUNTER_RELATIONS)
    claim_keys = set(index.claims)
    pr_edges: list[tuple[Any, Any]] = []
    for e in index.edges:
        if e.relation not in rel_set:
            continue
        if e.src_key in claim_keys and e.dst_key in claim_keys:
            pr_edges.append((e.src_key, e.dst_key))
    pr = _pagerank(pr_edges)
    norm = _max_normalize(pr)
    return {k: norm.get(k, 0.0) for k in claim_keys}

key_claims

key_claims(index: ClaimIndex, *, importance_map: dict[str, float] | None = None, centrality: dict[Key, float] | None = None, theme_assignment: dict[Key, str] | None = None, top_per_theme: int | None = None, limit: int | None = None) -> list[KeyClaim]

Rank claims by salience = support span × contestedness × centrality.

Salience multiplies cross-source support span (how many distinct sources back it), contestedness (a contested claim is more salient — it is where the field is live), and centrality (its standing among other claims). Each factor uses a (1 + …) / (0.5 + …) form so no single zero factor annihilates the rest, keeping the ranking well-behaved on sparse graphs.

With a theme_assignment and top_per_theme the result is the union of each theme's top band (still globally sorted); otherwise it is the global ranking, optionally truncated to limit. Ordering is fully deterministic (salience desc, then key).

Source code in zettelkasten/claims.py

def key_claims(
    index: ClaimIndex,
    *,
    importance_map: dict[str, float] | None = None,
    centrality: dict[Key, float] | None = None,
    theme_assignment: dict[Key, str] | None = None,
    top_per_theme: int | None = None,
    limit: int | None = None,
) -> list[KeyClaim]:
    """Rank claims by salience = support span × contestedness × centrality.

    Salience multiplies *cross-source support span* (how many distinct sources
    back it), *contestedness* (a contested claim is more salient — it is where
    the field is live), and *centrality* (its standing among other claims). Each
    factor uses a ``(1 + …)`` / ``(0.5 + …)`` form so no single zero factor
    annihilates the rest, keeping the ranking well-behaved on sparse graphs.

    With a ``theme_assignment`` and ``top_per_theme`` the result is the union of
    each theme's top band (still globally sorted); otherwise it is the global
    ranking, optionally truncated to ``limit``. Ordering is fully deterministic
    (salience desc, then key).
    """
    cen = centrality if centrality is not None else claim_centrality(index)
    ranked: list[KeyClaim] = []
    for key, rc in index.resolved.items():
        span = len(rc.support_sources)
        # Opposition breadth, mirroring support_span: distinct *paper* source
        # graphs that contradict the claim (the _cross / _citations folders are
        # evidence text, not independent opposing papers, so they're excluded —
        # exactly as support_sources excludes them on the supporting side).
        oppose_span = len({
            e.src_graph for e in rc.counter_in if e.src_graph not in _NON_PAPER_GRAPHS
        })
        contested = _counter_strength(rc)
        c = cen.get(key, 0.0)
        salience = (1.0 + span) * (1.0 + contested) * (0.5 + c)
        ranked.append(KeyClaim(
            key=key,
            graph=key[0],
            id=key[1],
            title=rc.note.title,
            type=rc.note.type,
            salience=round(salience, 6),
            strength=round(claim_strength(rc, importance_map=importance_map), 4),
            status=claim_status(rc),
            support_span=span,
            oppose_span=oppose_span,
            contested=round(contested, 4),
            theme=(theme_assignment or {}).get(key, ""),
        ))

    ranked.sort(key=lambda kc: (kc.salience, kc.key), reverse=True)

    if theme_assignment and top_per_theme:
        per_theme: dict[str, int] = {}
        banded: list[KeyClaim] = []
        for kc in ranked:  # already globally sorted, so bands keep top members
            # Claims absent from theme_assignment carry "" — route them into an
            # explicit untagged band (its own top_per_theme cap) rather than
            # dropping them, so unthemed claims still surface in key_claims.
            theme = kc.theme
            if per_theme.get(theme, 0) >= top_per_theme:
                continue
            per_theme[theme] = per_theme.get(theme, 0) + 1
            banded.append(kc)
        ranked = banded

    if limit is not None and limit >= 0:
        ranked = ranked[:limit]
    return ranked

core_paper_atom

core_paper_atom(rc: ResolvedClaim, index: ClaimIndex, *, works: list[Work] | None = None, citations: dict[str, dict] | None = None, importance_map: dict[str, float] | None = None) -> CorePaper | None

Pick the single best paper for a claim.

Candidates are the claim's home source, every cross-source supporter's home source, and the citation targets the claim's notes point at. Each is scored on lineage (it extends/replicates/responds-to/derives the claim), explanatory depth (it carries method/definition/model notes for the claim), paper importance, and earliness (an earlier high-importance supporter is preferred as the origin). The home/local supporters that are real source folders are owned; a citation-only winner is flagged unowned (an acquisition recommendation). When no single owner dominates a distributed claim, a surveying review is preferred and distributed set.

Returns None only when the claim has no candidate paper at all.

Source code in zettelkasten/claims.py

def core_paper_atom(
    rc: ResolvedClaim,
    index: ClaimIndex,
    *,
    works: list[Work] | None = None,
    citations: dict[str, dict] | None = None,
    importance_map: dict[str, float] | None = None,
) -> CorePaper | None:
    """Pick the single best paper for a claim.

    Candidates are the claim's home source, every cross-source supporter's home
    source, and the citation targets the claim's notes point at. Each is scored
    on **lineage** (it extends/replicates/responds-to/derives the claim),
    **explanatory depth** (it carries method/definition/model notes for the
    claim), **paper importance**, and **earliness** (an earlier high-importance
    supporter is preferred as the origin). The home/local supporters that are
    real source folders are *owned*; a citation-only winner is flagged
    ``unowned`` (an acquisition recommendation). When no single owner dominates
    a distributed claim, a surveying review is preferred and ``distributed`` set.

    Returns ``None`` only when the claim has no candidate paper at all.
    """
    works = works or []
    citations = citations or {}
    imp = importance_map if importance_map is not None else _paper_importance(works)
    source_to_work = {w.source_graph: w for w in works if w.source_graph}
    works_by_id = {w.id: w for w in works}

    # Which sources carry depth notes (method/definition/model) for this claim,
    # and which sources/citations stand in a lineage relation to it.
    depth_sources = {
        e.src_graph for e in index.by_target.get(rc.key, [])
        if e.src_type in DEPTH_TYPES
    }
    # Lineage is derived from *all* edges incident to the claim, not just the
    # supports/replicates partition: an incoming extends/replicates/responds-to
    # edge (some other paper built on this claim) or an outgoing derives-from
    # edge (this claim derives from another paper) both mark a lineage paper.
    # Restricting to rc.support_edges would miss extends/responds-to/derives-from
    # entirely, since support_edges only holds SUPPORT_RELATIONS.
    lineage_sources: set[str] = set()
    for e in index.by_target.get(rc.key, []):
        if e.relation in LINEAGE_RELATIONS and e.src_graph not in _NON_PAPER_GRAPHS:
            lineage_sources.add(e.src_graph)
    for e in index.by_source.get(rc.key, []):
        if e.relation in LINEAGE_RELATIONS and e.dst_graph not in _NON_PAPER_GRAPHS:
            lineage_sources.add(e.dst_graph)
    lineage_citations = {
        e.dst_id for e in index.by_source.get(rc.key, [])
        if e.dst_graph == "_citations" and e.relation in LINEAGE_RELATIONS
    }

    candidates: dict[str, dict[str, Any]] = {}

    def _add_source_candidate(sg: str) -> None:
        if sg in _NON_PAPER_GRAPHS or sg in candidates:
            return
        if sg not in index.real_sources and sg not in source_to_work:
            return
        w = source_to_work.get(sg)
        candidates[sg] = {
            "paper_id": sg,
            "source_graph": sg,
            "title": w.title if w else sg,
            "year": w.year if w else None,
            "is_review": bool(w.is_review) if w else False,
            "importance": imp.get(sg, 0.0),
            "lineage": sg in lineage_sources,
            "depth": sg in depth_sources,
            "unowned": False,
        }

    # Home paper (the source the claim lives in) is always a candidate.
    _add_source_candidate(rc.home_graph)
    for sg in rc.support_sources:
        _add_source_candidate(sg)
    # Lineage- and depth-bearing sources are candidates in their own right: a
    # paper that extends/derives/responds-to the claim, or carries its
    # method/definition/model, is a plausible origin even with no supports edge.
    for sg in lineage_sources:
        _add_source_candidate(sg)
    for sg in depth_sources:
        _add_source_candidate(sg)

    # Citation targets referenced by the claim's own outgoing links.
    for e in index.by_source.get(rc.key, []):
        if e.dst_graph != "_citations" or e.dst_id in candidates:
            continue
        cit = citations.get(e.dst_id, {})
        candidates[e.dst_id] = {
            "paper_id": e.dst_id,
            "source_graph": None,
            "title": str(cit.get("title") or e.dst_id),
            "year": _coerce_year(cit.get("year")),
            "is_review": str(cit.get("doc_type") or cit.get("type") or "").lower() == "review",
            "importance": imp.get(e.dst_id, 0.0),
            "lineage": e.relation in LINEAGE_RELATIONS or e.dst_id in lineage_citations,
            "depth": False,
            "unowned": True,
        }

    if not candidates:
        return None

    cands = list(candidates.values())
    owned = [c for c in cands if not c["unowned"]]
    # A claim is "distributed" when no owned candidate stands out: either there
    # is no owned candidate, or several sources support it with none carrying
    # lineage or depth to mark it as the origin.
    distributed = (not owned) or (
        len(rc.support_sources) >= 2
        and not any(c["lineage"] or c["depth"] for c in cands)
    )

    if distributed:
        reviews = [c for c in cands if c["is_review"]]
        pool = reviews or cands
    else:
        pool = cands

    # Stable id tie-break first, then the real ranking (Python sort is stable):
    # lineage, then depth, then importance, then earliest year (None last).
    pool.sort(key=lambda c: c["paper_id"])
    pool.sort(
        key=lambda c: (
            1 if c["lineage"] else 0,
            1 if c["depth"] else 0,
            round(c["importance"], 4),
            -(c["year"] if c["year"] is not None else 9999),
        ),
        reverse=True,
    )
    best = pool[0]

    reason_bits = []
    if best["lineage"]:
        reason_bits.append("lineage origin")
    if best["depth"]:
        reason_bits.append("explanatory depth")
    if distributed:
        reason_bits.append("distributed claim → " + ("survey" if best["is_review"] else "top supporter"))
    if best["unowned"]:
        reason_bits.append("not owned locally (acquire)")
    if not reason_bits:
        reason_bits.append("home paper" if best["paper_id"] == rc.home_graph else "top supporter")

    _ = works_by_id  # kept for parity / future depth-by-id lookups
    return CorePaper(
        paper_id=best["paper_id"],
        title=best["title"],
        year=best["year"],
        source_graph=best["source_graph"],
        unowned=best["unowned"],
        distributed=distributed,
        reason=", ".join(reason_bits),
        importance=round(best["importance"], 4),
    )

claim_uid

claim_uid(key: Key) -> str

A single string identity for a claim, stable across graphs.

Claims are keyed by (graph, note_id) (two graphs may reuse a note id), so the cover universe and the Papers payload need a flattened, collision-free id to join on. "<graph>::<id>" is that id; deterministic and reversible.

Source code in zettelkasten/claims.py

def claim_uid(key: Key) -> str:
    """A single string identity for a claim, stable across graphs.

    Claims are keyed by ``(graph, note_id)`` (two graphs may reuse a note id), so
    the cover universe and the Papers payload need a flattened, collision-free id
    to join on. ``"<graph>::<id>"`` is that id; deterministic and reversible.
    """
    return f"{key[0]}::{key[1]}"

enrich_works_with_claims

enrich_works_with_claims(works: list[Work], index: ClaimIndex, *, key_claim_keys: Iterable[Key] | None = None, citations: dict[str, dict] | None = None, importance_map: dict[str, float] | None = None) -> tuple[dict[Key, CorePaper | None], dict[Key, str | None]]

Populate the two :class:~zettelkasten.syllabus.Work claim hooks in place.

For each key claim in scope (key_claim_keys — the ranked :func:key_claims; defaults to every resolved claim when omitted) resolves its :func:core_paper_atom and the :class:Work that paper maps to, then sets

• Work.core_claim_count — how many key claims the work is the core paper for (feeds salience's claim-coreness term), and
• Work.unlocks_uncovered_claim — True iff the work is the core paper of a key claim not yet covered (its core paper not human-read "full"), else False (feeds read_priority's claim-coverage factor).

Claim-free scope is a no-op. With no key claims the counts stay 0 and unlocks_uncovered_claim stays None, so the salience claim term is dropped and the read_priority factor is a neutral 1.0 — every downstream score is byte-identical to the pre-claim behavior.

importance_map should be the base (pre-claim) paper importance so the core-paper tie-break does not feed back into the salience it is about to change. Returns (core_map, core_work_id) over the key claims — the per-claim core paper and the OWNED work id it resolves to (None when the core is unowned/unresolvable, so the claim becomes a permanent acquisition gap in the cover rather than a read recommendation) — for reuse by the cover and payload. Pure given the index + works; deterministic.

Source code in zettelkasten/claims.py

def enrich_works_with_claims(
    works: list[Work],
    index: ClaimIndex,
    *,
    key_claim_keys: Iterable[Key] | None = None,
    citations: dict[str, dict] | None = None,
    importance_map: dict[str, float] | None = None,
) -> tuple[dict[Key, CorePaper | None], dict[Key, str | None]]:
    """Populate the two :class:`~zettelkasten.syllabus.Work` claim hooks in place.

    For each key claim in scope (``key_claim_keys`` — the ranked
    :func:`key_claims`; defaults to every resolved claim when omitted) resolves
    its :func:`core_paper_atom` and the :class:`Work` that paper maps to, then sets

    * ``Work.core_claim_count`` — how many key claims the work is the *core* paper
      for (feeds salience's claim-coreness term), and
    * ``Work.unlocks_uncovered_claim`` — ``True`` iff the work is the core paper of
      a key claim not yet covered (its core paper not human-read ``"full"``), else
      ``False`` (feeds read_priority's claim-coverage factor).

    **Claim-free scope is a no-op.** With no key claims the counts stay ``0`` and
    ``unlocks_uncovered_claim`` stays ``None``, so the salience claim term is
    dropped and the read_priority factor is a neutral ``1.0`` — every downstream
    score is byte-identical to the pre-claim behavior.

    ``importance_map`` should be the **base** (pre-claim) paper importance so the
    core-paper tie-break does not feed back into the salience it is about to
    change. Returns ``(core_map, core_work_id)`` over the key claims — the
    per-claim core paper and the OWNED work id it resolves to (``None`` when the
    core is unowned/unresolvable, so the claim becomes a permanent acquisition gap
    in the cover rather than a read recommendation) — for reuse by the cover and
    payload. Pure given the index + works; deterministic.
    """
    keys = list(key_claim_keys) if key_claim_keys is not None else list(index.resolved)
    core_map: dict[Key, CorePaper | None] = {}
    core_work_id: dict[Key, str | None] = {}
    if not keys:
        return core_map, core_work_id

    imp = importance_map if importance_map is not None else _paper_importance(works)
    cits = citations or {}
    by_source_graph = {w.source_graph: w for w in works if w.source_graph}
    by_id = {w.id: w for w in works}

    count: dict[str, int] = {}
    unlocks: dict[str, bool] = {}
    for key in keys:
        rc = index.resolved.get(key)
        if rc is None:
            continue
        core = core_paper_atom(rc, index, works=works, citations=cits, importance_map=imp)
        core_map[key] = core
        w = _work_for_core_paper(core, by_source_graph, by_id)
        # Only an OWNED, readable core — a promoted source (``source_graph`` set) —
        # can cover a claim by being read. An unowned citation core cannot be marked
        # read, so it must NOT feed the read hooks (``core_claim_count`` /
        # ``unlocks_uncovered_claim``) — doing so would inflate read_priority for an
        # unacquirable paper. Such claims have no readable coverer: ``core_work_id``
        # is left None so they fall into the acquisition track (a permanent claim
        # gap in the cover) rather than the suggested reading list.
        owned = w is not None and w.source_graph is not None
        wid = w.id if owned else None
        core_work_id[key] = wid
        if owned:
            count[wid] = count.get(wid, 0) + 1
            # A key claim is "covered" once its core paper is fully read; reading
            # an unread core paper newly covers it, so that work "unlocks".
            if (w.coverage_human or "none").lower() != "full":
                unlocks[wid] = True

    for w in works:
        w.core_claim_count = count.get(w.id, 0)
        # Claims exist in scope → resolve the boolean hook (None is reserved for
        # the claim-free path, where this function returns early above).
        w.unlocks_uncovered_claim = bool(unlocks.get(w.id, False))

    return core_map, core_work_id

claim_path

claim_path(rc: ResolvedClaim, index: ClaimIndex, *, core: CorePaper | None = None) -> ClaimPath

Ordered reading path to a claim: prerequisite claims first, then itself.

Prerequisites come from two sources, unioned: the note's prerequisites frontmatter (resolved within the home graph, recursively — the index already holds the notes, so this reimplements :meth:get_prerequisites_chain over the index rather than holding live graph handles) and the claim-to-claim depends-on (outgoing → target is a prerequisite) / prerequisite-of (incoming → source is a prerequisite) edges. A depth-first post-order with a visiting set yields the dependency order and breaks any cycle safely.

Source code in zettelkasten/claims.py

def claim_path(
    rc: ResolvedClaim,
    index: ClaimIndex,
    *,
    core: CorePaper | None = None,
) -> ClaimPath:
    """Ordered reading path to a claim: prerequisite claims first, then itself.

    Prerequisites come from two sources, unioned: the note's ``prerequisites``
    frontmatter (resolved within the home graph, recursively — the index already
    holds the notes, so this reimplements :meth:`get_prerequisites_chain` over
    the index rather than holding live graph handles) and the claim-to-claim
    ``depends-on`` (outgoing → target is a prerequisite) / ``prerequisite-of``
    (incoming → source is a prerequisite) edges. A depth-first post-order with a
    visiting set yields the dependency order and breaks any cycle safely.
    """
    claims = index.claims
    notes_by_key = index.notes_by_key

    def _prereqs_of(key: Key) -> list[Key]:
        out: set[Key] = set()
        note = notes_by_key.get(key)
        if note is not None:
            for pid in note.prerequisites:
                pk = (key[0], pid)
                if pk in claims:
                    out.add(pk)
        for e in index.by_source.get(key, []):
            if e.relation == "depends-on" and e.dst_key in claims:
                out.add(e.dst_key)
        for e in index.by_target.get(key, []):
            if e.relation == "prerequisite-of" and e.src_key in claims:
                out.add(e.src_key)
        out.discard(key)
        return sorted(out)

    order: list[Key] = []
    done: set[Key] = set()
    visiting: set[Key] = set()

    def _visit(key: Key) -> None:
        if key in done or key in visiting:
            return  # cycle / already-placed → skip safely
        visiting.add(key)
        for pk in _prereqs_of(key):
            _visit(pk)
        visiting.discard(key)
        done.add(key)
        order.append(key)

    _visit(rc.key)
    steps = tuple(_claim_ref(index, k) for k in order)
    return ClaimPath(claim=rc.key, steps=steps, core_paper=core)

claim_relations

claim_relations(index: ClaimIndex) -> dict[str, Any]

Read the typed claim-to-claim structural edges as a dashboard adjacency.

READ ONLY — every edge already exists in the graph; nothing is proposed or inferred. nodes are the claims; edges are the structural relations whose both endpoints are claims, deduplicated and deterministically ordered.

Source code in zettelkasten/claims.py

def claim_relations(index: ClaimIndex) -> dict[str, Any]:
    """Read the typed claim-to-claim structural edges as a dashboard adjacency.

    READ ONLY — every edge already exists in the graph; nothing is proposed or
    inferred. ``nodes`` are the claims; ``edges`` are the structural relations
    whose both endpoints are claims, deduplicated and deterministically ordered.
    """
    claim_keys = set(index.claims)
    nodes = [_claim_ref(index, k) for k in sorted(claim_keys)]
    seen: set[tuple[str, str, str, str, str]] = set()
    edges: list[dict[str, Any]] = []
    for e in index.edges:
        if e.relation not in STRUCTURAL_RELATIONS:
            continue
        if e.src_key not in claim_keys or e.dst_key not in claim_keys:
            continue
        sig = (e.src_graph, e.src_id, e.relation, e.dst_graph, e.dst_id)
        if sig in seen:
            continue
        seen.add(sig)
        edges.append({
            "source_graph": e.src_graph,
            "source_id": e.src_id,
            "target_graph": e.dst_graph,
            "target_id": e.dst_id,
            "relation": e.relation,
        })
    edges.sort(key=lambda d: (d["source_graph"], d["source_id"], d["relation"],
                              d["target_graph"], d["target_id"]))
    return {"nodes": nodes, "edges": edges}

claim_anatomy

claim_anatomy(get_graph: GetGraph, *, claim_id: str, source_graph: str, project: str = '', graph: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None, _context: 'tuple | None' = None) -> dict[str, Any]

Assemble the full anatomy of a single claim.

Returns the claim (title/body/source locator), its evidence (supports/replicates quotes with page + grounding, strongest on top), its strongest counterclaim(s) with their own evidence, its caveats (qualifies), any supersedes edges, the derived status (established / contested / superseded), the bounded :func:claim_strength, and the :func:core_paper_atom + :func:claim_path. Returns {"error": ...} when the claim is not found in the scope.

_context lets :func:assemble_claims thread an already-built context in so a batch render does not rebuild the index per claim; external callers omit it and a fresh context is built.

Source code in zettelkasten/claims.py

def claim_anatomy(
    get_graph: GetGraph,
    *,
    claim_id: str,
    source_graph: str,
    project: str = "",
    graph: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
    _context: "tuple | None" = None,
) -> dict[str, Any]:
    """Assemble the full anatomy of a single claim.

    Returns the claim (title/body/source locator), its evidence
    (supports/replicates quotes with page + grounding, strongest on top), its
    strongest counterclaim(s) *with their own evidence*, its caveats
    (``qualifies``), any ``supersedes`` edges, the derived status
    (``established`` / ``contested`` / ``superseded``), the bounded
    :func:`claim_strength`, and the :func:`core_paper_atom` + :func:`claim_path`.
    Returns ``{"error": ...}`` when the claim is not found in the scope.

    ``_context`` lets :func:`assemble_claims` thread an already-built context in
    so a batch render does not rebuild the index per claim; external callers omit
    it and a fresh context is built.
    """
    if _context is None:
        ctx = _build_context(
            get_graph, project=project, graph=graph,
            graphs_dir=graphs_dir, namespace=namespace, localize=localize,
        )
    else:
        ctx = _context
    index, works, citations, importance_map, _centrality = ctx

    loc = localize or (lambda s: s)
    key = (loc(source_graph), claim_id)
    rc = index.resolved.get(key)
    if rc is None:
        return {
            "error": f"Claim '{claim_id}' not found in graph '{source_graph}'.",
            "type": "NotFound",
        }

    strength = claim_strength(rc, importance_map=importance_map)
    status = claim_status(rc)
    core = core_paper_atom(
        rc, index, works=works, citations=citations, importance_map=importance_map
    )
    path = claim_path(rc, index, core=core)

    evidence = [
        {**row, "contribution": _evidence_contribution(row, importance_map)}
        for row in rc.supporting_quotes
    ]
    evidence.sort(key=lambda r: (r["contribution"], r["id"]), reverse=True)

    counterclaims = _counterclaims(rc, index, importance_map)
    caveats = [_edge_ref(index, e, side="src") for e in rc.caveats]
    superseded_by = [_edge_ref(index, e, side="src") for e in rc.superseded_by]
    corroborations = [_edge_ref(index, e, side="src") for e in rc.corroborations]

    src = rc.note.source if isinstance(rc.note.source, dict) else {}
    return {
        "claim": {
            "id": rc.key[1],
            "graph": rc.key[0],
            "title": rc.note.title,
            "type": rc.note.type,
            "status": status,
            "body": rc.note.body,
            "source": {
                "book": src.get("book"),
                "chapter": src.get("chapter"),
                "page": src.get("page"),
            },
        },
        "status": status,
        "claim_strength": round(strength, 4),
        "support_span": len(rc.support_sources),
        "support_sources": list(rc.support_sources),
        "replication_count": rc.replication_count,
        "evidence": evidence,
        "corroborations": corroborations,
        "counterclaims": counterclaims,
        "caveats": caveats,
        "superseded_by": superseded_by,
        "core_paper": _core_paper_payload(core),
        "claim_path": {
            "steps": list(path.steps),
            "core_paper": _core_paper_payload(core),
        },
        "relations": [
            _edge_ref(index, e, side="dst") for e in rc.structural_out
        ] + [
            _edge_ref(index, e, side="src") for e in rc.structural_in
        ],
    }

assemble_claims

assemble_claims(get_graph: GetGraph, *, project: str = '', graph: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None, limit: int | None = None, theme_assignment: dict[Key, str] | None = None, top_per_theme: int | None = None, scope_name: str = '') -> dict[str, Any]

Assemble the claim-view payload: scope, counts, and ranked key claims.

The claim analogue of :func:papers.assemble_papers — pure given get_graph, scoped by a single graph or a project (or global). Each key claim carries its salience, strength, status, support span, and the core paper picked for it. Later waves wrap this in the MCP tool / routes.

Source code in zettelkasten/claims.py

def assemble_claims(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
    limit: int | None = None,
    theme_assignment: dict[Key, str] | None = None,
    top_per_theme: int | None = None,
    scope_name: str = "",
) -> dict[str, Any]:
    """Assemble the claim-view payload: scope, counts, and ranked key claims.

    The claim analogue of :func:`papers.assemble_papers` — pure given
    ``get_graph``, scoped by a single ``graph`` or a ``project`` (or global).
    Each key claim carries its salience, strength, status, support span, and the
    core paper picked for it. Later waves wrap this in the MCP tool / routes.
    """
    ctx = _build_context(
        get_graph, project=project, graph=graph,
        graphs_dir=graphs_dir, namespace=namespace, localize=localize,
    )
    index, works, citations, importance_map, centrality = ctx

    ranked = key_claims(
        index,
        importance_map=importance_map,
        centrality=centrality,
        theme_assignment=theme_assignment,
        top_per_theme=top_per_theme,
        limit=limit,
    )

    status_summary = {"established": 0, "contested": 0, "superseded": 0}
    for rc in index.resolved.values():
        status_summary[claim_status(rc)] += 1

    key_rows: list[dict[str, Any]] = []
    for kc in ranked:
        rc = index.resolved[kc.key]
        core = core_paper_atom(
            rc, index, works=works, citations=citations, importance_map=importance_map
        )
        key_rows.append({
            "id": kc.id,
            "graph": kc.graph,
            "title": kc.title,
            "type": kc.type,
            "salience": kc.salience,
            "strength": kc.strength,
            "status": kc.status,
            "support_span": kc.support_span,
            "oppose_span": kc.oppose_span,
            "contested": kc.contested,
            "theme": kc.theme,
            "core_paper": _core_paper_payload(core),
        })

    name = scope_name or graph or project or "all"
    return {
        "scope": {"type": "graph" if graph else "project", "name": name},
        "claim_count": len(index.claims),
        "source_count": len(index.real_sources),
        "status_summary": status_summary,
        "key_claims": key_rows,
    }

analyze_gaps

analyze_gaps(get_graph: GetGraph, *, project: str = '', graph: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None, limit: int | None = None, gap_types: Iterable[str] | None = None, per_type_limit: bool = False, _context: 'tuple | None' = None) -> dict[str, Any]

Typed, ranked research-agenda scan over the claim graph (supersedes find_gaps).

Pure and LLM-free. Builds the claim index + paper signals (or reuses a threaded _context) and emits typed gaps, each scored by a bounded severity (salience × type-weight × actionability) and wired to the producer-loop tool that would close it. The gap TYPES:

• evidential — a thin claim (few distinct supporting sources / low bounded strength) or a replication gap → expand_citations / claim('test').
• dialectical — an unresolved debate (a contested, non-superseded claim) or an unarticulated counterclaim (an outgoing contradicts whose target is not itself an active claim) → claim('test').
• structural — a foundational claim (many dependents) that is thin, a bridge claim (high centrality), or a frontier claim (no structural neighbours) → claim('test') / expand_citations.
• coverage — a thinly-developed landscape hub (few surveys/cites) → claim('propose', theme=hub).
• comprehensiveness — a high-value unread (co-cited) citation not promoted to a local source → promote_citation.
• temporal — a stale claim whose newest supporting paper lags the citation frontier → expand_citations (forward).

Plus six NEGATIVE-SPACE "opportunity" families (implemented in the leaf module :mod:zettelkasten.opportunities, all read-only / propose-only) that scan the MISSING edges and unmatched nodes rather than a weakness on an existing claim. They are OFF by default: the lazy import + scan fire ONLY when gap_types explicitly requests one of them, so the default report is unchanged.

• asymmetry — a practice memory-node with no aligned canon claim, or a canon claim with no aligned practice node (read off the cross-store overlay).
• bridge — two dense clusters with (near-)zero connecting edges but high inter-cluster embedding similarity (a synthesis opportunity).
• crux — an unresolved central debate (a contested camp from the debate map, ranked by strength × centrality).
• orphaned_question — a question note with no incoming answering edge.
• void — a sparse-but-surrounded region of embedding space.
• transfer — a method/model note that structurally fits an untried adjacent cluster.

Scope is a single graph, a project, or everything. gap_types filters the emitted types; limit caps the ranked output. Returns {scope, gaps, summary, message} with gaps sorted by severity desc.

Per-type budgeting auto-enables (no caller flag required) whenever the REQUESTED gap_types include a negative-space family: each such family is then capped to limit rows INDEPENDENTLY, while the claim-anchored families keep sharing the SINGLE global top-N slice. This exists because some negative-space families are unbounded (find_asymmetry emits O(#claims) rows), so co-requesting them would otherwise let a large family bury a rare one (a lone contested crux) below the global cut. Centralizing the decision here means correctness no longer depends on any caller remembering to pass a flag.

per_type_limit (opt-in, off by default) is retained as a backward- compatible explicit override that forces the same per-type budgeting on; it is now redundant with the auto-enable above (callers requesting a negative-space family get per-type budgeting either way). The budgeting only ever applies to the negative-space families — the claim-anchored families always share the one global top-N slice, so a MIXED request never changes their ranking/count. The DEFAULT scan (gap_types=None) requests only claim-anchored families, so no per-type logic runs and the single global slice stays byte-for-byte unchanged.

Source code in zettelkasten/claims.py

def analyze_gaps(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
    limit: int | None = None,
    gap_types: Iterable[str] | None = None,
    per_type_limit: bool = False,
    _context: "tuple | None" = None,
) -> dict[str, Any]:
    """Typed, ranked research-agenda scan over the claim graph (supersedes find_gaps).

    Pure and LLM-free. Builds the claim index + paper signals (or reuses a
    threaded ``_context``) and emits typed gaps, each scored by a bounded
    severity (``salience × type-weight × actionability``) and wired to the
    producer-loop tool that would close it. The gap TYPES:

    * ``evidential`` — a thin claim (few distinct supporting sources / low
      bounded strength) or a replication gap → ``expand_citations`` /
      ``claim('test')``.
    * ``dialectical`` — an unresolved debate (a contested, non-superseded claim)
      or an unarticulated counterclaim (an outgoing ``contradicts`` whose target
      is not itself an active claim) → ``claim('test')``.
    * ``structural`` — a foundational claim (many dependents) that is thin, a
      bridge claim (high centrality), or a frontier claim (no structural
      neighbours) → ``claim('test')`` / ``expand_citations``.
    * ``coverage`` — a thinly-developed landscape hub (few ``surveys``/``cites``)
      → ``claim('propose', theme=hub)``.
    * ``comprehensiveness`` — a high-value unread (co-cited) citation not promoted
      to a local source → ``promote_citation``.
    * ``temporal`` — a stale claim whose newest supporting paper lags the citation
      frontier → ``expand_citations`` (forward).

    Plus six NEGATIVE-SPACE "opportunity" families (implemented in the leaf module
    :mod:`zettelkasten.opportunities`, all read-only / propose-only) that scan the
    MISSING edges and unmatched nodes rather than a weakness on an existing claim.
    They are OFF by default: the lazy import + scan fire ONLY when ``gap_types``
    explicitly requests one of them, so the default report is unchanged.

    * ``asymmetry`` — a practice memory-node with no aligned canon claim, or a
      canon claim with no aligned practice node (read off the cross-store overlay).
    * ``bridge`` — two dense clusters with (near-)zero connecting edges but high
      inter-cluster embedding similarity (a synthesis opportunity).
    * ``crux`` — an unresolved central debate (a contested camp from the debate
      map, ranked by strength × centrality).
    * ``orphaned_question`` — a ``question`` note with no incoming answering edge.
    * ``void`` — a sparse-but-surrounded region of embedding space.
    * ``transfer`` — a method/model note that structurally fits an untried
      adjacent cluster.

    Scope is a single ``graph``, a ``project``, or everything. ``gap_types``
    filters the emitted types; ``limit`` caps the ranked output. Returns
    ``{scope, gaps, summary, message}`` with ``gaps`` sorted by severity desc.

    Per-type budgeting auto-enables (no caller flag required) whenever the
    REQUESTED ``gap_types`` include a negative-space family: each such family is
    then capped to ``limit`` rows INDEPENDENTLY, while the claim-anchored families
    keep sharing the SINGLE global top-N slice. This exists because some
    negative-space families are unbounded (``find_asymmetry`` emits O(#claims)
    rows), so co-requesting them would otherwise let a large family bury a rare
    one (a lone contested ``crux``) below the global cut. Centralizing the
    decision here means correctness no longer depends on any caller remembering to
    pass a flag.

    ``per_type_limit`` (opt-in, off by default) is retained as a backward-
    compatible explicit override that forces the same per-type budgeting on; it is
    now redundant with the auto-enable above (callers requesting a negative-space
    family get per-type budgeting either way). The budgeting only ever applies to
    the negative-space families — the claim-anchored families always share the one
    global top-N slice, so a MIXED request never changes their ranking/count. The
    DEFAULT scan (``gap_types=None``) requests only claim-anchored families, so no
    per-type logic runs and the single global slice stays byte-for-byte unchanged.
    """
    if _context is None:
        ctx = _build_context(
            get_graph, project=project, graph=graph,
            graphs_dir=graphs_dir, namespace=namespace, localize=localize,
        )
    else:
        ctx = _context
    index, works, citations, importance_map, centrality = ctx

    wanted = set(gap_types) if gap_types is not None else None

    def _want(t: str) -> bool:
        return wanted is None or t in wanted

    ranked = key_claims(index, importance_map=importance_map, centrality=centrality)
    kc_by_key = {kc.key: kc for kc in ranked}
    max_sal = max((kc.salience for kc in ranked), default=0.0) or 1.0

    src_year = {w.source_graph: w.year for w in works if w.source_graph}
    frontier_year = max((w.year for w in works if w.year is not None), default=None)
    real_sources = index.real_sources

    gaps: list[dict[str, Any]] = []

    # ── claim-anchored gaps (evidential / dialectical / structural / temporal) ──
    for key, rc in index.resolved.items():
        kc = kc_by_key.get(key)
        norm_sal = (kc.salience / max_sal) if kc else 0.0
        ref = _claim_ref(index, key)
        title = rc.note.title or key[1]
        status = claim_status(rc)
        strength = claim_strength(rc, importance_map=importance_map)
        core = core_paper_atom(
            rc, index, works=works, citations=citations, importance_map=importance_map
        )

        def _strengthen_action() -> tuple[str, dict[str, Any]]:
            """Prefer acquiring corroboration when the claim owns a source paper."""
            if core is not None and core.source_graph:
                return "expand_citations", {
                    "tool": "expand_citations",
                    "args": {"source": core.source_graph, "direction": "both"},
                }
            if core is not None and core.unowned:
                return "promote_citation", {
                    "tool": "promote_citation", "args": {"citation_id": core.paper_id},
                }
            return "test", {"tool": "claim", "action": "test", "args": {"text": title}}

        # evidential: thin support or low bounded strength (or a replication gap).
        if _want("evidential"):
            n_src = len(rc.support_sources)
            thin = n_src < _GAP_EVIDENTIAL_MIN_SOURCES
            weak = strength < _GAP_EVIDENTIAL_STRENGTH_FLOOR
            replication_gap = (
                n_src >= _GAP_EVIDENTIAL_MIN_SOURCES and rc.replication_count == 0
            )
            if thin or weak or replication_gap:
                tool, act = _strengthen_action()
                if replication_gap and not (thin or weak):
                    pred = "replication gap (multi-source but never replicated)"
                else:
                    pred = "thin claim (sparse evidence / low strength)"
                gaps.append(_gap(
                    "evidential", salience=norm_sal, action_tool=tool, action=act,
                    predicate=pred,
                    detail=(f"support_span={n_src}, strength={round(strength, 3)}, "
                            f"replications={rc.replication_count}"),
                    anchor=ref,
                ))

        # dialectical: an unresolved live debate, or an unarticulated counterclaim.
        if _want("dialectical"):
            if status == "contested":
                gaps.append(_gap(
                    "dialectical", salience=norm_sal, action_tool="test",
                    action={"tool": "claim", "action": "test", "args": {"text": title}},
                    predicate="unresolved debate (contested, not superseded)",
                    detail=(f"counter_in={len(rc.counter_in)}, "
                            f"caveats={len(rc.caveats)}"),
                    anchor=ref,
                ))
            dangling = [e for e in rc.counter_out if e.dst_key not in index.claims]
            if dangling:
                tgt = dangling[0]
                gaps.append(_gap(
                    "dialectical", salience=norm_sal, action_tool="test",
                    action={"tool": "claim", "action": "test",
                            "args": {"text": f"Counterclaim to: {title}"}},
                    predicate="unarticulated counterclaim (opposes a non-claim target)",
                    detail=(f"{tgt.relation} → {tgt.dst_graph}/{tgt.dst_id} "
                            "is not an active claim"),
                    anchor=ref,
                ))

        # structural: foundational (load-bearing & thin), bridge (central), or
        # frontier (no structural neighbours — the edge of the articulated graph).
        if _want("structural"):
            cen = centrality.get(key, 0.0)
            dependents = len(rc.structural_in)
            has_structure = bool(rc.structural_in or rc.structural_out)
            if dependents >= _GAP_FOUNDATIONAL_MIN_DEPENDENTS and strength < 0.5:
                gaps.append(_gap(
                    "structural", salience=norm_sal, action_tool="test",
                    action={"tool": "claim", "action": "test", "args": {"text": title}},
                    predicate="foundational claim is under-supported",
                    detail=f"dependents={dependents}, strength={round(strength, 3)}",
                    anchor=ref,
                ))
            elif cen >= _GAP_BRIDGE_CENTRALITY:
                gaps.append(_gap(
                    "structural", salience=norm_sal, action_tool="test",
                    action={"tool": "claim", "action": "test", "args": {"text": title}},
                    predicate="bridge claim (high centrality)",
                    detail=f"centrality={round(cen, 3)}",
                    anchor=ref,
                ))
            elif not has_structure:
                tool, act = _strengthen_action()
                gaps.append(_gap(
                    "structural", salience=norm_sal * 0.5, action_tool=tool, action=act,
                    predicate="frontier claim (no structural relations)",
                    detail="no depends-on / prerequisite / structural edges",
                    anchor=ref,
                ))

        # temporal: the claim's newest supporting paper lags the citation frontier.
        if _want("temporal") and frontier_year is not None:
            cand_years = [src_year.get(s) for s in rc.support_sources]
            cand_years.append(src_year.get(rc.home_graph))
            years = [y for y in cand_years if y is not None]
            if years:
                newest = max(years)
                age = frontier_year - newest
                if age >= _GAP_STALE_YEARS:
                    src = (core.source_graph if core and core.source_graph
                           else (rc.home_graph if rc.home_graph in real_sources else ""))
                    if src:
                        act = {"tool": "expand_citations",
                               "args": {"source": src, "direction": "forward"}}
                    else:
                        act = {"tool": "claim", "action": "test", "args": {"text": title}}
                    gaps.append(_gap(
                        "temporal",
                        salience=norm_sal * _saturate(age, float(_GAP_STALE_YEARS)),
                        action_tool=("expand_citations" if src else "test"),
                        action=act,
                        predicate="stale claim (evidence lags the citation frontier)",
                        detail=f"newest supporting year={newest}, frontier={frontier_year}",
                        anchor=ref,
                    ))

    # ── coverage: thinly-developed landscape hubs ──────────────────────────────
    if _want("coverage"):
        for nkey, note in index.notes_by_key.items():
            if (note.type or "") != "concept" or "landscape" not in (note.tags or []):
                continue
            outgoing = sum(
                1 for l in note.links if l.relation in ("surveys", "cites")
            )
            if outgoing < _GAP_THIN_HUB_LINKS:
                sal = max(0.0, (_GAP_THIN_HUB_LINKS - outgoing) / _GAP_THIN_HUB_LINKS)
                gaps.append(_gap(
                    "coverage", salience=sal, action_tool="propose",
                    action={"tool": "claim", "action": "propose",
                            "args": {"theme": nkey[1]}},
                    predicate="thinly-developed landscape hub",
                    detail=f"{outgoing} surveyed/cited work(s) (< {_GAP_THIN_HUB_LINKS})",
                    anchor=_claim_ref(index, nkey),
                ))

    # ── comprehensiveness: high-value unread (co-cited) works to acquire ───────
    if _want("comprehensiveness"):
        for cid, meta in citations.items():
            if cid in real_sources:
                continue
            cited_by = int(meta.get("cited_by_count") or 0)
            if cited_by < _GAP_MIN_UNREAD_CITED_BY:
                continue
            gaps.append(_gap(
                "comprehensiveness",
                salience=_saturate(cited_by, 20.0),
                action_tool="promote_citation",
                action={"tool": "promote_citation", "args": {"citation_id": cid}},
                predicate="high-value unread work (not promoted to a source)",
                detail=f"cited_by_count={cited_by}",
                anchor={"id": cid, "graph": "_citations",
                        "title": str(meta.get("title") or cid), "type": "citation"},
            ))

    # ── negative-space "opportunity" families (asymmetry / bridge / crux /
    #    orphaned_question / void / transfer) ─────────────────────────────────────
    # These scan MISSING edges + unmatched nodes across the graph and the
    # memory<->zettelkasten boundary. They live in the leaf module
    # :mod:`zettelkasten.opportunities` (which imports ``synapse``); ``claims`` must
    # NOT import ``synapse`` at load time (``synapse`` imports ``claims``), so the
    # import is LAZY and fires ONLY when one of the six families is actually
    # requested — the default scan (``gap_types=None``) is byte-for-byte unchanged.
    if wanted is not None and (wanted & _NEGATIVE_SPACE_GAP_TYPES):
        from zettelkasten import opportunities
        gaps.extend(opportunities.detect_gaps(
            index=index, ctx=ctx, get_graph=get_graph,
            centrality=centrality, importance_map=importance_map,
            kc_by_key=kc_by_key, max_sal=max_sal,
            wanted=(wanted & _NEGATIVE_SPACE_GAP_TYPES),
            localize=localize, namespace=namespace,
            graphs_dir=(graphs_dir or GRAPHS_DIR),
            project=project, graph=graph,
        ))

    # Auto-enable per-type budgeting for whichever REQUESTED families are
    # negative-space, so correctness never depends on a caller passing the flag;
    # ``per_type_limit`` stays as a backward-compatible explicit override.
    per_type = per_type_limit or (
        wanted is not None and not _NEGATIVE_SPACE_GAP_TYPES.isdisjoint(wanted)
    )

    if per_type and limit is not None and limit >= 0:
        # PER-TYPE budgeting scoped to the negative-space "opportunity" families
        # ONLY: slice EACH such family to its own ``limit`` rather than competing
        # in one global top-N, so no family can crowd another out of the report.
        # Some are unbounded — ``opportunities.find_asymmetry`` emits one gap per
        # unaligned canon claim (O(#claims)) — and would otherwise fill the global
        # slice and bury a rarer family (a lone contested ``crux``). The
        # claim-anchored families are NOT split: they share the ONE global top-N
        # slice, exactly as the default path, so a MIXED request never changes
        # their ranking/count. Everything is merged and re-sorted into one
        # severity-ordered list below.
        by_type: dict[str, list[dict[str, Any]]] = {}
        claim_anchored: list[dict[str, Any]] = []
        for g in gaps:
            if g["type"] in _NEGATIVE_SPACE_GAP_TYPES:
                by_type.setdefault(g["type"], []).append(g)
            else:
                claim_anchored.append(g)
        kept: list[dict[str, Any]] = []
        for group in by_type.values():
            group.sort(
                key=lambda g: (g["severity"], g["type"], g["anchor"].get("id", "")),
                reverse=True,
            )
            kept.extend(group[:limit])
        # Claim-anchored families share the single global top-N slice (unchanged).
        claim_anchored.sort(
            key=lambda g: (g["severity"], g["type"], g["anchor"].get("id", "")),
            reverse=True,
        )
        kept.extend(claim_anchored[:limit])
        gaps = kept

    gaps.sort(key=lambda g: (g["severity"], g["type"], g["anchor"].get("id", "")),
              reverse=True)
    if not per_type and limit is not None and limit >= 0:
        gaps = gaps[:limit]

    summary: dict[str, int] = {}
    for g in gaps:
        summary[g["type"]] = summary.get(g["type"], 0) + 1

    name = graph or project or "all"
    return {
        "scope": {"type": "graph" if graph else "project", "name": name},
        "gaps": gaps,
        "summary": summary,
        "message": (f"Found {len(gaps)} ranked gap(s) across {len(index.claims)} "
                    f"claim(s) in {len(index.real_sources)} source(s)."),
    }

rank_missing_papers

rank_missing_papers(get_graph: GetGraph, *, project: str = '', graph: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None, limit: int | None = None, _context: 'tuple | None' = None) -> dict[str, Any]

Rank cited-but-unowned works as acquisition targets (the M4 acquisition companion to :func:analyze_gaps).

Where :func:analyze_gaps names what's missing in the articulated graph, this names which works to acquire to close it. A target is any cited-but-unowned work in the corpus — a citation with no writable local source folder (Work.source_graph is None). Each is scored on a bounded 0–1 composite (weights sum to 1):

• claim-unlock (leads) — the salience-weighted key claims this work is the CORE paper of. Such a claim is uncovered by construction: an unowned core cannot be read, so acquiring the work is the only way to cover it (this is the unowned twin of :attr:Work.unlocks_uncovered_claim, which the read layer sets only for OWNED cores). Resolved via :func:core_paper_atom (unowned cores), salience via :func:key_claims.
• influence — the work's corpus importance (graph-derived), so a well-connected citation outranks an isolated one at equal unlock mass.
• citation mass — external cited_by_count (saturating), the same acquisition signal the comprehensiveness gap uses.

Pure and LLM-free; reuses the threaded _context when given (so a caller that already built the index/corpus does not rebuild it). Degenerate scopes are no-ops, not errors: with no claims every unlock term is 0 and targets fall back to influence/citation order; with no owned sources every work is a target. Returns {scope, targets, summary, message} sorted by score desc, each target wired to the promote_citation acquisition action.

Source code in zettelkasten/claims.py

def rank_missing_papers(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
    limit: int | None = None,
    _context: "tuple | None" = None,
) -> dict[str, Any]:
    """Rank cited-but-unowned works as acquisition targets (the M4 acquisition
    companion to :func:`analyze_gaps`).

    Where :func:`analyze_gaps` names what's *missing* in the articulated graph,
    this names which *works to acquire* to close it. A target is any
    cited-but-unowned work in the corpus — a citation with no writable local
    source folder (``Work.source_graph is None``). Each is scored on a bounded
    0–1 composite (weights sum to 1):

    * **claim-unlock** (leads) — the salience-weighted key claims this work is
      the CORE paper of. Such a claim is *uncovered by construction*: an unowned
      core cannot be read, so acquiring the work is the only way to cover it
      (this is the unowned twin of :attr:`Work.unlocks_uncovered_claim`, which
      the read layer sets only for OWNED cores). Resolved via
      :func:`core_paper_atom` (``unowned`` cores), salience via :func:`key_claims`.
    * **influence** — the work's corpus importance (graph-derived), so a
      well-connected citation outranks an isolated one at equal unlock mass.
    * **citation mass** — external ``cited_by_count`` (saturating), the same
      acquisition signal the ``comprehensiveness`` gap uses.

    Pure and LLM-free; reuses the threaded ``_context`` when given (so a caller
    that already built the index/corpus does not rebuild it). Degenerate scopes
    are no-ops, not errors: with no claims every unlock term is 0 and targets
    fall back to influence/citation order; with no owned sources every work is a
    target. Returns ``{scope, targets, summary, message}`` sorted by score desc,
    each target wired to the ``promote_citation`` acquisition action.
    """
    if _context is None:
        ctx = _build_context(
            get_graph, project=project, graph=graph,
            graphs_dir=graphs_dir, namespace=namespace, localize=localize,
        )
    else:
        ctx = _context
    index, works, citations, importance_map, centrality = ctx

    name = graph or project or "all"
    scope = {"type": "graph" if graph else "project", "name": name}

    # Candidate universe: cited-but-unowned works (no writable source folder).
    unowned = [w for w in works if w.source_graph is None]
    if not unowned:
        return {
            "scope": scope,
            "targets": [],
            "summary": {"unowned_total": 0, "unlocking": 0, "returned": 0},
            "message": "No cited-but-unowned works in scope.",
        }

    # Which unowned works are the CORE paper of a key claim (→ uncovered claim it
    # would unlock), with the claim's normalized salience as the unlock weight.
    ranked = key_claims(index, importance_map=importance_map, centrality=centrality)
    max_sal = max((kc.salience for kc in ranked), default=0.0) or 1.0
    unlocks: dict[str, list[dict[str, Any]]] = {}
    for kc in ranked:
        rc = index.resolved.get(kc.key)
        if rc is None:
            continue
        core = core_paper_atom(
            rc, index, works=works, citations=citations, importance_map=importance_map
        )
        if core is None or not core.unowned:
            continue
        unlocks.setdefault(core.paper_id, []).append({
            "uid": claim_uid(kc.key),
            "id": kc.id,
            "graph": kc.graph,
            "title": kc.title,
            "salience": round(kc.salience / max_sal, 4),
        })

    targets: list[dict[str, Any]] = []
    for w in unowned:
        meta = citations.get(w.id, {})
        cited_by = int(meta.get("cited_by_count") or 0)
        claim_rows = sorted(
            unlocks.get(w.id, []), key=lambda r: (-r["salience"], r["uid"])
        )
        unlock_mass = sum(r["salience"] for r in claim_rows)
        unlock_term = _saturate(unlock_mass, 1.0)
        influence = importance_map.get(w.id, 0.0)
        cited_term = _saturate(float(cited_by), _MISSING_CITED_HALF)
        score = (
            _MISSING_UNLOCK_WEIGHT * unlock_term
            + _MISSING_INFLUENCE_WEIGHT * influence
            + _MISSING_CITED_WEIGHT * cited_term
        )

        reason_bits: list[str] = []
        if claim_rows:
            reason_bits.append(f"unlocks {len(claim_rows)} otherwise-uncovered claim(s)")
        if influence > 0:
            reason_bits.append("influential in the corpus")
        if cited_by:
            reason_bits.append(f"{cited_by} external citation(s)")
        reason = ", ".join(reason_bits) or "cited but not owned locally"

        targets.append({
            "id": w.id,
            "title": w.title or str(meta.get("title") or w.id),
            "year": w.year,
            "score": round(score, 6),
            "influence": round(influence, 4),
            "cited_by_count": cited_by,
            "unlocks_count": len(claim_rows),
            "unlocks": claim_rows,
            "reason": reason,
            "action": _missing_paper_action(w.id, meta),
        })

    targets.sort(
        key=lambda t: (t["score"], t["unlocks_count"], t["cited_by_count"], t["id"]),
        reverse=True,
    )
    unlocking = sum(1 for t in targets if t["unlocks_count"] > 0)
    if limit is not None and limit >= 0:
        targets = targets[:limit]

    return {
        "scope": scope,
        "targets": targets,
        "summary": {
            "unowned_total": len(unowned),
            "unlocking": unlocking,
            "returned": len(targets),
        },
        "message": (
            f"Ranked {len(targets)} cited-but-unowned acquisition target(s) "
            f"across {len(index.claims)} claim(s) in {len(index.real_sources)} source(s)."
        ),
    }

discover_contradictions

discover_contradictions(get_graph: GetGraph, *, project: str = '', graph: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None, classify_fn: 'Callable[[str, str], str] | None' = None, embed_fn: 'Callable[[str, str], Any] | None' = None, cluster_threshold: float = DISCOVER_CLUSTER_THRESHOLD, max_k: int = DISCOVER_MAX_K, max_pairs_per_cluster: int = DISCOVER_MAX_PAIRS_PER_CLUSTER, max_pairs: int = DISCOVER_MAX_PAIRS, band_low: float = DISCOVER_BAND_LOW, near_duplicate_floor: float = DISCOVER_NEAR_DUP_FLOOR, confidence_floor: float = DISCOVER_CONFIDENCE_FLOOR, require_symmetry: bool = DISCOVER_REQUIRE_SYMMETRY, _context: 'tuple | None' = None) -> dict[str, Any]

Discover candidate contradictions between claims — PROPOSE-ONLY, precision-first.

Pure and write-free. Candidate recall clusters the claim/finding note vectors (:func:embeddings.propose_clusters); within each cluster candidate pairs are formed and severity-pre-ranked (by the two claims' :func:key_claims salience plus a cheap lexical-polarity nudge) so the bounded classifier budget goes to the most salient pairs first. THREE deterministic guards run BEFORE any LLM call: a SUBJECT-OVERLAP gate (shared concept tag / linked concept / cited work), a NEAR-DUPLICATE exclusion (cosine >= near_duplicate_floor is a paraphrase, not a contradiction), and a similarity BAND (band_low <= cosine < near_duplicate_floor — contradictions live related-but-not-identical). The survivors are handed to an INJECTABLE, write-free classify_fn whose contract restricts the relation to contradicts/qualifies/responds-to/none and says "answer none unless the two claims make OPPOSING assertions about the same thing". A pair is emitted only when classified contradicts with confidence

= confidence_floor; require_symmetry (default ON) additionally demands the reverse-order (A<->B) classification agree. A :class:synapse.typing_llm.TypingBackendError from the classifier FAST-ABORTS (the whole backend is dead — do not retry per pair).

classify_fn / embed_fn are injected by tests so no LLM/embedder runs; the defaults are the lazy write-free agent and the box's EmbeddingIndex. Returns {scope, proposals, summary, message} and WRITES NOTHING — each proposal carries {pair, relation, confidence, rationale, shared_subject, similarity, evidence, promote} and the promote descriptor wires a confirmed edge through the gated claim write path.

Source code in zettelkasten/claims.py

def discover_contradictions(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
    classify_fn: "Callable[[str, str], str] | None" = None,
    embed_fn: "Callable[[str, str], Any] | None" = None,
    cluster_threshold: float = DISCOVER_CLUSTER_THRESHOLD,
    max_k: int = DISCOVER_MAX_K,
    max_pairs_per_cluster: int = DISCOVER_MAX_PAIRS_PER_CLUSTER,
    max_pairs: int = DISCOVER_MAX_PAIRS,
    band_low: float = DISCOVER_BAND_LOW,
    near_duplicate_floor: float = DISCOVER_NEAR_DUP_FLOOR,
    confidence_floor: float = DISCOVER_CONFIDENCE_FLOOR,
    require_symmetry: bool = DISCOVER_REQUIRE_SYMMETRY,
    _context: "tuple | None" = None,
) -> dict[str, Any]:
    """Discover candidate contradictions between claims — PROPOSE-ONLY, precision-first.

    Pure and write-free. Candidate recall clusters the claim/finding note vectors
    (:func:`embeddings.propose_clusters`); within each cluster candidate pairs are
    formed and severity-pre-ranked (by the two claims' :func:`key_claims` salience
    plus a cheap lexical-polarity nudge) so the bounded classifier budget goes to
    the most salient pairs first. THREE deterministic guards run BEFORE any LLM
    call: a SUBJECT-OVERLAP gate (shared concept tag / linked concept / cited
    work), a NEAR-DUPLICATE exclusion (cosine >= ``near_duplicate_floor`` is a
    paraphrase, not a contradiction), and a similarity BAND (``band_low`` <= cosine
    < ``near_duplicate_floor`` — contradictions live related-but-not-identical).
    The survivors are handed to an INJECTABLE, write-free ``classify_fn`` whose
    contract restricts the relation to contradicts/qualifies/responds-to/none and
    says "answer none unless the two claims make OPPOSING assertions about the same
    thing". A pair is emitted only when classified ``contradicts`` with confidence
    >= ``confidence_floor``; ``require_symmetry`` (default ON) additionally demands
    the reverse-order (A<->B) classification agree. A
    :class:`synapse.typing_llm.TypingBackendError` from the classifier FAST-ABORTS
    (the whole backend is dead — do not retry per pair).

    ``classify_fn`` / ``embed_fn`` are injected by tests so no LLM/embedder runs;
    the defaults are the lazy write-free agent and the box's ``EmbeddingIndex``.
    Returns ``{scope, proposals, summary, message}`` and WRITES NOTHING — each
    proposal carries ``{pair, relation, confidence, rationale, shared_subject,
    similarity, evidence, promote}`` and the ``promote`` descriptor wires a
    confirmed edge through the gated ``claim`` write path.
    """
    from itertools import combinations

    from zettelkasten.embeddings import propose_clusters
    from zettelkasten.synapse.typing_llm import TypingBackendError

    if _context is None:
        ctx = _build_context(
            get_graph, project=project, graph=graph,
            graphs_dir=graphs_dir, namespace=namespace, localize=localize,
        )
    else:
        ctx = _context
    index, works, citations, importance_map, centrality = ctx

    name = graph or project or "all"
    scope = {"type": "graph" if graph else "project", "name": name}

    # ── pool: active claim/finding notes (index.claims already excludes drafts) ──
    pool = {k: rc for k, rc in index.resolved.items()
            if (rc.note.type or "") not in _DISCOVER_EXCLUDED_TYPES}

    # ── candidate recall: cluster the claim vectors (missing vecs → unclustered) ──
    get_vec = _discover_vector_provider(get_graph, embed_fn)
    vectors: dict[str, list[float] | None] = {}
    vec_by_uid: dict[str, Any] = {}
    uid_to_key: dict[str, Key] = {}
    for key in pool:
        uid = claim_uid(key)
        uid_to_key[uid] = key
        vec = get_vec(key[0], key[1])
        vectors[uid] = vec if _guarded_cosine(vec, vec) is not None else None
        vec_by_uid[uid] = vec

    clustered = propose_clusters(vectors, threshold=cluster_threshold, max_k=max_k)

    # Salience for the severity pre-rank (normalized to the scope's max).
    ranked = key_claims(index, importance_map=importance_map, centrality=centrality)
    max_sal = max((kc.salience for kc in ranked), default=0.0) or 1.0
    sal_by_key = {kc.key: (kc.salience / max_sal) for kc in ranked}

    # ── form + guard candidate pairs within each cluster ────────────────────────
    candidates: list[dict[str, Any]] = []
    for cluster in clustered["clusters"]:
        cluster_pairs: list[dict[str, Any]] = []
        for uid_a, uid_b in combinations(sorted(cluster), 2):
            key_a, key_b = uid_to_key[uid_a], uid_to_key[uid_b]
            sim = _guarded_cosine(vec_by_uid.get(uid_a), vec_by_uid.get(uid_b))
            # similarity BAND + near-duplicate exclusion (deterministic guards).
            if sim is None or sim < band_low or sim >= near_duplicate_floor:
                continue
            # SUBJECT-OVERLAP gate: require a shared concept/entity/cite signal.
            shared = _subject_signals(index, key_a) & _subject_signals(index, key_b)
            if not shared:
                continue
            a_text = _claim_text(pool[key_a].note)
            b_text = _claim_text(pool[key_b].note)
            severity = (sal_by_key.get(key_a, 0.0) + sal_by_key.get(key_b, 0.0)) * (
                1.0 + _polarity_bonus(a_text, b_text)
            )
            cluster_pairs.append({
                "key_a": key_a, "key_b": key_b,
                "a_text": a_text, "b_text": b_text,
                "similarity": round(sim, 6),
                "shared_subject": sorted(shared),
                "severity": round(severity, 6),
            })
        cluster_pairs.sort(key=lambda p: (p["severity"], p["similarity"]), reverse=True)
        candidates.extend(cluster_pairs[: max(0, max_pairs_per_cluster)])

    # Global severity pre-rank + cap, so the LLM budget hits the most salient pairs.
    candidates.sort(
        key=lambda p: (p["severity"], p["similarity"], claim_uid(p["key_a"]), claim_uid(p["key_b"])),
        reverse=True,
    )
    candidates = candidates[: max(0, max_pairs)]

    # ── conservative stance classification (injectable, write-free) ─────────────
    # The forward/reverse gate is the SHARED classifier (:func:`stance.
    # classify_contradiction`) — the same one the cross-store claim-aligned synapse
    # build uses, so there is no second fork to keep in sync.
    fn = classify_fn or stance.default_stance_classify_fn
    proposals: list[dict[str, Any]] = []
    considered = 0
    aborted = False
    for cand in candidates:
        considered += 1
        try:
            verdict = stance.classify_contradiction(
                cand["a_text"], cand["b_text"], fn,
                confidence_floor=confidence_floor, require_symmetry=require_symmetry,
            )
        except TypingBackendError:
            # The whole classification backend is dead — stop hammering it.
            aborted = True
            break
        if verdict is None:
            continue
        confidence = verdict["confidence"]
        rationale = verdict["rationale"]

        key_a, key_b = cand["key_a"], cand["key_b"]
        proposals.append({
            "pair": [_claim_ref(index, key_a), _claim_ref(index, key_b)],
            "relation": "contradicts",
            "confidence": round(confidence, 4),
            "rationale": rationale,
            "shared_subject": cand["shared_subject"],
            "similarity": cand["similarity"],
            "evidence": {
                claim_uid(key_a): _pair_evidence(index.resolved.get(key_a)),
                claim_uid(key_b): _pair_evidence(index.resolved.get(key_b)),
            },
            # A confirmed contradiction is authored through the gated producer as a
            # ``contradicts`` stance edge (propose-only; honors ZK_PROPOSE_ONLY).
            "promote": {
                "tool": "claim",
                "action": "create",
                "args": {
                    "title": pool[key_a].note.title or key_a[1],
                    "contradicts": [{"id": key_b[1], "graph": key_b[0], "confidence": round(confidence, 4)}],
                },
                "note": ("Promotion is the only write path; it runs through the gated "
                         "claim producer (propose-only). discover_contradictions writes nothing."),
            },
        })

    proposals.sort(key=lambda p: (p["confidence"], p["similarity"]), reverse=True)
    return {
        "scope": scope,
        "proposals": proposals,
        "summary": {
            "claims": len(pool),
            "candidate_pairs": len(candidates),
            "considered": considered,
            "proposed": len(proposals),
            "aborted": aborted,
        },
        "message": (
            f"Proposed {len(proposals)} candidate contradiction(s) from "
            f"{len(candidates)} guarded pair(s) across {len(pool)} claim(s)."
            + (" Classification aborted early (backend unavailable)." if aborted else "")
        ),
    }

debate_map

debate_map(get_graph: GetGraph, *, project: str = '', graph: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None, cross_store_contradictions: 'list[dict[str, Any]] | None' = None, _context: 'tuple | None' = None) -> dict[str, Any]

Aggregate the claim graph's stance edges into a dashboard tension graph.

Pure and read-only. NODES are the claims, each sized by :func:claim_strength and colored by :func:claim_centrality, carrying its derived :func:claim_status. EDGES are the stance relations whose both endpoints are claims — the counter relations (contradicts / responds-to / qualifies) plus supports / replicates for context and any supersedes — deduplicated and deterministically ordered. CAMPS are the connected components over the contradicts edges (the actual debates), skipping contradicts SELF-LOOPS (src == dst data artifacts) so they never seed a phantom size-1 camp. Each camp carries a status derived from the same dialectical logic :func:analyze_gaps uses: resolved when a member is superseded by ANOTHER member of the SAME camp (an intra-camp supersedes — an external supersedes from outside the debate does NOT resolve it), thin when the debate is evidentially weak, else contested.

cross_store_contradictions (from :func:zettelkasten.synapse.synthesis.cross_store_contradictions) folds CROSS-STORE counter-evidence into the SAME structure: each item {memory_id, zk_id, zk_source, confidence, ...} where the canon claim is in scope adds a synthetic PRACTICE node and a contradicts edge practice --contradicts--> claim, joining that claim to a debate camp. So a strong canon claim contradicted only by real-world practice still shows as contested — its contested status reflects practice, not just the literature.

Returns {scope, nodes, edges, camps, summary} — JSON ready for a tension-graph view. Reuses the existing claim index and scorers; writes nothing.

Source code in zettelkasten/claims.py

def debate_map(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
    cross_store_contradictions: "list[dict[str, Any]] | None" = None,
    _context: "tuple | None" = None,
) -> dict[str, Any]:
    """Aggregate the claim graph's stance edges into a dashboard tension graph.

    Pure and read-only. NODES are the claims, each sized by :func:`claim_strength`
    and colored by :func:`claim_centrality`, carrying its derived
    :func:`claim_status`. EDGES are the stance relations whose both endpoints are
    claims — the counter relations (contradicts / responds-to / qualifies) plus
    supports / replicates for context and any     ``supersedes`` — deduplicated and
    deterministically ordered. CAMPS are the connected components over the
    ``contradicts`` edges (the actual debates), skipping ``contradicts``
    SELF-LOOPS (``src == dst`` data artifacts) so they never seed a phantom
    size-1 camp. Each camp carries a status derived from the same dialectical
    logic :func:`analyze_gaps` uses: ``resolved`` when a member is superseded by
    ANOTHER member of the SAME camp (an intra-camp ``supersedes`` — an external
    ``supersedes`` from outside the debate does NOT resolve it), ``thin`` when the
    debate is evidentially weak, else ``contested``.

    ``cross_store_contradictions`` (from
    :func:`zettelkasten.synapse.synthesis.cross_store_contradictions`) folds
    CROSS-STORE counter-evidence into the SAME structure: each item
    ``{memory_id, zk_id, zk_source, confidence, ...}`` where the canon claim is
    in scope adds a synthetic PRACTICE node and a ``contradicts`` edge
    ``practice --contradicts--> claim``, joining that claim to a debate camp. So a
    strong canon claim contradicted only by real-world practice still shows as
    contested — its contested status reflects practice, not just the literature.

    Returns ``{scope, nodes, edges, camps, summary}`` — JSON ready for a
    tension-graph view. Reuses the existing claim index and scorers; writes
    nothing.
    """
    if _context is None:
        ctx = _build_context(
            get_graph, project=project, graph=graph,
            graphs_dir=graphs_dir, namespace=namespace, localize=localize,
        )
    else:
        ctx = _context
    index, works, citations, importance_map, centrality = ctx

    name = graph or project or "all"
    scope = {"type": "graph" if graph else "project", "name": name}
    claim_keys = set(index.claims)

    # ── nodes: claims sized by strength, colored by centrality ──────────────────
    nodes: list[dict[str, Any]] = []
    strength_by_key: dict[Key, float] = {}
    for key, rc in index.resolved.items():
        strength = claim_strength(rc, importance_map=importance_map)
        status = claim_status(rc)
        strength_by_key[key] = strength
        oppose_span = len({
            e.src_graph for e in rc.counter_in if e.src_graph not in _NON_PAPER_GRAPHS
        })
        nodes.append({
            "uid": claim_uid(key),
            "id": key[1],
            "graph": key[0],
            "title": rc.note.title or key[1],
            "type": rc.note.type,
            "strength": round(strength, 4),
            "centrality": round(centrality.get(key, 0.0), 4),
            "status": status,
            "support_span": len(rc.support_sources),
            "oppose_span": oppose_span,
        })
    nodes.sort(key=lambda n: n["uid"])

    # ── edges: claim-claim stance relations (counter + support + supersedes) ────
    stance_rels = set(COUNTER_RELATIONS) | set(SUPPORT_RELATIONS) | {"supersedes"}
    seen: set[tuple[str, str, str, str, str]] = set()
    edges: list[dict[str, Any]] = []
    # Undirected adjacency over ``contradicts`` edges → debate camps.
    adjacency: dict[Key, set[Key]] = {}
    for e in index.edges:
        if e.relation not in stance_rels:
            continue
        if e.src_key not in claim_keys or e.dst_key not in claim_keys:
            continue
        # Skip SELF-LOOPS (src == dst data artifacts): a self-``contradicts`` edge
        # is not a debate — it would otherwise seed a phantom size-1 camp.
        if e.src_key == e.dst_key:
            continue
        sig = (e.src_graph, e.src_id, e.relation, e.dst_graph, e.dst_id)
        if sig in seen:
            continue
        seen.add(sig)
        edges.append({
            "source": claim_uid(e.src_key),
            "target": claim_uid(e.dst_key),
            "source_graph": e.src_graph,
            "source_id": e.src_id,
            "target_graph": e.dst_graph,
            "target_id": e.dst_id,
            "relation": e.relation,
        })
        if e.relation == "contradicts":
            adjacency.setdefault(e.src_key, set()).add(e.dst_key)
            adjacency.setdefault(e.dst_key, set()).add(e.src_key)

    # ── cross-store counter-evidence: practice --contradicts--> canon claim ──────
    # A practice node contradicting an in-scope canon claim joins that claim to a
    # debate camp (so its contested status reflects practice, not just literature).
    # Practice nodes get a synthetic ``_memory::<id>`` uid and cannot resolve a
    # camp (they never appear in ``index.resolved``).
    cross_count = 0
    for c in cross_store_contradictions or []:
        zk_source, zk_id = c.get("zk_source"), c.get("zk_id")
        mem_id = c.get("memory_id")
        if not zk_source or not zk_id or not mem_id:
            continue
        ckey = (zk_source, zk_id)
        if ckey not in claim_keys:
            continue
        pkey = (_MEMORY_STORE, str(mem_id))
        conf = max(0.0, min(1.0, float(c.get("confidence", 0.0) or 0.0)))
        if pkey not in strength_by_key:
            strength_by_key[pkey] = conf
            nodes.append({
                "uid": claim_uid(pkey),
                "id": str(mem_id),
                "graph": _MEMORY_STORE,
                "title": c.get("memory_title") or str(mem_id),
                "type": c.get("memory_type") or "practice",
                "store": "memory",
                "tier": "practice",
                "strength": round(conf, 4),
                "centrality": 0.0,
                "status": "counter-evidence",
                "support_span": 0,
                "oppose_span": 0,
            })
        sig = (_MEMORY_STORE, str(mem_id), "contradicts", zk_source, zk_id)
        if sig not in seen:
            seen.add(sig)
            edges.append({
                "source": claim_uid(pkey),
                "target": claim_uid(ckey),
                "source_graph": _MEMORY_STORE,
                "source_id": str(mem_id),
                "target_graph": zk_source,
                "target_id": zk_id,
                "relation": "contradicts",
                "cross_store": True,
            })
        adjacency.setdefault(pkey, set()).add(ckey)
        adjacency.setdefault(ckey, set()).add(pkey)
        cross_count += 1

    nodes.sort(key=lambda n: n["uid"])
    edges.sort(key=lambda d: (d["source"], d["relation"], d["target"]))

    # ── camps: connected components over the contradicts adjacency ──────────────
    camps: list[dict[str, Any]] = []
    visited: set[Key] = set()
    for start in sorted(adjacency):
        if start in visited:
            continue
        component: list[Key] = []
        stack = [start]
        while stack:
            cur = stack.pop()
            if cur in visited:
                continue
            visited.add(cur)
            component.append(cur)
            stack.extend(sorted(adjacency.get(cur, set())))
        member_keys = sorted(component)
        member_set = set(member_keys)
        # A camp is ``resolved`` only when the superseding claim is INSIDE the
        # camp: a member superseded by ANOTHER member of the same contradicts-
        # component. ``claim_status`` returns "superseded" for ANY incoming
        # supersedes edge — including one from an EXTERNAL claim not in this
        # debate — so scoping to intra-camp edges keeps an unrelated
        # ``C supersedes A`` from mislabelling a genuinely contested A-vs-B camp.
        superseded = False
        for k in member_keys:
            rc_k = index.resolved.get(k)
            if rc_k is None:
                continue
            if any(e.src_key in member_set for e in rc_k.superseded_by):
                superseded = True
                break
        strengths = [strength_by_key.get(k, 0.0) for k in member_keys]
        mean_strength = sum(strengths) / len(strengths) if strengths else 0.0
        if superseded:
            camp_status = "resolved"
        elif mean_strength < _GAP_EVIDENTIAL_STRENGTH_FLOOR:
            camp_status = "thin"
        else:
            camp_status = "contested"
        camps.append({
            "id": claim_uid(member_keys[0]),
            "members": [claim_uid(k) for k in member_keys],
            "status": camp_status,
            "size": len(member_keys),
            "mean_strength": round(mean_strength, 4),
        })
    camps.sort(key=lambda c: (c["size"], c["id"]), reverse=True)

    status_summary = {"resolved": 0, "contested": 0, "thin": 0}
    for c in camps:
        status_summary[c["status"]] += 1

    return {
        "scope": scope,
        "nodes": nodes,
        "edges": edges,
        "camps": camps,
        "summary": {
            "claims": len(nodes),
            "stance_edges": len(edges),
            "debates": len(camps),
            "camp_status": status_summary,
            "cross_store_contradictions": cross_count,
        },
        "message": (
            f"Mapped {len(camps)} debate(s) over {len(edges)} stance edge(s) "
            f"across {len(nodes)} claim(s)."
            + (f" ({cross_count} cross-store counter-evidence edge(s))" if cross_count else "")
        ),
    }

find_supersessions

find_supersessions(get_graph: GetGraph, *, project: str = '', graph: str = '', graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None, margin: float = 0.15, _context: 'tuple | None' = None) -> dict[str, Any]

Propose belief-revision (supersedes) edges — PROPOSE-ONLY.

Pure and write-free. For every contradicts edge between two claims, B supersedes A when ALL hold: (a) claim_strength(B) - claim_strength(A) >= margin (default 0.15); (b) their SUPPORT SETS OVERLAP (share >= 1 supporting source OR the same subject signal); and (c) B is NEWER. Recency is decided on ONE axis: if EITHER side has a datable supporting-paper year we compare paper-years ALONE (the same src_year / frontier logic :func:analyze_gaps uses) — a stronger-but-undated claim never supersedes an evidence-dated one on note-id — and only when NEITHER side has a paper-year do we fall back to the note-id creation date. Also skips contradicts SELF-LOOPS (src == dst data artifacts). Each contradicting pair yields AT MOST one proposal (the stronger-and-newer side supersedes the other); a tie or a straddling pair (each newer/stronger on a different axis) yields none.

Returns {scope, proposals, summary, message}. Each proposal is a PROPOSED supersedes edge {winner, loser, rationale, strength_delta, recency} — never auto-written (a confirmed edge is authored through the gated claim write path).

Source code in zettelkasten/claims.py

def find_supersessions(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
    margin: float = 0.15,
    _context: "tuple | None" = None,
) -> dict[str, Any]:
    """Propose belief-revision (``supersedes``) edges — PROPOSE-ONLY.

    Pure and write-free. For every ``contradicts`` edge between two claims, B
    supersedes A when ALL hold: (a) ``claim_strength(B) - claim_strength(A) >=
    margin`` (default 0.15); (b) their SUPPORT SETS OVERLAP (share >= 1 supporting
    source OR the same subject signal); and (c) B is NEWER. Recency is decided on
    ONE axis: if EITHER side has a datable supporting-paper year we compare
    paper-years ALONE (the same ``src_year`` / frontier logic :func:`analyze_gaps`
    uses) — a stronger-but-undated claim never supersedes an evidence-dated one on
    note-id — and only when NEITHER side has a paper-year do we fall back to the
    note-id creation date. Also skips ``contradicts`` SELF-LOOPS (``src == dst``
    data artifacts). Each contradicting pair yields AT MOST one proposal (the
    stronger-and-newer side supersedes the other); a tie or a straddling pair
    (each newer/stronger on a different axis) yields none.

    Returns ``{scope, proposals, summary, message}``. Each proposal is a PROPOSED
    ``supersedes`` edge ``{winner, loser, rationale, strength_delta, recency}`` —
    never auto-written (a confirmed edge is authored through the gated ``claim``
    write path).
    """
    if _context is None:
        ctx = _build_context(
            get_graph, project=project, graph=graph,
            graphs_dir=graphs_dir, namespace=namespace, localize=localize,
        )
    else:
        ctx = _context
    index, works, citations, importance_map, centrality = ctx

    name = graph or project or "all"
    scope = {"type": "graph" if graph else "project", "name": name}
    claim_keys = set(index.claims)
    src_year = {w.source_graph: w.year for w in works if w.source_graph}

    # Collect UNORDERED contradicting pairs (both counter directions), deduped.
    pairs: set[tuple[Key, Key]] = set()
    for e in index.edges:
        if e.relation != "contradicts":
            continue
        if e.src_key not in claim_keys or e.dst_key not in claim_keys:
            continue
        # Skip ``contradicts`` SELF-LOOPS (src == dst data artifacts): a claim
        # cannot supersede itself, and the pair would be a degenerate (a, a).
        if e.src_key == e.dst_key:
            continue
        a, b = sorted((e.src_key, e.dst_key))
        pairs.add((a, b))

    proposals: list[dict[str, Any]] = []
    for key_a, key_b in sorted(pairs):
        rc_a, rc_b = index.resolved.get(key_a), index.resolved.get(key_b)
        if rc_a is None or rc_b is None:
            continue
        s_a = claim_strength(rc_a, importance_map=importance_map)
        s_b = claim_strength(rc_b, importance_map=importance_map)

        # support-set overlap: a shared supporting source OR a shared subject.
        shared_sources = set(rc_a.support_sources) & set(rc_b.support_sources)
        shared_subject = _subject_signals(index, key_a) & _subject_signals(index, key_b)
        if not shared_sources and not shared_subject:
            continue

        # recency: newest supporting-paper year, else the note-id creation date.
        y_a = _newest_support_year(rc_a, src_year)
        y_b = _newest_support_year(rc_b, src_year)
        d_a = _note_date_key(key_a)
        d_b = _note_date_key(key_b)

        def _newer(win: Key, lose: Key, wy, ly, wd, ld) -> "str | None":
            # Paper-year evidence and note-id/creation dates are INCOMPARABLE
            # axes. If EITHER side carries a datable supporting-paper year, the
            # comparison is decided on paper-year alone: a stronger-but-undated
            # claim must not be declared "newer" than an evidence-dated one on
            # note-id. Only when NEITHER side has a paper-year do we fall back to
            # the note-id creation date.
            if wy is not None or ly is not None:
                if wy is None or ly is None:
                    # Exactly one side has a paper-year — the side WITHOUT one
                    # cannot win on note-id. Block this supersession direction.
                    return None
                if wy != ly:
                    return f"year {wy} > {ly}" if wy > ly else None
                return None
            if wd is not None and ld is not None and wd != ld:
                return "newer note" if wd > ld else None
            return None

        # Decide the winner: stronger by >= margin AND strictly newer.
        winner = loser = None
        reason_recency = None
        if (s_b - s_a) >= margin:
            reason_recency = _newer(key_b, key_a, y_b, y_a, d_b, d_a)
            if reason_recency is not None:
                winner, loser = key_b, key_a
                s_win, s_lose = s_b, s_a
        if winner is None and (s_a - s_b) >= margin:
            reason_recency = _newer(key_a, key_b, y_a, y_b, d_a, d_b)
            if reason_recency is not None:
                winner, loser = key_a, key_b
                s_win, s_lose = s_a, s_b
        if winner is None:
            continue

        overlap_bits = []
        if shared_sources:
            overlap_bits.append(f"{len(shared_sources)} shared source(s)")
        if shared_subject:
            overlap_bits.append(f"{len(shared_subject)} shared subject signal(s)")
        proposals.append({
            "relation": "supersedes",
            "winner": _claim_ref(index, winner),
            "loser": _claim_ref(index, loser),
            "strength_delta": round(s_win - s_lose, 4),
            "recency": reason_recency,
            "support_overlap": sorted(shared_sources),
            "shared_subject": sorted(shared_subject),
            "rationale": (
                f"stronger (Δstrength={round(s_win - s_lose, 3)} ≥ {margin}), "
                f"{reason_recency}, and support overlaps ({', '.join(overlap_bits)})"
            ),
            # A confirmed supersession is a claim→claim ``supersedes`` edge; it is
            # authored through the gated write path (propose-only), never here.
            "promote": {
                "tool": "claim",
                "action": "create",
                "args": {"claim_id": winner[1], "title": index.notes_by_key[winner].title or winner[1]},
                "note": ("PROPOSED supersedes edge. Authoring it is the only write path and "
                         "runs through the gated claim producer. find_supersessions writes nothing."),
            },
        })

    proposals.sort(key=lambda p: (p["strength_delta"], p["winner"]["id"]), reverse=True)
    return {
        "scope": scope,
        "proposals": proposals,
        "summary": {
            "claims": len(claim_keys),
            "contradicting_pairs": len(pairs),
            "proposed": len(proposals),
        },
        "message": (
            f"Proposed {len(proposals)} supersession(s) from {len(pairs)} "
            f"contradicting pair(s) across {len(claim_keys)} claim(s)."
        ),
    }