zettelkasten.synapse.overlay

zettelkasten.synapse.overlay

The synapse link overlay: persisted, bipartite, typed cross-store edges.

The overlay is synapse's ONLY writable artifact — and it writes only to its own .synapse/links/ tree, never to .memory/ or .zettelkasten/. Every edge is bipartite (one memory endpoint, one ZK endpoint) and typed (memory --relation--> zk). It is committed (durable knowledge, expensive to recompute because typing costs LLM calls) with a manifest recording the store signatures it was built against, so :func:is_stale can detect drift and :func:build_connections can refresh.

Public surface:

• :func:build_connections — full pipeline (candidates -> gate -> LLM type -> threshold -> persist). Accepts an injected typer so it runs without an LLM.
• :func:get_connections — typed edges touching a node (either store).
• :func:high_conf_neighbors — 1-hop high-confidence cross-store neighbours, for retrieval expansion.
• :func:is_stale / :func:refresh_if_stale — cache lifecycle.

zk_signature

zk_signature(zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None) -> str

A freshness token over the in-scope ZK source dirs (newest *.md mtimes).

Each source dir is resolved through :func:resolve_source_dir so a federated <repo_id>:<graph> source scans the PEER repo's notes (not a same-named local box), and an edit to a peer note advances the token → is_stale fires.

Source code in zettelkasten/synapse/overlay.py

def zk_signature(zk_get_graph: GetGraph, projects: list[str] | None = None,
                 graphs_dir: "Path | None" = None) -> str:
    """A freshness token over the in-scope ZK source dirs (newest ``*.md`` mtimes).

    Each source dir is resolved through :func:`resolve_source_dir` so a federated
    ``<repo_id>:<graph>`` source scans the PEER repo's notes (not a same-named
    local box), and an edit to a peer note advances the token → ``is_stale`` fires.
    """
    from zettelkasten.graph import GRAPHS_DIR

    base = graphs_dir or GRAPHS_DIR
    sources, _ = resolve_scope(zk_get_graph, projects=projects, graphs_dir=graphs_dir)
    parts: list[str] = []
    for source in sorted(s for s in sources if s != MEMORY_SOURCE):
        parts.append(f"{source}|{_newest_md_mtime(resolve_source_dir(source, base))}")
    return "\x00".join(parts)

load_overlay

load_overlay(kind: str = 'generic') -> dict[str, Any]

Load the overlay of kind, returning an empty skeleton when absent/malformed.

Source code in zettelkasten/synapse/overlay.py

def load_overlay(kind: str = "generic") -> dict[str, Any]:
    """Load the overlay of ``kind``, returning an empty skeleton when absent/malformed."""
    path = overlay_path(kind)
    if not path.exists():
        return {"version": FORMAT_VERSION, "manifest": {}, "edges": []}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        logger.warning("synapse: could not read overlay %s: %s", path, exc)
        return {"version": FORMAT_VERSION, "manifest": {}, "edges": []}
    if not isinstance(data, dict):
        return {"version": FORMAT_VERSION, "manifest": {}, "edges": []}
    data.setdefault("edges", [])
    data.setdefault("manifest", {})
    return data

load_overlay_from

load_overlay_from(path: 'Path | str') -> dict[str, Any]

Load an overlay from an EXPLICIT links.json path (no env mutation).

