zettelkasten.synapse.retrieval

zettelkasten.synapse.retrieval

Unified cross-store retrieval: memory tree + intersecting ZK projects.

Everything here reuses the ZK framing machinery (:func:zettelkasten.framing. project_query / :func:~zettelkasten.framing.frame_question) unchanged — the only new ingredient is a COMBINED grapher that routes the special _memory source to :func:zettelkasten.synapse.memory_source.get_memory_source and every other name to the caller's ZK grapher. Hits are then tagged by store (memory vs zk) and tier (practice vs canon) so a consumer can weight private practice against public canon.

Read-only: no store is mutated. The scope is the intersection config (which ZK projects intersect this repo's memory), overridable per call.

clear_federated_cache

clear_federated_cache() -> None

Drop the process-global federated graph + peer-ANN caches (test/maintenance hook).

Source code in zettelkasten/synapse/retrieval.py

def clear_federated_cache() -> None:
    """Drop the process-global federated graph + peer-ANN caches (test/maintenance hook)."""
    with _FED_GRAPH_LOCK:
        _FED_GRAPH_CACHE.clear()
    with _PEER_GI_LOCK:
        _PEER_GI_CACHE.clear()

scope_tokens

scope_tokens(projects: list[str] | None) -> list[str]

Round-trippable scope tokens for reporting/manifest (local + federated).

Normalizes the scope arg (config or override) to project / <repo_id>:<project> tokens that re-parse to the same specs.

Source code in zettelkasten/synapse/retrieval.py

def scope_tokens(projects: list[str] | None) -> list[str]:
    """Round-trippable scope tokens for reporting/manifest (local + federated).

    Normalizes the scope arg (config or override) to ``project`` /
    ``<repo_id>:<project>`` tokens that re-parse to the same specs.
    """
    return [spec.token() for spec in _scope_specs(projects)]

unresolved_federated_repos

unresolved_federated_repos(projects: list[str] | None = None) -> set[str]

Repo ids that ARE in the current scope but cannot be resolved right now.

A federated scope whose peer repo is unreachable via :func:zettelkasten.federation.find_repo (deleted, moved, or otherwise offline) — as distinct from a peer the user has intentionally removed from the scope, which simply isn't among the specs at all. The distinction is what lets the overlay builder PRESERVE existing edges to a temporarily-missing peer (rather than pruning expensive LLM-typed edges) while still dropping edges to a peer that was deliberately dropped from intersects.

Source code in zettelkasten/synapse/retrieval.py

def unresolved_federated_repos(projects: list[str] | None = None) -> set[str]:
    """Repo ids that ARE in the current scope but cannot be resolved right now.

    A federated scope whose peer repo is unreachable via
    :func:`zettelkasten.federation.find_repo` (deleted, moved, or otherwise
    offline) — as distinct from a peer the user has intentionally *removed* from
    the scope, which simply isn't among the specs at all. The distinction is what
    lets the overlay builder PRESERVE existing edges to a temporarily-missing peer
    (rather than pruning expensive LLM-typed edges) while still dropping edges to a
    peer that was deliberately dropped from ``intersects``.
    """
    from zettelkasten import federation

    missing: set[str] = set()
    for spec in _scope_specs(projects):
        if spec.repo is not None and federation.find_repo(spec.repo) is None:
            missing.add(spec.repo)
    return missing

resolve_source_dir

resolve_source_dir(source_name: str, default_base: 'Path') -> 'Path'

Filesystem dir backing a (possibly namespaced) ZK source name.

A federated <repo_id>:<graph> resolves to <peer .zettelkasten/>/<graph>; a bare name to default_base/<name>. Used for freshness/staleness scans that must reach the peer's notes, not a same-named local box.

Source code in zettelkasten/synapse/retrieval.py

def resolve_source_dir(source_name: str, default_base: "Path") -> "Path":
    """Filesystem dir backing a (possibly namespaced) ZK source name.

    A federated ``<repo_id>:<graph>`` resolves to ``<peer .zettelkasten/>/<graph>``;
    a bare name to ``default_base/<name>``. Used for freshness/staleness scans that
    must reach the peer's notes, not a same-named local box.
    """
    local, repo = _resolve_federated(source_name)
    base = repo.zettel_dir if repo is not None else default_base
    return base / local

combined_grapher

combined_grapher(zk_get_graph: GetGraph) -> GetGraph

Wrap a ZK grapher so _memory and federated names also route.

• _memory -> the read-only memory-source adapter.
• a namespaced <repo_id>:<graph> -> a read-only :class:ZettelGraph pointed at the peer repo's .zettelkasten/ (structural load so its embedding index stays lazy and rebuilds locally on first .embeddings access). Cached in the PROCESS-GLOBAL :data:_FED_GRAPH_CACHE (mtime-gated), not per-grapher, so the index stays warm across queries in a long-lived worker. The peer's graphs: allowlist is enforced fail-closed.
• every other name -> the caller's local grapher, unchanged.

Resolved graphs are memoized for the LIFE OF THIS GRAPHER (one grapher is built per query). A single query resolves _memory and each box hundreds of times — via pruning, the keyword/semantic channels, and the reranker's per-candidate vector fetch — and each unmemoized _memory hit recomputed storage.source_fileset_signature() (a full glob+stat of every .memory/ file), the dominant warm-query cost at scale. Memoizing per-grapher collapses that to one resolution per distinct source per query while preserving freshness: the next query builds a new grapher (and the underlying :func:get_memory_source / :data:_FED_GRAPH_CACHE still revalidate on their own), so a within-query snapshot never masks a between-query change.

Source code in zettelkasten/synapse/retrieval.py

def combined_grapher(zk_get_graph: GetGraph) -> GetGraph:
    """Wrap a ZK grapher so ``_memory`` and federated names also route.

    * ``_memory`` -> the read-only memory-source adapter.
    * a namespaced ``<repo_id>:<graph>`` -> a read-only :class:`ZettelGraph`
      pointed at the peer repo's ``.zettelkasten/`` (structural load so its
      embedding index stays lazy and rebuilds locally on first ``.embeddings``
      access). Cached in the PROCESS-GLOBAL :data:`_FED_GRAPH_CACHE` (mtime-gated),
      not per-grapher, so the index stays warm across queries in a long-lived
      worker. The peer's ``graphs:`` allowlist is enforced fail-closed.
    * every other name -> the caller's local grapher, unchanged.

    Resolved graphs are memoized for the LIFE OF THIS GRAPHER (one grapher is
    built per query). A single query resolves ``_memory`` and each box hundreds of
    times — via pruning, the keyword/semantic channels, and the reranker's
    per-candidate vector fetch — and each unmemoized ``_memory`` hit recomputed
    ``storage.source_fileset_signature()`` (a full glob+stat of every ``.memory/``
    file), the dominant warm-query cost at scale. Memoizing per-grapher collapses
    that to one resolution per distinct source per query while preserving
    freshness: the next query builds a new grapher (and the underlying
    :func:`get_memory_source` / :data:`_FED_GRAPH_CACHE` still revalidate on their
    own), so a within-query snapshot never masks a between-query change.
    """
    from zettelkasten import federation

    memo: dict[str, Any] = {}

    def get_graph(name: str) -> Any:
        if name in memo:
            return memo[name]
        if name == MEMORY_SOURCE:
            resolved = get_memory_source()
        else:
            resolved = None
            split = federation.split_namespace(name)
            if split is not None:
                repo = federation.find_repo(split[0])
                if repo is not None:
                    local = split[1]
                    if not federation.graph_allowed(repo, local):
                        raise KeyError(
                            f"federated graph {name!r} is not exposed by its peer"
                        )
                    resolved = _federated_graph(name, repo, local)
            if resolved is None:
                resolved = zk_get_graph(name)
        memo[name] = resolved
        return resolved

    return get_graph

resolve_scope

resolve_scope(zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None, include_memory: bool = True) -> tuple[list[str], GetGraph]

Resolve (search_sources, get_graph) for the cross-store scope.

projects=None uses the intersection config; an explicit list overrides it ([] means "no ZK, memory only"), where each entry is a local project name or a federated <repo_id>:<project> token. ZK sources are the union of each scope's :func:~zettelkasten.framing.frame_search_sources, de-duplicated with order preserved; federated sources are surfaced under <repo_id>:<graph> names (honoring the peer's allowlist). The _memory source is appended unless include_memory is False.

