zettelkasten.coverage

zettelkasten.coverage

Zettelkasten coverage, concept-hub, and project-render computation.

Deterministic coverage/grounding audits, cross-source concept-hub and link-mesh suggestions, and the composite project-graph render — carved out of :mod:zettelkasten.server for clarity. server re-exports every name here so existing importers keep working unchanged. Server-level helpers and the graph cache are reached through the server module object (not value-imported) so everything resolves at CALL time against the live module, preserving behavior and test monkeypatching. The import is safe against cycles because server performs its re-export import of this module at end-of-file.

project_graph

project_graph(project: str) -> str

Get the composite graph for a project: all source notes, cross-links, and citations.

Loads every source graph in the project plus the _cross graph, and returns merged notes (with source attribution), all links, and citations.

Parameters:

Name	Type	Description	Default
project	str	Project name	required

Source code in zettelkasten/coverage.py

def project_graph(project: str) -> str:
    """Get the composite graph for a project: all source notes, cross-links, and citations.

    Loads every source graph in the project plus the _cross graph,
    and returns merged notes (with source attribution), all links, and citations.

    Args:
        project: Project name
    """
    project_data = server.load_project(project)
    if not project_data:
        return json.dumps({"error": f"Project '{project}' not found.", "type": "NotFound"})

    sources = project_data.get("sources", [])
    source_set = set(sources)
    source_nodes: list[dict] = []
    all_notes: list[dict] = []
    all_links: list[dict] = []
    missing_sources: list[str] = []

    # Track which note ids live in the project's sources, plus citations and
    # _cross notes the project actually references, so the composite stays
    # self-contained instead of dumping every global entity.
    project_note_ids: set[str] = set()
    referenced_citations: set[str] = set()
    referenced_cross: set[str] = set()

    def _record_links(note, owner_graph: str) -> None:
        for link in note.links:
            target_graph = link.graph or owner_graph
            all_links.append({
                "source_id": note.id,
                "target_id": link.target,
                "relation": link.relation,
                "source_graph": owner_graph,
                "target_graph": target_graph,
            })
            if target_graph == "_citations":
                referenced_citations.add(link.target)
            elif target_graph == "_cross":
                referenced_cross.add(link.target)

    for source_name in sources:
        # Special folders (``_cross``/``_citations``/...) are not real sources;
        # ignore a legacy manifest that listed ``_cross`` in ``sources`` (the old
        # workaround) so it does not drag the entire global cross folder in.
        if source_name.startswith("_"):
            continue
        source_dir = server.GRAPHS_DIR / source_name
        if not source_dir.is_dir():
            missing_sources.append(source_name)
            continue
        zg = server._get_graph(source_name)
        meta = server.load_source_meta(source_name)
        source_id = f"source:{source_name}"
        source_nodes.append({
            "id": source_id,
            "type": "source",
            "name": source_name,
            "title": meta.get("title") or source_name,
            "doc_type": meta.get("doc_type", ""),
            "authors": meta.get("authors", []),
            "year": meta.get("year"),
            "note_count": len(zg.notes),
        })
        for note in zg.notes.values():
            project_note_ids.add(note.id)
            all_notes.append({
                "id": note.id,
                "title": note.title,
                "type": note.type,
                "tags": note.tags,
                "source_graph": source_name,
            })
            # Membership edge: note belongs to its source hub. Flagged so
            # consumers can style/route it separately from semantic relations.
            all_links.append({
                "source_id": source_id,
                "target_id": note.id,
                "relation": "contains",
                "source_graph": source_name,
                "target_graph": source_name,
                "membership": True,
            })
            _record_links(note, source_name)

        # Source-level relations from _meta.yaml become edges between source
        # hubs (extends, responds-to, replicates, ...). Dead data until now.
        for rel in meta.get("relations", []) or []:
            if not isinstance(rel, dict):
                continue
            target = rel.get("target")
            relation = rel.get("relation")
            if not target or not relation:
                continue
            all_links.append({
                "source_id": source_id,
                "target_id": f"source:{target}",
                "relation": relation,
                "source_graph": source_name,
                "target_graph": target,
                "source_level": True,
                "external": target not in source_set,
            })

    # _cross synthesis notes: EXPLICIT membership is authoritative — include a
    # cross note iff its id is in the manifest's ``cross:`` list. The legacy
    # link-inference is demoted to a *suggestion*: a cross note connected to the
    # project's sources but not yet claimed is surfaced in ``suggested_cross`` for
    # one-click curation, never auto-included.
    claimed_cross = server.project_cross_ids(project_data)
    claimed_set = set(claimed_cross)
    suggested_cross: list[dict] = []
    cross_dir = server.GRAPHS_DIR / "_cross"
    if cross_dir.is_dir():
        cross_graph = server._get_graph("_cross")
        for note in cross_graph.notes.values():
            if note.id in claimed_set:
                all_notes.append({
                    "id": note.id,
                    "title": note.title,
                    "type": note.type,
                    "tags": note.tags,
                    "source_graph": "_cross",
                })
                _record_links(note, "_cross")
            elif server.cross_note_connects(note, source_set, project_note_ids, referenced_cross):
                suggested_cross.append({
                    "id": note.id,
                    "title": note.title,
                    "type": note.type,
                    "tags": note.tags,
                })

    # Citations: only those actually referenced by an included note.
    all_citations = server.load_citations()
    citations = {cid: all_citations[cid] for cid in referenced_citations if cid in all_citations}

    result: dict = {
        "project": project,
        "description": project_data.get("description", ""),
        "sources": sources,
        "cross": claimed_cross,
        "suggested_cross": suggested_cross,
        "source_nodes": source_nodes,
        "notes": all_notes,
        "links": all_links,
        "citations": citations,
        "summary": {
            "total_sources": len(source_nodes),
            "total_notes": len(all_notes),
            "total_links": len(all_links),
            "total_citations": len(citations),
            "total_cross": len(claimed_cross),
            "suggested_cross_count": len(suggested_cross),
            "source_count": len(sources),
        },
    }
    if missing_sources:
        result["missing_sources"] = missing_sources
    return json.dumps(result)