A path-parametrized twin of :func:load_overlay for callers that must read a SPECIFIC overlay file rather than the workspace default (e.g. the Iceberg hosted broker, which resolves the owner's source overlay path itself and must not depend on / mutate os.environ or the process-global synapse dir).

Returns the SAME empty skeleton ({"version", "manifest", "edges"}) as :func:load_overlay when the file is absent, unreadable, malformed, or not a JSON object, so a caller always receives a well-formed overlay dict.

Source code in zettelkasten/synapse/overlay.py

def load_overlay_from(path: "Path | str") -> dict[str, Any]:
    """Load an overlay from an EXPLICIT ``links.json`` path (no env mutation).

    A path-parametrized twin of :func:`load_overlay` for callers that must read a
    SPECIFIC overlay file rather than the workspace default (e.g. the Iceberg
    hosted broker, which resolves the owner's source overlay path itself and must
    not depend on / mutate ``os.environ`` or the process-global synapse dir).

    Returns the SAME empty skeleton (``{"version", "manifest", "edges"}``) as
    :func:`load_overlay` when the file is absent, unreadable, malformed, or not a
    JSON object, so a caller always receives a well-formed overlay dict.
    """
    path = Path(path)
    if not path.exists():
        return {"version": FORMAT_VERSION, "manifest": {}, "edges": []}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        logger.warning("synapse: could not read overlay %s: %s", path, exc)
        return {"version": FORMAT_VERSION, "manifest": {}, "edges": []}
    if not isinstance(data, dict):
        return {"version": FORMAT_VERSION, "manifest": {}, "edges": []}
    data.setdefault("edges", [])
    data.setdefault("manifest", {})
    return data

save_overlay

save_overlay(overlay: dict[str, Any], kind: str = 'generic') -> Path

Persist the overlay of kind atomically to .synapse/links/.

Source code in zettelkasten/synapse/overlay.py

def save_overlay(overlay: dict[str, Any], kind: str = "generic") -> Path:
    """Persist the overlay of ``kind`` atomically to ``.synapse/links/``."""
    path = overlay_path(kind)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(overlay, indent=2, ensure_ascii=False), encoding="utf-8")
    tmp.replace(path)
    return path

is_stale

is_stale(zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None, kind: str = 'generic') -> bool

Whether the overlay was built against a now-changed view of the stores.

Source code in zettelkasten/synapse/overlay.py

def is_stale(zk_get_graph: GetGraph, projects: list[str] | None = None,
             graphs_dir: "Path | None" = None, kind: str = "generic") -> bool:
    """Whether the overlay was built against a now-changed view of the stores."""
    overlay = load_overlay(kind)
    manifest = overlay.get("manifest") or {}
    if not manifest:
        return True
    mem_sig = storage.source_fileset_signature()
    zk_sig = zk_signature(zk_get_graph, projects=projects, graphs_dir=graphs_dir)
    return manifest.get("memory_sig") != mem_sig or manifest.get("zk_sig") != zk_sig

build_connections

build_connections(zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None, typer: Typer | None = None, per_node_k: int = 5, sim_threshold: float = 0.45, min_confidence: float = 0.55, max_pairs: int = 200, limit: int | None = None, kind: str = 'generic', canon_types: 'tuple[str, ...] | None' = None, practice_types: 'tuple[str, ...] | None' = None, overfetch: int = 1, enrich_edges: 'Callable[[list[dict[str, Any]], GetGraph, list[str] | None], None] | None' = None) -> dict[str, Any]

Run the full connection pipeline and persist the overlay of kind.

Candidates (semantic NN + shared provenance) are gated to the top max_pairs by score, each is typed (LLM by default; inject typer to avoid the LLM), and edges surviving min_confidence with a real relation are written. Returns the persisted overlay dict.

kind selects the persisted file (generic → links.json, claim → claim_links.json) so the claim-aligned overlay is SEPARATE from the generic note-note one; offline-peer preservation and drift are per-kind. canon_types/practice_types/overfetch are forwarded to :func:candidates.generate_candidates (the claim path restricts registers and overfetches to offset the type filter). enrich_edges is an optional hook invoked ONCE on the surviving edge list (before peer-merge/persist) so a caller can annotate edges — e.g. the claim path attaches claim_strength and a blended synthesis priority — WITHOUT changing the confidence gate; it must never drop edges.

Source code in zettelkasten/synapse/overlay.py

