zettelkasten.claim_producer

zettelkasten.claim_producer

Claim producer: the agent-driven write loop that creates claims into the graph.

Where :mod:zettelkasten.claims is the pure, read-only claim ENGINE (it never writes, infers an edge, or calls an LLM), this module is its WRITE counterpart — the grounded producer that materializes agent-ratified claims, landscape themes, and their verbatim evidence back into the _cross graph. It mirrors the writer half of :mod:zettelkasten.review: every function is parametrized by a get_graph callable and routes ALL write-back through the M3a durable substrate — atomic crash-safe write (:func:graph.atomic_write_text), a per-path lock (:func:commit.review_write_lock), and a path-scoped debounced git commit (:func:commit._schedule_zettel_commit). Authored claims therefore survive crashes and are version-controlled exactly like _reviews/ manifests.

GROUNDING DISCIPLINE — this module never fabricates claim text, quotes, or stances. The caller supplies the grounded sentence (title + body), the member / evidence ids, the verbatim quote text, and the stance confidence; the producer only WIRES them into the graph's existing data model, reusing the exact conventions the read layer already reads:

• a claim is a claim note living in _cross (multi-source synthesis);
• evidence is a quote note linking --supports--> claim (the support edge is OUTGOING FROM THE QUOTE), so :func:materialize_claim writes quote notes that point at the claim;
• a stance from a member note is a supports / contradicts edge stored ON the member note and pointing at the claim (so it is an INCOMING edge the read layer counts), carrying an agent-authored confidence;
• a landscape theme is a concept note tagged landscape in _cross with outgoing surveys edges to the works it bundles (the same hubs :func:papers._derive_landscape_hubs and analyze_gaps surface).

The functions are UNDECORATED: they are dispatched by the claim(action=...) MCP tool in :mod:zettelkasten.server and will be called directly by the dashboard routes in a later wave. Federation is honored — a write whose target graph resolves to a read-only (federated) repo is refused before any file is touched.

ClaimCandidate dataclass

One ranked candidate claim cluster (a pure, serializable record).

The GATHER half fills everything except sentence / stances (the agent half writes those). members are the clustered note refs, quotes the RETRIEVED verbatim evidence notes incident to them (never generated), provenance the distinct paper sources spanned, and scores the component ranking signals (support_span × contestedness × novelty, plus the coherence the cluster cleared the floor with). is_counterclaim / counter_of mark a contradicting sub-cluster the agent split out.

Source code in zettelkasten/claim_producer.py

@dataclass
class ClaimCandidate:
    """One ranked candidate claim cluster (a pure, serializable record).

    The GATHER half fills everything except ``sentence`` / ``stances`` (the agent
    half writes those). ``members`` are the clustered note refs, ``quotes`` the
    RETRIEVED verbatim evidence notes incident to them (never generated),
    ``provenance`` the distinct paper sources spanned, and ``scores`` the
    component ranking signals (support_span × contestedness × novelty, plus the
    coherence the cluster cleared the floor with). ``is_counterclaim`` /
    ``counter_of`` mark a contradicting sub-cluster the agent split out.
    """

    cluster_id: str
    members: list[dict[str, Any]] = field(default_factory=list)
    quotes: list[dict[str, Any]] = field(default_factory=list)
    provenance: list[str] = field(default_factory=list)
    scores: dict[str, float] = field(default_factory=dict)
    sentence: str = ""
    stances: list[dict[str, Any]] = field(default_factory=list)
    is_counterclaim: bool = False
    counter_of: str = ""

ProposalMaterial dataclass

The deterministic GATHER result for a theme's mining pass.

candidates are the top-K surviving clusters in rank order; gaps record why a theme yielded nothing (unresolvable theme, too few embedded notes) so a miss is explicit, never silent. JSON-serializable via :meth:to_dict.

Source code in zettelkasten/claim_producer.py

@dataclass
class ProposalMaterial:
    """The deterministic GATHER result for a theme's mining pass.

    ``candidates`` are the top-K surviving clusters in rank order; ``gaps`` record
    why a theme yielded nothing (unresolvable theme, too few embedded notes) so a
    miss is explicit, never silent. JSON-serializable via :meth:`to_dict`.
    """

    scope: dict[str, Any]
    theme: str
    candidates: list[ClaimCandidate] = field(default_factory=list)
    gaps: list[dict[str, Any]] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

create_claim

create_claim(get_graph: GetGraph, *, claim_id: str = '', title: str = '', body: 'str | None' = None, supports: 'list[Any] | None' = None, contradicts: 'list[Any] | None' = None, tags: 'list[str] | None' = None, source: 'dict[str, Any] | None' = None, status: str = '', confidence: 'float | None' = None, graphs_dir: 'Path | str | None' = None) -> dict[str, Any]

Create or overwrite a grounded _cross claim note by id.

The GENERAL creator that :func:materialize_claim (and, later, the test_claim promotion) build on. The caller supplies the grounded sentence (title + body) and the member ids for the supports / contradicts stance edges; this never fabricates claim text or members. An omitted claim_id mints a fresh unique id from the title (a brand-new claim); an explicit id makes the write idempotent (create-or-overwrite). All write-back goes through the per-claim lock + atomic write + scheduled commit.

body is a None sentinel: None means "unset" (an overwrite preserves the existing body; a brand-new claim gets ''), while an explicit '' is an intentional blank body. An empty status likewise means "unset" — a brand-new claim defaults to complete and an overwrite preserves the existing claim's status (see the read-merge in :func:_create_claim_locked).

Refuses (PermissionError) when the _cross (or any member) graph is a federated, read-only target. confidence, when given, must be a number in [0, 1] (else ValueError).

Source code in zettelkasten/claim_producer.py

def create_claim(
    get_graph: GetGraph,
    *,
    claim_id: str = "",
    title: str = "",
    body: "str | None" = None,
    supports: "list[Any] | None" = None,
    contradicts: "list[Any] | None" = None,
    tags: "list[str] | None" = None,
    source: "dict[str, Any] | None" = None,
    status: str = "",
    confidence: "float | None" = None,
    graphs_dir: "Path | str | None" = None,
) -> dict[str, Any]:
    """Create or overwrite a grounded ``_cross`` claim note by id.

    The GENERAL creator that :func:`materialize_claim` (and, later, the
    ``test_claim`` promotion) build on. The caller supplies the grounded sentence
    (``title`` + ``body``) and the member ids for the ``supports`` /
    ``contradicts`` stance edges; this never fabricates claim text or members. An
    omitted ``claim_id`` mints a fresh unique id from the title (a brand-new
    claim); an explicit id makes the write idempotent (create-or-overwrite). All
    write-back goes through the per-claim lock + atomic write + scheduled commit.

    ``body`` is a ``None`` sentinel: ``None`` means "unset" (an overwrite
    preserves the existing body; a brand-new claim gets ``''``), while an explicit
    ``''`` is an intentional blank body. An empty ``status`` likewise means
    "unset" — a brand-new claim defaults to ``complete`` and an overwrite
    preserves the existing claim's status (see the read-merge in
    :func:`_create_claim_locked`).

    Refuses (``PermissionError``) when the ``_cross`` (or any member) graph is a
    federated, read-only target. ``confidence``, when given, must be a number in
    ``[0, 1]`` (else ``ValueError``).
    """
    title = (title or "").strip()
    if not title:
        raise ValueError("create_claim requires a non-empty title.")
    confidence = _coerce_confidence(confidence)
    base = _base(graphs_dir)
    if claim_id:
        cid = claim_id.strip()
        validate_id(cid, kind="claim id", for_filename=True)
        with review_write_lock(f"claim::{cid}", graphs_dir=base):
            _assert_writable(get_graph, CROSS_GRAPH)
            return _create_claim_locked(
                base, get_graph, claim_id=cid, title=title, body=body,
                supports=supports, contradicts=contradicts, tags=tags,
                source=source, status=status, confidence=confidence,
            )
    # Omitted id: mint a fresh unique id under the shared mint lock so the
    # existence check and the write are atomic (no TOCTOU between two minters).
    with review_write_lock(_MINT_LOCK_KEY, graphs_dir=base):
        cid = _unique_cross_id(base, title)
        validate_id(cid, kind="claim id", for_filename=True)
        _assert_writable(get_graph, CROSS_GRAPH)
        return _create_claim_locked(
            base, get_graph, claim_id=cid, title=title, body=body,
            supports=supports, contradicts=contradicts, tags=tags,
            source=source, status=status, confidence=confidence,
        )

delete_claim

delete_claim(get_graph: GetGraph, *, claim_id: str, graphs_dir: 'Path | str | None' = None) -> dict[str, Any]

Soft-delete a _cross claim: status=discarded + tombstoned edges.

Unlike the hard delete_note, the claim note is RETAINED on disk (so the delete is reversible and git-committed) but marked discarded — which drops it from every active read-layer view (its status is no longer in ACTIVE_STATUSES). Every incident supports / contradicts edge across the scoped graphs is tombstoned (Link.tombstoned=True) so the stance data is preserved and restorable rather than hard-removed. Returns {"error", "type": "NotFound"} when no such claim exists.

Refuses (PermissionError) when _cross is a federated, read-only target.

Source code in zettelkasten/claim_producer.py

def delete_claim(
    get_graph: GetGraph,
    *,
    claim_id: str,
    graphs_dir: "Path | str | None" = None,
) -> dict[str, Any]:
    """Soft-delete a ``_cross`` claim: ``status=discarded`` + tombstoned edges.

    Unlike the hard ``delete_note``, the claim note is RETAINED on disk (so the
    delete is reversible and git-committed) but marked ``discarded`` — which drops
    it from every active read-layer view (its status is no longer in
    ACTIVE_STATUSES). Every incident ``supports`` / ``contradicts`` edge across
    the scoped graphs is tombstoned (``Link.tombstoned=True``) so the stance data
    is preserved and restorable rather than hard-removed. Returns
    ``{"error", "type": "NotFound"}`` when no such claim exists.

    Refuses (``PermissionError``) when ``_cross`` is a federated, read-only target.
    """
    cid = (claim_id or "").strip()
    if not cid:
        raise ValueError("delete_claim requires a claim id.")
    validate_id(cid, kind="claim id", for_filename=True)
    base = _base(graphs_dir)
    with review_write_lock(f"claim::{cid}", graphs_dir=base):
        _assert_writable(get_graph, CROSS_GRAPH)
        # The claim body RMW (read, mark discarded, write) under the FILE's own
        # per-note lock — the SAME key the stance-member path uses — so a _cross
        # note that is both a claim and a stance member is never mutated under two
        # disjoint keys. Released before the tombstone scan (which re-acquires
        # per-note locks itself, including for this file).
        with _note_write_lock(base, CROSS_GRAPH, cid):
            note = _read_note(base, CROSS_GRAPH, cid)
            if note is None:
                return {
                    "error": f"Claim '{cid}' not found in {CROSS_GRAPH}.",
                    "type": "NotFound",
                }
            if (note.type or "") != "claim":
                raise ValueError(
                    f"refusing to discard '{CROSS_GRAPH}/{cid}': it is a "
                    f"'{note.type}' note, not a claim. delete_claim only soft-deletes "
                    "claim notes (the _cross namespace also holds quotes and hubs)."
                )
            note.status = DISCARDED_STATUS
            claim_path = _write_note_durable(base, CROSS_GRAPH, note)
        tombstoned = _tombstone_incident_edges(base, get_graph, claim_id=cid)
        return {
            "id": cid,
            "graph": CROSS_GRAPH,
            "status": DISCARDED_STATUS,
            "path": str(claim_path),
            "tombstoned_edges": tombstoned,
        }

delete_theme

delete_theme(get_graph: GetGraph, *, theme_id: str, graphs_dir: 'Path | str | None' = None) -> dict[str, Any]

Soft-delete a _cross landscape concept hub: status=discarded.

The reverse of :func:materialize_theme — used to undo a review's promote-theme. Like :func:delete_claim, the concept note is RETAINED on disk (reversible, git-committed) but marked discarded so it drops out of every active read-layer view, including :func:papers._derive_landscape_hubs (which skips non-active hubs). The hub's surveys edges are left intact — they ride along on the retained note and matter only if the hub is later restored. Returns {"error", "type": "NotFound"} when no such concept hub exists.

Refuses (PermissionError) when _cross is a federated, read-only target, and (ValueError) when the target note is not a concept hub.

Source code in zettelkasten/claim_producer.py

def delete_theme(
    get_graph: GetGraph,
    *,
    theme_id: str,
    graphs_dir: "Path | str | None" = None,
) -> dict[str, Any]:
    """Soft-delete a ``_cross`` landscape concept hub: ``status=discarded``.

    The reverse of :func:`materialize_theme` — used to undo a review's
    promote-theme. Like :func:`delete_claim`, the concept note is RETAINED on disk
    (reversible, git-committed) but marked ``discarded`` so it drops out of every
    active read-layer view, including :func:`papers._derive_landscape_hubs` (which
    skips non-active hubs). The hub's ``surveys`` edges are left intact — they ride
    along on the retained note and matter only if the hub is later restored.
    Returns ``{"error", "type": "NotFound"}`` when no such concept hub exists.

    Refuses (``PermissionError``) when ``_cross`` is a federated, read-only target,
    and (``ValueError``) when the target note is not a ``concept`` hub.
    """
    tid = (theme_id or "").strip()
    if not tid:
        raise ValueError("delete_theme requires a theme id.")
    validate_id(tid, kind="theme id", for_filename=True)
    base = _base(graphs_dir)
    with review_write_lock(f"claim::{tid}", graphs_dir=base):
        _assert_writable(get_graph, CROSS_GRAPH)
        with _note_write_lock(base, CROSS_GRAPH, tid):
            note = _read_note(base, CROSS_GRAPH, tid)
            if note is None:
                return {
                    "error": f"Theme hub '{tid}' not found in {CROSS_GRAPH}.",
                    "type": "NotFound",
                }
            if (note.type or "") != "concept":
                raise ValueError(
                    f"refusing to discard '{CROSS_GRAPH}/{tid}': it is a "
                    f"'{note.type}' note, not a concept hub. delete_theme only "
                    "soft-deletes landscape concept hubs."
                )
            note.status = DISCARDED_STATUS
            hub_path = _write_note_durable(base, CROSS_GRAPH, note)
        return {
            "id": tid,
            "graph": CROSS_GRAPH,
            "status": DISCARDED_STATUS,
            "path": str(hub_path),
        }

materialize_claim

materialize_claim(get_graph: GetGraph, *, claim_id: str, title: str = '', body: 'str | None' = None, supports: 'list[Any] | None' = None, contradicts: 'list[Any] | None' = None, quotes: 'list[Any] | None' = None, tags: 'list[str] | None' = None, source: 'dict[str, Any] | None' = None, confidence: 'float | None' = None, graphs_dir: 'Path | str | None' = None) -> dict[str, Any]

Idempotently accept a claim proposal: claim note + stance edges + quotes.

A WRAPPER over :func:create_claim that additionally writes the verbatim evidence quote notes (each linking supports → the claim). The claim_id is REQUIRED and stable so re-accepting the same proposal overwrites the claim, its stance edges, and its quote notes in place — never duplicating a note or an edge. Stance edges carry the agent-authored confidence; quote bodies are the caller's retrieved verbatim text (never generated). All writes share a single per-claim lock.

body is a None sentinel (None = preserve existing on overwrite; '' = intentional blank). Each quotes entry is a verbatim string or a dict {body|text, id?, title?, page?, source?, grounding?, confidence?}; a quote id defaults to a deterministic <claim_id>-q<N> so re-accepts are stable. confidence, when given, must be a number in [0, 1].

Source code in zettelkasten/claim_producer.py

def materialize_claim(
    get_graph: GetGraph,
    *,
    claim_id: str,
    title: str = "",
    body: "str | None" = None,
    supports: "list[Any] | None" = None,
    contradicts: "list[Any] | None" = None,
    quotes: "list[Any] | None" = None,
    tags: "list[str] | None" = None,
    source: "dict[str, Any] | None" = None,
    confidence: "float | None" = None,
    graphs_dir: "Path | str | None" = None,
) -> dict[str, Any]:
    """Idempotently accept a claim proposal: claim note + stance edges + quotes.

    A WRAPPER over :func:`create_claim` that additionally writes the verbatim
    evidence ``quote`` notes (each linking ``supports`` → the claim). The
    ``claim_id`` is REQUIRED and stable so re-accepting the same proposal
    overwrites the claim, its stance edges, and its quote notes in place — never
    duplicating a note or an edge. Stance edges carry the agent-authored
    ``confidence``; quote bodies are the caller's retrieved verbatim text (never
    generated). All writes share a single per-claim lock.

    ``body`` is a ``None`` sentinel (``None`` = preserve existing on overwrite;
    ``''`` = intentional blank). Each ``quotes`` entry is a verbatim string or a
    dict ``{body|text, id?, title?, page?, source?, grounding?, confidence?}``; a
    quote id defaults to a deterministic ``<claim_id>-q<N>`` so re-accepts are
    stable. ``confidence``, when given, must be a number in ``[0, 1]``.
    """
    title = (title or "").strip()
    if not title:
        raise ValueError("materialize_claim requires a non-empty title.")
    cid = (claim_id or "").strip()
    if not cid:
        raise ValueError("materialize_claim requires a stable claim id for idempotency.")
    validate_id(cid, kind="claim id", for_filename=True)
    confidence = _coerce_confidence(confidence)
    base = _base(graphs_dir)
    with review_write_lock(f"claim::{cid}", graphs_dir=base):
        _assert_writable(get_graph, CROSS_GRAPH)
        # status="" → the read-merge restores a discarded claim to ``complete``
        # but preserves an existing active status (never flips ``revised`` back).
        result = _create_claim_locked(
            base, get_graph, claim_id=cid, title=title, body=body,
            supports=supports, contradicts=contradicts, tags=tags,
            source=source, status="", confidence=confidence,
        )
        result["quotes"] = _materialize_quotes_locked(
            base, claim_id=cid, quotes=quotes, default_confidence=confidence,
        )
        result["materialized"] = True
        return result