set_coverage

set_coverage(source: str, agent: str = '', human: str = '') -> str

Set a source's processing coverage level.

Coverage tracks how thoroughly a source has been processed. coverage.agent auto-advances none -> skimmed when the first note is added; use this tool to record deeper progress (e.g. "full") or to set the human level.

Parameters:

Name	Type	Description	Default
source	str	Source graph name	required
agent	str	Agent coverage: none | skimmed | full (omit to leave unchanged)	''
human	str	Human coverage: none | skimmed | full (omit to leave unchanged)	''

Source code in zettelkasten/coverage.py

def set_coverage(source: str, agent: str = "", human: str = "") -> str:
    """Set a source's processing coverage level.

    Coverage tracks how thoroughly a source has been processed. `coverage.agent`
    auto-advances none -> skimmed when the first note is added; use this tool to
    record deeper progress (e.g. "full") or to set the human level.

    Args:
        source: Source graph name
        agent: Agent coverage: none | skimmed | full (omit to leave unchanged)
        human: Human coverage: none | skimmed | full (omit to leave unchanged)
    """
    # Advancing to "full" asserts the source is thoroughly and faithfully
    # extracted, so gate it on the mechanical grounding audit. force=True is not
    # a kwarg here (the dashboard shares this seam); deeper coverage must be
    # clean. Skimmed/none are unaffected.
    if agent == "full":
        audit = server._verify_source_extraction(source)
        if "error" not in audit and audit.get("blocking_count", 0) > 0:
            return json.dumps({
                "error": (
                    f"Cannot set agent coverage to 'full': {audit['blocking_count']} "
                    "grounding violation(s) unresolved. Fix unverified quotes and add "
                    "supporting quotes to claims/findings, then retry. "
                    "Run verify_extraction for details."
                ),
                "type": "GroundingError",
                "counts": audit.get("counts", {}),
            })

    try:
        coverage = server.write_coverage(source, agent=agent, human=human)
    except ValueError as e:
        return json.dumps({"error": str(e), "type": "ValidationError"})
    except FileNotFoundError:
        return json.dumps({"error": f"Source '{source}' not found.", "type": "NotFound"})
    return json.dumps({"source": source, "coverage": coverage})

coverage_matrix

coverage_matrix(source: str, schema: str = '', project: str = '', synthesis_graph: str = '') -> str

Coverage matrix for a source's notes against its schema dimensions.

Deterministic Covered/Thin/Missing per dimension tag — the model-independent backstop the auditor reads instead of hand-building the matrix. Dimension tags come from schema (a registry schema name) when given, else from the rubric the extraction run persisted onto the source meta.

This is ADVISORY, not a hard gate: a source legitimately may not cover every dimension. ok is true only when no dimension is Missing, but coverage is not enforced anywhere — the auditor uses judgement on the matrix.

Parameters:

Name	Type	Description	Default
source	str	Source graph name (folder under .zettelkasten/)	required
schema	str	Optional registry schema name; overrides the persisted tags	''

Source code in zettelkasten/coverage.py

def coverage_matrix(
    source: str,
    schema: str = "",
    project: str = "",
    synthesis_graph: str = "",
) -> str:
    """Coverage matrix for a source's notes against its schema dimensions.

    Deterministic Covered/Thin/Missing per dimension tag — the model-independent
    backstop the auditor reads instead of hand-building the matrix. Dimension
    tags come from ``schema`` (a registry schema name) when given, else from the
    rubric the extraction run persisted onto the source meta.

    This is ADVISORY, not a hard gate: a source legitimately may not cover every
    dimension. ``ok`` is true only when no dimension is Missing, but coverage is
    not enforced anywhere — the auditor uses judgement on the matrix.

    Args:
        source: Source graph name (folder under .zettelkasten/)
        schema: Optional registry schema name; overrides the persisted tags
    """
    tags: list[str] | None = None
    if schema:
        try:
            from zettelkasten.extraction_schemas import expand_schema

            expanded = expand_schema(schema)
        except Exception as exc:  # noqa: BLE001 — surface as a clean error
            return json.dumps({"error": str(exc), "type": "ValidationError"})
        if expanded is None:
            return json.dumps({
                "error": f"Schema '{schema}' is not in the registry.",
                "type": "NotFound",
            })
        tags = list(expanded.get("tags") or [])
        return json.dumps(server._coverage_matrix(source, tags=tags, grounded=expanded.get("grounded")))
    return json.dumps(server._coverage_matrix(
        source, project=project, synthesis_graph=synthesis_graph))