Source code in zettelkasten/synapse/retrieval.py

def resolve_scope(
    zk_get_graph: GetGraph,
    projects: list[str] | None = None,
    graphs_dir: "Path | None" = None,
    include_memory: bool = True,
) -> tuple[list[str], GetGraph]:
    """Resolve ``(search_sources, get_graph)`` for the cross-store scope.

    ``projects=None`` uses the intersection config; an explicit list overrides
    it (``[]`` means "no ZK, memory only"), where each entry is a local project
    name or a federated ``<repo_id>:<project>`` token. ZK sources are the union
    of each scope's :func:`~zettelkasten.framing.frame_search_sources`,
    de-duplicated with order preserved; federated sources are surfaced under
    ``<repo_id>:<graph>`` names (honoring the peer's allowlist). The ``_memory``
    source is appended unless ``include_memory`` is False.
    """
    from zettelkasten import federation

    sources: list[str] = []
    seen: set[str] = set()
    for spec in _scope_specs(projects):
        if spec.repo is None:
            names = framing.frame_search_sources(spec.project, graphs_dir=graphs_dir)
        else:
            repo = federation.find_repo(spec.repo)
            if repo is None:
                logger.warning(
                    "synapse: federated repo %r not found; skipping scope %s",
                    spec.repo, spec.token(),
                )
                continue
            namespace = lambda s, _rid=repo.id: federation.namespace_id(_rid, s)
            raw = framing.frame_search_sources(
                spec.project, graphs_dir=repo.zettel_dir, namespace=namespace
            )
            names = [n for n in raw if _fed_source_allowed(repo, n)]
        for name in names:
            if name not in seen:
                seen.add(name)
                sources.append(name)
    if include_memory and MEMORY_SOURCE not in seen:
        sources.append(MEMORY_SOURCE)

    return sources, combined_grapher(zk_get_graph)