def build_connections(
    zk_get_graph: GetGraph,
    projects: list[str] | None = None,
    graphs_dir: "Path | None" = None,
    typer: Typer | None = None,
    per_node_k: int = 5,
    sim_threshold: float = 0.45,
    min_confidence: float = 0.55,
    max_pairs: int = 200,
    limit: int | None = None,
    kind: str = "generic",
    canon_types: "tuple[str, ...] | None" = None,
    practice_types: "tuple[str, ...] | None" = None,
    overfetch: int = 1,
    enrich_edges: "Callable[[list[dict[str, Any]], GetGraph, list[str] | None], None] | None" = None,
) -> dict[str, Any]:
    """Run the full connection pipeline and persist the overlay of ``kind``.

    Candidates (semantic NN + shared provenance) are gated to the top
    ``max_pairs`` by score, each is typed (LLM by default; inject ``typer`` to
    avoid the LLM), and edges surviving ``min_confidence`` with a real relation
    are written. Returns the persisted overlay dict.

    ``kind`` selects the persisted file (``generic`` → ``links.json``, ``claim`` →
    ``claim_links.json``) so the claim-aligned overlay is SEPARATE from the
    generic note-note one; offline-peer preservation and drift are per-kind.
    ``canon_types``/``practice_types``/``overfetch`` are forwarded to
    :func:`candidates.generate_candidates` (the claim path restricts registers and
    overfetches to offset the type filter). ``enrich_edges`` is an optional hook
    invoked ONCE on the surviving edge list (before peer-merge/persist) so a caller
    can annotate edges — e.g. the claim path attaches ``claim_strength`` and a
    blended synthesis ``priority`` — WITHOUT changing the confidence gate; it must
    never drop edges.
    """
    from zettelkasten.synapse import candidates as cand

    typer = typer or llm_typer
    sources, get_graph = resolve_scope(zk_get_graph, projects=projects, graphs_dir=graphs_dir)
    mem = get_graph(MEMORY_SOURCE)

    cands = cand.generate_candidates(
        zk_get_graph, projects=projects, graphs_dir=graphs_dir,
        per_node_k=per_node_k, sim_threshold=sim_threshold, limit=limit,
        canon_types=canon_types, practice_types=practice_types, overfetch=overfetch,
    )[:max_pairs]

    edges: list[dict[str, Any]] = []
    now = _now()
    typing_error: str | None = None
    typed_any = False
    for c in cands:
        mnote = mem.notes.get(c.memory_id)
        try:
            zg = get_graph(c.zk_source)
            znote = getattr(zg, "notes", {}).get(c.zk_id)
        except Exception:
            znote = None
        if mnote is None or znote is None:
            continue
        msum = {
            "id": c.memory_id, "store": "memory", "type": mnote.type,
            "title": mnote.title, "text": cand._node_text(mnote),
        }
        zsum = {
            "id": c.zk_id, "store": "zk", "type": getattr(znote, "type", ""),
            "title": getattr(znote, "title", ""), "text": cand._node_text(znote),
        }
        try:
            typed = typer(msum, zsum, c.signals())
        except (TypingBackendError, TimeoutError) as exc:
            # An UNAVAILABLE typing backend (bridge never came up / timed out), not
            # a per-pair 'none'. If nothing has typed yet the whole runtime is dead:
            # abort now rather than eat one timeout per candidate. If we HAD typed
            # earlier, treat it as a transient blip and skip just this pair.
            if not typed_any:
                typing_error = str(exc) or exc.__class__.__name__
                logger.error("synapse: typing backend unavailable — aborting build: %s", exc)
                break
            logger.warning("synapse: typing failed for one candidate, skipping: %s", exc)
            continue
        typed_any = True
        relation = typed.get("relation", "none")
        confidence = float(typed.get("confidence", 0.0) or 0.0)
        if relation not in TYPED_RELATIONS or confidence < min_confidence:
            continue
        edges.append({
            "memory_id": c.memory_id,
            "zk_id": c.zk_id,
            "zk_source": c.zk_source,
            "relation": relation,
            "confidence": round(confidence, 4),
            "rationale": typed.get("rationale", ""),
            "signals": c.signals(),
            "created_at": now,
        })

    if typing_error is not None and not edges:
        # The typing backend was unavailable and nothing was classified. Return an
        # error WITHOUT persisting, so a previously-good overlay is left intact
        # rather than clobbered by an empty rebuild (and its manifest signatures
        # not falsely advanced to "fresh").
        logger.warning("synapse: build aborted — overlay left unchanged (%s)", typing_error)
        return {
            "version": FORMAT_VERSION,
            "manifest": {
                "generated_at": now,
                "kind": kind,
                "candidate_count": len(cands),
                "edge_count": 0,
                "typing_error": typing_error,
            },
            "edges": [],
            "error": typing_error,
        }

    # Optional per-kind edge annotation (e.g. the claim path attaches
    # ``claim_strength`` + a blended ``priority``). Purely additive: the caller
    # must not drop edges (the confidence gate above is the ONLY gate).
    if enrich_edges is not None and edges:
        enrich_edges(edges, get_graph, projects)

    # Carry forward edges to a federated peer that is IN scope but currently
    # unresolvable (deleted / moved / offline). Such a peer drops out of
    # ``resolve_scope``, so a naive rebuild would silently prune its edges — the
    # exact ones that were expensive to LLM-type. We preserve them so a temporary
    # disappearance is lossless (they reactivate when the peer returns), while a
    # peer intentionally removed from ``intersects`` is NOT in scope, so its edges
    # are correctly dropped. New edges are for locally-resolvable sources, so they
    # never overlap a preserved (namespaced) peer edge.
    preserved = _preserved_peer_edges(projects, kind)
    if preserved:
        edges = edges + preserved

    overlay = {
        "version": FORMAT_VERSION,
        "manifest": {
            "generated_at": now,
            "kind": kind,
            "projects": scope_tokens(projects),
            "memory_sig": storage.source_fileset_signature(),
            "zk_sig": zk_signature(zk_get_graph, projects=projects, graphs_dir=graphs_dir),
            "params": {
                "per_node_k": per_node_k,
                "sim_threshold": sim_threshold,
                "min_confidence": min_confidence,
                "max_pairs": max_pairs,
                "overfetch": overfetch,
            },
            "candidate_count": len(cands),
            "edge_count": len(edges),
            "preserved_edge_count": len(preserved),
        },
        "edges": edges,
    }
    save_overlay(overlay, kind)
    logger.info(
        "synapse: built %s overlay — %d edges from %d candidates (%d preserved from offline peers)",
        kind, len(edges), len(cands), len(preserved),
    )
    return overlay