verify_extraction

verify_extraction(source: str) -> str

Audit a source for grounding violations (anti-invention check).

Scans every note in a source and reports where extraction may have invented or unmoored content: quotes not confirmed verbatim against the source, claims/findings with no supporting quote, page mismatches, and orphan quotes. Run this before marking a source's agent coverage as "full" — that level is gated on a clean audit (blocking_count == 0).

Parameters:

Name	Type	Description	Default
source	str	Source graph name (folder under .zettelkasten/)	required

Source code in zettelkasten/coverage.py

def verify_extraction(source: str) -> str:
    """Audit a source for grounding violations (anti-invention check).

    Scans every note in a source and reports where extraction may have invented
    or unmoored content: quotes not confirmed verbatim against the source,
    claims/findings with no supporting quote, page mismatches, and orphan quotes.
    Run this before marking a source's agent coverage as "full" — that level is
    gated on a clean audit (`blocking_count == 0`).

    Args:
        source: Source graph name (folder under .zettelkasten/)
    """
    return json.dumps(server._verify_source_extraction(source))

suggest_concept_hubs

suggest_concept_hubs(project: str, similarity_threshold: float = 0.7, min_sources: int = 2, since_source: str = '', max_suggestions: int = 10) -> str

Find implicit concept clusters across project sources.

Identifies groups of semantically similar notes from different sources that could benefit from a synthesis (concept hub) note in _cross/.

Each suggestion carries an action so the caller knows whether to grow the _cross graph or just attach to it:

• attach: an existing _cross hub already covers this theme — link the related note into that hub instead of minting a new one. This is the main guard against _cross growing without bound, so it is always preferred.
• new_hub: no existing hub covers the cluster — a new synthesis note in _cross/ is warranted.

Parameters:

Name	Type	Description	Default
project	str	Project name	required
similarity_threshold	float	Minimum similarity to consider a pair related (default 0.7)	0.7
min_sources	int	Minimum number of distinct sources a cluster must span (default 2)	2
since_source	str	When set, only evaluate clusters involving this one source (incremental mode). This keeps the cost O(sources) per call instead of the full O(sources^2) sweep, so it is cheap enough to run on every source addition. It also follows the source's own links: when a note it links to is already covered by a hub, that hub is surfaced as an attach with no embedding work at all.	''
max_suggestions	int	Hard cap on returned suggestions (keeps the payload light regardless of how large the project or _cross graph grows).	10

Source code in zettelkasten/coverage.py