prune_sources

prune_sources(text: str, sources: list[str], get_graph: GetGraph, cap: int | None, *, always_keep: 'tuple[str, ...]' = (MEMORY_SOURCE,)) -> list[str]

Prune sources to the cap most keyword-relevant for text.

Cross-store reads otherwise build one embedding index per in-scope source on every cold query — the dominant cost and the kglite crash surface. This bounds the fan-out with a cheap, embedding-free keyword prescan: sources are ranked by :func:_keyword_relevance and only the top cap survive (ties and signal-less queries fall back to the original scope order, so the cap always bounds the fan-out even when keyword signal is weak/absent). Sources in always_keep (the memory tree by default — small, private practice worth keeping) are retained and count toward the cap.

Returns the surviving sources in their ORIGINAL scope order (deterministic; order only breaks reranker ties downstream). cap None/<= 0, or a scope already within the cap, returns sources unchanged.

Source code in zettelkasten/synapse/retrieval.py

def prune_sources(
    text: str,
    sources: list[str],
    get_graph: GetGraph,
    cap: int | None,
    *,
    always_keep: "tuple[str, ...]" = (MEMORY_SOURCE,),
) -> list[str]:
    """Prune ``sources`` to the ``cap`` most keyword-relevant for ``text``.

    Cross-store reads otherwise build one embedding index per in-scope source on
    every cold query — the dominant cost and the kglite crash surface. This bounds
    the fan-out with a cheap, embedding-free keyword prescan: sources are ranked
    by :func:`_keyword_relevance` and only the top ``cap`` survive (ties and
    signal-less queries fall back to the original scope order, so the cap always
    bounds the fan-out even when keyword signal is weak/absent). Sources in
    ``always_keep`` (the memory tree by default — small, private practice worth
    keeping) are retained and count toward the cap.

    Returns the surviving sources in their ORIGINAL scope order (deterministic;
    order only breaks reranker ties downstream). ``cap`` ``None``/``<= 0``, or a
    scope already within the cap, returns ``sources`` unchanged.
    """
    if cap is None or cap <= 0 or len(sources) <= cap:
        return sources

    keep_set = {s for s in sources if s in always_keep}
    budget = max(cap - len(keep_set), 0)

    ranked: list[tuple[int, int, str]] = []
    for pos, sname in enumerate(sources):
        if sname in keep_set:
            continue
        # -score so a higher relevance sorts first; pos keeps the original order
        # as a stable tiebreaker (and the sole signal when the query is tokenless).
        ranked.append((-_keyword_relevance(text, get_graph, sname), pos, sname))
    ranked.sort()
    selected = keep_set | {sname for _, _, sname in ranked[:budget]}

    pruned = [s for s in sources if s in selected]
    if len(pruned) < len(sources):
        logger.info(
            "synapse: pruned scope %d -> %d sources (cap=%d) for query prescan",
            len(sources), len(pruned), cap,
        )
    return pruned