materialize_theme

materialize_theme(get_graph: GetGraph, *, theme_id: str = '', title: str = '', members: 'list[Any] | None' = None, body: 'str | None' = None, tags: 'list[str] | None' = None, graphs_dir: 'Path | str | None' = None) -> dict[str, Any]

Idempotently materialize a landscape concept hub in _cross.

Writes a concept note tagged landscape whose outgoing surveys edges bundle the member works it maps (source folder names and/or citation ids) — the same theme-hub shape :func:papers._derive_landscape_hubs reads. The hub note is create-or-overwritten by id and ONLY its surveys edges are rebuilt (deduped) from members on every accept; everything else is READ-MERGED on overwrite — the existing hub body (unless a new body is given), its non-landscape tags, and its non-surveys links are all preserved — so re-accepting the same proposal yields an identical note with no duplicate edges and never blanks the hub's prose or drops a hand-added edge.

body is a None sentinel (None = preserve existing on overwrite; '' = intentional blank). Each members entry is a work-id string or {id, graph?} (graph is the target's home graph — a source folder or _citations — or empty for a plain work id). Refuses on a federated, read-only _cross, and refuses (type guard) to clobber a non-concept note sharing the id.

Source code in zettelkasten/claim_producer.py

def materialize_theme(
    get_graph: GetGraph,
    *,
    theme_id: str = "",
    title: str = "",
    members: "list[Any] | None" = None,
    body: "str | None" = None,
    tags: "list[str] | None" = None,
    graphs_dir: "Path | str | None" = None,
) -> dict[str, Any]:
    """Idempotently materialize a landscape concept hub in ``_cross``.

    Writes a ``concept`` note tagged ``landscape`` whose outgoing ``surveys``
    edges bundle the member works it maps (source folder names and/or citation
    ids) — the same theme-hub shape :func:`papers._derive_landscape_hubs` reads.
    The hub note is create-or-overwritten by id and ONLY its ``surveys`` edges are
    rebuilt (deduped) from ``members`` on every accept; everything else is
    READ-MERGED on overwrite — the existing hub ``body`` (unless a new ``body`` is
    given), its non-``landscape`` ``tags``, and its non-``surveys`` links are all
    preserved — so re-accepting the same proposal yields an identical note with no
    duplicate edges and never blanks the hub's prose or drops a hand-added edge.

    ``body`` is a ``None`` sentinel (``None`` = preserve existing on overwrite;
    ``''`` = intentional blank). Each ``members`` entry is a work-id string or
    ``{id, graph?}`` (``graph`` is the target's home graph — a source folder or
    ``_citations`` — or empty for a plain work id). Refuses on a federated,
    read-only ``_cross``, and refuses (type guard) to clobber a non-``concept``
    note sharing the id.
    """
    title = (title or "").strip()
    if not title:
        raise ValueError("materialize_theme requires a non-empty title (the concept label).")
    base = _base(graphs_dir)

    survey_links: list[Link] = []
    surveys_out: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    for ref in members or []:
        member = _coerce_ref(ref)
        if not member["id"]:
            continue
        key = (member["id"], member["graph"])
        if key in seen:
            continue
        seen.add(key)
        survey_links.append(
            Link(target=member["id"], relation=_SURVEYS_RELATION, graph=member["graph"])
        )
        surveys_out.append({"id": member["id"], "graph": member["graph"]})

    def _write_hub(tid: str) -> dict[str, Any]:
        validate_id(tid, kind="theme id", for_filename=True)
        _assert_writable(get_graph, CROSS_GRAPH)
        # Read-merge + write under the FILE's own per-note lock (Fix 4: the same
        # key any stance-member path uses for this file).
        with _note_write_lock(base, CROSS_GRAPH, tid):
            existing = _assert_overwritable(base, CROSS_GRAPH, tid, expected_type="concept")
            # body: None = preserve existing (or "" brand-new); "" = intentional blank.
            if body is not None:
                merged_body = body
            else:
                merged_body = existing.body if existing else ""
            # tags: preserve existing tags, add the caller's, guarantee landscape;
            # order-stable + deduped.
            merged_tags: list[str] = []
            for t in ([LANDSCAPE_TAG] + list(existing.tags if existing else [])
                      + list(tags or [])):
                if t and t not in merged_tags:
                    merged_tags.append(t)
            # links: rebuild ONLY the surveys edges; preserve every other link the
            # hub carried (hand-added relations, provenance edges, etc.).
            preserved_links = [
                l for l in (existing.links if existing else [])
                if l.relation != _SURVEYS_RELATION
            ]
            merged_links = preserved_links + survey_links
            hub = Note(
                id=tid,
                title=title,
                type="concept",
                source=dict(existing.source) if (existing and isinstance(existing.source, dict)) else {},
                body=merged_body,
                tags=merged_tags,
                links=merged_links,
                status=(existing.status if existing else "complete") or "complete",
            )
            path = _write_note_durable(base, CROSS_GRAPH, hub)
        return {
            "id": tid,
            "graph": CROSS_GRAPH,
            "type": "concept",
            "tags": merged_tags,
            "path": str(path),
            "surveys": surveys_out,
        }

    if theme_id:
        tid = theme_id.strip()
        with review_write_lock(f"claim::{tid}", graphs_dir=base):
            return _write_hub(tid)
    # Omitted id: mint under the shared mint lock so the existence check and the
    # write are atomic (no TOCTOU between two minters).
    with review_write_lock(_MINT_LOCK_KEY, graphs_dir=base):
        return _write_hub(_unique_cross_id(base, title))

gather_claim_candidates

gather_claim_candidates(get_graph: GetGraph, *, project: str = '', graph: str = '', theme: str = '', top_k: int = DEFAULT_TOP_K, min_sources: int = MIN_DISTINCT_SOURCES, coherence_floor: float = COHERENCE_FLOOR, max_quotes: int = MAX_CANDIDATE_QUOTES, embed_fn: 'Callable[[str, str], Any] | None' = None, graphs_dir: 'Path | str | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> ProposalMaterial

Deterministic GATHER: the grounded candidate claim clusters for a theme.

Pure (no LLM, no network, no writes). Builds the read-engine claim index over the scope (:func:claims.build_claim_index), resolves the theme's note pool (the assertion-bearing notes in the works the landscape hub surveys), clusters that pool by embedding (:func:_connected_clusters over the locked cosine primitive), and keeps only clusters that clear the COHERENCE FLOOR (:func:_cluster_coherence ≥ coherence_floor) AND span ≥ min_sources distinct paper sources. Survivors are RANKED by cross-source support span × contestedness × novelty-vs-existing-claims and DEDUPED against already-accepted _cross claims (a cluster whose members are a subset of an accepted claim's, or whose centroid is near-identical to one, is dropped). Returns the top top_k as a serializable :class:ProposalMaterial; each candidate carries its member ids, retrieved verbatim quotes, provenance, and component scores.

Source code in zettelkasten/claim_producer.py

def gather_claim_candidates(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    theme: str = "",
    top_k: int = DEFAULT_TOP_K,
    min_sources: int = MIN_DISTINCT_SOURCES,
    coherence_floor: float = COHERENCE_FLOOR,
    max_quotes: int = MAX_CANDIDATE_QUOTES,
    embed_fn: "Callable[[str, str], Any] | None" = None,
    graphs_dir: "Path | str | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> ProposalMaterial:
    """Deterministic GATHER: the grounded candidate claim clusters for a theme.

    Pure (no LLM, no network, no writes). Builds the read-engine claim index over
    the scope (:func:`claims.build_claim_index`), resolves the theme's note pool
    (the assertion-bearing notes in the works the landscape hub surveys), clusters
    that pool by embedding (:func:`_connected_clusters` over the locked cosine
    primitive), and keeps only clusters that clear the COHERENCE FLOOR
    (:func:`_cluster_coherence` ≥ ``coherence_floor``) AND span ≥ ``min_sources``
    distinct paper sources. Survivors are RANKED by cross-source support span ×
    contestedness × novelty-vs-existing-claims and DEDUPED against already-accepted
    ``_cross`` claims (a cluster whose members are a subset of an accepted claim's,
    or whose centroid is near-identical to one, is dropped). Returns the top
    ``top_k`` as a serializable :class:`ProposalMaterial`; each candidate carries
    its member ids, retrieved verbatim quotes, provenance, and component scores.
    """
    from zettelkasten.claims import COUNTER_RELATIONS, SUPPORT_RELATIONS, build_claim_index

    base = _base(graphs_dir)
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    theme = (theme or "").strip()
    scope = {
        "type": "graph" if graph else "project",
        "name": graph or project or "all",
        "theme": theme,
    }
    if not theme:
        return ProposalMaterial(scope=scope, theme=theme,
                                gaps=[_proposal_gap("theme", "no theme supplied")])

    member_graphs = _resolve_theme_pool_graphs(base, theme, loc)
    if not member_graphs:
        return ProposalMaterial(
            scope=scope, theme=theme,
            gaps=[_proposal_gap("theme", f"theme '{theme}' resolves to no landscape hub / member works")],
        )

    index = build_claim_index(
        get_graph, project=project, graph=graph, graphs_dir=base, namespace=ns, localize=loc,
    )
    get_vec = _vector_provider(get_graph, embed_fn)

    # Already-accepted claims (active _cross claim/finding notes) and the member
    # sets they were built from — the dedup baseline.
    stance_rels = tuple(SUPPORT_RELATIONS) + tuple(COUNTER_RELATIONS)
    accepted: dict[tuple[str, str], set[tuple[str, str]]] = {}
    for key in index.claims:
        if key[0] != CROSS_GRAPH:
            continue
        accepted[key] = {
            e.src_key for e in index.by_target.get(key, []) if e.relation in stance_rels
        }

    # Build the embedded note pool: assertion-bearing notes in the theme's member
    # graphs, excluding evidence/concept notes and the accepted claims themselves.
    gaps: list[dict[str, Any]] = []
    pool_keys: list[tuple[str, str]] = []
    vectors: dict[tuple[str, str], Any] = {}
    for key, note in index.notes_by_key.items():
        if key[0] not in member_graphs:
            continue
        ntype = note.type or ""
        if not ntype or ntype in _POOL_EXCLUDED_TYPES or key in accepted:
            continue
        vec = get_vec(key[0], key[1])
        # Exclude unusable vectors — missing/empty OR zero-norm (the bare
        # ``if not vec`` misses ``[0.0, 0.0, …]``). A zero-norm member would
        # otherwise score 1.0 to everything and join/false-pass every cluster.
        if not _is_usable_vector(vec):
            continue
        pool_keys.append(key)
        vectors[key] = vec
    pool_keys.sort()

    if len(pool_keys) < 2:
        gaps.append(_proposal_gap("pool", "fewer than 2 embedded notes in the theme pool",
                                  embedded=len(pool_keys)))
        return ProposalMaterial(scope=scope, theme=theme, gaps=gaps)

    clusters = _connected_clusters(pool_keys, vectors, CLUSTER_SIM_THRESHOLD)

    survivors: list[tuple[list[tuple[str, str]], list[str], dict[str, float]]] = []
    for cl in clusters:
        sources = _distinct_sources(cl)
        if len(sources) < min_sources:
            continue
        coherence = _cluster_coherence(cl, vectors)
        if coherence < coherence_floor:
            continue
        cl_set = set(cl)
        # DEDUP (membership): a cluster fully contained in an accepted claim's
        # member set is already synthesized.
        if any(cl_set and cl_set <= members for members in accepted.values()):
            continue
        # DEDUP (semantic) + novelty: distance of the cluster centroid to the
        # nearest accepted claim's embedding.
        from zettelkasten.syllabus import _centroid

        cen = _centroid([vectors[k] for k in cl])
        max_acc_sim = 0.0
        for akey in accepted:
            av = get_vec(akey[0], akey[1])
            # Guarded sim: an INCOMPARABLE accepted-claim vector (e.g. a stale,
            # wrong-dimension cached embedding) is NOT a duplicate — skip it
            # instead of scoring it 1.0 and silently suppressing every candidate.
            s = _guarded_cosine_sim(cen, av) if cen is not None else None
            if s is not None:
                max_acc_sim = max(max_acc_sim, s)
        if max_acc_sim >= ACCEPTED_DEDUP_SIM:
            continue
        novelty = round(max(0.0, 1.0 - max_acc_sim), 6)
        contested = _cluster_contestedness(cl, index)
        support_span = len(sources)
        rank_score = round(
            (1.0 + support_span) * (1.0 + contested) * (0.5 + novelty), 6
        )
        survivors.append((cl, sources, {
            "support_span": support_span,
            "contestedness": contested,
            "novelty": novelty,
            "coherence": round(coherence, 6),
            "rank_score": rank_score,
        }))

    # Rank: score desc, then provenance, then first member — fully deterministic.
    survivors.sort(key=lambda t: (-t[2]["rank_score"], t[1], t[0][0]))

    candidates: list[ClaimCandidate] = []
    for i, (cl, sources, scores) in enumerate(survivors[: max(0, top_k)], start=1):
        members = [
            {"id": k[1], "graph": k[0],
             "title": index.notes_by_key[k].title, "type": index.notes_by_key[k].type}
            for k in cl
        ]
        candidates.append(ClaimCandidate(
            cluster_id=f"cand-{i}",
            members=members,
            quotes=_candidate_quotes(index, cl, max_quotes),
            provenance=sources,
            scores=scores,
        ))

    if not candidates and not gaps:
        gaps.append(_proposal_gap("candidates", "no cluster cleared the coherence/source gate"))
    return ProposalMaterial(scope=scope, theme=theme, candidates=candidates, gaps=gaps)

phrase_proposals

phrase_proposals(material: ProposalMaterial, *, propose_fn: 'Callable[[str, str], str] | None' = None) -> ProposalMaterial

PHRASE each gathered candidate via an INJECTABLE agent (the agent half).

For each non-counterclaim candidate it calls propose_fn(system, prompt) -> str (defaulting to :func:_default_propose_fn, the lazy in-process dashboard agent) to (a) phrase ONE grounded sentence paraphrasing the cluster, (b) classify each member's stance (supports/contradicts/qualifies) + confidence, validated against the cluster's actual members and the allowed stances, and (c) when a coherent contradicting sub-cluster surfaces, SPLIT it into a separate counterclaim :class:ClaimCandidate (is_counterclaim / counter_of). It never invents facts/quotes — phrasing only. Mutates and returns material for convenience; propose_fn is injected by tests so no LLM runs.

Source code in zettelkasten/claim_producer.py

def phrase_proposals(
    material: ProposalMaterial,
    *,
    propose_fn: "Callable[[str, str], str] | None" = None,
) -> ProposalMaterial:
    """PHRASE each gathered candidate via an INJECTABLE agent (the agent half).

    For each non-counterclaim candidate it calls ``propose_fn(system, prompt) ->
    str`` (defaulting to :func:`_default_propose_fn`, the lazy in-process
    dashboard agent) to (a) phrase ONE grounded sentence paraphrasing the cluster,
    (b) classify each member's stance (supports/contradicts/qualifies) +
    confidence, validated against the cluster's actual members and the allowed
    stances, and (c) when a coherent contradicting sub-cluster surfaces, SPLIT it
    into a separate counterclaim :class:`ClaimCandidate` (``is_counterclaim`` /
    ``counter_of``). It never invents facts/quotes — phrasing only. Mutates and
    returns ``material`` for convenience; ``propose_fn`` is injected by tests so
    no LLM runs.
    """
    fn = propose_fn or _default_propose_fn
    out: list[ClaimCandidate] = []
    for cand in material.candidates:
        if cand.is_counterclaim:
            out.append(cand)
            continue
        member_keys = {(m["graph"], m["id"]) for m in cand.members}
        parsed = _parse_propose_response(fn(_PROPOSE_CONTRACT, _build_propose_prompt(material, cand)))
        cand.sentence = str(parsed.get("sentence") or "")
        stances: list[dict[str, Any]] = []
        for st in parsed.get("stances") or []:
            if not isinstance(st, dict):
                continue
            mid, mgraph = str(st.get("id") or ""), str(st.get("graph") or "")
            stance = str(st.get("stance") or "")
            if (mgraph, mid) not in member_keys or stance not in _PROPOSAL_STANCES:
                continue
            stances.append({
                "id": mid, "graph": mgraph, "stance": stance,
                "confidence": _coerce_confidence(st.get("confidence")),
            })

        cand.stances = stances
        # Split a coherent contradicting sub-cluster into a counterclaim candidate.
        contra_keys = {(s["graph"], s["id"]) for s in stances if s["stance"] == "contradicts"}
        if not contra_keys:
            out.append(cand)
            continue

        # The contradicting members (and their quotes) belong to the COUNTERCLAIM.
        # Capture them BEFORE pruning the main candidate's member list.
        counter_members = [m for m in cand.members if (m["graph"], m["id"]) in contra_keys]
        counter_quotes = [q for q in cand.quotes if _quote_supports_key(q) in contra_keys]
        # The counterclaim ASSERTS the opposite sentence, so its members SUPPORT
        # it — populate per-member ``supports`` stances so it is materializable
        # (the loop ``continue``s on is_counterclaim, so it is never re-phrased
        # to fill these in later).
        counter_stances = [
            {"id": s["id"], "graph": s["graph"], "stance": "supports",
             "confidence": s["confidence"]}
            for s in stances if s["stance"] == "contradicts"
        ]

        # Remove the contradicting members from the MAIN candidate's member list
        # so a member is not duplicated across the main claim AND its counterclaim
        # (the main claim's stances still record the contradicts classification
        # that drove the split). The members move WHOLLY into the counterclaim.
        cand.members = [m for m in cand.members if (m["graph"], m["id"]) not in contra_keys]
        # When EVERY member contradicted, the main candidate is now empty (its
        # members all moved into the counterclaim). Do not emit an empty/garbage
        # primary — only emit the counterclaim below.
        if cand.members:
            out.append(cand)

        out.append(ClaimCandidate(
            cluster_id=f"{cand.cluster_id}-counter",
            members=counter_members,
            quotes=counter_quotes,
            provenance=sorted({m["graph"] for m in counter_members
                               if m["graph"] not in _NON_SOURCE_GRAPHS}),
            scores=dict(cand.scores),
            sentence=str(parsed.get("counterclaim_sentence") or ""),
            stances=counter_stances,
            is_counterclaim=True,
            counter_of=cand.cluster_id,
        ))
    material.candidates = out
    return material

compute_proposal_signature

compute_proposal_signature(*, theme: str, project: str = '', graph: str = '', top_k: int = DEFAULT_TOP_K, min_sources: int = MIN_DISTINCT_SOURCES, coherence_floor: float = COHERENCE_FLOOR, max_quotes: int = MAX_CANDIDATE_QUOTES, agent_identity: 'str | None' = None, graphs_dir: 'Path | str | None' = None, localize: 'Callable[[str], str] | None' = None) -> str

hash(scope + theme-pool fingerprint + _cross fingerprint + contract + agent).

Mirrors :func:outline.compute_generation_signature / _scope_fingerprint: a cheap (folder, file, mtime_ns) fingerprint over the theme's member source folders PLUS _cross (the accepted-claim baseline the dedup reads), skipping _-prefixed files, folded with the scope descriptor and the embedded contract version. Any note add/edit/remove in the pool or any accepted-claim change shifts the signature and busts the cache; an unchanged graph hashes identically so the proposals are served from cache.

The scope ALSO folds in max_quotes (it changes the candidates' retrieved evidence) and the phrasing agent/model identity (agent_identity, defaulting to :func:_propose_agent_signature) — so changing how many quotes are carried, or swapping the agent/model, busts the cache rather than serving proposals phrased under the old parameters.

Source code in zettelkasten/claim_producer.py

def compute_proposal_signature(
    *,
    theme: str,
    project: str = "",
    graph: str = "",
    top_k: int = DEFAULT_TOP_K,
    min_sources: int = MIN_DISTINCT_SOURCES,
    coherence_floor: float = COHERENCE_FLOOR,
    max_quotes: int = MAX_CANDIDATE_QUOTES,
    agent_identity: "str | None" = None,
    graphs_dir: "Path | str | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> str:
    """``hash(scope + theme-pool fingerprint + _cross fingerprint + contract + agent)``.

    Mirrors :func:`outline.compute_generation_signature` / ``_scope_fingerprint``:
    a cheap ``(folder, file, mtime_ns)`` fingerprint over the theme's member
    source folders PLUS ``_cross`` (the accepted-claim baseline the dedup reads),
    skipping ``_``-prefixed files, folded with the scope descriptor and the
    embedded contract version. Any note add/edit/remove in the pool or any
    accepted-claim change shifts the signature and busts the cache; an unchanged
    graph hashes identically so the proposals are served from cache.

    The scope ALSO folds in ``max_quotes`` (it changes the candidates' retrieved
    evidence) and the phrasing ``agent``/model identity (``agent_identity``,
    defaulting to :func:`_propose_agent_signature`) — so changing how many quotes
    are carried, or swapping the agent/model, busts the cache rather than serving
    proposals phrased under the old parameters.
    """
    base = _base(graphs_dir)
    loc = localize or (lambda s: s)
    agent_id = agent_identity if agent_identity is not None else _propose_agent_signature()
    member_graphs = sorted(_resolve_theme_pool_graphs(base, theme, loc))
    fp: list[tuple[str, str, int]] = []
    for g in member_graphs + [CROSS_GRAPH]:
        d = base / g
        if not d.is_dir():
            continue
        for p in sorted(d.glob("*.md")):
            if p.name.startswith("_"):
                continue
            try:
                fp.append((g, p.name, p.stat().st_mtime_ns))
            except OSError:
                continue
    payload = json.dumps(
        {
            "scope": {"theme": theme, "project": project, "graph": graph,
                      "top_k": top_k, "min_sources": min_sources,
                      "coherence_floor": coherence_floor, "max_quotes": max_quotes},
            "fingerprint": fp,
            "contract": _PROPOSE_VERSION,
            "agent": agent_id,
        },
        sort_keys=True,
        default=str,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

propose_claims

propose_claims(get_graph: GetGraph, *, project: str = '', graph: str = '', theme: str = '', top_k: int = DEFAULT_TOP_K, min_sources: int = MIN_DISTINCT_SOURCES, coherence_floor: float = COHERENCE_FLOOR, max_quotes: int = MAX_CANDIDATE_QUOTES, force: bool = False, propose_fn: 'Callable[[str, str], str] | None' = None, embed_fn: 'Callable[[str, str], Any] | None' = None, graphs_dir: 'Path | str | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Orchestrate the per-theme mining loop: cache → GATHER → phrase.

The single entry point the claim('propose') dispatch calls. Computes the :func:compute_proposal_signature; on a cache hit (unchanged graph, not force) returns the stored proposals with NO agent call (cached=True). Otherwise it runs the DETERMINISTIC :func:gather_claim_candidates engine, PHRASES the candidates via the INJECTABLE :func:phrase_proposals (propose_fn defaults to the lazy in-process dashboard agent), caches the result, and returns it. READ/COMPUTE ONLY — it never writes a claim to the graph (acceptance is the separate :func:materialize_claim step).

Returns {theme, scope, candidates, gaps, signature, cached} (each candidate a plain serializable dict).

Source code in zettelkasten/claim_producer.py

def propose_claims(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    theme: str = "",
    top_k: int = DEFAULT_TOP_K,
    min_sources: int = MIN_DISTINCT_SOURCES,
    coherence_floor: float = COHERENCE_FLOOR,
    max_quotes: int = MAX_CANDIDATE_QUOTES,
    force: bool = False,
    propose_fn: "Callable[[str, str], str] | None" = None,
    embed_fn: "Callable[[str, str], Any] | None" = None,
    graphs_dir: "Path | str | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Orchestrate the per-theme mining loop: cache → GATHER → phrase.

    The single entry point the ``claim('propose')`` dispatch calls. Computes the
    :func:`compute_proposal_signature`; on a cache hit (unchanged graph, not
    ``force``) returns the stored proposals with NO agent call (``cached=True``).
    Otherwise it runs the DETERMINISTIC :func:`gather_claim_candidates` engine,
    PHRASES the candidates via the INJECTABLE :func:`phrase_proposals`
    (``propose_fn`` defaults to the lazy in-process dashboard agent), caches the
    result, and returns it. READ/COMPUTE ONLY — it never writes a claim to the
    graph (acceptance is the separate :func:`materialize_claim` step).

    Returns ``{theme, scope, candidates, gaps, signature, cached}`` (each
    candidate a plain serializable dict).
    """
    base = _base(graphs_dir)
    loc = localize or (lambda s: s)
    theme = (theme or "").strip()

    signature = compute_proposal_signature(
        theme=theme, project=project, graph=graph, top_k=top_k,
        min_sources=min_sources, coherence_floor=coherence_floor,
        max_quotes=max_quotes, graphs_dir=base, localize=loc,
    )
    if not force:
        cached = _PROPOSAL_CACHE.get(signature)
        if cached is not None:
            return {**cached, "cached": True}

    material = gather_claim_candidates(
        get_graph, project=project, graph=graph, theme=theme, top_k=top_k,
        min_sources=min_sources, coherence_floor=coherence_floor,
        max_quotes=max_quotes, embed_fn=embed_fn, graphs_dir=base,
        namespace=namespace, localize=loc,
    )
    phrased = phrase_proposals(material, propose_fn=propose_fn)

    result: dict[str, Any] = {
        "theme": theme,
        "scope": phrased.scope,
        "candidates": [asdict(c) for c in phrased.candidates],
        "gaps": phrased.gaps,
        "signature": signature,
        "cached": False,
    }
    if len(_PROPOSAL_CACHE) >= _PROPOSAL_CACHE_MAX:
        _PROPOSAL_CACHE.clear()
    _PROPOSAL_CACHE[signature] = result
    return result

test_claim

test_claim(get_graph: GetGraph, *, text: str = '', project: str = '', graph: str = '', top_k: int = TEST_MAX_CANDIDATES, relevance_floor: float = TEST_RELEVANCE_FLOOR, classify_fn: 'Callable[[str, str], str] | None' = None, summarize_fn: 'Callable[[str, str], str] | None' = None, embed_fn: 'Callable[[str, str], Any] | None' = None, embed_text_fn: 'Callable[[str], Any] | None' = None, graphs_dir: 'Path | str | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Test a hypothesis against the corpus — READ-ONLY, falsification-disciplined.

The hypothesis text is embedded as an EPHEMERAL scratch claim (NEVER written to the graph). The engine then GATHERS candidate evidence deterministically — a semantic search over the corpus notes/quotes/claims (cosine similarity ≥ relevance_floor via the GUARDED :func:_guarded_cosine_sim) plus a one-hop citation/relation-graph walk from the top hits — and an INJECTABLE, WRITE-FREE classify_fn judges each member's STANCE (supports / contradicts / qualifies / irrelevant). The verdict is scored through the read engine's :func:claims.claim_strength, with disconfirming evidence weighted SYMMETRICALLY (so refuted and untested-in-corpus are first-class). Quotes are RETRIEVED from the graph's quote notes — NEVER generated.

Returns the verdict + tally, the classified evidence (with retrieved verbatim quotes + provenance), the top :func:claims.analyze_gaps findings (what's missing), and a promote descriptor that wires the hypothesis into the gated :func:materialize_claim write path (the ONLY write path — this tool itself writes nothing).

classify_fn / embed_fn / embed_text_fn are injected by tests so no LLM/embedder runs; the defaults are the lazy in-process (write-free) agent and the Model2VecAdapter. summarize_fn produces the optional natural-language assessment of how the hypothesis sizes up; unlike classify_fn it has NO agent default — a caller opts in by passing :func:_default_summarize_fn (the dashboard route does), so it stays None (assessment omitted) under tests and the deterministic /actions path without an explicit LLM call.

Source code in zettelkasten/claim_producer.py

def test_claim(
    get_graph: GetGraph,
    *,
    text: str = "",
    project: str = "",
    graph: str = "",
    top_k: int = TEST_MAX_CANDIDATES,
    relevance_floor: float = TEST_RELEVANCE_FLOOR,
    classify_fn: "Callable[[str, str], str] | None" = None,
    summarize_fn: "Callable[[str, str], str] | None" = None,
    embed_fn: "Callable[[str, str], Any] | None" = None,
    embed_text_fn: "Callable[[str], Any] | None" = None,
    graphs_dir: "Path | str | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Test a hypothesis against the corpus — READ-ONLY, falsification-disciplined.

    The hypothesis ``text`` is embedded as an EPHEMERAL scratch claim (NEVER
    written to the graph). The engine then GATHERS candidate evidence
    deterministically — a semantic search over the corpus notes/quotes/claims
    (cosine similarity ≥ ``relevance_floor`` via the GUARDED
    :func:`_guarded_cosine_sim`) plus a one-hop citation/relation-graph walk from
    the top hits — and an INJECTABLE, WRITE-FREE ``classify_fn`` judges each
    member's STANCE (supports / contradicts / qualifies / irrelevant). The verdict
    is scored through the read engine's :func:`claims.claim_strength`, with
    disconfirming evidence weighted SYMMETRICALLY (so ``refuted`` and
    ``untested-in-corpus`` are first-class). Quotes are RETRIEVED from the graph's
    quote notes — NEVER generated.

    Returns the verdict + tally, the classified evidence (with retrieved
    verbatim quotes + provenance), the top :func:`claims.analyze_gaps` findings
    (what's missing), and a ``promote`` descriptor that wires the hypothesis into
    the gated :func:`materialize_claim` write path (the ONLY write path — this
    tool itself writes nothing).

    ``classify_fn`` / ``embed_fn`` / ``embed_text_fn`` are injected by tests so no
    LLM/embedder runs; the defaults are the lazy in-process (write-free) agent and
    the Model2VecAdapter. ``summarize_fn`` produces the optional natural-language
    ``assessment`` of how the hypothesis sizes up; unlike ``classify_fn`` it has
    NO agent default — a caller opts in by passing :func:`_default_summarize_fn`
    (the dashboard route does), so it stays ``None`` (assessment omitted) under
    tests and the deterministic ``/actions`` path without an explicit LLM call.
    """
    from zettelkasten.claims import (
        STRUCTURAL_RELATIONS,
        SUPPORT_RELATIONS,
        COUNTER_RELATIONS,
        _build_context,
        analyze_gaps,
    )

    text = (text or "").strip()
    base = _base(graphs_dir)
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    scope = {"type": "graph" if graph else "project",
             "name": graph or project or "all", "hypothesis": text}
    if not text:
        raise ValueError("test_claim requires a non-empty hypothesis text.")

    ctx = _build_context(
        get_graph, project=project, graph=graph,
        graphs_dir=base, namespace=ns, localize=loc,
    )
    index, works, citations, importance_map, _centrality = ctx

    embed_text = embed_text_fn or _default_embed_text
    hvec = embed_text(text)
    get_vec = _vector_provider(get_graph, embed_fn)

    # ── deterministic GATHER: semantic search over the corpus note pool ─────────
    scored: list[tuple[float, tuple[str, str]]] = []
    if _is_usable_vector(hvec):
        for key, note in index.notes_by_key.items():
            if (note.type or "") in _TEST_EXCLUDED_TYPES:
                continue
            vec = get_vec(key[0], key[1])
            if not _is_usable_vector(vec):
                continue
            sim = _guarded_cosine_sim(hvec, vec)
            if sim is None or sim < relevance_floor:
                continue
            scored.append((round(sim, 6), key))
    scored.sort(key=lambda t: (-t[0], t[1]))
    cap = max(0, top_k)
    semantic = scored[:cap]
    sim_by_key = {k: s for s, k in semantic}

    # ── citation/relation-graph walk: one hop out from the semantic hits ────────
    walk_relations = set(SUPPORT_RELATIONS) | set(COUNTER_RELATIONS) | set(STRUCTURAL_RELATIONS)
    selected: list[tuple[str, str]] = [k for _s, k in semantic]
    selected_set = set(selected)
    walked: list[tuple[str, str]] = []
    for _s, key in semantic:
        for e in index.by_source.get(key, []):
            if e.relation in walk_relations and e.dst_key in index.notes_by_key \
                    and e.dst_key not in selected_set:
                nt = (index.notes_by_key[e.dst_key].type or "")
                if nt and nt not in _TEST_EXCLUDED_TYPES:
                    selected_set.add(e.dst_key)
                    walked.append(e.dst_key)
        for e in index.by_target.get(key, []):
            if e.relation in walk_relations and e.src_key in index.notes_by_key \
                    and e.src_key not in selected_set and e.src_type != "quote":
                nt = (index.notes_by_key[e.src_key].type or "")
                if nt and nt not in _TEST_EXCLUDED_TYPES:
                    selected_set.add(e.src_key)
                    walked.append(e.src_key)
    selected = selected + walked
    selected = selected[: max(cap, len(semantic))]

    # ── build the candidate evidence members (retrieved quotes attached) ────────
    members: list[dict[str, Any]] = []
    for key in selected:
        note = index.notes_by_key.get(key)
        if note is None:
            continue
        members.append({
            "id": key[1], "graph": key[0],
            "title": note.title or key[1], "type": note.type or "",
            "similarity": sim_by_key.get(key),
            "via": "semantic" if key in sim_by_key else "citation-walk",
            "quotes": _retrieve_quotes_for(index, key),
        })

    # ── INJECTABLE stance classification (write-free agent; determinism gated) ──
    member_keys = {(m["graph"], m["id"]) for m in members}
    stance_by_key: dict[tuple[str, str], dict[str, Any]] = {}
    if members:
        fn = classify_fn or _default_classify_fn
        parsed = _parse_propose_response(fn(_CLASSIFY_CONTRACT, _build_classify_prompt(text, members)))
        for st in parsed.get("stances") or []:
            if not isinstance(st, dict):
                continue
            mk = (str(st.get("graph") or ""), str(st.get("id") or ""))
            stance = str(st.get("stance") or "")
            if mk not in member_keys or stance not in _TEST_STANCES:
                continue
            stance_by_key[mk] = {
                "stance": stance,
                "confidence": _coerce_confidence(st.get("confidence")),
            }
    # A member the classifier did not rate (or rated out of range) is irrelevant.
    for m in members:
        rated = stance_by_key.get((m["graph"], m["id"]))
        m["stance"] = rated["stance"] if rated else "irrelevant"
        m["confidence"] = rated["confidence"] if rated else None

    verdict = _score_verdict(members, importance_map)

    # ── what's MISSING: the typed gap scan, reusing the already-built context ───
    try:
        gap_report = analyze_gaps(
            get_graph, project=project, graph=graph, limit=TEST_GAP_LIMIT, _context=ctx,
        )
        gaps = gap_report.get("gaps", [])
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("test_claim gap scan failed: %s", exc)
        gaps = []

    # ── optional natural-language assessment of how the hypothesis sizes up ─────
    # Opt-in: a caller passes summarize_fn (the dashboard route passes the default
    # write-free agent). Tests / the deterministic /actions path leave it None, so
    # no LLM runs and the field is simply omitted. Best-effort — a failure or an
    # empty (no rated evidence) case yields no assessment rather than aborting.
    assessment = ""
    if summarize_fn is not None and any(
        m.get("stance") and m["stance"] != "irrelevant" for m in members
    ):
        try:
            assessment = _clean_assessment(
                summarize_fn(_ASSESS_CONTRACT, _build_assess_prompt(text, verdict, members))
            )
        except Exception as exc:  # pragma: no cover - defensive (agent/LLM failure)
            logger.warning("test_claim assessment failed: %s", exc)
            assessment = ""

    # ── promote-to-graph descriptor (the ONLY write path; gated producer) ───────
    supports_refs = [
        {"id": m["id"], "graph": m["graph"], "confidence": m["confidence"]}
        for m in members if m["stance"] == "supports"
    ]
    contradicts_refs = [
        {"id": m["id"], "graph": m["graph"], "confidence": m["confidence"]}
        for m in members if m["stance"] == "contradicts"
    ]
    promote_quotes = [
        {"body": q["body"], "page": q.get("page")}
        for m in members if m["stance"] == "supports" for q in m["quotes"]
    ]
    promote = {
        "tool": "claim",
        "action": "materialize",
        "available": bool(supports_refs or contradicts_refs),
        "args": {
            "claim_id": generate_id(text),
            "title": text,
            "supports": supports_refs,
            "contradicts": contradicts_refs,
            "quotes": promote_quotes,
        },
        "note": ("Promotion is the only write path; it runs through the gated "
                 "materialize_claim producer. test_claim writes nothing."),
    }

    return {
        "scope": scope,
        "hypothesis": text,
        "assessment": assessment,
        "verdict": verdict["verdict"],
        "strength": verdict["strength"],
        "support_strength": verdict["support_strength"],
        "counter_strength": verdict["counter_strength"],
        "tally": verdict["tally"],
        "support_sources": verdict["support_sources"],
        "evidence": [
            {"id": m["id"], "graph": m["graph"], "title": m["title"],
             "type": m["type"], "stance": m["stance"], "confidence": m["confidence"],
             "similarity": m["similarity"], "via": m["via"],
             "quotes": m["quotes"]}
            for m in members
        ],
        "gaps": gaps,
        "promote": promote,
        "embedded": _is_usable_vector(hvec),
    }

suggest_claims_for_topic

suggest_claims_for_topic(get_graph: GetGraph, *, text: str = '', project: str = '', graph: str = '', top_k: int = SUGGEST_MAX_CLAIMS, relevance_floor: float = SUGGEST_RELEVANCE_FLOOR, embed_fn: 'Callable[[str, str], Any] | None' = None, embed_text_fn: 'Callable[[str], Any] | None' = None, graphs_dir: 'Path | str | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Rank EXISTING corpus claims by relevance to a pitched topic — read-only.

Embeds text as an EPHEMERAL scratch vector (NEVER written to the graph), then scores every claim note in scope by the GUARDED cosine similarity (:func:_guarded_cosine_sim) of its cached vector to the topic, keeping those at or above relevance_floor and returning the top top_k in descending similarity. Deterministic and write-free: it proposes NO new claims and mutates nothing — it only surfaces claims that already exist so the caller can place them under a freshly-pitched section.

embed_fn / embed_text_fn are injectable so tests run without an embedder; the defaults are the in-process Model2Vec adapter.

Returns {scope, topic, claims, embedded} where each claim is {uid, id, graph, title, type, similarity}.

Source code in zettelkasten/claim_producer.py

def suggest_claims_for_topic(
    get_graph: GetGraph,
    *,
    text: str = "",
    project: str = "",
    graph: str = "",
    top_k: int = SUGGEST_MAX_CLAIMS,
    relevance_floor: float = SUGGEST_RELEVANCE_FLOOR,
    embed_fn: "Callable[[str, str], Any] | None" = None,
    embed_text_fn: "Callable[[str], Any] | None" = None,
    graphs_dir: "Path | str | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Rank EXISTING corpus claims by relevance to a pitched topic — read-only.

    Embeds ``text`` as an EPHEMERAL scratch vector (NEVER written to the graph),
    then scores every claim note in scope by the GUARDED cosine similarity
    (:func:`_guarded_cosine_sim`) of its cached vector to the topic, keeping those
    at or above ``relevance_floor`` and returning the top ``top_k`` in descending
    similarity. Deterministic and write-free: it proposes NO new claims and
    mutates nothing — it only surfaces claims that already exist so the caller can
    place them under a freshly-pitched section.

    ``embed_fn`` / ``embed_text_fn`` are injectable so tests run without an
    embedder; the defaults are the in-process Model2Vec adapter.

    Returns ``{scope, topic, claims, embedded}`` where each claim is
    ``{uid, id, graph, title, type, similarity}``.
    """
    from zettelkasten.claims import build_claim_index, claim_uid

    text = (text or "").strip()
    base = _base(graphs_dir)
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    scope = {"type": "graph" if graph else "project",
             "name": graph or project or "all", "topic": text}
    if not text:
        raise ValueError("suggest_claims_for_topic requires a non-empty topic text.")

    index = build_claim_index(
        get_graph, project=project, graph=graph, graphs_dir=base,
        namespace=ns, localize=loc,
    )
    embed_text = embed_text_fn or _default_embed_text
    hvec = embed_text(text)
    get_vec = _vector_provider(get_graph, embed_fn)

    scored: list[tuple[float, tuple[str, str], Any]] = []
    if _is_usable_vector(hvec):
        for key, note in index.claims.items():
            vec = get_vec(key[0], key[1])
            if not _is_usable_vector(vec):
                continue
            sim = _guarded_cosine_sim(hvec, vec)
            if sim is None or sim < relevance_floor:
                continue
            scored.append((round(sim, 6), key, note))
    # Descending similarity; ties broken by key for a stable, deterministic order.
    scored.sort(key=lambda t: (-t[0], t[1]))

    claims = [
        {
            "uid": claim_uid(key),
            "id": key[1],
            "graph": key[0],
            "title": note.title or key[1],
            "type": note.type or "",
            "similarity": sim,
        }
        for sim, key, note in scored[: max(0, top_k)]
    ]
    return {
        "scope": scope,
        "topic": text,
        "claims": claims,
        "embedded": _is_usable_vector(hvec),
    }

cluster_threshold_for_granularity

cluster_threshold_for_granularity(granularity: int) -> float

Map the wizard granularity slider (10..90) to a leftover-cluster cosine cut.

Mirrors the Concepts-graph slider's metaphor: a LOW granularity yields fewer, broader topics (a looser similarity cut) and a HIGH granularity yields more, narrower topics (a tighter cut). The input is clamped to [10, 90] and mapped LINEARLY onto [CLUSTER_THRESHOLD_GRANULARITY_MIN, CLUSTER_THRESHOLD_GRANULARITY_MAX], so the midpoint (50) lands exactly on :data:CLUSTER_SIM_THRESHOLD and the default reproduces the prior behavior.

Source code in zettelkasten/claim_producer.py

def cluster_threshold_for_granularity(granularity: int) -> float:
    """Map the wizard granularity slider (10..90) to a leftover-cluster cosine cut.

    Mirrors the Concepts-graph slider's metaphor: a LOW granularity yields fewer,
    broader topics (a looser similarity cut) and a HIGH granularity yields more,
    narrower topics (a tighter cut). The input is clamped to ``[10, 90]`` and
    mapped LINEARLY onto ``[CLUSTER_THRESHOLD_GRANULARITY_MIN,
    CLUSTER_THRESHOLD_GRANULARITY_MAX]``, so the midpoint (50) lands exactly on
    :data:`CLUSTER_SIM_THRESHOLD` and the default reproduces the prior behavior.
    """
    g = max(10, min(90, int(granularity)))
    frac = (g - 10) / 80.0
    span = CLUSTER_THRESHOLD_GRANULARITY_MAX - CLUSTER_THRESHOLD_GRANULARITY_MIN
    return round(CLUSTER_THRESHOLD_GRANULARITY_MIN + frac * span, 4)

propose_placements

propose_placements(get_graph: GetGraph, *, themes: 'list[dict[str, Any]]', unplaced: 'list[dict[str, Any]]', question: str = '', embed_fn: 'Callable[[str, str], Any] | None' = None, embed_text_fn: 'Callable[[str], Any] | None' = None, placement_floor: float = PLACEMENT_RELEVANCE_FLOOR, cluster_threshold: float = CLUSTER_SIM_THRESHOLD, coherence_floor: float = COHERENCE_FLOOR, min_cluster_size: int = NEW_THEME_MIN_CLAIMS, scope: 'dict[str, Any] | None' = None) -> dict[str, Any]

Propose a home for each unplaced claim — deterministic, read-only.

For every existing theme (real OR proposed) with at least one vectorized claim, a CENTROID is taken over its placed claims' cached vectors (:func:syllabus._centroid). Each unplaced claim with a usable vector is scored by GUARDED cosine similarity (:func:_guarded_cosine_sim) against every centroid; its best theme at or above placement_floor becomes a PLACEMENT proposal. Claims that match no theme are LEFTOVERS — clustered among themselves (:func:_connected_clusters at cluster_threshold) and, for each component of at least min_cluster_size claims that clears the coherence_floor (:func:_cluster_coherence), pitched as a NEW-theme proposal titled after its most-central claim.

Question-aware ranking. When the outline's research question is given (and embeddable), it is embedded into an EPHEMERAL scratch vector and every proposed NEW theme is scored by the guarded cosine similarity of its cluster centroid to that question (question_relevance). New themes are then ordered by question relevance first (then coherence), so the structures most aligned with what the outline is trying to ask surface at the top — the question STEERS the proposal without suppressing any coherent cluster. With no question the order is coherence-driven, exactly as before, and question_relevance is None.

Write-free: proposes no claims, creates no theme, mutates nothing. The caller applies approved placements via set_placement and approved new themes via add_proposed_theme. embed_fn / embed_text_fn are injectable so tests run without an embedder; the defaults pull each note's cached vector and embed the question via the in-process Model2Vec adapter.

Returns {scope, question, placements, new_themes, embedded} where each placement is {uid, id, graph, title, theme_id, theme_label, similarity} and each new theme is {suggested_title, coherence, question_relevance, claims:[{uid, id, graph, title, similarity_to_centroid}]}.

Source code in zettelkasten/claim_producer.py

def propose_placements(
    get_graph: GetGraph,
    *,
    themes: "list[dict[str, Any]]",
    unplaced: "list[dict[str, Any]]",
    question: str = "",
    embed_fn: "Callable[[str, str], Any] | None" = None,
    embed_text_fn: "Callable[[str], Any] | None" = None,
    placement_floor: float = PLACEMENT_RELEVANCE_FLOOR,
    cluster_threshold: float = CLUSTER_SIM_THRESHOLD,
    coherence_floor: float = COHERENCE_FLOOR,
    min_cluster_size: int = NEW_THEME_MIN_CLAIMS,
    scope: "dict[str, Any] | None" = None,
) -> dict[str, Any]:
    """Propose a home for each unplaced claim — deterministic, read-only.

    For every existing theme (real OR proposed) with at least one vectorized
    claim, a CENTROID is taken over its placed claims' cached vectors
    (:func:`syllabus._centroid`). Each unplaced claim with a usable vector is
    scored by GUARDED cosine similarity (:func:`_guarded_cosine_sim`) against
    every centroid; its best theme at or above ``placement_floor`` becomes a
    PLACEMENT proposal. Claims that match no theme are LEFTOVERS — clustered
    among themselves (:func:`_connected_clusters` at ``cluster_threshold``) and,
    for each component of at least ``min_cluster_size`` claims that clears the
    ``coherence_floor`` (:func:`_cluster_coherence`), pitched as a NEW-theme
    proposal titled after its most-central claim.

    **Question-aware ranking.** When the outline's research ``question`` is given
    (and embeddable), it is embedded into an EPHEMERAL scratch vector and every
    proposed NEW theme is scored by the guarded cosine similarity of its cluster
    centroid to that question (``question_relevance``). New themes are then
    ordered by question relevance first (then coherence), so the structures most
    aligned with what the outline is trying to ask surface at the top — the
    question STEERS the proposal without suppressing any coherent cluster. With no
    question the order is coherence-driven, exactly as before, and
    ``question_relevance`` is ``None``.

    Write-free: proposes no claims, creates no theme, mutates nothing. The caller
    applies approved placements via ``set_placement`` and approved new themes via
    ``add_proposed_theme``. ``embed_fn`` / ``embed_text_fn`` are injectable so
    tests run without an embedder; the defaults pull each note's cached vector and
    embed the question via the in-process Model2Vec adapter.

    Returns ``{scope, question, placements, new_themes, embedded}`` where each
    placement is ``{uid, id, graph, title, theme_id, theme_label, similarity}``
    and each new theme is ``{suggested_title, coherence, question_relevance,
    claims:[{uid, id, graph, title, similarity_to_centroid}]}``.
    """
    from zettelkasten.claims import claim_uid
    from zettelkasten.syllabus import _centroid

    get_vec = _vector_provider(get_graph, embed_fn)
    vec_cache: dict[tuple[str, str], Any] = {}

    # The outline's question as an ephemeral scratch vector (never persisted), used
    # only to RANK the proposed new themes by how well they answer it.
    question = (question or "").strip()
    q_vec: Any = None
    if question:
        embed_text = embed_text_fn or _default_embed_text
        cand = embed_text(question)
        if _is_usable_vector(cand):
            q_vec = cand

    def vec_for(graph: str, nid: str) -> Any:
        key = (graph, nid)
        if key not in vec_cache:
            vec_cache[key] = get_vec(graph, nid)
        return vec_cache[key]

    # Theme centroids over each theme's currently-placed, vectorized claims. A
    # theme with no usable claim vector (e.g. a freshly-pitched proposed section
    # holding only overlay-only drafts) is not a placement target — skipped.
    theme_centroids: list[tuple[str, str, Any]] = []
    for t in themes:
        tid = str(t.get("id") or "")
        if not tid:
            continue
        member_vecs = [
            v
            for c in (t.get("claims") or [])
            if _is_usable_vector(v := vec_for(str(c.get("graph") or ""), str(c.get("id") or "")))
        ]
        if not member_vecs:
            continue
        cen = _centroid(member_vecs)
        if _is_usable_vector(cen):
            theme_centroids.append((tid, str(t.get("label") or tid), cen))

    placements: list[dict[str, Any]] = []
    leftover_keys: list[tuple[str, str]] = []
    leftover_rows: dict[tuple[str, str], dict[str, Any]] = {}
    leftover_vecs: dict[tuple[str, str], Any] = {}

    for c in unplaced:
        graph = str(c.get("graph") or "")
        nid = str(c.get("id") or "")
        key = (graph, nid)
        v = vec_for(graph, nid)
        if not _is_usable_vector(v):
            continue
        # Best theme above the floor; ties keep the earlier (deterministic) theme.
        best: tuple[float, str, str] | None = None
        for tid, label, cen in theme_centroids:
            sim = _guarded_cosine_sim(v, cen)
            if sim is None or sim < placement_floor:
                continue
            if best is None or sim > best[0]:
                best = (sim, tid, label)
        if best is not None:
            placements.append({
                "uid": c.get("uid") or claim_uid(key),
                "id": nid,
                "graph": graph,
                "title": c.get("title") or nid,
                "theme_id": best[1],
                "theme_label": best[2],
                "similarity": round(best[0], 6),
            })
        else:
            leftover_keys.append(key)
            leftover_rows[key] = c
            leftover_vecs[key] = v

    # Sort placements strongest-match first, ties by uid for a stable order.
    placements.sort(key=lambda p: (-p["similarity"], p["uid"]))

    # Cluster the leftovers among themselves: a coherent component of >= the
    # minimum size becomes a NEW-theme proposal, titled after its most-central
    # claim (the author edits the title before approving).
    new_themes: list[dict[str, Any]] = []
    for comp in _connected_clusters(leftover_keys, leftover_vecs, cluster_threshold):
        if len(comp) < min_cluster_size:
            continue
        if _cluster_coherence(comp, leftover_vecs) < coherence_floor:
            continue
        cen = _centroid([leftover_vecs[k] for k in comp])
        scored: list[tuple[float, tuple[str, str]]] = []
        for k in comp:
            s = _guarded_cosine_sim(leftover_vecs[k], cen) if cen is not None else None
            scored.append((s if s is not None else -1.0, k))
        scored.sort(key=lambda t: (-t[0], t[1]))
        members = [
            {
                "uid": leftover_rows[k].get("uid") or claim_uid(k),
                "id": k[1],
                "graph": k[0],
                "title": leftover_rows[k].get("title") or k[1],
                "similarity_to_centroid": round(s, 6) if s >= 0.0 else None,
            }
            for s, k in scored
        ]
        # How well this cluster answers the outline's question (None without one).
        q_rel: float | None = None
        if q_vec is not None and cen is not None:
            qs = _guarded_cosine_sim(q_vec, cen)
            if qs is not None:
                q_rel = round(qs, 6)
        new_themes.append({
            "suggested_title": members[0]["title"] if members else "",
            "coherence": round(_cluster_coherence(comp, leftover_vecs), 6),
            "question_relevance": q_rel,
            "claims": members,
        })

    # With a question, surface the most question-relevant new themes first (then
    # coherence); without one, keep the coherence-driven discovery order. Missing
    # relevance (un-embeddable cluster) sorts last so it never outranks a scored one.
    if q_vec is not None:
        new_themes.sort(
            key=lambda nt: (
                -(nt["question_relevance"] if nt["question_relevance"] is not None else -1.0),
                -nt["coherence"],
                nt["suggested_title"],
            )
        )

    return {
        "scope": scope or {},
        "question": question,
        "placements": placements,
        "new_themes": new_themes,
        "embedded": any(_is_usable_vector(v) for v in vec_cache.values()),
    }

partition_structural_themes

partition_structural_themes(get_graph: GetGraph, *, kind: str, themes: 'list[dict[str, Any]]', unplaced: 'list[dict[str, Any]]', spine: str = '', spine_mode: str = '', include_unplaced: bool = False, project: str = '', graph: str = '', scope: 'dict[str, Any] | None' = None, graphs_dir: 'Path | str | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Partition a review's claim pool into structural themes — read-only.

Reuses :func:propose_placements' whole-pool construction (the deduped union of every themed claim + the unplaced worklist) so the structural partition sees the SAME in-scope claim pool the semantic proposer clusters, then groups that pool by kind:

• source — by the claim's graph (the supporting note's source graph); each group's title is that source's _meta title (falling back to the graph name), matching :func:tables_gather.list_row_values' source axis.
• note_type / tag — by the claim note's frontmatter type / tags (read from disk and passed through the matrix's own :func:tables_common._value_keys, so non-value tags like hub are dropped identically). A claim may land in MULTIPLE tag groups.
• spine — by the dimension of the named spine organization the claim's supporting note is a member of, using the SAME spine-side membership index the matrix cells and the outline spine sections read (:func:outline.gather._resolve_spine_partition). spine (an org id) is REQUIRED for this axis; every dimension becomes a theme (in the matrix's column order) even when it has no member claims, so the returned skeleton mirrors the spine structure. A claim may be a member of multiple dimensions.

Deterministic and write-free: no embedder, no clustering, no LLM. Each theme is {key, title, claims:[{uid, id, graph, title}], count}; themes are sorted by count descending (ties by title then key) — EXCEPT the spine axis, which keeps the matrix's column order (empty dimensions included) so its skeleton mirrors the matrix/outline 1:1. When include_unplaced is set, claims that matched no group are appended as a trailing "Unplaced" theme (kept LAST regardless of count, mirroring the outline's unplaced bucket).

The result ALSO carries an always-present unplaced (the leftover claim members) and unplaced_count — the claims that matched NO group (no tag, missing note_type, an unreadable note, an empty graph, or a spine non-member). This is additive and independent of include_unplaced (which only controls the trailing theme), so those claims are never silently dropped.

Raises ValueError for an unknown kind, or (for the spine axis) a missing spine id or one that does not resolve to a materialized spine in the review's scope — the route maps these to 400.

Source code in zettelkasten/claim_producer.py

def partition_structural_themes(
    get_graph: GetGraph,
    *,
    kind: str,
    themes: "list[dict[str, Any]]",
    unplaced: "list[dict[str, Any]]",
    spine: str = "",
    spine_mode: str = "",
    include_unplaced: bool = False,
    project: str = "",
    graph: str = "",
    scope: "dict[str, Any] | None" = None,
    graphs_dir: "Path | str | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Partition a review's claim pool into structural themes — read-only.

    Reuses :func:`propose_placements`' whole-pool construction (the deduped union
    of every themed claim + the unplaced worklist) so the structural partition
    sees the SAME in-scope claim pool the semantic proposer clusters, then groups
    that pool by ``kind``:

    * ``source`` — by the claim's ``graph`` (the supporting note's source graph);
      each group's title is that source's ``_meta`` title (falling back to the
      graph name), matching :func:`tables_gather.list_row_values`' ``source`` axis.
    * ``note_type`` / ``tag`` — by the claim note's frontmatter ``type`` / ``tags``
      (read from disk and passed through the matrix's own
      :func:`tables_common._value_keys`, so non-value tags like ``hub`` are
      dropped identically). A claim may land in MULTIPLE ``tag`` groups.
    * ``spine`` — by the dimension of the named ``spine`` organization the claim's
      supporting note is a member of, using the SAME spine-side membership index
      the matrix cells and the outline spine sections read
      (:func:`outline.gather._resolve_spine_partition`). ``spine`` (an org id) is
      REQUIRED for this axis; every dimension becomes a theme (in the matrix's
      column order) even when it has no member claims, so the returned skeleton
      mirrors the spine structure. A claim may be a member of multiple dimensions.

    Deterministic and write-free: no embedder, no clustering, no LLM. Each theme is
    ``{key, title, claims:[{uid, id, graph, title}], count}``; themes are sorted by
    ``count`` descending (ties by title then key) — EXCEPT the ``spine`` axis, which
    keeps the matrix's column order (empty dimensions included) so its skeleton
    mirrors the matrix/outline 1:1. When ``include_unplaced`` is set, claims that
    matched no group are appended as a trailing ``"Unplaced"`` theme (kept LAST
    regardless of count, mirroring the outline's unplaced bucket).

    The result ALSO carries an always-present ``unplaced`` (the leftover claim
    members) and ``unplaced_count`` — the claims that matched NO group (no tag,
    missing note_type, an unreadable note, an empty graph, or a spine non-member).
    This is additive and independent of ``include_unplaced`` (which only controls
    the trailing theme), so those claims are never silently dropped.

    Raises ``ValueError`` for an unknown ``kind``, or (for the ``spine`` axis) a
    missing ``spine`` id or one that does not resolve to a materialized spine in
    the review's scope — the route maps these to 400.
    """
    from zettelkasten.claims import claim_uid
    from zettelkasten.tables_common import _value_keys

    kind = (kind or "").strip()
    if kind not in STRUCTURAL_AXES:
        raise ValueError(
            f"Unknown structural axis '{kind}'. "
            f"Expected one of: {', '.join(sorted(STRUCTURAL_AXES))}."
        )
    base = _base(graphs_dir)
    loc = localize or (lambda s: s)

    # The whole in-scope claim pool: every themed claim + the unplaced worklist,
    # deduped by uid (identical to propose-placements' ``whole_pool`` construction)
    # so a claim that is both themed and (defensively) unplaced is partitioned once.
    seen_uids: set[str] = set()
    pool: list[dict[str, Any]] = []
    for c in (*(cl for t in themes for cl in (t.get("claims") or [])), *unplaced):
        uid = str(c.get("uid") or "")
        if uid and uid in seen_uids:
            continue
        if uid:
            seen_uids.add(uid)
        pool.append(c)

    def _member(c: dict[str, Any]) -> dict[str, Any]:
        g = str(c.get("graph") or "")
        nid = str(c.get("id") or "")
        return {
            "uid": c.get("uid") or claim_uid((g, nid)),
            "id": nid,
            "graph": g,
            "title": c.get("title") or nid,
        }

    # group key -> {"title": str, "claims": [member, ...]}. Insertion order is
    # preserved (dict, 3.7+) so a stable secondary sort is possible under ties.
    groups: dict[str, dict[str, Any]] = {}
    leftovers: list[dict[str, Any]] = []

    def _bucket(key: str, title: str, member: dict[str, Any]) -> None:
        grp = groups.get(key)
        if grp is None:
            grp = {"title": title, "claims": []}
            groups[key] = grp
        grp["claims"].append(member)

    if kind == "source":
        for c in pool:
            g = str(c.get("graph") or "")
            if not g:
                leftovers.append(c)
                continue
            _bucket(g, g, _member(c))
        # Resolve each group's title from its source ``_meta`` (mirrors the matrix
        # ``source`` label: the source title, falling back to the graph name).
        for key, grp in groups.items():
            try:
                meta = load_source_meta(loc(key), graphs_dir=base)
            except Exception:  # noqa: BLE001 — a bad/absent meta degrades to the id
                meta = {}
            grp["title"] = str(meta.get("title") or key)

    elif kind in ("note_type", "tag"):
        for c in pool:
            g = str(c.get("graph") or "")
            nid = str(c.get("id") or "")
            try:
                note = _read_note(base, loc(g), nid) if g and nid else None
            except Exception:  # noqa: BLE001 — an unreadable note contributes nothing
                note = None
            values = _value_keys(note, kind) if note is not None else []
            if not values:
                leftovers.append(c)
                continue
            member = _member(c)
            for value in values:
                _bucket(value, value, member)

    else:  # spine
        spine = (spine or "").strip()
        if not spine:
            raise ValueError("A 'spine' organization id is required for the spine axis.")
        from zettelkasten.outline.gather import _resolve_spine_partition_ex

        part = _resolve_spine_partition_ex(
            get_graph, spine, mode=spine_mode, project=project, graph=graph,
            base=base, loc=loc,
        )
        if part is None:
            raise ValueError(
                f"Spine '{spine}' does not resolve to a materialized spine in this scope."
            )
        # Pre-create a theme per dimension (in the matrix's column order) so the
        # returned partition mirrors the spine skeleton even where a dimension has
        # no member claims. Membership is keyed on the SAME spine-side uid index
        # the matrix cells and the outline sections read.
        for node_id, label in part.dimensions:
            if node_id not in groups:
                groups[node_id] = {"title": label or node_id, "claims": []}
        for c in pool:
            member = _member(c)
            uid = member["uid"]
            placed = False
            for node_id, _label in part.dimensions:
                if uid in part.membership.get(node_id, set()):
                    groups[node_id]["claims"].append(member)
                    placed = True
            if not placed:
                leftovers.append(c)

    themes_out: list[dict[str, Any]] = [
        {
            "key": key,
            "title": grp["title"],
            "claims": sorted(grp["claims"], key=lambda m: m["uid"]),
            "count": len(grp["claims"]),
        }
        for key, grp in groups.items()
    ]
    # The SPINE axis mirrors the matrix/outline skeleton: its dimensions were
    # pre-created in the matrix's COLUMN order (including empty ones), so it must
    # PRESERVE that order — a count-sort would drop empty/low-count dimensions out
    # of skeleton position and break matrix↔outline parity. Insertion order is the
    # column order (dict is insertion-ordered), so the spine axis is left unsorted.
    # Every other axis has no skeleton to honor, so it ranks strongest (most
    # populated) first; ties broken by title then key for a fully deterministic
    # order.
    if kind != "spine":
        themes_out.sort(key=lambda t: (-t["count"], t["title"].lower(), t["key"]))

    # The ungrouped remainder (claims with no tag / missing note_type / an
    # unreadable note / an empty graph, or a spine non-member) as members, kept in
    # a stable uid order — computed once and reused for both the trailing theme and
    # the always-present ``unplaced`` signal below.
    leftover_members = sorted((_member(c) for c in leftovers), key=lambda m: m["uid"])

    # The ungrouped remainder rides LAST (never sorted into the ranked themes) when
    # requested, mirroring the outline's trailing unplaced bucket.
    if include_unplaced and leftover_members:
        themes_out.append({
            "key": _UNPLACED_THEME_KEY,
            "title": "Unplaced",
            "claims": leftover_members,
            "count": len(leftover_members),
        })

    return {
        "scope": scope or {},
        "axis": {
            "kind": kind,
            "spine": spine if kind == "spine" else "",
            "include_unplaced": bool(include_unplaced),
        },
        "themes": themes_out,
        # ALWAYS-present leftover signal (additive, independent of
        # ``include_unplaced``): the claims that matched NO group, so the caller can
        # show "N unplaced" and offer them for placement rather than letting them
        # vanish silently. ``include_unplaced`` additionally folds these same claims
        # into ``themes`` as a trailing "Unplaced" section; this signal exposes them
        # even when it is not set.
        "unplaced": leftover_members,
        "unplaced_count": len(leftover_members),
    }

propose_inbound_claims

propose_inbound_claims(get_graph: GetGraph, *, target_theme_id: str, themes: 'list[dict[str, Any]]', unplaced: 'list[dict[str, Any]]', include_placed: bool = True, require_better_fit: bool = True, embed_fn: 'Callable[[str, str], Any] | None' = None, embed_text_fn: 'Callable[[str], Any] | None' = None, placement_floor: float = PLACEMENT_RELEVANCE_FLOOR, poach_margin: float = INBOUND_POACH_MARGIN, scope: 'dict[str, Any] | None' = None) -> dict[str, Any]

Propose claims to pull INTO one target theme — deterministic, read-only.

Builds the TARGET theme's claim CENTROID (over its placed, vectorized claims) and scores candidate claims against it by guarded cosine similarity:

• UNPLACED claims at or above placement_floor become inbound candidates (source "unplaced").
• When include_placed is set, claims currently placed in OTHER themes are considered too. With require_better_fit (the default) such a claim is proposed ONLY when it clears the floor AND beats its CURRENT theme's centroid by at least poach_margin — a comparative gate so a claim that already fits where it lives is never suggested for a move. With require_better_fit=False (the "all claims" scope) EVERY placed claim that clears the floor is surfaced regardless of how well it fits its current theme — its current_similarity is still reported so the author can see, and decide, that it may already be better off where it sits.

Write-free: proposes nothing the author cannot reject and mutates nothing; the caller commits an approved move via set_placement. embed_fn is injectable so tests run without an embedder. The target theme's own claims are never proposed back into it. A target with no usable claim centroid (e.g. an empty or draft section holding only un-vectorized claims) falls back to embedding the theme's OWN text (label + body) via embed_text_fn and scores candidates against THAT query vector, so a new theme can still pull in fitting claims. has_centroid stays False in that fallback (no real claim centroid) but groups may be non-empty; query_source records which query vector was used ("centroid", "theme_text", or "none"). Only when neither a centroid nor a usable theme-text vector exists is the early has_centroid=False / empty-groups result returned.

Returns {scope, target_theme_id, target_theme_label, has_centroid, query_source, groups, embedded} where each group is {source, source_id, source_label, claims} and each claim is {uid, id, graph, title, similarity, current_similarity, current_theme_id, current_theme_label} (the current_* fields are null for an unplaced candidate).

Source code in zettelkasten/claim_producer.py

def propose_inbound_claims(
    get_graph: GetGraph,
    *,
    target_theme_id: str,
    themes: "list[dict[str, Any]]",
    unplaced: "list[dict[str, Any]]",
    include_placed: bool = True,
    require_better_fit: bool = True,
    embed_fn: "Callable[[str, str], Any] | None" = None,
    embed_text_fn: "Callable[[str], Any] | None" = None,
    placement_floor: float = PLACEMENT_RELEVANCE_FLOOR,
    poach_margin: float = INBOUND_POACH_MARGIN,
    scope: "dict[str, Any] | None" = None,
) -> dict[str, Any]:
    """Propose claims to pull INTO one target theme — deterministic, read-only.

    Builds the TARGET theme's claim CENTROID (over its placed, vectorized claims)
    and scores candidate claims against it by guarded cosine similarity:

    * UNPLACED claims at or above ``placement_floor`` become inbound candidates
      (source ``"unplaced"``).
    * When ``include_placed`` is set, claims currently placed in OTHER themes are
      considered too. With ``require_better_fit`` (the default) such a claim is
      proposed ONLY when it clears the floor AND beats its CURRENT theme's centroid
      by at least ``poach_margin`` — a comparative gate so a claim that already fits
      where it lives is never suggested for a move. With ``require_better_fit=False``
      (the "all claims" scope) EVERY placed claim that clears the floor is surfaced
      regardless of how well it fits its current theme — its ``current_similarity``
      is still reported so the author can see, and decide, that it may already be
      better off where it sits.

    Write-free: proposes nothing the author cannot reject and mutates nothing; the
    caller commits an approved move via ``set_placement``. ``embed_fn`` is
    injectable so tests run without an embedder. The target theme's own claims are
    never proposed back into it. A target with no usable claim centroid (e.g. an
    empty or draft section holding only un-vectorized claims) falls back to
    embedding the theme's OWN text (``label`` + ``body``) via ``embed_text_fn`` and
    scores candidates against THAT query vector, so a new theme can still pull in
    fitting claims. ``has_centroid`` stays False in that fallback (no real claim
    centroid) but ``groups`` may be non-empty; ``query_source`` records which query
    vector was used (``"centroid"``, ``"theme_text"``, or ``"none"``). Only when
    neither a centroid nor a usable theme-text vector exists is the early
    ``has_centroid=False`` / empty-groups result returned.

    Returns ``{scope, target_theme_id, target_theme_label, has_centroid,
    query_source, groups, embedded}`` where each group is
    ``{source, source_id, source_label, claims}``
    and each claim is ``{uid, id, graph, title, similarity, current_similarity,
    current_theme_id, current_theme_label}`` (the ``current_*`` fields are null for
    an unplaced candidate).
    """
    from zettelkasten.claims import claim_uid
    from zettelkasten.syllabus import _centroid

    target_theme_id = str(target_theme_id or "")
    get_vec = _vector_provider(get_graph, embed_fn)
    vec_cache: dict[tuple[str, str], Any] = {}

    def vec_for(graph: str, nid: str) -> Any:
        key = (graph, nid)
        if key not in vec_cache:
            vec_cache[key] = get_vec(graph, nid)
        return vec_cache[key]

    def centroid_for(theme: "dict[str, Any]") -> Any:
        member_vecs = [
            v
            for c in (theme.get("claims") or [])
            if _is_usable_vector(v := vec_for(str(c.get("graph") or ""), str(c.get("id") or "")))
        ]
        if not member_vecs:
            return None
        cen = _centroid(member_vecs)
        return cen if _is_usable_vector(cen) else None

    def make_row(
        c: "dict[str, Any]",
        sim: float,
        *,
        current_id: "str | None" = None,
        current_label: "str | None" = None,
        current_sim: "float | None" = None,
    ) -> dict[str, Any]:
        graph = str(c.get("graph") or "")
        nid = str(c.get("id") or "")
        return {
            "uid": c.get("uid") or claim_uid((graph, nid)),
            "id": nid,
            "graph": graph,
            "title": c.get("title") or nid,
            "similarity": round(sim, 6),
            "current_similarity": round(current_sim, 6) if current_sim is not None else None,
            "current_theme_id": current_id,
            "current_theme_label": current_label,
        }

    target = next((t for t in themes if str(t.get("id") or "") == target_theme_id), None)
    target_label = str((target or {}).get("label") or target_theme_id)
    target_centroid = centroid_for(target) if target is not None else None

    # Whether the target has a real claim centroid drives ``has_centroid`` below —
    # existing callers/tests rely on it being False when the theme holds no placed,
    # vectorized claims, even though the theme-text fallback may still find matches.
    has_centroid = _is_usable_vector(target_centroid)

    # The vector candidate claims are scored against. Normally the target's claim
    # centroid; for an empty/draft theme with no usable centroid we fall back to
    # embedding the theme's own text (label + body) so a new theme can still pull
    # in fitting claims. ``query_source`` records which was used.
    query_vec = target_centroid
    query_source = "centroid" if has_centroid else "none"
    if not has_centroid and target is not None:
        embed_text = embed_text_fn or _default_embed_text
        parts = [
            str(target.get("label") or "").strip(),
            str(target.get("body") or "").strip(),
        ]
        theme_text = ". ".join(p for p in parts if p).strip()
        if theme_text:
            try:
                fallback_vec = embed_text(theme_text)
            except Exception:  # pragma: no cover - defensive (offline / no model)
                fallback_vec = None
            if _is_usable_vector(fallback_vec):
                query_vec = fallback_vec
                query_source = "theme_text"

    # No vector to match against (empty/draft theme with no usable centroid and no
    # usable theme-text fallback): nothing to pull in. The ``embedded`` flag still
    # reports whether any vector was usable so the UI can tell "offline" apart from
    # "this theme has no grounded claims yet".
    if not _is_usable_vector(query_vec):
        return {
            "scope": scope or {},
            "target_theme_id": target_theme_id,
            "target_theme_label": target_label,
            "has_centroid": False,
            "query_source": "none",
            "groups": [],
            "embedded": any(_is_usable_vector(v) for v in vec_cache.values()),
        }

    # Claims that already have a theme home. Reconcile can surface a theme-homed
    # but un-ordered claim in BOTH its theme and the ``unplaced`` worklist (the
    # worklist flags it as still needing attention); without this guard such a
    # claim would appear twice here — once under "Unplaced" and again under its
    # theme group. Exclude any unplaced candidate already placed in a theme so it
    # is only ever offered from where it actually sits.
    placed_uids: set[str] = set()
    for t in themes:
        for c in (t.get("claims") or []):
            uid = c.get("uid") or claim_uid((str(c.get("graph") or ""), str(c.get("id") or "")))
            placed_uids.add(uid)

    # Unplaced candidates: any usable claim at/above the floor (no current theme,
    # so the comparative margin does not apply).
    unplaced_rows: list[dict[str, Any]] = []
    for c in unplaced:
        uid = c.get("uid") or claim_uid((str(c.get("graph") or ""), str(c.get("id") or "")))
        if uid in placed_uids:
            continue
        v = vec_for(str(c.get("graph") or ""), str(c.get("id") or ""))
        if not _is_usable_vector(v):
            continue
        sim = _guarded_cosine_sim(v, query_vec)
        if sim is None or sim < placement_floor:
            continue
        unplaced_rows.append(make_row(c, sim))
    unplaced_rows.sort(key=lambda r: (-r["similarity"], r["uid"]))

    groups: list[dict[str, Any]] = []
    if unplaced_rows:
        groups.append({
            "source": "unplaced",
            "source_id": None,
            "source_label": "Unplaced",
            "claims": unplaced_rows,
        })

    # Placed candidates: claims currently sitting in OTHER themes. In poach mode
    # (``require_better_fit``) only those that fit the target better than their
    # current theme by the margin are surfaced — the current theme's centroid
    # INCLUDES the claim itself, which only inflates its current-fit, a conservative
    # bias that makes poaching harder, never easier. In literal "all" mode the gate
    # is dropped: every placed claim clearing the floor is surfaced, with its
    # current-theme fit reported so the author can judge whether to move it.
    if include_placed:
        poach_groups: list[dict[str, Any]] = []
        for t in themes:
            tid = str(t.get("id") or "")
            if not tid or tid == target_theme_id:
                continue
            cur_centroid = centroid_for(t)
            cur_usable = _is_usable_vector(cur_centroid)
            # Poaching needs a current-theme centroid to beat; without one there is
            # nothing to compare against, so skip the theme. Literal "all" mode has
            # no such requirement — surface its claims with an unknown current fit.
            if require_better_fit and not cur_usable:
                continue
            cur_label = str(t.get("label") or tid)
            rows: list[dict[str, Any]] = []
            for c in (t.get("claims") or []):
                v = vec_for(str(c.get("graph") or ""), str(c.get("id") or ""))
                if not _is_usable_vector(v):
                    continue
                sim = _guarded_cosine_sim(v, query_vec)
                if sim is None or sim < placement_floor:
                    continue
                cur_sim = _guarded_cosine_sim(v, cur_centroid) if cur_usable else None
                if require_better_fit:
                    cur_val = cur_sim if cur_sim is not None else -1.0
                    if sim < cur_val + poach_margin:
                        continue
                rows.append(make_row(c, sim, current_id=tid, current_label=cur_label, current_sim=cur_sim))
            if rows:
                rows.sort(key=lambda r: (-r["similarity"], r["uid"]))
                poach_groups.append({
                    "source": "theme",
                    "source_id": tid,
                    "source_label": cur_label,
                    "claims": rows,
                })
        # Strongest source first (by its best candidate), ties by label.
        poach_groups.sort(key=lambda g: (-(g["claims"][0]["similarity"]), g["source_label"]))
        groups.extend(poach_groups)

    return {
        "scope": scope or {},
        "target_theme_id": target_theme_id,
        "target_theme_label": target_label,
        "has_centroid": has_centroid,
        "query_source": query_source,
        "groups": groups,
        "embedded": any(_is_usable_vector(v) for v in vec_cache.values()),
    }

mine_claims_for_topic

mine_claims_for_topic(get_graph: GetGraph, *, text: str = '', project: str = '', graph: str = '', top_k: int = MINE_MAX_MEMBERS, relevance_floor: float = MINE_RELEVANCE_FLOOR, body_chars: int = MINE_BODY_CHARS, fulltext_chars: int = MINE_FULLTEXT_CHARS, include_fulltext: bool = True, propose_fn: 'Callable[[str, str], str] | None' = None, embed_fn: 'Callable[[str, str], Any] | None' = None, embed_text_fn: 'Callable[[str], Any] | None' = None, fulltext_fn: 'Callable[[str], str | None] | None' = None, graphs_dir: 'Path | str | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Mine NEW grounded claim candidates for a pitched topic — full-text, read-only.

GATHERS the corpus notes most similar to text (ephemeral topic embed + guarded cosine, relevance_floor gate, top top_k), enriches them with their FULL note bodies AND each source's extracted PDF fulltext (when a local PDF is available — controlled by include_fulltext / fulltext_fn), then asks the WRITE-FREE phrasing agent (propose_fn) to paraphrase ONE grounded claim sentence (plus an optional counterclaim) from that material. It NEVER invents facts and writes NOTHING — it returns candidate SENTENCES the caller can land as overlay draft claims. Already-synthesized _cross claims and concept hubs are excluded from the pool (mining produces fresh synthesis).

All model/IO seams are injectable so tests run with no embedder, no LLM, and no PDF stack. Returns {scope, topic, candidates, members, sources_read, fulltext_used, embedded}.

Source code in zettelkasten/claim_producer.py

def mine_claims_for_topic(
    get_graph: GetGraph,
    *,
    text: str = "",
    project: str = "",
    graph: str = "",
    top_k: int = MINE_MAX_MEMBERS,
    relevance_floor: float = MINE_RELEVANCE_FLOOR,
    body_chars: int = MINE_BODY_CHARS,
    fulltext_chars: int = MINE_FULLTEXT_CHARS,
    include_fulltext: bool = True,
    propose_fn: "Callable[[str, str], str] | None" = None,
    embed_fn: "Callable[[str, str], Any] | None" = None,
    embed_text_fn: "Callable[[str], Any] | None" = None,
    fulltext_fn: "Callable[[str], str | None] | None" = None,
    graphs_dir: "Path | str | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Mine NEW grounded claim candidates for a pitched topic — full-text, read-only.

    GATHERS the corpus notes most similar to ``text`` (ephemeral topic embed +
    guarded cosine, ``relevance_floor`` gate, top ``top_k``), enriches them with
    their FULL note bodies AND each source's extracted PDF fulltext (when a local
    PDF is available — controlled by ``include_fulltext`` / ``fulltext_fn``), then
    asks the WRITE-FREE phrasing agent (``propose_fn``) to paraphrase ONE grounded
    claim sentence (plus an optional counterclaim) from that material. It NEVER
    invents facts and writes NOTHING — it returns candidate SENTENCES the caller
    can land as overlay draft claims. Already-synthesized ``_cross`` claims and
    ``concept`` hubs are excluded from the pool (mining produces fresh synthesis).

    All model/IO seams are injectable so tests run with no embedder, no LLM, and
    no PDF stack. Returns ``{scope, topic, candidates, members, sources_read,
    fulltext_used, embedded}``.
    """
    from zettelkasten.claims import build_claim_index

    text = (text or "").strip()
    base = _base(graphs_dir)
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    scope = {"type": "graph" if graph else "project",
             "name": graph or project or "all", "topic": text}
    if not text:
        raise ValueError("mine_claims_for_topic requires a non-empty topic text.")

    index = build_claim_index(
        get_graph, project=project, graph=graph, graphs_dir=base,
        namespace=ns, localize=loc,
    )
    embed_text = embed_text_fn or _default_embed_text
    hvec = embed_text(text)
    embedded = _is_usable_vector(hvec)
    get_vec = _vector_provider(get_graph, embed_fn)

    scored: list[tuple[float, tuple[str, str], Any]] = []
    if embedded:
        for key, note in index.notes_by_key.items():
            g, nid = key
            if (note.type or "") in _MINE_EXCLUDED_TYPES:
                continue
            # Skip already-synthesized cross claims — mining produces NEW synthesis.
            if g == CROSS_GRAPH:
                continue
            vec = get_vec(g, nid)
            if not _is_usable_vector(vec):
                continue
            sim = _guarded_cosine_sim(hvec, vec)
            if sim is None or sim < relevance_floor:
                continue
            scored.append((round(sim, 6), key, note))
    scored.sort(key=lambda t: (-t[0], t[1]))
    chosen = scored[: max(0, top_k)]

    empty = {
        "scope": scope, "topic": text, "candidates": [], "members": [],
        "sources_read": [], "fulltext_used": [], "embedded": embedded,
    }
    if not chosen:
        return empty

    members: list[dict[str, Any]] = []
    sources_read: list[str] = []
    for sim, key, note in chosen:
        g, nid = key
        members.append({
            "id": nid, "graph": g, "title": note.title or nid,
            "type": note.type or "", "body": (note.body or "").strip()[:body_chars],
            "similarity": sim,
        })
        if g not in sources_read and g not in _NON_SOURCE_GRAPHS:
            sources_read.append(g)

    source_texts: list[dict[str, Any]] = []
    fulltext_used: list[str] = []
    if include_fulltext:
        ft_fn = fulltext_fn or _default_fulltext_fn
        for g in sources_read:
            ft = ft_fn(g)
            if ft and ft.strip():
                source_texts.append({"graph": g, "text": ft.strip()[:fulltext_chars]})
                fulltext_used.append(g)

    fn = propose_fn or _default_propose_fn
    parsed = _parse_propose_response(fn(_PROPOSE_CONTRACT, _build_mine_prompt(text, members, source_texts)))
    member_keys = {(m["graph"], m["id"]) for m in members}
    stance_by_key: dict[tuple[str, str], dict[str, Any]] = {}
    for st in parsed.get("stances") or []:
        if not isinstance(st, dict):
            continue
        mid, mgraph = str(st.get("id") or ""), str(st.get("graph") or "")
        stance = str(st.get("stance") or "")
        if (mgraph, mid) not in member_keys or stance not in _PROPOSAL_STANCES:
            continue
        stance_by_key[(mgraph, mid)] = {
            "id": mid, "graph": mgraph, "stance": stance,
            "confidence": _coerce_confidence(st.get("confidence")),
        }

    def _prov(keys: "set[tuple[str, str]]") -> list[str]:
        return sorted({g for (g, _i) in keys if g not in _NON_SOURCE_GRAPHS})

    candidates: list[dict[str, Any]] = []
    sentence = str(parsed.get("sentence") or "").strip()
    if sentence:
        primary_keys = {
            k for k in member_keys
            if stance_by_key.get(k, {}).get("stance") != "contradicts"
        } or set(member_keys)
        candidates.append({
            "sentence": sentence,
            "members": [stance_by_key.get(k, {"id": k[1], "graph": k[0], "stance": "supports", "confidence": None})
                        for k in sorted(primary_keys)],
            "provenance": _prov(primary_keys),
            "is_counterclaim": False,
        })
    contra_keys = {k for k, s in stance_by_key.items() if s["stance"] == "contradicts"}
    counter = str(parsed.get("counterclaim_sentence") or "").strip()
    if counter and contra_keys:
        candidates.append({
            "sentence": counter,
            "members": [{"id": k[1], "graph": k[0], "stance": "supports",
                         "confidence": stance_by_key[k]["confidence"]}
                        for k in sorted(contra_keys)],
            "provenance": _prov(contra_keys),
            "is_counterclaim": True,
        })

    return {
        "scope": scope, "topic": text, "candidates": candidates,
        "members": members, "sources_read": sources_read,
        "fulltext_used": fulltext_used, "embedded": embedded,
    }

propose_mine_sources

propose_mine_sources(get_graph: GetGraph, *, target_theme_id: str, themes: 'list[dict[str, Any]]', project: str = '', graph: str = '', relevance_floor: float = MINE_RELEVANCE_FLOOR, top_n: int = MINE_SOURCES_TOP_N, embed_fn: 'Callable[[str, str], Any] | None' = None, embed_text_fn: 'Callable[[str], Any] | None' = None, graphs_dir: 'Path | str | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None, scope: 'dict[str, Any] | None' = None) -> dict[str, Any]

Propose candidate SOURCE papers to mine for one theme — read-only.

The first half of the interactive "Mine from sources" flow. Returns two flavours of source, ranked for the user to pick from:

• in_theme — the source graphs ALREADY backing the target theme's existing claims/papers (its claim provenance + holding pen).
• proposed — OTHER source papers WITHIN THE CURRENT PROJECT/GRAPH SCOPE whose notes are similar to the theme (the theme's label + body are embedded, the scope's notes scored by guarded cosine, aggregated to the source-graph level, gated at relevance_floor, and capped to top_n). Sources already in_theme are excluded from this list.

Each returned source is {graph, title, similarity, tag} where tag is "in_theme" or "proposed". similarity is the source's best note similarity to the theme (None for an in_theme source when no vector is available). Write-free; embed_fn / embed_text_fn are injectable so tests run with no embedder. Returns {scope, theme_id, theme_label, sources, embedded}.

Source code in zettelkasten/claim_producer.py

def propose_mine_sources(
    get_graph: GetGraph,
    *,
    target_theme_id: str,
    themes: "list[dict[str, Any]]",
    project: str = "",
    graph: str = "",
    relevance_floor: float = MINE_RELEVANCE_FLOOR,
    top_n: int = MINE_SOURCES_TOP_N,
    embed_fn: "Callable[[str, str], Any] | None" = None,
    embed_text_fn: "Callable[[str], Any] | None" = None,
    graphs_dir: "Path | str | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
    scope: "dict[str, Any] | None" = None,
) -> dict[str, Any]:
    """Propose candidate SOURCE papers to mine for one theme — read-only.

    The first half of the interactive "Mine from sources" flow. Returns two
    flavours of source, ranked for the user to pick from:

    * ``in_theme`` — the source graphs ALREADY backing the target theme's existing
      claims/papers (its claim provenance + holding pen).
    * ``proposed`` — OTHER source papers WITHIN THE CURRENT PROJECT/GRAPH SCOPE
      whose notes are similar to the theme (the theme's ``label`` + ``body`` are
      embedded, the scope's notes scored by guarded cosine, aggregated to the
      source-graph level, gated at ``relevance_floor``, and capped to ``top_n``).
      Sources already ``in_theme`` are excluded from this list.

    Each returned source is ``{graph, title, similarity, tag}`` where ``tag`` is
    ``"in_theme"`` or ``"proposed"``. ``similarity`` is the source's best note
    similarity to the theme (``None`` for an ``in_theme`` source when no vector is
    available). Write-free; ``embed_fn`` / ``embed_text_fn`` are injectable so
    tests run with no embedder. Returns ``{scope, theme_id, theme_label, sources,
    embedded}``.
    """
    from zettelkasten.claims import build_claim_index

    base = _base(graphs_dir)
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    scope = scope or {"type": "graph" if graph else "project",
                      "name": graph or project or "all"}
    target_theme_id = str(target_theme_id or "")

    target = next(
        (t for t in (themes or []) if str(t.get("id") or "") == target_theme_id),
        None,
    )
    theme_label = str((target or {}).get("label") or target_theme_id)
    theme_body = str((target or {}).get("body") or "")

    title_by_graph = _source_title_map(themes or [])
    in_theme = _theme_source_graphs(target)

    # Embed the theme text once; score the scope's notes against it and aggregate
    # to a per-source-graph best similarity. This drives both the ``proposed`` set
    # and the ``in_theme`` rows' similarity display.
    parts = [theme_label.strip(), theme_body.strip()]
    theme_text = ". ".join(p for p in parts if p).strip()
    embed_text = embed_text_fn or _default_embed_text
    qvec = None
    if theme_text:
        try:
            qvec = embed_text(theme_text)
        except Exception:  # pragma: no cover - defensive (offline / no model)
            qvec = None
    embedded = _is_usable_vector(qvec)

    per_graph_sim: dict[str, float] = {}
    if embedded:
        index = build_claim_index(
            get_graph, project=project, graph=graph, graphs_dir=base,
            namespace=ns, localize=loc,
        )
        get_vec = _vector_provider(get_graph, embed_fn)
        for key, note in index.notes_by_key.items():
            g, nid = key
            if g in _NON_SOURCE_GRAPHS:
                continue
            if (note.type or "") in _MINE_EXCLUDED_TYPES:
                continue
            vec = get_vec(g, nid)
            if not _is_usable_vector(vec):
                continue
            sim = _guarded_cosine_sim(qvec, vec)
            if sim is None:
                continue
            if sim > per_graph_sim.get(g, -1.0):
                per_graph_sim[g] = sim

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

    # in_theme first (strongest theme-similarity first, then slug), each tagged so
    # the picker can group/label them.
    in_rows: list[tuple[float, str]] = []
    for g in in_theme:
        sim = per_graph_sim.get(g)
        in_rows.append((sim if sim is not None else -1.0, g))
    in_rows.sort(key=lambda t: (-t[0], t[1]))
    for sim_val, g in in_rows:
        sim = per_graph_sim.get(g)
        sources.append({
            "graph": g,
            "title": title_by_graph.get(g, g),
            "similarity": round(sim, 6) if sim is not None else None,
            "tag": "in_theme",
        })

    # proposed: in-scope sources above the floor, not already in the theme, ranked.
    proposed: list[tuple[float, str]] = [
        (sim, g) for g, sim in per_graph_sim.items()
        if g not in in_theme and sim >= relevance_floor
    ]
    proposed.sort(key=lambda t: (-t[0], t[1]))
    for sim, g in proposed[: max(0, top_n)]:
        sources.append({
            "graph": g,
            "title": title_by_graph.get(g, g),
            "similarity": round(sim, 6),
            "tag": "proposed",
        })

    return {
        "scope": scope,
        "theme_id": target_theme_id,
        "theme_label": theme_label,
        "sources": sources,
        "embedded": embedded,
    }

mine_atomic_claims

mine_atomic_claims(get_graph: GetGraph, *, target_theme_id: str, themes: 'list[dict[str, Any]]', graphs: 'list[str]', project: str = '', graph: str = '', max_per_source: int = MINE_ATOMIC_MAX_PER_SOURCE, max_members: int = MINE_MAX_MEMBERS, body_chars: int = MINE_BODY_CHARS, fulltext_chars: int = MINE_ATOMIC_FULLTEXT_CHARS, extract_fn: 'Callable[[str, str], str] | None' = None, embed_fn: 'Callable[[str, str], Any] | None' = None, embed_text_fn: 'Callable[[str], Any] | None' = None, fulltext_fn: 'Callable[[str], str | None] | None' = None, fuzzy_floor: 'float | None' = None, graphs_dir: 'Path | str | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None, scope: 'dict[str, Any] | None' = None) -> dict[str, Any]

Extract MULTIPLE atomic, grounded claims per selected source — read-only.

The second half of the interactive "Mine from sources" flow. For each SELECTED source graph (from :func:propose_mine_sources), it gathers that source's notes (the most theme-similar first when an embedder is available), RE-READS the source's extracted PDF fulltext (fulltext_fn), and asks the WRITE-FREE extraction agent (extract_fn) for a LIST of ATOMIC, theme-focused claim sentences, each with a VERBATIM supporting quote. Every quote is mechanically VERIFIED against that source's fulltext via :func:grounding.verify_quote — any claim whose quote does not verify is DROPPED, so the agent can surface but never fabricate. Up to max_per_source verified claims per source survive.

Writes NOTHING — it returns candidate claims the caller lands as overlay draft claims (each carrying its verified quote + the source's member note ids for a later promote). All model/IO seams are injectable so tests run with no embedder, no LLM, and no PDF stack. Returns {scope, theme_id, theme_label, claims, sources_read, sources_skipped, rejected, embedded} where each claim is {sentence, quote, confidence, source_graph, source_title, supports, method, score, page}.

Source code in zettelkasten/claim_producer.py

def mine_atomic_claims(
    get_graph: GetGraph,
    *,
    target_theme_id: str,
    themes: "list[dict[str, Any]]",
    graphs: "list[str]",
    project: str = "",
    graph: str = "",
    max_per_source: int = MINE_ATOMIC_MAX_PER_SOURCE,
    max_members: int = MINE_MAX_MEMBERS,
    body_chars: int = MINE_BODY_CHARS,
    fulltext_chars: int = MINE_ATOMIC_FULLTEXT_CHARS,
    extract_fn: "Callable[[str, str], str] | None" = None,
    embed_fn: "Callable[[str, str], Any] | None" = None,
    embed_text_fn: "Callable[[str], Any] | None" = None,
    fulltext_fn: "Callable[[str], str | None] | None" = None,
    fuzzy_floor: "float | None" = None,
    graphs_dir: "Path | str | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
    scope: "dict[str, Any] | None" = None,
) -> dict[str, Any]:
    """Extract MULTIPLE atomic, grounded claims per selected source — read-only.

    The second half of the interactive "Mine from sources" flow. For each SELECTED
    source graph (from :func:`propose_mine_sources`), it gathers that source's
    notes (the most theme-similar first when an embedder is available), RE-READS
    the source's extracted PDF fulltext (``fulltext_fn``), and asks the WRITE-FREE
    extraction agent (``extract_fn``) for a LIST of ATOMIC, theme-focused claim
    sentences, each with a VERBATIM supporting quote. Every quote is mechanically
    VERIFIED against that source's fulltext via :func:`grounding.verify_quote` —
    any claim whose quote does not verify is DROPPED, so the agent can surface but
    never fabricate. Up to ``max_per_source`` verified claims per source survive.

    Writes NOTHING — it returns candidate claims the caller lands as overlay draft
    claims (each carrying its verified quote + the source's member note ids for a
    later promote). All model/IO seams are injectable so tests run with no
    embedder, no LLM, and no PDF stack. Returns ``{scope, theme_id, theme_label,
    claims, sources_read, sources_skipped, rejected, embedded}`` where each claim is
    ``{sentence, quote, confidence, source_graph, source_title, supports, method,
    score, page}``.
    """
    from zettelkasten import grounding
    from zettelkasten.claims import build_claim_index

    base = _base(graphs_dir)
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    scope = scope or {"type": "graph" if graph else "project",
                      "name": graph or project or "all"}
    target_theme_id = str(target_theme_id or "")

    target = next(
        (t for t in (themes or []) if str(t.get("id") or "") == target_theme_id),
        None,
    )
    theme_label = str((target or {}).get("label") or target_theme_id)
    theme_body = str((target or {}).get("body") or "")
    title_by_graph = _source_title_map(themes or [])

    # Only real source graphs may be mined (never _cross / _citations).
    selected = [g for g in (graphs or []) if g and g not in _NON_SOURCE_GRAPHS]

    empty = {
        "scope": scope, "theme_id": target_theme_id, "theme_label": theme_label,
        "claims": [], "sources_read": [], "sources_skipped": [],
        "rejected": 0, "embedded": False,
    }
    if not selected:
        return empty

    index = build_claim_index(
        get_graph, project=project, graph=graph, graphs_dir=base,
        namespace=ns, localize=loc,
    )

    # Theme query vector for ranking members within a source (best-effort — without
    # an embedder we fall back to the source's notes in index order).
    parts = [theme_label.strip(), theme_body.strip()]
    theme_text = ". ".join(p for p in parts if p).strip()
    embed_text = embed_text_fn or _default_embed_text
    qvec = None
    if theme_text:
        try:
            qvec = embed_text(theme_text)
        except Exception:  # pragma: no cover - defensive (offline / no model)
            qvec = None
    embedded = _is_usable_vector(qvec)
    get_vec = _vector_provider(get_graph, embed_fn)

    ft_fn = fulltext_fn or _default_fulltext_fn
    ef = extract_fn or _default_extract_fn

    claims_out: list[dict[str, Any]] = []
    sources_read: list[str] = []
    sources_skipped: list[dict[str, Any]] = []
    rejected = 0

    for g in selected:
        # Gather this source's mineable notes, ranked by theme similarity when an
        # embedder is available (un-embedded notes sort last but still participate).
        pool: list[tuple[float, tuple[str, str], Any]] = []
        for key, note in index.notes_by_key.items():
            if key[0] != g:
                continue
            if (note.type or "") in _MINE_EXCLUDED_TYPES:
                continue
            sim = -1.0
            if embedded:
                vec = get_vec(g, key[1])
                s = _guarded_cosine_sim(qvec, vec) if _is_usable_vector(vec) else None
                if s is not None:
                    sim = s
            pool.append((sim, key, note))
        if not pool:
            sources_skipped.append({"graph": g, "reason": "no notes in scope"})
            continue
        pool.sort(key=lambda t: (-t[0], t[1]))
        chosen = pool[: max(0, max_members)]
        members = [
            {"id": k[1], "graph": k[0], "title": note.title or k[1],
             "type": note.type or "", "body": (note.body or "").strip()[:body_chars]}
            for _s, k, note in chosen
        ]

        fulltext = ft_fn(g)
        if not fulltext or not fulltext.strip():
            sources_skipped.append({"graph": g, "reason": "no local full text"})
            continue
        sources_read.append(g)

        try:
            raw = ef(
                _MINE_ATOMIC_CONTRACT,
                _build_mine_atomic_prompt(
                    theme_label, theme_body, members, fulltext.strip()[:fulltext_chars]
                ),
            )
        except Exception as exc:  # pragma: no cover - defensive (agent/LLM failure)
            logger.warning("mine_atomic_claims extraction failed for %s: %s", g, exc)
            sources_skipped.append({"graph": g, "reason": "extraction failed"})
            continue

        # The member note ids ground a surviving claim's provenance (its ``supports``
        # for a later promote). Carried whole — a mined draft remembers the source's
        # notes it was extracted alongside.
        supports = [{"id": m["id"], "graph": m["graph"]} for m in members]

        seen: set[str] = set()
        kept = 0
        for cand in _parse_atomic_response(raw):
            if kept >= max(0, max_per_source):
                break
            quote = cand["quote"]
            norm = grounding.normalize_for_match(quote)
            if not norm or norm in seen:
                continue
            if fuzzy_floor is None:
                verdict = grounding.verify_quote(quote, fulltext, None)
            else:
                verdict = grounding.verify_quote(quote, fulltext, None, fuzzy_floor=fuzzy_floor)
            # HARD verbatim gate: an atomic claim must carry a TRUE verbatim quote.
            # ``verify_quote``'s fuzzy fallback accepts a SequenceMatcher near-match
            # (e.g. "50 mg" vs "90 mg" at ratio ~0.97), which is NOT verbatim and
            # would let the agent's edited string land as "verified". Accept ONLY a
            # genuine excerpt — an exact normalized substring of the source text
            # ("exact") or a curated highlight match ("annotation") — and DROP a
            # merely-fuzzy (or unverified) quote outright.
            if not verdict.get("verified") or verdict.get("method") not in ("exact", "annotation"):
                rejected += 1
                continue
            seen.add(norm)
            claims_out.append({
                "sentence": cand["sentence"],
                "quote": quote,
                "confidence": cand["confidence"],
                "source_graph": g,
                "source_title": title_by_graph.get(g, g),
                "supports": supports,
                "method": verdict.get("method"),
                "score": verdict.get("score"),
                "page": verdict.get("page"),
            })
            kept += 1

    return {
        "scope": scope,
        "theme_id": target_theme_id,
        "theme_label": theme_label,
        "claims": claims_out,
        "sources_read": sources_read,
        "sources_skipped": sources_skipped,
        "rejected": rejected,
        "embedded": embedded,
    }

ground_claim

ground_claim(get_graph: GetGraph, *, claim_id: str, project: str = '', graph: str = '', extract_fn: 'Callable[[str, str], str] | None' = None, corpus_fn: 'Callable[[str], tuple[str | None, list[dict[str, Any]]]] | None' = None, max_per_source: int = GROUND_MAX_QUOTES_PER_SOURCE, max_sources: int = GROUND_MAX_SOURCES, fulltext_chars: int = GROUND_FULLTEXT_CHARS, fuzzy_floor: 'float | None' = None, confidence: 'float | None' = None, graphs_dir: 'Path | str | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Close an evidential gap by SCRAPING a claim's own sources for quotes — WRITES.

The actionable counterpart to :func:analyze_gaps' evidential finding ("thin claim — sparse evidence"). Resolves claim_id in the scoped claim index, walks its DISTINCT supporting source documents, RE-READS each source's extracted PDF fulltext, and asks the WRITE-FREE extraction agent (extract_fn) for VERBATIM sentences that support the claim. Each candidate is mechanically VERIFIED against the same source text via :func:grounding.verify_quote — anything that is not a real (exact/fuzzy/annotation) substring is REJECTED, so the agent can surface but never fabricate evidence. Accepted, deduped quotes are appended as grounded quote notes (supports → the claim) under a distinct -g* id namespace.

All model/IO seams are injectable (extract_fn / corpus_fn) so tests run with no LLM and no PDF stack. Writes to _cross → refused on a federated / read-only review. Returns {scope, claim_id, title, sources_scanned, sources_skipped, quotes_added, rejected, already_present, message}.

Source code in zettelkasten/claim_producer.py

def ground_claim(
    get_graph: GetGraph,
    *,
    claim_id: str,
    project: str = "",
    graph: str = "",
    extract_fn: "Callable[[str, str], str] | None" = None,
    corpus_fn: "Callable[[str], tuple[str | None, list[dict[str, Any]]]] | None" = None,
    max_per_source: int = GROUND_MAX_QUOTES_PER_SOURCE,
    max_sources: int = GROUND_MAX_SOURCES,
    fulltext_chars: int = GROUND_FULLTEXT_CHARS,
    fuzzy_floor: "float | None" = None,
    confidence: "float | None" = None,
    graphs_dir: "Path | str | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Close an evidential gap by SCRAPING a claim's own sources for quotes — WRITES.

    The actionable counterpart to :func:`analyze_gaps`' ``evidential`` finding
    ("thin claim — sparse evidence"). Resolves ``claim_id`` in the scoped claim
    index, walks its DISTINCT supporting source documents, RE-READS each source's
    extracted PDF fulltext, and asks the WRITE-FREE extraction agent (``extract_fn``)
    for VERBATIM sentences that support the claim. Each candidate is mechanically
    VERIFIED against the same source text via :func:`grounding.verify_quote` —
    anything that is not a real (exact/fuzzy/annotation) substring is REJECTED, so
    the agent can surface but never fabricate evidence. Accepted, deduped quotes
    are appended as grounded ``quote`` notes (``supports`` → the claim) under a
    distinct ``-g*`` id namespace.

    All model/IO seams are injectable (``extract_fn`` / ``corpus_fn``) so tests run
    with no LLM and no PDF stack. Writes to ``_cross`` → refused on a federated /
    read-only review. Returns ``{scope, claim_id, title, sources_scanned,
    sources_skipped, quotes_added, rejected, already_present, message}``.
    """
    from zettelkasten import grounding
    from zettelkasten.claims import _build_context

    cid = (claim_id or "").strip()
    if not cid:
        raise ValueError("ground_claim requires a claim id.")
    validate_id(cid, kind="claim id", for_filename=True)
    base = _base(graphs_dir)
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    scope = {"type": "graph" if graph else "project", "name": graph or project or "all"}

    ctx = _build_context(
        get_graph, project=project, graph=graph,
        graphs_dir=base, namespace=ns, localize=loc,
    )
    index = ctx[0]

    target = None
    for key, rc in index.resolved.items():
        if rc.note.id == cid or key[1] == cid:
            target = rc
            break
    if target is None:
        raise ValueError(f"ground_claim: claim '{cid}' not found in scope.")

    title = (target.note.title or cid).strip()
    body = target.note.body or ""

    # Existing evidence (any id namespace) — dedupe so re-grounding is idempotent
    # and we never re-add a passage the claim already carries.
    seen: set[str] = set()
    for q in target.supporting_quotes:
        norm = grounding.normalize_for_match(q.get("body") or "")
        if norm:
            seen.add(norm)
    already_present = len(seen)

    # The claim's strongest-evidence sources first; bound the scan.
    sources = list(target.support_sources)[: max(0, max_sources)]

    cf = corpus_fn or _default_corpus_fn
    ef = extract_fn or _default_extract_fn

    sources_scanned: list[str] = []
    sources_skipped: list[dict[str, Any]] = []
    accepted: list[dict[str, Any]] = []
    rejected = 0

    for g in sources:
        fulltext, annotations = cf(g)
        if not fulltext or not fulltext.strip():
            sources_skipped.append({"graph": g, "reason": "no local full text"})
            continue
        sources_scanned.append(g)
        try:
            parsed = _parse_propose_response(
                ef(_EXTRACT_CONTRACT, _build_extract_prompt(title, body, fulltext[:fulltext_chars]))
            )
        except Exception as exc:  # pragma: no cover - defensive (agent/LLM failure)
            logger.warning("ground_claim extraction failed for %s: %s", g, exc)
            sources_skipped.append({"graph": g, "reason": "extraction failed"})
            continue
        candidates = parsed.get("quotes") if isinstance(parsed, dict) else None
        kept_here = 0
        for raw in candidates or []:
            if kept_here >= max(0, max_per_source):
                break
            cand = raw if isinstance(raw, str) else (raw.get("body") or raw.get("text") if isinstance(raw, dict) else "")
            cand = (cand or "").strip()
            if not cand:
                continue
            norm = grounding.normalize_for_match(cand)
            if not norm or norm in seen:
                continue
            if fuzzy_floor is None:
                verdict = grounding.verify_quote(cand, fulltext, annotations)
            else:
                verdict = grounding.verify_quote(cand, fulltext, annotations, fuzzy_floor=fuzzy_floor)
            if not verdict.get("verified"):
                rejected += 1
                continue
            seen.add(norm)
            accepted.append({
                "body": cand,
                "source_graph": g,
                "page": verdict.get("page"),
                "method": verdict.get("method"),
                "score": verdict.get("score"),
            })
            kept_here += 1

    quotes_added: list[dict[str, Any]] = []
    if accepted:
        with review_write_lock(f"claim::{cid}", graphs_dir=base):
            _assert_writable(get_graph, CROSS_GRAPH)
            conf = _coerce_confidence(confidence)
            for spec in accepted:
                norm = grounding.normalize_for_match(spec["body"])
                qid = f"{cid}-g{hashlib.sha1(norm.encode('utf-8')).hexdigest()[:10]}"
                validate_id(qid, kind="quote id", for_filename=True)
                src: dict[str, Any] = {"source": spec["source_graph"]}
                if spec.get("page") is not None:
                    src["page"] = spec["page"]
                note = Note(
                    id=qid,
                    title=f"Quote: {cid}",
                    type="quote",
                    source=src,
                    body=spec["body"],
                    tags=[],
                    links=[
                        Link(
                            target=cid,
                            relation="supports",
                            graph="",
                            confidence=conf,
                            origin=_PRODUCER_ORIGIN,
                        )
                    ],
                    grounding={
                        "verified": True,
                        "method": spec.get("method"),
                        "score": spec.get("score"),
                        "source": spec["source_graph"],
                    },
                )
                with _note_write_lock(base, CROSS_GRAPH, qid):
                    _assert_overwritable(base, CROSS_GRAPH, qid, expected_type="quote")
                    path = _write_note_durable(base, CROSS_GRAPH, note)
                quotes_added.append({
                    "id": qid, "graph": CROSS_GRAPH, "source": spec["source_graph"],
                    "body": spec["body"], "page": spec.get("page"),
                    "method": spec.get("method"), "score": spec.get("score"),
                    "path": str(path),
                })

    if quotes_added:
        msg = (f"Grounded '{title}' with {len(quotes_added)} verified quote(s) from "
               f"{len(sources_scanned)} source(s).")
    elif not sources:
        msg = "Claim has no supporting source documents to scrape."
    elif not sources_scanned:
        msg = "No local full text available for this claim's source(s)."
    else:
        msg = (f"Scanned {len(sources_scanned)} source(s) but found no new verbatim "
               f"supporting passages ({rejected} candidate(s) failed verification).")

    return {
        "scope": scope,
        "claim_id": cid,
        "title": title,
        "sources_scanned": sources_scanned,
        "sources_skipped": sources_skipped,
        "quotes_added": quotes_added,
        "rejected": rejected,
        "already_present": already_present,
        "message": msg,
    }

ground_paper

ground_paper(get_graph: GetGraph, *, paper_id: str, project: str = '', graph: str = '', extract_fn: 'Callable[[str, str], str] | None' = None, corpus_fn: 'Callable[[str], tuple[str | None, list[dict[str, Any]]]] | None' = None, max_per_slot: int = GROUND_PAPER_MAX_PER_SLOT, fulltext_chars: int = GROUND_FULLTEXT_CHARS, fuzzy_floor: 'float | None' = None, confidence: 'float | None' = None, graphs_dir: 'Path | str | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Fill a paper's missing CCC notes by SCRAPING its own source — WRITES.

The actionable side of the outline's per-paper [GAP:<paper-id>] marker. Resolves paper_id to its OWNED source graph, RE-READS that source's extracted PDF fulltext, and asks the WRITE-FREE extraction agent (extract_fn) for VERBATIM passages bucketed by the CCC slots the paper is MISSING (definition / mechanism / numbers / quotes). Each candidate is mechanically VERIFIED against the same source text via :func:grounding.verify_quote — anything that is not a real (exact/fuzzy/annotation) substring is REJECTED, so the agent surfaces but never fabricates. Accepted, deduped passages are written as TYPED notes into the paper's own source graph (definition→definition, mechanism→method, numbers→example, quotes→quote) and linked to the claims the paper supports (related for the depth notes, supports for quotes) so the slot the gap flagged is populated and a regenerate drops the marker.

All model/IO seams are injectable (extract_fn / corpus_fn) so tests run with no LLM and no PDF stack. Writes to the source graph → refused on a federated / read-only target. Returns {scope, paper_id, source_graph, title, notes_added, rejected, slots_already_filled, message}.

Source code in zettelkasten/claim_producer.py

def ground_paper(
    get_graph: GetGraph,
    *,
    paper_id: str,
    project: str = "",
    graph: str = "",
    extract_fn: "Callable[[str, str], str] | None" = None,
    corpus_fn: "Callable[[str], tuple[str | None, list[dict[str, Any]]]] | None" = None,
    max_per_slot: int = GROUND_PAPER_MAX_PER_SLOT,
    fulltext_chars: int = GROUND_FULLTEXT_CHARS,
    fuzzy_floor: "float | None" = None,
    confidence: "float | None" = None,
    graphs_dir: "Path | str | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Fill a paper's missing CCC notes by SCRAPING its own source — WRITES.

    The actionable side of the outline's per-paper ``[GAP:<paper-id>]`` marker.
    Resolves ``paper_id`` to its OWNED source graph, RE-READS that source's
    extracted PDF fulltext, and asks the WRITE-FREE extraction agent (``extract_fn``)
    for VERBATIM passages bucketed by the CCC slots the paper is MISSING
    (definition / mechanism / numbers / quotes). Each candidate is mechanically
    VERIFIED against the same source text via :func:`grounding.verify_quote` —
    anything that is not a real (exact/fuzzy/annotation) substring is REJECTED, so
    the agent surfaces but never fabricates. Accepted, deduped passages are written
    as TYPED notes into the paper's own source graph (definition→``definition``,
    mechanism→``method``, numbers→``example``, quotes→``quote``) and linked to the
    claims the paper supports (``related`` for the depth notes, ``supports`` for
    quotes) so the slot the gap flagged is populated and a regenerate drops the
    marker.

    All model/IO seams are injectable (``extract_fn`` / ``corpus_fn``) so tests run
    with no LLM and no PDF stack. Writes to the source graph → refused on a
    federated / read-only target. Returns ``{scope, paper_id, source_graph, title,
    notes_added, rejected, slots_already_filled, message}``.
    """
    from zettelkasten import grounding
    from zettelkasten.claims import SUPPORT_RELATIONS, _build_context

    pid = (paper_id or "").strip()
    if not pid:
        raise ValueError("ground_paper requires a paper id.")
    validate_id(pid, kind="paper id", for_filename=True)
    base = _base(graphs_dir)
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    scope = {"type": "graph" if graph else "project", "name": graph or project or "all"}

    ctx = _build_context(
        get_graph, project=project, graph=graph,
        graphs_dir=base, namespace=ns, localize=loc,
    )
    index, works = ctx[0], ctx[1]

    source_graph, title = _resolve_paper_source(index, works, pid)
    if source_graph is None:
        return {
            "scope": scope, "paper_id": pid, "source_graph": None, "title": title,
            "notes_added": [], "rejected": 0, "slots_already_filled": [],
            "message": f"Paper '{pid}' is not owned locally — nothing to scrape.",
        }

    # Claims the paper grounds (its support-source claims) — the new notes link to
    # these so a claim-anchored CCC slot is populated. A paper that grounds no
    # active claim still gets its notes (closing the claim-sparse fallback gap).
    target_claims = [
        key for key, rc in index.resolved.items()
        if source_graph in rc.support_sources
    ]

    # Which CCC slots already have a note for this paper that is INCIDENT to one of
    # its claims (mirrors outline._slot_notes' detection) — skip those so a re-run
    # never re-mines a slot the paper already carries.
    type_to_slot = {t: slot for slot, t in GROUND_PAPER_SLOT_TYPES.items()}
    filled: set[str] = set()
    incident_keys: set[tuple[str, str]] = set()
    for ck in target_claims:
        for e in index.by_target.get(ck, []):
            incident_keys.add(e.src_key)
        for e in index.by_source.get(ck, []):
            incident_keys.add(e.dst_key)
    for nk in incident_keys:
        if nk[0] != source_graph:
            continue
        note = index.notes_by_key.get(nk)
        slot = type_to_slot.get(note.type if note else "")
        if slot:
            filled.add(slot)
    wanted_slots = [s for s in GROUND_PAPER_SLOT_TYPES if s not in filled]

    cf = corpus_fn or _default_corpus_fn
    ef = extract_fn or _default_paper_extract_fn

    if not wanted_slots:
        return {
            "scope": scope, "paper_id": pid, "source_graph": source_graph, "title": title,
            "notes_added": [], "rejected": 0, "slots_already_filled": sorted(filled),
            "message": f"'{title}' already has notes for every CCC slot.",
        }

    fulltext, annotations = cf(source_graph)
    if not fulltext or not fulltext.strip():
        return {
            "scope": scope, "paper_id": pid, "source_graph": source_graph, "title": title,
            "notes_added": [], "rejected": 0, "slots_already_filled": sorted(filled),
            "message": f"No local full text available for '{title}' to extract from.",
        }

    try:
        parsed = _parse_propose_response(
            ef(_EXTRACT_PAPER_CONTRACT,
               _build_extract_paper_prompt(title, wanted_slots, fulltext[:fulltext_chars]))
        )
    except Exception as exc:  # pragma: no cover - defensive (agent/LLM failure)
        logger.warning("ground_paper extraction failed for %s: %s", source_graph, exc)
        return {
            "scope": scope, "paper_id": pid, "source_graph": source_graph, "title": title,
            "notes_added": [], "rejected": 0, "slots_already_filled": sorted(filled),
            "message": "Extraction agent failed — please retry.",
        }

    # Verify each candidate verbatim passage against the source text BEFORE writing.
    seen: set[str] = set()
    accepted: list[dict[str, Any]] = []
    rejected = 0
    for slot in wanted_slots:
        cands = parsed.get(slot) if isinstance(parsed, dict) else None
        kept_here = 0
        for raw in cands or []:
            if kept_here >= max(0, max_per_slot):
                break
            cand = raw if isinstance(raw, str) else (
                (raw.get("body") or raw.get("text")) if isinstance(raw, dict) else ""
            )
            cand = (cand or "").strip()
            if not cand:
                continue
            norm = grounding.normalize_for_match(cand)
            if not norm or norm in seen:
                continue
            if fuzzy_floor is None:
                verdict = grounding.verify_quote(cand, fulltext, annotations)
            else:
                verdict = grounding.verify_quote(cand, fulltext, annotations, fuzzy_floor=fuzzy_floor)
            if not verdict.get("verified"):
                rejected += 1
                continue
            seen.add(norm)
            accepted.append({
                "slot": slot,
                "type": GROUND_PAPER_SLOT_TYPES[slot],
                "body": cand,
                "page": verdict.get("page"),
                "method": verdict.get("method"),
                "score": verdict.get("score"),
            })
            kept_here += 1

    notes_added: list[dict[str, Any]] = []
    if accepted:
        conf = _coerce_confidence(confidence)
        # A neutral ``related`` edge makes a depth note incident to the claim (so the
        # CCC slot is detected) WITHOUT counting as stance evidence — only
        # ``supports``/``replicates`` inflate support span/strength. Quotes ARE
        # evidence, so they carry ``supports`` like ground_claim.
        with review_write_lock(f"paper::{source_graph}", graphs_dir=base):
            _assert_writable(get_graph, source_graph)
            for spec in accepted:
                ntype = spec["type"]
                norm = grounding.normalize_for_match(spec["body"])
                nid = f"g-{spec['slot']}-{hashlib.sha1(norm.encode('utf-8')).hexdigest()[:10]}"
                validate_id(nid, kind="note id", for_filename=True)
                src: dict[str, Any] = {"source": source_graph}
                if spec.get("page") is not None:
                    src["page"] = spec["page"]
                links: list[Link] = []
                rel = "supports" if ntype == "quote" else "related"
                edge_conf = conf if rel in SUPPORT_RELATIONS else None
                for ck in target_claims:
                    links.append(Link(
                        target=ck[1],
                        relation=rel,
                        graph="" if ck[0] == source_graph else ck[0],
                        confidence=edge_conf,
                        origin=_PRODUCER_ORIGIN,
                    ))
                note = Note(
                    id=nid,
                    title=f"{spec['slot'].capitalize()}: {title}",
                    type=ntype,
                    source=src,
                    body=spec["body"],
                    tags=[],
                    links=links,
                    grounding={
                        "verified": True,
                        "method": spec.get("method"),
                        "score": spec.get("score"),
                        "source": source_graph,
                    },
                )
                with _note_write_lock(base, source_graph, nid):
                    _assert_overwritable(base, source_graph, nid, expected_type=ntype)
                    path = _write_note_durable(base, source_graph, note)
                notes_added.append({
                    "id": nid, "graph": source_graph, "slot": spec["slot"],
                    "type": ntype, "body": spec["body"], "page": spec.get("page"),
                    "method": spec.get("method"), "score": spec.get("score"),
                    "path": str(path),
                })

    if notes_added:
        by_slot = ", ".join(
            f"{n} {s}" for s, n in sorted(
                {sl: sum(1 for a in notes_added if a["slot"] == sl) for sl in {a["slot"] for a in notes_added}}.items()
            )
        )
        msg = f"Filled '{title}' with {len(notes_added)} verified note(s) ({by_slot})."
    else:
        msg = (f"Scanned the source for '{title}' but found no new verbatim passages "
               f"for the missing slot(s) ({rejected} candidate(s) failed verification).")

    return {
        "scope": scope,
        "paper_id": pid,
        "source_graph": source_graph,
        "title": title,
        "notes_added": notes_added,
        "rejected": rejected,
        "slots_already_filled": sorted(filled),
        "message": msg,
    }