def suggest_concept_hubs(
    project: str,
    similarity_threshold: float = 0.7,
    min_sources: int = 2,
    since_source: str = "",
    max_suggestions: int = 10,
) -> str:
    """Find implicit concept clusters across project sources.

    Identifies groups of semantically similar notes from different sources
    that could benefit from a synthesis (concept hub) note in _cross/.

    Each suggestion carries an ``action`` so the caller knows whether to grow the
    _cross graph or just attach to it:

    - ``attach``: an existing _cross hub already covers this theme — link the
      related note into that hub instead of minting a new one. This is the main
      guard against _cross growing without bound, so it is always preferred.
    - ``new_hub``: no existing hub covers the cluster — a new synthesis note in
      _cross/ is warranted.

    Args:
        project: Project name
        similarity_threshold: Minimum similarity to consider a pair related (default 0.7)
        min_sources: Minimum number of distinct sources a cluster must span (default 2)
        since_source: When set, only evaluate clusters involving this one source
            (incremental mode). This keeps the cost O(sources) per call instead of
            the full O(sources^2) sweep, so it is cheap enough to run on every
            source addition. It also follows the source's own links: when a note it
            links to is already covered by a hub, that hub is surfaced as an
            ``attach`` with no embedding work at all.
        max_suggestions: Hard cap on returned suggestions (keeps the payload light
            regardless of how large the project or _cross graph grows).
    """
    project_data = server.load_project(project)
    if not project_data:
        return json.dumps({"error": f"Project '{project}' not found.", "type": "NotFound"})

    source_names = project_data.get("sources", [])
    all_notes_by_source: dict[str, list[str]] = {}

    for source_name in source_names:
        source_dir = server.GRAPHS_DIR / source_name
        if not source_dir.is_dir():
            continue
        zg = server._get_graph(source_name)
        if zg.notes:
            all_notes_by_source[source_name] = list(zg.notes.keys())

    if len(all_notes_by_source) < min_sources:
        return json.dumps({"project": project, "suggestions": [], "count": 0, "message": "Not enough sources with notes to find cross-source clusters."})

    if since_source and since_source not in all_notes_by_source:
        return json.dumps({"project": project, "since_source": since_source, "suggestions": [], "count": 0, "message": f"Source '{since_source}' has no notes in project '{project}'."})

    # Map every source note already represented by a _cross hub: note_id -> hubs.
    # A note is "covered" when a hub links to it (hub -> note). This map is what
    # lets us recommend attaching to an existing hub instead of creating a new
    # one, which keeps _cross growth tied to genuinely new themes.
    coverage: dict[str, list[tuple[str, str]]] = {}
    existing_cross_targets: set[str] = set()
    cross_dir = server.GRAPHS_DIR / "_cross"
    if cross_dir.is_dir():
        cross_graph = server._get_graph("_cross")
        for hub in cross_graph.notes.values():
            for link in hub.links:
                coverage.setdefault(link.target, []).append((hub.id, hub.title))
                existing_cross_targets.add(link.target)

    import numpy as np

    suggestions: list[dict] = []

    # --- Cheap link-following pass (incremental only). If the new source links to
    # a note that is already covered by a hub, the new note almost certainly
    # belongs in that same hub. No embeddings needed — O(links), high precision.
    if since_source:
        new_zg = server._get_graph(since_source)
        for note in new_zg.notes.values():
            already_linked_hubs = {l.target for l in note.links if (l.graph or "") == "_cross"}
            seen_hubs: set[str] = set()
            for link in note.links:
                for hub_id, hub_title in coverage.get(link.target, []):
                    if hub_id in already_linked_hubs or hub_id in seen_hubs:
                        continue
                    seen_hubs.add(hub_id)
                    suggestions.append({
                        "action": "attach",
                        "via": "link",
                        "hub": {"id": hub_id, "title": hub_title},
                        "note": {"id": note.id, "source_graph": since_source, "title": note.title},
                        "related_to": link.target,
                    })
                    if len(suggestions) >= max_suggestions:
                        break
                if len(suggestions) >= max_suggestions:
                    break
            if len(suggestions) >= max_suggestions:
                break

    # --- Semantic pass: similar note pairs across sources. Instead of the former
    # O(S²·cap²) all-pairs sweep, every project note's normalized vector is stacked
    # into one matrix and each note queries its top-k CROSS-source nearest
    # neighbors (exact masked block matmul + argpartition, C-level). In incremental
    # mode only the new source's notes originate queries (O(new · N)); otherwise
    # every note does. The former per-source cap was a recall limiter; it is now a
    # high safety valve (:data:`_MAX_NOTES_PER_SOURCE_SAFETY`) so no note — and no
    # cross-source edge it would contribute — is silently dropped.
    pairs: list[tuple[str, str, str, str, float]] = []
    if len(suggestions) < max_suggestions:
        recs, matrix = _project_note_matrix(all_notes_by_source, np)
        if matrix is not None:
            # Source-order index → canonical pair ordering (a = earlier source),
            # matching the former sweep so suggested titles/anchors are stable.
            src_order = {name: i for i, name in enumerate(all_notes_by_source)}
            if since_source:
                source_rows = [i for i, (s, _n) in enumerate(recs) if s == since_source]
            else:
                source_rows = list(range(len(recs)))
            neighbor_pairs = _cross_source_neighbor_pairs(
                recs, matrix, source_rows, _HUB_NEIGHBOR_TOPK, similarity_threshold, np,
            )
            for pair, sim in neighbor_pairs.items():
                p, q = tuple(pair)
                (sa, na_id), (sb, nb_id) = recs[p], recs[q]
                if since_source:
                    # a = the new source's note; b = the note in the other source.
                    if sb == since_source:
                        (sa, na_id), (sb, nb_id) = (sb, nb_id), (sa, na_id)
                elif src_order[sb] < src_order[sa]:
                    # a = earlier source (canonical), b = later.
                    (sa, na_id), (sb, nb_id) = (sb, nb_id), (sa, na_id)
                pairs.append((na_id, sa, nb_id, sb, sim))

    used: set[str] = set()
    # Deterministic strongest-first order (ties broken by ids/sources).
    pairs.sort(key=lambda x: (-x[4], x[0], x[1], x[2], x[3]))
    for note_a, src_a, note_b, src_b, sim in pairs:
        if len(suggestions) >= max_suggestions:
            break
        if note_a in used and note_b in used:
            continue

        cluster_notes = [{
            "id": note_a,
            "source_graph": src_a,
            "title": server._get_graph(src_a).notes[note_a].title,
        }, {
            "id": note_b,
            "source_graph": src_b,
            "title": server._get_graph(src_b).notes[note_b].title,
        }]

        # If either side is already covered by a hub, this is an attach, not a new
        # hub. Skip entirely when both are already synthesized.
        anchor_hub = None
        for cand in (note_b, note_a):
            if cand in coverage:
                hub_id, hub_title = coverage[cand][0]
                anchor_hub = {"id": hub_id, "title": hub_title}
                break

        used.add(note_a)
        used.add(note_b)

        if anchor_hub is not None:
            if note_a in existing_cross_targets and note_b in existing_cross_targets:
                continue
            suggestions.append({
                "action": "attach",
                "via": "similarity",
                "hub": anchor_hub,
                "notes": cluster_notes,
                "max_similarity": round(sim, 3),
            })
        else:
            suggestions.append({
                "action": "new_hub",
                "via": "similarity",
                "suggested_title": cluster_notes[0]["title"],
                "notes": cluster_notes,
                "max_similarity": round(sim, 3),
                "source_count": 2,
            })

    result: dict = {
        "project": project,
        "similarity_threshold": similarity_threshold,
        "min_sources": min_sources,
        "suggestions": suggestions,
        "count": len(suggestions),
        "counts": {
            "attach": sum(1 for s in suggestions if s["action"] == "attach"),
            "new_hub": sum(1 for s in suggestions if s["action"] == "new_hub"),
        },
    }
    if since_source:
        result["since_source"] = since_source
    return json.dumps(result)