frame_across_stores

frame_across_stores(question: str, zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None, max_sources: int | None = None, **frame_kwargs: Any) -> dict[str, Any]

Frame a question across memory + intersecting ZK, tagging hits by store/tier.

Returns the same shape as :func:zettelkasten.framing.frame_question, with each finding/claim/other entry annotated with store and tier and a top-level scope block describing what was searched.

max_sources caps how many in-scope sources are framed against (default: the synapse config, :func:zettelkasten.synapse.config.max_sources). The scope is pruned to the most keyword-relevant boxes for question BEFORE framing so the expensive per-box embedding index build is bounded — the dominant cold-query cost and the kglite crash surface. 0/negative disables pruning.

Source code in zettelkasten/synapse/retrieval.py

def frame_across_stores(
    question: str,
    zk_get_graph: GetGraph,
    projects: list[str] | None = None,
    graphs_dir: "Path | None" = None,
    max_sources: int | None = None,
    **frame_kwargs: Any,
) -> dict[str, Any]:
    """Frame a question across memory + intersecting ZK, tagging hits by store/tier.

    Returns the same shape as :func:`zettelkasten.framing.frame_question`, with
    each finding/claim/other entry annotated with ``store`` and ``tier`` and a
    top-level ``scope`` block describing what was searched.

    ``max_sources`` caps how many in-scope sources are framed against (default:
    the synapse config, :func:`zettelkasten.synapse.config.max_sources`). The
    scope is pruned to the most keyword-relevant boxes for ``question`` BEFORE
    framing so the expensive per-box embedding index build is bounded — the
    dominant cold-query cost and the kglite crash surface. ``0``/negative
    disables pruning.
    """
    sources, get_graph = resolve_scope(zk_get_graph, projects=projects, graphs_dir=graphs_dir)
    if not sources:
        return {"error": "No sources in scope to frame against.", "type": "NotFound"}

    full_scope = sources
    # Keep every ANN-servable source (local + federated peers) and prune only the
    # uncovered remainder to the cap — see search_across_stores / _select_sources.
    sources = _select_sources(question, sources, get_graph, graphs_dir, max_sources)

    result = framing.frame_question(
        question, get_graph=get_graph, project="", search_sources=sources,
        extra_recall=_federated_ann_recall, **frame_kwargs
    )
    if "error" in result:
        return result

    for facet in result.get("facets", []):
        for bucket in ("findings", "claims", "other"):
            for entry in facet.get(bucket, []):
                _annotate_entry(entry)

    result["scope"] = {
        "projects": scope_tokens(projects),
        "sources": sources,
        "memory_included": MEMORY_SOURCE in sources,
        "scope_size": len(full_scope),
        "pruned": len(sources) < len(full_scope),
    }
    return result

search_across_stores

search_across_stores(query: str, zk_get_graph: GetGraph, projects: list[str] | None = None, top_k: int = 15, graphs_dir: 'Path | None' = None, expand: bool = False, expand_min_conf: float = 0.6, max_sources: int | None = None) -> dict[str, Any]

Hybrid search across memory + intersecting ZK, fused via RRF.

Delegates to :func:zettelkasten.framing.project_query (per-source keyword + semantic channels fused with rrf_order, then merged across sources) and resolves each hit to a result dict tagged by store/tier.