refresh_if_stale

refresh_if_stale(zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None, kind: str = 'generic', **build_kwargs: Any) -> dict[str, Any]

Rebuild the overlay of kind only when the stores have drifted; else load it.

Source code in zettelkasten/synapse/overlay.py

def refresh_if_stale(zk_get_graph: GetGraph, projects: list[str] | None = None,
                     graphs_dir: "Path | None" = None, kind: str = "generic",
                     **build_kwargs: Any) -> dict[str, Any]:
    """Rebuild the overlay of ``kind`` only when the stores have drifted; else load it."""
    if is_stale(zk_get_graph, projects=projects, graphs_dir=graphs_dir, kind=kind):
        return build_connections(zk_get_graph, projects=projects, graphs_dir=graphs_dir,
                                  kind=kind, **build_kwargs)
    return load_overlay(kind)

get_connections

get_connections(node_id: str, overlay: dict[str, Any] | None = None, kind: str = 'generic') -> list[dict[str, Any]]

Typed edges touching node_id on either endpoint (in the kind overlay).

Source code in zettelkasten/synapse/overlay.py

def get_connections(node_id: str, overlay: dict[str, Any] | None = None,
                    kind: str = "generic") -> list[dict[str, Any]]:
    """Typed edges touching ``node_id`` on either endpoint (in the ``kind`` overlay)."""
    if overlay is None:
        overlay = load_overlay(kind)
    node_id = node_id.strip()
    return [e for e in overlay.get("edges", []) if e.get("memory_id") == node_id or e.get("zk_id") == node_id]

high_conf_neighbors

high_conf_neighbors(node_id: str, min_conf: float = 0.6, overlay: dict[str, Any] | None = None, kind: str = 'generic') -> list[dict[str, Any]]

1-hop high-confidence cross-store neighbours of node_id.

Returns [{neighbor_id, store, relation, confidence}] — the OTHER endpoint of each qualifying edge. Used to expand retrieval with strongly-linked nodes from the opposite store.

Source code in zettelkasten/synapse/overlay.py

def high_conf_neighbors(
    node_id: str, min_conf: float = 0.6, overlay: dict[str, Any] | None = None,
    kind: str = "generic",
) -> list[dict[str, Any]]:
    """1-hop high-confidence cross-store neighbours of ``node_id``.

    Returns ``[{neighbor_id, store, relation, confidence}]`` — the OTHER endpoint
    of each qualifying edge. Used to expand retrieval with strongly-linked nodes
    from the opposite store.
    """
    out: list[dict[str, Any]] = []
    for e in get_connections(node_id, overlay=overlay, kind=kind):
        if float(e.get("confidence", 0.0)) < min_conf:
            continue
        if e.get("memory_id") == node_id:
            out.append({"neighbor_id": e.get("zk_id"), "store": "zk",
                        "relation": e.get("relation"), "confidence": e.get("confidence")})
        else:
            out.append({"neighbor_id": e.get("memory_id"), "store": "memory",
                        "relation": e.get("relation"), "confidence": e.get("confidence")})
    return out