link_mesh

link_mesh(project: str, similarity_threshold: 'float | None' = None, per_note_top_k: 'int | None' = None, max_links: int = 200, max_notes_per_source: int = _MAX_NOTES_PER_SOURCE_SAFETY, dry_run: bool = True) -> str

Weave a dense cross-source related mesh across a project's sources.

The alternative to :func:suggest_concept_hubs (hub-and-spoke synthesis): instead of minting _cross hubs, this authors DIRECT related edges between semantically similar notes that live in different source graphs of the project. When a source is folded into a project it has no edges to the project's other sources yet — this creates them so the composite project graph is actually connected.

Every authored edge is stamped origin='link-mesh' (provable ownership for a later prune → reversible) and carries the pair's cosine similarity as its confidence. Authoring is idempotent: a pair already linked with a related edge across the same graphs is left untouched and counted under skipped_existing.

Parameters:

Name	Type	Description	Default
project	str	Project name.	required
similarity_threshold	'float | None'	Minimum cosine similarity for a pair to be linked. None (default) → auto: a sensible quality floor (0.7).	None
per_note_top_k	'int | None'	Cap on new mesh edges incident to any single note, so a hub-like note doesn't monopolize the budget (strongest pairs win). None (default) → auto: DENSITY-MATCHED to the project's own sources — the target degree is the average intra-source out-degree (clamped to 1..5), so the woven mesh is about as dense as the existing graphs rather than a magic number.	None
max_links	int	Hard cap on total edges considered/authored in one pass.	200
max_notes_per_source	int	Per-source SAFETY VALVE on how many notes enter the neighbor search (guards a pathologically huge single source). Defaults high enough never to bind on a real project — it no longer caps recall the way the former small cap did.	_MAX_NOTES_PER_SOURCE_SAFETY
dry_run	bool	When true (default), compute and return the pairs that WOULD be linked without writing anything. When false, author the edges.	True

Source code in zettelkasten/coverage.py