When expand is set, each hit is expanded via the synapse link overlay: its high-confidence 1-hop cross-store neighbours are appended (tagged expanded_from + relation) — so a canon hit pulls in the practice that applies it, and vice versa.

max_sources caps how many in-scope sources are searched (default: the synapse config, :func:zettelkasten.synapse.config.max_sources). The scope is pruned to the most keyword-relevant boxes for query BEFORE the hybrid pass so the expensive per-box embedding index build is bounded — the dominant cold-query cost and the kglite crash surface. 0/negative disables pruning.

Source code in zettelkasten/synapse/retrieval.py

def search_across_stores(
    query: str,
    zk_get_graph: GetGraph,
    projects: list[str] | None = None,
    top_k: int = 15,
    graphs_dir: "Path | None" = None,
    expand: bool = False,
    expand_min_conf: float = 0.6,
    max_sources: int | None = None,
) -> dict[str, Any]:
    """Hybrid search across memory + intersecting ZK, fused via RRF.

    Delegates to :func:`zettelkasten.framing.project_query` (per-source keyword +
    semantic channels fused with ``rrf_order``, then merged across sources) and
    resolves each hit to a result dict tagged by store/tier.

    When ``expand`` is set, each hit is expanded via the synapse link overlay:
    its high-confidence 1-hop cross-store neighbours are appended (tagged
    ``expanded_from`` + ``relation``) — so a canon hit pulls in the practice that
    applies it, and vice versa.

    ``max_sources`` caps how many in-scope sources are searched (default: the
    synapse config, :func:`zettelkasten.synapse.config.max_sources`). The scope is
    pruned to the most keyword-relevant boxes for ``query`` BEFORE the hybrid pass
    so the expensive per-box embedding index build is bounded — the dominant
    cold-query cost and the kglite crash surface. ``0``/negative disables pruning.
    """
    sources, get_graph = resolve_scope(zk_get_graph, projects=projects, graphs_dir=graphs_dir)
    if not sources:
        return {"query": query, "results": [], "count": 0, "scope": {"sources": []}}

    full_scope = sources
    # Keep every ANN-servable source (this repo's global index covers local boxes;
    # each peer's global index covers its federated boxes) so its recall is one
    # pre-filtered vector search, and keyword-prune only the uncovered remainder to
    # the cap — bounding the cold per-box builds without dropping ANN-covered recall.
    sources = _select_sources(query, sources, get_graph, graphs_dir, max_sources)

    hits = framing.project_query(
        query, sources, top_k, get_graph, graphs_dir=graphs_dir,
        extra_recall=_federated_ann_recall,
    )
    results: list[dict[str, Any]] = []
    seen: set[str] = set()
    for nid, sim, sname, tier in hits:
        try:
            note = get_graph(sname).notes.get(nid)
        except Exception:
            note = None
        if note is None:
            continue
        results.append(_result_for(nid, sname, note, sim=sim, keyword_tier=tier))
        seen.add(nid)

    if expand:
        from zettelkasten.synapse import overlay as _overlay

        ov = _overlay.load_overlay()
        expanded: list[dict[str, Any]] = []
        for base in list(results):
            for hop in _overlay.high_conf_neighbors(base["id"], min_conf=expand_min_conf, overlay=ov):
                nb_id = hop.get("neighbor_id")
                if not nb_id or nb_id in seen:
                    continue
                # Resolve overlay neighbours against the FULL scope (a specific
                # node id, structural lookup only — no embeddings), so pruning the
                # semantic fan-out never drops a high-confidence expansion target.
                found = _find_note(get_graph, full_scope, nb_id)
                if found is None:
                    continue
                sname, note = found
                row = _result_for(nb_id, sname, note)
                row["expanded_from"] = base["id"]
                row["relation"] = hop.get("relation")
                row["connection_confidence"] = hop.get("confidence")
                expanded.append(row)
                seen.add(nb_id)
        results.extend(expanded)

    return {
        "query": query,
        "results": results,
        "count": len(results),
        "scope": {
            "projects": scope_tokens(projects),
            "sources": sources,
            "memory_included": MEMORY_SOURCE in sources,
            "expanded": bool(expand),
            "scope_size": len(full_scope),
            "pruned": len(sources) < len(full_scope),
        },
    }