def link_mesh(
    project: str,
    similarity_threshold: "float | None" = None,
    per_note_top_k: "int | None" = None,
    max_links: int = 200,
    max_notes_per_source: int = _MAX_NOTES_PER_SOURCE_SAFETY,
    dry_run: bool = True,
) -> str:
    """Weave a dense cross-source ``related`` mesh across a project's sources.

    The alternative to :func:`suggest_concept_hubs` (hub-and-spoke synthesis):
    instead of minting ``_cross`` hubs, this authors DIRECT ``related`` edges
    between semantically similar notes that live in *different* source graphs of
    the project. When a source is folded into a project it has no edges to the
    project's other sources yet — this creates them so the composite project
    graph is actually connected.

    Every authored edge is stamped ``origin='link-mesh'`` (provable ownership for
    a later prune → reversible) and carries the pair's cosine similarity as its
    ``confidence``. Authoring is idempotent: a pair already linked with a
    ``related`` edge across the same graphs is left untouched and counted under
    ``skipped_existing``.

    Args:
        project: Project name.
        similarity_threshold: Minimum cosine similarity for a pair to be linked.
            ``None`` (default) → auto: a sensible quality floor (0.7).
        per_note_top_k: Cap on new mesh edges incident to any single note, so a
            hub-like note doesn't monopolize the budget (strongest pairs win).
            ``None`` (default) → auto: DENSITY-MATCHED to the project's own
            sources — the target degree is the average intra-source out-degree
            (clamped to 1..5), so the woven mesh is about as dense as the existing
            graphs rather than a magic number.
        max_links: Hard cap on total edges considered/authored in one pass.
        max_notes_per_source: Per-source SAFETY VALVE on how many notes enter the
            neighbor search (guards a pathologically huge single source). Defaults
            high enough never to bind on a real project — it no longer caps recall
            the way the former small cap did.
        dry_run: When true (default), compute and return the pairs that WOULD be
            linked without writing anything. When false, author the edges.
    """
    project_data = server.load_project(project)
    if not project_data:
        return json.dumps({"error": f"Project '{project}' not found.", "type": "NotFound"})

    source_names = project_data.get("sources", [])
    all_notes_by_source: dict[str, list[str]] = {}
    for source_name in source_names:
        if not (server.GRAPHS_DIR / source_name).is_dir():
            continue
        zg = server._get_graph(source_name)
        if zg.notes:
            all_notes_by_source[source_name] = list(zg.notes.keys())

    if len(all_notes_by_source) < 2:
        return json.dumps({
            "project": project,
            "mode": "mesh",
            "dry_run": dry_run,
            "links": [],
            "count": 0,
            "message": "Need at least two sources with notes to weave a mesh.",
        })

    # Resolve the auto knobs. The threshold is just a quality floor; the real
    # density lever is per_note_top_k, which we match to how densely the project's
    # own sources are already linked (average out-degree, clamped) so the mesh
    # feels native rather than arbitrarily sparse/dense.
    if similarity_threshold is None:
        similarity_threshold = 0.7
    auto_degree = per_note_top_k is None
    avg_source_degree = 0.0
    if auto_degree:
        total_notes = sum(len(v) for v in all_notes_by_source.values())
        total_links = 0
        for src in all_notes_by_source:
            zg = server._get_graph(src)
            for nid in all_notes_by_source[src]:
                note = zg.notes.get(nid)
                if note is not None:
                    total_links += len(note.links)
        avg_source_degree = (total_links / total_notes) if total_notes else 0.0
        # No existing links at all → default to 2 (create some connectivity);
        # otherwise match the rounded average, clamped to a sane 1..5.
        per_note_top_k = max(1, min(5, round(avg_source_degree))) if total_links else 2

    import numpy as np

    # Per-note top-k nearest-neighbor candidate generation, replacing the former
    # O(S²·cap²) all-pairs sweep. Every project note's normalized vector is stacked
    # into one matrix; each note then queries its top-k CROSS-source neighbors that
    # clear the threshold (exact masked block matmul + argpartition, C-level). The
    # over-fetch is a comfortable margin over ``per_note_top_k`` so the downstream
    # greedy degree budget sees every edge it could select (an edge survives greedy
    # only if it is among a note's strongest, and the buffer covers the rank drift
    # from partner-saturation skips). ``max_notes_per_source`` is now a high safety
    # valve, not a recall limiter.
    src_order = {name: i for i, name in enumerate(all_notes_by_source)}
    recs, matrix = _project_note_matrix(all_notes_by_source, np, max_notes_per_source)
    candidates: list[tuple[float, str, str, str, str]] = []
    if matrix is not None:
        want = min(len(recs), per_note_top_k + 25)
        neighbor_pairs = _cross_source_neighbor_pairs(
            recs, matrix, list(range(len(recs))), want, similarity_threshold, np,
        )
        for pair, sim in neighbor_pairs.items():
            p, q = tuple(pair)
            (sa, na_id), (sb, nb_id) = recs[p], recs[q]
            # Canonical ordering: src_a is the earlier source (matches the former
            # sweep, so edges are authored on the same side).
            if src_order[sb] < src_order[sa]:
                (sa, na_id), (sb, nb_id) = (sb, nb_id), (sa, na_id)
            candidates.append((sim, sa, na_id, sb, nb_id))

    # Deterministic strongest-first order (ties broken by sources/ids).
    candidates.sort(key=lambda x: (-x[0], x[1], x[2], x[3], x[4]))

    # Greedy per-note-degree budget: strongest pairs first, but no note may gain
    # more than ``per_note_top_k`` new mesh edges.
    from collections import defaultdict
    degree: dict[tuple[str, str], int] = defaultdict(int)
    selected: list[tuple[float, str, str, str, str]] = []
    for sim, src_a, nid_a, src_b, nid_b in candidates:
        if len(selected) >= max_links:
            break
        ka, kb = (src_a, nid_a), (src_b, nid_b)
        if degree[ka] >= per_note_top_k or degree[kb] >= per_note_top_k:
            continue
        selected.append((sim, src_a, nid_a, src_b, nid_b))
        degree[ka] += 1
        degree[kb] += 1

    links_out: list[dict] = []
    created = 0
    skipped_existing = 0
    dirty: dict[str, server.ZettelGraph] = {}

    for sim, src_a, nid_a, src_b, nid_b in selected:
        zg_a = server._get_graph(src_a)
        note_a = zg_a.notes.get(nid_a)
        zg_b = server._get_graph(src_b)
        note_b = zg_b.notes.get(nid_b)
        if note_a is None or note_b is None:
            continue

        already = any(
            l.target == nid_b and l.relation == "related" and (l.graph or "") == src_b
            for l in note_a.links
        )
        entry = {
            "a": {"source_graph": src_a, "id": nid_a, "title": note_a.title},
            "b": {"source_graph": src_b, "id": nid_b, "title": note_b.title},
            "similarity": round(sim, 3),
            "existing": already,
        }
        links_out.append(entry)

        if already:
            skipped_existing += 1
            continue
        if not dry_run:
            note_a.links.append(server.Link(
                target=nid_b,
                relation="related",
                direction="bidirectional",
                graph=src_b,
                origin="link-mesh",
                confidence=round(sim, 3),
            ))
            dirty[src_a] = zg_a
        created += 1

    if not dry_run and dirty:
        # Persist each mutated source once (a source note may gain several edges).
        touched_notes: dict[str, set[str]] = defaultdict(set)
        for _sim, src_a, nid_a, _sb, _nb in selected:
            touched_notes[src_a].add(nid_a)
        for src, zg in dirty.items():
            for nid in touched_notes.get(src, set()):
                note = zg.notes.get(nid)
                if note is not None:
                    zg.save_note(note)

    return json.dumps({
        "project": project,
        "mode": "mesh",
        "dry_run": dry_run,
        "similarity_threshold": similarity_threshold,
        "per_note_top_k": per_note_top_k,
        # Transparency for the auto density-match: whether top-k was derived and
        # the intra-source degree it matched.
        "auto_density": auto_degree,
        "avg_source_degree": round(avg_source_degree, 2),
        "links": links_out,
        "count": len(links_out),
        "created": 0 if dry_run else created,
        "would_create": created if dry_run else 0,
        "skipped_existing": skipped_existing,
    })

suggest_cross_connections

suggest_cross_connections(project: str = '', graph: str = '', note_id: str = '', top_k: int = 8, similarity_threshold: float = 0.5, target_scope: str = 'project', include_types: 'list[str] | None' = None, min_cross_degree: int = 1, max_candidates: int = 500, max_notes_per_source: int = 200) -> str

Cross-graph connectivity engine: candidate world edges + island audit.

Surfaces, for the content notes in scope, semantically related notes in a DIFFERENT graph that are not yet linked -- the job-2 (flat cross-graph connectivity) candidates suggest(kind="connections") cannot find because it is graph-local -- plus a connectivity audit flagging island graphs and under-connected notes. READ-ONLY: the agent writes the edges it judges real via note(action="link", ..., target_graph=).

Scope (choose one): - note_id (+ graph): one note -> the rest of the target scope. - graph: one source graph's notes -> the rest of the target scope. - project: every content note in the project's sources <-> each other. - none: the whole corpus. target_scope ('project' default | 'corpus') controls where target notes may live; with a project it restricts targets to that project's sources.

Idempotent: already-linked pairs (a direct link in either direction, or a prerequisite) are excluded, so a re-run after the agent wires edges returns only what is still missing -- islands should trend to zero.

Source code in zettelkasten/coverage.py

def suggest_cross_connections(
    project: str = "",
    graph: str = "",
    note_id: str = "",
    top_k: int = 8,
    similarity_threshold: float = 0.5,
    target_scope: str = "project",
    include_types: "list[str] | None" = None,
    min_cross_degree: int = 1,
    max_candidates: int = 500,
    max_notes_per_source: int = 200,
) -> str:
    """Cross-graph connectivity engine: candidate world edges + island audit.

    Surfaces, for the content notes in scope, semantically related notes in a
    DIFFERENT graph that are not yet linked -- the job-2 (flat cross-graph
    connectivity) candidates suggest(kind="connections") cannot find because it is
    graph-local -- plus a connectivity audit flagging island graphs and
    under-connected notes. READ-ONLY: the agent writes the edges it judges real
    via note(action="link", ..., target_graph=<b.graph>).

    Scope (choose one):
    - ``note_id`` (+ ``graph``): one note -> the rest of the target scope.
    - ``graph``: one source graph's notes -> the rest of the target scope.
    - ``project``: every content note in the project's sources <-> each other.
    - none: the whole corpus.
    ``target_scope`` (``'project'`` default | ``'corpus'``) controls where target
    notes may live; with a project it restricts targets to that project's sources.

    Idempotent: already-linked pairs (a direct link in either direction, or a
    prerequisite) are excluded, so a re-run after the agent wires edges returns
    only what is still missing -- islands should trend to zero.
    """
    import numpy as np

    inc_types = frozenset(include_types) if include_types else frozenset(_CROSS_CONN_CONTENT_TYPES)

    # ── Resolve scope → source graphs (propose FROM) + target graphs (link TO).
    if note_id and not graph:
        return json.dumps({
            "error": "scope note_id requires graph (the note's source graph).",
            "type": "BadRequest",
        })
    target_graphs = _cross_conn_scope_graphs(project, target_scope)
    if graph:
        source_graphs = [graph]
        # A single graph / note still needs the wider target set to link INTO.
        if not target_graphs:
            target_graphs = [g for g in server.note_graph_names() if g != "_cross"]
        if graph not in target_graphs:
            target_graphs = [graph] + target_graphs
    else:
        source_graphs = list(target_graphs)

    if not source_graphs or not target_graphs:
        return json.dumps({
            "error": (
                "no graphs in scope -- pass a project with sources, a graph, or a "
                "note_id+graph (or run against the whole corpus)."
            ),
            "type": "BadRequest",
        })

    scope_key = {
        "project": project, "graph": graph, "note_id": note_id,
        "target_scope": target_scope, "similarity_threshold": similarity_threshold,
        "top_k": top_k,
    }

    # ── Cache coherence: a long-lived MCP server's _get_graph cache is refreshed
    # only by the THROTTLED auto-reload (_AUTO_SYNC_INTERVAL), so a graph edited on
    # disk within that window — edges another process (or a synthesizer) just wrote —
    # is served stale. The island audit + already-linked/candidate reads below then
    # report freshly-connected sources as still-islands. Force a per-graph staleness
    # check (bypassing the throttle) for every in-scope graph so all reads reflect
    # current link state; reloads only fire for graphs that actually changed.
    server.refresh_graphs(list(source_graphs) + list(target_graphs))

    # ── Connectivity audit (cheap, embedding-free). Reported even if no vectors.
    islands, under_connected = _cross_conn_audit(
        source_graphs, inc_types, min_cross_degree, max_report=max_candidates,
    )

    # ── Collect normalized content-note vectors across the union of scope graphs.
    union_graphs: list[str] = []
    for g in list(source_graphs) + list(target_graphs):
        if g not in union_graphs:
            union_graphs.append(g)

    recs: list[dict] = []
    vlist: list = []
    for g in union_graphs:
        if not (server.GRAPHS_DIR / g).is_dir():
            continue
        try:
            zg = server._get_graph(g)
        except Exception:  # noqa: BLE001 — skip an unreadable graph
            continue
        kept = 0
        for nid, note in zg.notes.items():
            if kept >= max_notes_per_source:
                break
            if not _cross_conn_is_content(note, inc_types):
                continue
            raw = zg.embeddings.get_vector(nid)
            if raw is None:
                continue
            v = np.asarray(raw, dtype=np.float32)
            norm = float(np.linalg.norm(v))
            if not norm:
                continue
            recs.append({"graph": g, "id": nid, "title": note.title, "type": note.type,
                         "tags": list(note.tags or [])})
            vlist.append(v / norm)
            kept += 1

    stats = {
        "notes_in_scope": len(recs),
        "source_graphs": len(source_graphs),
        "target_graphs": len(target_graphs),
        "graphs_flagged": len(islands),
    }
    if len(recs) < 2:
        return json.dumps({
            "scope": scope_key, "candidates": [], "islands": islands,
            "under_connected": under_connected,
            "stats": {**stats, "pairs_proposed": 0}, "truncated": False,
        })

    M = np.vstack(vlist)  # (N, D) normalized float32
    n = len(recs)
    # Integer graph ids for vectorized same-graph masking.
    graph_names = sorted({r["graph"] for r in recs})
    gid = {name: i for i, name in enumerate(graph_names)}
    graph_id_arr = np.array([gid[r["graph"]] for r in recs], dtype=np.int32)
    target_set = set(target_graphs)
    is_target = np.array([r["graph"] in target_set for r in recs], dtype=bool)

    source_set = set(source_graphs)
    if note_id:
        source_pos = [i for i, r in enumerate(recs)
                      if r["graph"] == graph and r["id"] == note_id]
        if not source_pos:
            return json.dumps({
                "error": (
                    f"note '{note_id}' in graph '{graph}' is not an indexed content "
                    f"note (missing, excluded type/tag, or no embedding)."
                ),
                "type": "NotFound",
            })
    else:
        source_pos = [i for i, r in enumerate(recs) if r["graph"] in source_set]

    linked = _cross_conn_linked_pairs(union_graphs)

    # ── Block-wise neighbor search: masked matmul + top-k per row (C-level).
    # Over-fetch a small buffer so already-linked hits don't crowd out real
    # candidates before the Python-side link/threshold filter.
    want = top_k + 10
    best: dict = {}  # frozenset pair -> (sim, a_rec, b_rec)
    block = 512
    for start in range(0, len(source_pos), block):
        idx = source_pos[start:start + block]
        S = M[idx]                       # (b, D)
        sims = S @ M.T                   # (b, N)
        row_g = graph_id_arr[idx]        # (b,)
        # Mask: only target columns, different graph, above threshold, not self.
        valid = is_target[None, :] & (graph_id_arr[None, :] != row_g[:, None])
        sims = np.where(valid, sims, -1.0)
        k = min(want, n)
        part = np.argpartition(sims, -k, axis=1)[:, -k:]  # (b, k) unsorted top-k cols
        for r, si in enumerate(idx):
            a = recs[si]
            cols = part[r]
            # Sort this row's k candidates by similarity desc, then filter.
            order = cols[np.argsort(sims[r, cols])[::-1]]
            taken = 0
            for ci in order:
                sim = float(sims[r, ci])
                if sim < similarity_threshold:
                    break
                b = recs[ci]
                key = frozenset({(a["graph"], a["id"]), (b["graph"], b["id"])})
                if key in linked:
                    continue
                prev = best.get(key)
                if prev is None or sim > prev[0]:
                    best[key] = (sim, a, b)
                taken += 1
                if taken >= top_k:
                    break

    ranked = sorted(best.values(), key=lambda x: x[0], reverse=True)
    truncated = len(ranked) > max_candidates
    ranked = ranked[:max_candidates]

    candidates: list[dict] = []
    for sim, a, b in ranked:
        shared = [t for t in a["tags"] if t in set(b["tags"])]
        why_bits = []
        if shared:
            why_bits.append(f"shared tags {shared}")
        why_bits.append("cross-source")
        candidates.append({
            "a": {"graph": a["graph"], "note_id": a["id"], "title": a["title"], "type": a["type"]},
            "b": {"graph": b["graph"], "note_id": b["id"], "title": b["title"], "type": b["type"]},
            "similarity": round(sim, 3),
            "suggested_relation": _cross_conn_suggest_relation(a, b, shared),
            "why": "; ".join(why_bits),
        })

    return json.dumps({
        "scope": scope_key,
        "candidates": candidates,
        "islands": islands,
        "under_connected": under_connected,
        "stats": {**stats, "pairs_proposed": len(candidates)},
        "truncated": truncated,
    })