zettelkasten.synapse

zettelkasten.synapse

Synapse: a read-only layer linking the memory tree and the zettelkasten.

Synapse never writes to .memory/ or .zettelkasten/ — both stores stay canonical, written only by their own workflows. On top of them synapse provides:

• Unified retrieval across the memory tree + intersecting ZK projects, fused in the shared 512-d model2vec space (both stores embed with the same model).
• Typed cross-store connections discovered hands-off (semantic NN + shared provenance, LLM-typed) and stored in synapse's own bipartite overlay — every edge has one memory endpoint and one ZK endpoint, never within-store.
• A matrix/apex synthesis lens projecting the overlay (deferred module).

It is application-agnostic substrate; a domain agent (e.g. a quant research assistant) is just one consumer.

Layout:

• .synapse/config.yaml — committed, user-authored intersections.
• .synapse/links/links.json — committed link overlay (typed cross-store edges). Durable because typing costs LLM calls; a manifest tracks the store signatures it was built against so it can be refreshed when either store drifts.
• .angelo/synapse/ — gitignored scratch space for any recomputable caches.

This subpackage lives inside zettelkasten (which already imports memory utilities) so the cross-store code respects the one-way dependency direction: zettelkasten may import memory; memory never imports zettelkasten.

IntersectSpec dataclass

One resolved intersects entry: a ZK project + optional federated repo.

repo is None is a LOCAL scope (project in this repo's .zettelkasten/); a non-None repo names a peer id declared under federated_repos in .zettelkasten/config.yaml.

Source code in zettelkasten/synapse/config.py

@dataclass(frozen=True)
class IntersectSpec:
    """One resolved ``intersects`` entry: a ZK project + optional federated repo.

    ``repo is None`` is a LOCAL scope (project in this repo's ``.zettelkasten/``);
    a non-None ``repo`` names a peer id declared under ``federated_repos`` in
    ``.zettelkasten/config.yaml``.
    """

    project: str
    repo: str | None = None

    def token(self) -> str:
        """Round-trippable scope token: ``project`` or ``<repo>:<project>``.

        The federated form mirrors :func:`zettelkasten.federation.namespace_id`
        so a token stored in a build manifest re-parses to the same spec.
        """
        return f"{self.repo}:{self.project}" if self.repo else self.project

token

token() -> str

Round-trippable scope token: project or <repo>:<project>.

The federated form mirrors :func:zettelkasten.federation.namespace_id so a token stored in a build manifest re-parses to the same spec.

Source code in zettelkasten/synapse/config.py

def token(self) -> str:
    """Round-trippable scope token: ``project`` or ``<repo>:<project>``.

    The federated form mirrors :func:`zettelkasten.federation.namespace_id`
    so a token stored in a build manifest re-parses to the same spec.
    """
    return f"{self.repo}:{self.project}" if self.repo else self.project

MemoryNote dataclass

A memory entry adapted to the note surface project_query reads.

links/backlinks are intentionally empty: cross-store relationships live in synapse's own overlay, and within-memory edges (related_to) aren't part of retrieval. entry keeps the raw record so downstream (connection typing, provenance) can read files/parent_id/timestamps.

Source code in zettelkasten/synapse/memory_source.py

@dataclass
class MemoryNote:
    """A memory entry adapted to the note surface ``project_query`` reads.

    ``links``/backlinks are intentionally empty: cross-store relationships live
    in synapse's own overlay, and within-memory edges (``related_to``) aren't
    part of retrieval. ``entry`` keeps the raw record so downstream (connection
    typing, provenance) can read ``files``/``parent_id``/timestamps.
    """

    id: str
    title: str
    type: str
    tags: list[str]
    body: str
    aliases: list[str] = field(default_factory=list)
    links: list[Any] = field(default_factory=list)
    entry: dict[str, Any] = field(default_factory=dict)

MemorySourceGraph

The .memory/ tree adapted to the ZK source-graph surface (read-only).

Source code in zettelkasten/synapse/memory_source.py

class MemorySourceGraph:
    """The ``.memory/`` tree adapted to the ZK source-graph surface (read-only)."""

    def __init__(self, entries: list[dict[str, Any]] | None = None, embedder: Any = None) -> None:
        if entries is None:
            entries = _read_entries()
        self.name = MEMORY_SOURCE
        self.notes: dict[str, MemoryNote] = {}
        for e in entries:
            eid = e.get("id")
            if not eid:
                continue
            self.notes[str(eid)] = MemoryNote(
                id=str(eid),
                title=str(e.get("title", "")),
                type=str(e.get("type", "note")),
                tags=_coerce_tags(e.get("tags")),
                body=str(e.get("body", "")),
                entry=e,
            )
        self.embeddings: MemorySemanticIndex | None = (
            MemorySemanticIndex(entries, embedder) if embedder is not None else None
        )

    def get_backlinks(self, note_id: str) -> list[dict[str, Any]]:
        # Memory has no supports/replicates edges; corroboration is a ZK concept.
        return []

Candidate dataclass

A possible cross-store link (one memory node, one ZK node) + why.

Source code in zettelkasten/synapse/candidates.py

@dataclass
class Candidate:
    """A possible cross-store link (one memory node, one ZK node) + why."""

    memory_id: str
    zk_id: str
    zk_source: str
    semantic_sim: float = 0.0
    shared_provenance: list[str] = field(default_factory=list)
    # A small additive score prior (default 0.0 → no effect on the generic
    # note-note path). The claim-aligned build sets it to prefer ``_cross``
    # synthesis claims over single-source findings.
    prefer_bonus: float = 0.0

    def score(self) -> float:
        """Combined candidate strength (provenance overlap is a strong prior)."""
        return self.semantic_sim + (0.5 if self.shared_provenance else 0.0) + self.prefer_bonus

    def signals(self) -> dict[str, Any]:
        return {
            "semantic_sim": round(self.semantic_sim, 4),
            "shared_provenance": list(self.shared_provenance),
        }

score

score() -> float

Combined candidate strength (provenance overlap is a strong prior).

Source code in zettelkasten/synapse/candidates.py

def score(self) -> float:
    """Combined candidate strength (provenance overlap is a strong prior)."""
    return self.semantic_sim + (0.5 if self.shared_provenance else 0.0) + self.prefer_bonus

Navigator

The deterministic action executor that steers the LLM to ANSWER or ABSTAIN.

Construct with the injected seams (all optional except the LLM) and call :meth:navigate. Every source of non-determinism — the model, tether verification, verifier grounding, neighborhood expansion, and the prune walk — is injected, so a run is reproducible and testable with no live model or store. The navigator NEVER decides grounding itself: the terminal ANSWER/ ABSTAIN verdict is always delegated to :func:zettelkasten.synapse.grounding_router.route.

Source code in zettelkasten/synapse/navigation.py

class Navigator:
    """The deterministic action executor that steers the LLM to ANSWER or ABSTAIN.

    Construct with the injected seams (all optional except the LLM) and call
    :meth:`navigate`. Every source of non-determinism — the model, tether
    verification, verifier grounding, neighborhood expansion, and the prune walk
    — is injected, so a run is reproducible and testable with no live model or
    store. The navigator NEVER decides grounding itself: the terminal ANSWER/
    ABSTAIN verdict is always delegated to
    :func:`zettelkasten.synapse.grounding_router.route`.
    """

    def __init__(
        self,
        llm: LLM,
        *,
        registry: VerifierRegistry | None = None,
        resolver: TetherResolver | None = None,
        config: NavConfig | None = None,
        expander: Expander | None = None,
        grounder: Grounder | None = None,
        walk_fn: WalkFn | None = None,
        voi_model: VOIModel | None = None,
        relevance: Mapping[str, float] | None = None,
        system: str = NAV_SYSTEM,
    ) -> None:
        """Bind the executor to its injected seams.

        ``llm`` is the required model seam (see :data:`LLM`). ``registry`` is the
        verifier registry the ``ground``/``answer`` verdicts route through
        (defaults to the router's production registry). ``resolver`` is the
        :class:`~zettelkasten.synapse.epistemics.TetherResolver` used to derive
        node statuses (defaults to the maximally conservative one). ``expander``
        / ``grounder`` / ``walk_fn`` back the ``expand`` / ``ground`` / ``prune``
        moves; ``voi_model`` scores handles (default :func:`minimal_voi`);
        ``relevance`` optionally supplies per-token relevance for VOI.
        """
        self.llm = llm
        self.registry = registry
        self.resolver = resolver or TetherResolver()
        self.config = config or NavConfig()
        self.expander = expander or _default_expander
        self.grounder = grounder or _default_grounder
        self.walk_fn = walk_fn
        if voi_model is not None:
            self.voi_model = voi_model
        elif self.config.voi_policy == "resolvability_cost":
            self.voi_model = resolvability_cost_voi
        else:
            self.voi_model = minimal_voi
        self.relevance = dict(relevance or {})
        self.system = system

    # -- public entry point ---------------------------------------------------

    def navigate(
        self,
        question: str,
        neighborhood: Neighborhood,
        *,
        groundings: Mapping[str, GroundingSpec] | None = None,
    ) -> NavResult:
        """Run the navigation loop over ``neighborhood`` to a verdict.

        ``question`` is the user's query; ``neighborhood`` is the assembled
        substrate the run starts from (further ``expand`` moves may grow it).
        ``groundings`` optionally seeds grounding specs (the ``ground`` action
        adds more). Returns a :class:`NavResult` whose ``decision`` is the
        router's ANSWER/ABSTAIN verdict (or a structurally-forced ABSTAIN on a
        budget/parse failure).
        """
        state = _NavState(
            neighborhood=neighborhood,
            node_status=apply_node_statuses(neighborhood.node_list(), self.resolver),
            groundings=dict(groundings or {}),
        )

        spent = 0
        hops = 0
        renav = 0
        parse_repairs = 0
        feedback = ""
        last_output: AgentOutput | None = None

        while True:
            if hops >= self.config.max_hops:
                state.trace.append(f"halt: hop cap {self.config.max_hops} reached")
                return self._finish(
                    self._fallback_abstain(state, "hop cap reached before an answer"),
                    "hops", state, last_output, hops, spent, renav, parse_repairs,
                    loop_break="voi-max",
                )

            self._render(state, question)
            prompt = self._build_prompt(
                question,
                state,
                feedback,
                spent=spent,
                hops=hops,
            )

            # HARD budget guardrail: check the next turn's cost BEFORE the call,
            # so an over-long loop is halted deterministically. VOI-max fallback
            # then breaks the loop rather than looping forever.
            est = _estimate_tokens(prompt) + _estimate_tokens(self.system)
            if spent + est > self.config.token_budget:
                state.trace.append(
                    f"halt: token budget {self.config.token_budget} would be exceeded "
                    f"(spent={spent}, next~{est})"
                )
                return self._finish(
                    self._fallback_abstain(state, "token budget exhausted"),
                    "budget", state, last_output, hops, spent, renav, parse_repairs,
                    loop_break="voi-max",
                )

            raw = self.llm(self.system, prompt)
            spent += est + _estimate_tokens(raw)

            try:
                output = parse_output(raw)
            except MemDSLParseError as exc:
                # Log the raw pre-parse text so a near-miss the parser could not
                # recover is diagnosable later. DEBUG + logger (never stdout): the
                # RESULT line lives on stdout and must stay uncontaminated.
                logger.debug(
                    "MemDSL OUTPUT parse-fail (repair %d/%d): %s\nraw pre-parse text:\n%s",
                    parse_repairs,
                    self.config.max_parse_repairs,
                    exc,
                    raw,
                )
                if parse_repairs < self.config.max_parse_repairs:
                    parse_repairs += 1
                    # Embed BOTH terminal exemplars on repair turns so the repair
                    # feedback is EVEN-HANDED: a lone @answer exemplar biases a turn
                    # that SHOULD abstain toward emitting an @answer (which the
                    # grounder then catches, but at the cost of a wasted repair turn
                    # and a nudge the wrong way). Showing the @abstain form alongside
                    # it makes clear that abstaining is a first-class, equally-valid
                    # OUTPUT. One concise exemplar of each keeps the message bounded
                    # against the per-turn token floor the budget guardrail counts.
                    feedback = (
                        f"PARSE ERROR: {exc}. Re-emit a VALID MemDSL OUTPUT artifact "
                        "(correct header, well-formed actions, one terminal decision — "
                        "an @answer with a cited DAG, OR an @abstain <reason> :: <detail>; "
                        "abstaining is an equally-valid answer when nothing grounds it). "
                        "Common mistakes to avoid: do NOT wrap the output in markdown "
                        "code fences (no ``` / ```memdsl); emit NOTHING before the "
                        "'@memdsl.out/2' header or after the terminal decision (no preamble, "
                        "no sign-off); use straight ASCII quotes (\") in structural "
                        "tokens, not curly quotes. Match one of these exact forms:\n"
                        f"{NAV_EXEMPLAR_ANSWER}"
                        "\nor, to abstain:\n"
                        f"{NAV_EXEMPLAR_ABSTAIN}"
                    )
                    state.trace.append(f"parse-fail: repair retry {parse_repairs}")
                    continue
                state.trace.append("parse-fail: repair cap reached; abstaining")
                return self._finish(
                    self._structural_abstain("unparseable agent output after repair retry"),
                    "parse-fail", state, last_output, hops, spent, renav, parse_repairs,
                )

            feedback = ""
            last_output = output
            hops += 1
            self._execute_actions(state, output.actions)

            # Terminal: an explicit agent abstain is honored as an abstain.
            if output.abstain is not None:
                state.trace.append(f"terminal: agent @abstain ({output.abstain.reason})")
                return self._finish(
                    self._structural_abstain(
                        output.abstain.detail or "agent abstained", reason=output.abstain.reason
                    ),
                    "abstain", state, last_output, hops, spent, renav, parse_repairs,
                )

            # Terminal: an @answer's verdict is ALWAYS delegated to the router.
            if output.reasoning is not None:
                decision = self._route(state, output.reasoning)
                if decision.verdict == ANSWER:
                    state.trace.append("terminal: router ANSWER")
                    return self._finish(
                        decision, "answer", state, last_output, hops, spent, renav, parse_repairs,
                    )
                # Bounded self-correction: a verifier refutation triggers ONE
                # re-navigate (pick a different path), never a double-down.
                if REASON_FALSE_PREMISE in decision.reasons and renav < self.config.max_renavigations:
                    renav += 1
                    feedback = (
                        f"RE-NAVIGATE: a verifier refuted a ground ({decision.detail}). "
                        "Do not repeat that claim; navigate to a different, grounded path "
                        "or @abstain."
                    )
                    state.trace.append(f"self-correct: re-navigate {renav} after false-premise")
                    continue
                # Bounded self-correction: a low-coverage abstain triggers ONE
                # re-navigate that NAMES the ungrounded nodes the conclusion rested
                # on, so the agent can rebuild on grounded ground or abstain honestly.
                if REASON_LOW_COVERAGE in decision.reasons and renav < self.config.max_renavigations:
                    renav += 1
                    feedback = self._low_coverage_feedback(output.reasoning, decision)
                    state.trace.append(f"self-correct: re-navigate {renav} after low-coverage")
                    continue
                state.trace.append(f"terminal: router ABSTAIN ({', '.join(decision.reasons)})")
                return self._finish(
                    decision, "abstain", state, last_output, hops, spent, renav, parse_repairs,
                )

            # No terminal this turn — only navigation actions. Loop again; the
            # budget/hop guardrail will eventually break a non-terminating loop.
            state.trace.append(f"hop {hops}: navigation actions, no terminal")

    # -- render + prompt ------------------------------------------------------

    def _render(self, state: _NavState, question: str) -> None:
        """Render the current neighborhood to an envelope and refresh VOI + maps."""
        envelope = render_neighborhood(
            state.neighborhood,
            state.node_status,
            expanded=state.expanded,
            max_body_chars=self.config.render_max_body_chars,
        )
        state.envelope = envelope
        state.ref_to_token = {n.ref: n.address.to_token() for n in envelope.nodes}
        state.token_to_ref = {tok: ref for ref, tok in state.ref_to_token.items()}
        state.voi = self._compute_voi(state)

    def _compute_voi(self, state: _NavState) -> dict[str, VOIAnnotation]:
        """Annotate every rendered handle with an advisory ``(voi, cost)``.

        For an ELIDED node the render body is a truncated, lossy view, but a
        ``ground`` re-fetches the FULL body — so the truncated view UNDERSTATES
        the real grounding cost. Estimating cost from the shown body would let
        the VOI heuristic wrongly prefer an elided node whose true cost is much
        larger. So an elided node's cost is estimated from its FULL substrate
        body (read back from the neighborhood by token), reflecting what
        ``ground()`` will actually re-fetch; a non-elided node's shown body IS
        its full body, so it is estimated directly.
        """
        annotations: dict[str, VOIAnnotation] = {}
        for node in (state.envelope.nodes if state.envelope else []):
            token = node.address.to_token()
            verdict = state.node_status.get(token, ep.INFERRED)
            body_for_cost = node.body
            if node.elided:
                raw = state.neighborhood.nodes.get(token)
                if raw is not None:
                    body_for_cost = raw.text
            candidate = VOICandidate(
                ref=node.ref,
                token=token,
                relevance=self.relevance.get(token, 1.0),
                status=verdict,
                cost=_estimate_tokens(body_for_cost),
            )
            annotations[node.ref] = self.voi_model(candidate)
        return annotations

    def _build_prompt(
        self,
        question: str,
        state: _NavState,
        feedback: str,
        *,
        spent: int = 0,
        hops: int = 0,
    ) -> str:
        """Assemble the deterministic turn prompt with bounded advisory hints."""
        parts = [f"QUESTION: {question}", "", "SUBSTRATE (MemDSL render):"]
        parts.append(serialize_envelope(state.envelope) if state.envelope else "")
        if state.voi:
            ranked = sorted(
                (
                    (ref, annotation)
                    for ref, annotation in state.voi.items()
                    if annotation.voi > 0
                ),
                key=lambda item: (-item[1].voi, item[1].cost, item[0]),
            )[: max(0, self.config.focus_handle_limit)]
            if ranked:
                parts.append(
                    "FOCUS HANDLES (advisory only; choose every action yourself):"
                )
            for ref, ann in ranked:
                parts.append(f"  {ref} voi={ann.voi:.6f} cost={ann.cost}")
        parts.append(
            "BUDGET HINT (advisory; hard bounds remain engine-owned): "
            f"tokens_remaining={max(0, self.config.token_budget - spent)} "
            f"hops_remaining={max(0, self.config.max_hops - hops)}"
        )
        if feedback:
            parts.extend(["", feedback])
        parts.extend([
            "",
            "Emit a MemDSL OUTPUT artifact: navigation actions then one terminal "
            "decision (@answer with a cited DAG, or @abstain <reason>).",
        ])
        return "\n".join(parts)

    # -- action execution -----------------------------------------------------

    def _execute_actions(self, state: _NavState, actions: Sequence[Action]) -> None:
        """Run the turn's non-terminal actions in the FIXED deterministic order."""
        ordered = sorted(
            enumerate(actions),
            key=lambda pair: (_ACTION_ORDER.get(pair[1].verb, 99), pair[0]),
        )
        for _idx, action in ordered:
            if action.verb == "expand":
                self._do_expand(state, action)
            elif action.verb == "join":
                self._do_join(state, action)
            elif action.verb == "ground":
                self._do_ground(state, action)
            elif action.verb == "prune":
                self._do_prune(state, action)
            # Unknown/terminal verbs never reach here (terminals are not actions).

    def _resolve(self, state: _NavState, handle: str) -> str:
        """Resolve an agent-supplied handle (an envelope ref or a token) to a token."""
        return state.ref_to_token.get(handle, handle)

    def _do_expand(self, state: _NavState, action: Action) -> None:
        """Execute ``expand(target)``: fetch + merge the target's neighborhood."""
        if not action.args:
            return
        target = self._resolve(state, action.args[0])
        state.expanded.add(target)
        try:
            nodes, edges = self.expander(state.neighborhood, target)
        except Exception:  # noqa: BLE001 - a bad expander never sinks the run
            state.trace.append(f"expand({target}): expander failed")
            return
        added = self._merge(state, nodes, edges)
        state.trace.append(f"expand({target}): +{added} node(s)")

    def _do_join(self, state: _NavState, action: Action) -> None:
        """Execute ``join(a, b, ...)``: record a weakest-link cross-hop binding.

        The LAST positional arg is the ``hop_to`` node the answer would rest on;
        the first is the bridge entity / ``hop_from``. The binding status is the
        weakest-link composition of the two endpoints (via
        :func:`~zettelkasten.synapse.epistemics.propagate`), and a within-view
        ``related`` edge is added so the join is visible on the next render.
        """
        if len(action.args) < 2:
            return
        hop_from = self._resolve(state, action.args[0])
        hop_to = self._resolve(state, action.args[-1])
        va = state.node_status.get(hop_from, ep.INFERRED)
        vb = state.node_status.get(hop_to, ep.INFERRED)
        binding_status = ep.propagate(InferenceForm.DEDUCTION, [va, vb])
        state.bindings.append(
            BridgeBinding(entity=hop_from, hop_from=hop_from, hop_to=hop_to, status=binding_status)
        )
        if hop_from in state.neighborhood.nodes and hop_to in state.neighborhood.nodes:
            state.neighborhood.edges.append(Edge(src=hop_from, dst=hop_to, relation="related"))
        state.trace.append(
            f"join({hop_from}->{hop_to}): binding status={binding_status.wire().value}"
        )

    def _do_ground(self, state: _NavState, action: Action) -> None:
        """Execute ``ground(ref, verifier=..., ...)``: re-fetch + register a spec.

        Re-fetches the target node's full body (an elided body a later verbatim
        check needs) via the injected grounder, and registers a
        :class:`~zettelkasten.synapse.grounding_router.GroundingSpec` for the ref
        so the terminal ``route`` runs the verifier against it. The verifier type
        comes from the ``verifier`` kwarg (default ``quote``); the remaining
        kwargs plus the node body form the opaque payload. The spec is keyed by
        BOTH the token and the envelope ref, so the DAG may cite either.

        A fetch-back of a node ALREADY in view updates its body IN PLACE (never a
        duplicate node): when the seed was rendered ELIDED (empty/truncated body),
        replacing the stale body with the fetched FULL substrate text is what lets
        a faithful restatement of the fetched content clear the router's fail-closed
        restatement floor instead of being floored on an empty body.
        """
        if not action.args:
            return
        token = self._resolve(state, action.args[0])
        fetched = None
        try:
            fetched = self.grounder(token)
        except Exception:  # noqa: BLE001 - grounding fetch-back is best-effort
            fetched = None
        if fetched is not None:
            existing = state.neighborhood.nodes.get(token)
            if existing is None:
                # A genuinely new node the neighborhood had not seen yet: admit it.
                self._merge(state, [fetched], [])
            elif fetched.token == token and fetched.text and fetched.text != existing.text:
                # ELIDED-SEED FETCH-BACK: the seed was rendered with an empty/
                # truncated body, so ``_merge`` (which skips an already-present
                # token) would leave the body empty — and under the fail-closed
                # restatement floor an empty body licenses nothing, flooring a
                # faithful restatement of the fetched text to ABSTAIN. Update the
                # existing node's body IN PLACE (no duplicate node) with the fetched
                # FULL substrate text, so the router sees the real body: both the
                # ``node_bodies`` map and the quote payload below read ``node.text``.
                existing.text = fetched.text

        verifier_type = action.kwargs.get("verifier", "quote")
        node = state.neighborhood.nodes.get(token)
        payload: dict[str, Any] = {k: v for k, v in action.kwargs.items() if k != "verifier"}
        if node is not None and "body" not in payload:
            payload["body"] = node.text
        spec = GroundingSpec(verifier_type=verifier_type, payload=payload)
        state.groundings[token] = spec
        ref = state.token_to_ref.get(token)
        if ref is not None:
            state.groundings[ref] = spec
        state.trace.append(f"ground({token}): verifier={verifier_type}")

    def _do_prune(self, state: _NavState, action: Action) -> None:
        """Execute ``prune(...)`` via a decision-tree walk (the prune move).

        Runs the injected walk (default
        :func:`zettelkasten.decision_tree.walk`, imported lazily) with the
        action's kwargs, then removes every ``excluded`` node from the
        neighborhood — the walk's exclusion IS the prune. Matching is by the
        node's bare id against the walk's namespaced ``graph::id`` uids and any
        bare ids, so a pruned subtree drops out of the next render. Best-effort:
        a walk failure prunes nothing rather than sinking the run.
        """
        walk_fn = self.walk_fn
        if walk_fn is None:
            from zettelkasten import decision_tree

            walk_fn = decision_tree.walk
        try:
            bundle = walk_fn(**action.kwargs) or {}
        except Exception:  # noqa: BLE001 - a bad walk prunes nothing
            state.trace.append("prune: walk failed")
            return
        excluded_ids: set[str] = set()
        for uid in bundle.get("excluded", []) or []:
            excluded_ids.add(str(uid))
            excluded_ids.add(str(uid).split("::")[-1])
        removed = 0
        for token, node in list(state.neighborhood.nodes.items()):
            if node.id in excluded_ids or token in excluded_ids:
                del state.neighborhood.nodes[token]
                state.node_status.pop(token, None)
                state.pruned.add(token)
                removed += 1
        if removed:
            state.neighborhood.edges = [
                e for e in state.neighborhood.edges
                if e.src in state.neighborhood.nodes and e.dst in state.neighborhood.nodes
            ]
        state.trace.append(f"prune: -{removed} node(s)")

    def _merge(self, state: _NavState, nodes: Sequence[Node], edges: Sequence[Edge]) -> int:
        """Merge newly discovered nodes/edges, deriving statuses for new nodes.

        The declared ``config.max_nodes`` (when set) is a HARD ceiling on the
        neighborhood: once it is reached, further NEW nodes are dropped so
        LLM-driven ``expand``/``ground`` can never grow the substrate past the
        budget. The cap is checked only AFTER the already-present/pruned skip, so
        a ``ground`` re-fetch of a node already in view is never blocked (it is a
        no-op merge regardless). Incoming nodes are admitted in order, so the
        truncation is deterministic; edges to dropped nodes are dangling and are
        filtered out below.
        """
        added = 0
        new_nodes: list[Node] = []
        cap = self.config.max_nodes
        for node in nodes:
            # Guard the token: a candidate from any adapter (more likely a
            # less-controlled federated peer) whose id is address-invalid would
            # raise :class:`MemDSLParseError` on ``node.token`` and sink the whole
            # turn to a ``navigator-error`` abstain. Reuse the assembler's
            # ``_safe_token`` so such a node is SKIPPED (never merged, a debug log)
            # and the surviving nodes still merge — a bad-id node can never sink
            # the turn. Because a bad-id node is thus never admitted here (nor at
            # the guarded assembly boundary), every node reaching the render/route
            # ``.address.to_token()`` sites is already addressable.
            token = _safe_token(node)
            if token is None:
                continue
            if token in state.neighborhood.nodes or token in state.pruned:
                continue
            if cap is not None and len(state.neighborhood.nodes) >= cap:
                break
            state.neighborhood.nodes[token] = node
            new_nodes.append(node)
            added += 1
        if new_nodes:
            state.node_status.update(apply_node_statuses(new_nodes, self.resolver))
        for edge in edges:
            if edge.src in state.neighborhood.nodes and edge.dst in state.neighborhood.nodes:
                state.neighborhood.edges.append(edge)
        return added

    # -- routing (the verdict is always the router's) -------------------------

    def _route(self, state: _NavState, dag: ReasoningDAG) -> RouterDecision:
        """Delegate an ``@answer`` DAG to the grounding router for the verdict.

        Builds the node-status map the router consumes, keyed by BOTH the
        substrate token and the envelope ref so the DAG may cite either. A ref
        that was explicitly grounded (a ``ground`` spec) is left OUT of the
        node-status map so the router's verifier verdict stands (the router lets
        an explicit ``node_status`` override a verifier). Sequential multi-hop
        bindings then cap each ``hop_to`` node by its weakest-link binding — a
        later hop can be no more grounded than the binding it rests on.

        A ``node_bodies`` map of RAW, un-elided node text (from
        ``state.neighborhood.nodes[token].text``, NOT the wrapped/elided envelope
        body) is also assembled — keyed by BOTH the substrate token and the envelope
        ref like ``status_map`` — and handed to the router so its restatement floor
        can verify each ``[restatement]`` claim's text against the body it cites.
        """
        grounded_keys = set(state.groundings)
        status_map: dict[str, StatusVerdict] = {}
        node_bodies: dict[str, str] = {}
        for node in (state.envelope.nodes if state.envelope else []):
            token = node.address.to_token()
            verdict = state.node_status.get(token, ep.INFERRED)
            for key in (token, node.ref):
                if key not in grounded_keys:
                    status_map[key] = verdict
            # The restatement floor always needs the RAW body — even for an
            # explicitly-grounded ref (the quote verifier checks a payload, not the
            # claim text) — so node_bodies is keyed for every rendered node.
            raw = state.neighborhood.nodes.get(token)
            if raw is not None:
                for key in (token, node.ref):
                    node_bodies[key] = raw.text

        for binding in state.bindings:
            for key in (binding.hop_to, state.token_to_ref.get(binding.hop_to, "")):
                if not key or key in grounded_keys:
                    continue
                current = status_map.get(key)
                status_map[key] = (
                    ep.propagate(InferenceForm.DEDUCTION, [current, binding.status])
                    if current is not None
                    else binding.status
                )

        return gr.route(
            dag,
            prose=dag.prose,
            node_bodies=node_bodies,
            node_status=status_map,
            groundings=state.groundings,
            registry=self.registry,
            edges=state.neighborhood.edges,
            trace_floor=self.config.trace_floor,
        )

    def _low_coverage_feedback(self, dag: ReasoningDAG, decision: RouterDecision) -> str:
        """Build a low-coverage RE-NAVIGATE hint naming the ungrounded cited nodes.

        Walks the conclusion's transitive premise closure and reports every cited
        node whose re-derived verdict is NOT verified (inferred) or is stale — read
        from ``decision.node_verdicts`` — so the agent knows exactly which grounds
        floored the conclusion. It is instructed to rebuild the conclusion (and its
        premise chain) resting ONLY on nodes rendered ``grounded`` / ``measured``, or
        to ``@abstain`` honestly. Mirrors the false-premise re-navigation hint.
        """
        by_id = {c.id: c for c in dag.claims}
        conclusion = decision.conclusion or (dag.claims[-1].id if dag.claims else "")
        closure: set[str] = set()
        stack = [conclusion]
        while stack:
            cid = stack.pop()
            if cid in closure or cid not in by_id:
                continue
            closure.add(cid)
            stack.extend(by_id[cid].premises)

        offending: set[str] = set()
        for cid in closure:
            for ref in by_id[cid].cites:
                verdict = decision.node_verdicts.get(ref)
                if verdict is None:
                    continue
                if verdict.wire() not in _VERIFIED_WIRE or verdict.stale:
                    offending.add(ref)

        if offending:
            named = ", ".join(sorted(offending))
            where = f"node(s) rendered inferred/stale ({named})"
        else:
            where = "node(s) not rendered grounded/measured"
        return (
            f"RE-NAVIGATE: your conclusion rested on {where}, which cannot ground a "
            "verified answer. Rebuild your conclusion and its premise chain citing "
            "ONLY nodes rendered 'grounded' or 'measured', or @abstain honestly."
        )

    # -- terminal helpers -----------------------------------------------------

    def _structural_abstain(self, detail: str, *, reason: str = REASON_LOW_COVERAGE) -> RouterDecision:
        """A structurally-forced ABSTAIN (parse failure or an explicit agent abstain).

        Used only when there is no DAG for the router to adjudicate; a broken
        ``@answer`` path still goes through the real router (see :meth:`_route`).
        """
        return RouterDecision(
            verdict=ABSTAIN,
            status=None,
            reasons=(reason,),
            abstention=Abstention(reason=reason, detail=detail),
            detail=detail,
        )

    def _fallback_abstain(self, state: _NavState, detail: str) -> RouterDecision:
        """The VOI-max loop-break fallback: abstain, noting the top-VOI handle.

        When a budget/hop guardrail breaks a non-terminating loop there is no
        grounded DAG to answer from, so the deterministic fallback is to ABSTAIN.
        The highest-VOI handle (the move the agent should have spent budget on) is
        recorded in the detail for inspection — the VOI-max tie-break is by ref so
        it is deterministic.
        """
        top = ""
        if state.voi:
            top = max(sorted(state.voi), key=lambda ref: state.voi[ref].voi)
            detail = f"{detail}; VOI-max fallback handle={top}"
        return self._structural_abstain(detail)

    def _finish(
        self,
        decision: RouterDecision,
        halted_reason: str,
        state: _NavState,
        output: AgentOutput | None,
        hops: int,
        spent: int,
        renav: int,
        parse_repairs: int,
        *,
        loop_break: str = "",
    ) -> NavResult:
        """Assemble the :class:`NavResult` from the final decision + run telemetry."""
        return NavResult(
            decision=decision,
            halted_reason=halted_reason,
            output=output,
            envelope=state.envelope,
            hops=hops,
            tokens_spent=spent,
            renavigations=renav,
            parse_repairs=parse_repairs,
            loop_break=loop_break,
            voi=dict(state.voi),
            bindings=list(state.bindings),
            trace=list(state.trace),
        )

navigate

navigate(question: str, neighborhood: Neighborhood, *, groundings: Mapping[str, GroundingSpec] | None = None) -> NavResult

Run the navigation loop over neighborhood to a verdict.

question is the user's query; neighborhood is the assembled substrate the run starts from (further expand moves may grow it). groundings optionally seeds grounding specs (the ground action adds more). Returns a :class:NavResult whose decision is the router's ANSWER/ABSTAIN verdict (or a structurally-forced ABSTAIN on a budget/parse failure).

Source code in zettelkasten/synapse/navigation.py

def navigate(
    self,
    question: str,
    neighborhood: Neighborhood,
    *,
    groundings: Mapping[str, GroundingSpec] | None = None,
) -> NavResult:
    """Run the navigation loop over ``neighborhood`` to a verdict.

    ``question`` is the user's query; ``neighborhood`` is the assembled
    substrate the run starts from (further ``expand`` moves may grow it).
    ``groundings`` optionally seeds grounding specs (the ``ground`` action
    adds more). Returns a :class:`NavResult` whose ``decision`` is the
    router's ANSWER/ABSTAIN verdict (or a structurally-forced ABSTAIN on a
    budget/parse failure).
    """
    state = _NavState(
        neighborhood=neighborhood,
        node_status=apply_node_statuses(neighborhood.node_list(), self.resolver),
        groundings=dict(groundings or {}),
    )

    spent = 0
    hops = 0
    renav = 0
    parse_repairs = 0
    feedback = ""
    last_output: AgentOutput | None = None

    while True:
        if hops >= self.config.max_hops:
            state.trace.append(f"halt: hop cap {self.config.max_hops} reached")
            return self._finish(
                self._fallback_abstain(state, "hop cap reached before an answer"),
                "hops", state, last_output, hops, spent, renav, parse_repairs,
                loop_break="voi-max",
            )

        self._render(state, question)
        prompt = self._build_prompt(
            question,
            state,
            feedback,
            spent=spent,
            hops=hops,
        )

        # HARD budget guardrail: check the next turn's cost BEFORE the call,
        # so an over-long loop is halted deterministically. VOI-max fallback
        # then breaks the loop rather than looping forever.
        est = _estimate_tokens(prompt) + _estimate_tokens(self.system)
        if spent + est > self.config.token_budget:
            state.trace.append(
                f"halt: token budget {self.config.token_budget} would be exceeded "
                f"(spent={spent}, next~{est})"
            )
            return self._finish(
                self._fallback_abstain(state, "token budget exhausted"),
                "budget", state, last_output, hops, spent, renav, parse_repairs,
                loop_break="voi-max",
            )

        raw = self.llm(self.system, prompt)
        spent += est + _estimate_tokens(raw)

        try:
            output = parse_output(raw)
        except MemDSLParseError as exc:
            # Log the raw pre-parse text so a near-miss the parser could not
            # recover is diagnosable later. DEBUG + logger (never stdout): the
            # RESULT line lives on stdout and must stay uncontaminated.
            logger.debug(
                "MemDSL OUTPUT parse-fail (repair %d/%d): %s\nraw pre-parse text:\n%s",
                parse_repairs,
                self.config.max_parse_repairs,
                exc,
                raw,
            )
            if parse_repairs < self.config.max_parse_repairs:
                parse_repairs += 1
                # Embed BOTH terminal exemplars on repair turns so the repair
                # feedback is EVEN-HANDED: a lone @answer exemplar biases a turn
                # that SHOULD abstain toward emitting an @answer (which the
                # grounder then catches, but at the cost of a wasted repair turn
                # and a nudge the wrong way). Showing the @abstain form alongside
                # it makes clear that abstaining is a first-class, equally-valid
                # OUTPUT. One concise exemplar of each keeps the message bounded
                # against the per-turn token floor the budget guardrail counts.
                feedback = (
                    f"PARSE ERROR: {exc}. Re-emit a VALID MemDSL OUTPUT artifact "
                    "(correct header, well-formed actions, one terminal decision — "
                    "an @answer with a cited DAG, OR an @abstain <reason> :: <detail>; "
                    "abstaining is an equally-valid answer when nothing grounds it). "
                    "Common mistakes to avoid: do NOT wrap the output in markdown "
                    "code fences (no ``` / ```memdsl); emit NOTHING before the "
                    "'@memdsl.out/2' header or after the terminal decision (no preamble, "
                    "no sign-off); use straight ASCII quotes (\") in structural "
                    "tokens, not curly quotes. Match one of these exact forms:\n"
                    f"{NAV_EXEMPLAR_ANSWER}"
                    "\nor, to abstain:\n"
                    f"{NAV_EXEMPLAR_ABSTAIN}"
                )
                state.trace.append(f"parse-fail: repair retry {parse_repairs}")
                continue
            state.trace.append("parse-fail: repair cap reached; abstaining")
            return self._finish(
                self._structural_abstain("unparseable agent output after repair retry"),
                "parse-fail", state, last_output, hops, spent, renav, parse_repairs,
            )

        feedback = ""
        last_output = output
        hops += 1
        self._execute_actions(state, output.actions)

        # Terminal: an explicit agent abstain is honored as an abstain.
        if output.abstain is not None:
            state.trace.append(f"terminal: agent @abstain ({output.abstain.reason})")
            return self._finish(
                self._structural_abstain(
                    output.abstain.detail or "agent abstained", reason=output.abstain.reason
                ),
                "abstain", state, last_output, hops, spent, renav, parse_repairs,
            )

        # Terminal: an @answer's verdict is ALWAYS delegated to the router.
        if output.reasoning is not None:
            decision = self._route(state, output.reasoning)
            if decision.verdict == ANSWER:
                state.trace.append("terminal: router ANSWER")
                return self._finish(
                    decision, "answer", state, last_output, hops, spent, renav, parse_repairs,
                )
            # Bounded self-correction: a verifier refutation triggers ONE
            # re-navigate (pick a different path), never a double-down.
            if REASON_FALSE_PREMISE in decision.reasons and renav < self.config.max_renavigations:
                renav += 1
                feedback = (
                    f"RE-NAVIGATE: a verifier refuted a ground ({decision.detail}). "
                    "Do not repeat that claim; navigate to a different, grounded path "
                    "or @abstain."
                )
                state.trace.append(f"self-correct: re-navigate {renav} after false-premise")
                continue
            # Bounded self-correction: a low-coverage abstain triggers ONE
            # re-navigate that NAMES the ungrounded nodes the conclusion rested
            # on, so the agent can rebuild on grounded ground or abstain honestly.
            if REASON_LOW_COVERAGE in decision.reasons and renav < self.config.max_renavigations:
                renav += 1
                feedback = self._low_coverage_feedback(output.reasoning, decision)
                state.trace.append(f"self-correct: re-navigate {renav} after low-coverage")
                continue
            state.trace.append(f"terminal: router ABSTAIN ({', '.join(decision.reasons)})")
            return self._finish(
                decision, "abstain", state, last_output, hops, spent, renav, parse_repairs,
            )

        # No terminal this turn — only navigation actions. Loop again; the
        # budget/hop guardrail will eventually break a non-terminating loop.
        state.trace.append(f"hop {hops}: navigation actions, no terminal")

config_path

config_path() -> Path

Absolute path to .synapse/config.yaml.

Source code in zettelkasten/synapse/config.py

def config_path() -> Path:
    """Absolute path to ``.synapse/config.yaml``."""
    return synapse_dir() / "config.yaml"

intersecting_projects

intersecting_projects(config: dict[str, Any] | None = None) -> list[str]

Return the ordered, de-duplicated list of intersecting LOCAL project names.

Backward-compatible view over :func:intersecting_scopes that keeps only local (non-federated) entries — used where a plain local project name is expected (e.g. legacy callers). Federated scopes are surfaced via :func:intersecting_scopes.

Source code in zettelkasten/synapse/config.py

def intersecting_projects(config: dict[str, Any] | None = None) -> list[str]:
    """Return the ordered, de-duplicated list of intersecting LOCAL project names.

    Backward-compatible view over :func:`intersecting_scopes` that keeps only
    local (non-federated) entries — used where a plain local project name is
    expected (e.g. legacy callers). Federated scopes are surfaced via
    :func:`intersecting_scopes`.
    """
    return [s.project for s in intersecting_scopes(config) if s.repo is None]

intersecting_scopes

intersecting_scopes(config: dict[str, Any] | None = None) -> list[IntersectSpec]

Return the ordered, de-duplicated list of intersecting scopes.

Reads .synapse/config.yaml when config is not supplied. Malformed or blank entries are dropped (best-effort, never raises) so a partial config still yields the valid subset. De-dup is keyed on (repo, project) so the same project can intersect from both a local and a federated source.

Source code in zettelkasten/synapse/config.py

def intersecting_scopes(config: dict[str, Any] | None = None) -> list[IntersectSpec]:
    """Return the ordered, de-duplicated list of intersecting scopes.

    Reads ``.synapse/config.yaml`` when ``config`` is not supplied. Malformed or
    blank entries are dropped (best-effort, never raises) so a partial config
    still yields the valid subset. De-dup is keyed on ``(repo, project)`` so the
    same project can intersect from both a local and a federated source.
    """
    if config is None:
        config = read_config()

    raw = config.get(CONFIG_KEY, [])
    if raw in (None, ""):
        return []
    if not isinstance(raw, list):
        logger.warning("synapse config %r must be a list, got %s", CONFIG_KEY, type(raw).__name__)
        return []

    scopes: list[IntersectSpec] = []
    seen: set[tuple[str | None, str]] = set()
    for item in raw:
        spec = _coerce_spec(item)
        if spec is None:
            continue
        key = (spec.repo, spec.project)
        if key in seen:
            continue
        seen.add(key)
        scopes.append(spec)
    return scopes

parse_scope_token

parse_scope_token(token: Any) -> IntersectSpec | None

Parse one override token (from synapse(projects=[...])) into a spec.

A <repo_id>:<project> token (the namespace separator : can never appear in a local project name) resolves to a FEDERATED spec; any other non-blank string is a local project. Returns None for blank/invalid input.

Source code in zettelkasten/synapse/config.py

def parse_scope_token(token: Any) -> IntersectSpec | None:
    """Parse one override token (from ``synapse(projects=[...])``) into a spec.

    A ``<repo_id>:<project>`` token (the namespace separator ``:`` can never
    appear in a local project name) resolves to a FEDERATED spec; any other
    non-blank string is a local project. Returns None for blank/invalid input.
    """
    if not isinstance(token, str):
        return None
    token = token.strip()
    if not token:
        return None
    from zettelkasten import federation

    split = federation.split_namespace(token)
    if split is not None:
        repo_id, project = split
        return IntersectSpec(project=project, repo=repo_id)
    return IntersectSpec(project=token)

read_config

read_config() -> dict[str, Any]

Read the synapse config, returning {} when missing or malformed.

Source code in zettelkasten/synapse/config.py

def read_config() -> dict[str, Any]:
    """Read the synapse config, returning ``{}`` when missing or malformed."""
    path = config_path()
    if not path.exists():
        return {}
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError) as exc:
        logger.warning("Could not read synapse config %s: %s", path, exc)
        return {}
    return data if isinstance(data, dict) else {}

synapse_dir

synapse_dir() -> Path

Absolute path to the committed .synapse/ directory.

Source code in zettelkasten/synapse/config.py

def synapse_dir() -> Path:
    """Absolute path to the committed ``.synapse/`` directory."""
    return _anchor(Path(os.environ.get("SYNAPSE_DIR", ".synapse")))

write_config

write_config(config: dict[str, Any]) -> Path

Write the synapse config atomically (temp file + rename).

Source code in zettelkasten/synapse/config.py

def write_config(config: dict[str, Any]) -> Path:
    """Write the synapse config atomically (temp file + rename)."""
    path = config_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".yaml.tmp")
    tmp.write_text(
        yaml.dump(config, default_flow_style=False, allow_unicode=True, sort_keys=False),
        encoding="utf-8",
    )
    tmp.replace(path)
    return path

get_memory_source

get_memory_source(force_refresh: bool = False) -> MemorySourceGraph

Return a cached :class:MemorySourceGraph, rebuilt when .memory/ changes.

The cache is keyed on storage.source_fileset_signature() so an edit, add, delete, or rename of any memory source file transparently rebuilds the .notes map and drops the stale semantic index.

Source code in zettelkasten/synapse/memory_source.py

def get_memory_source(force_refresh: bool = False) -> MemorySourceGraph:
    """Return a cached :class:`MemorySourceGraph`, rebuilt when ``.memory/`` changes.

    The cache is keyed on ``storage.source_fileset_signature()`` so an edit,
    add, delete, or rename of any memory source file transparently rebuilds the
    ``.notes`` map and drops the stale semantic index.
    """
    global _CACHED_SOURCE, _CACHED_SIG
    sig = storage.source_fileset_signature()
    with _CACHE_LOCK:
        if not force_refresh and _CACHED_SOURCE is not None and _CACHED_SIG == sig:
            return _CACHED_SOURCE
        entries = _read_entries()
        _CACHED_SOURCE = MemorySourceGraph(entries, embedder=_shared_embedder())
        _CACHED_SIG = sig
        return _CACHED_SOURCE

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

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

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)

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

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)]

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),
        },
    }

candidates_for

candidates_for(node_id: str, zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None, per_node_k: int = 8, sim_threshold: float = 0.45, canon_types: 'tuple[str, ...] | None' = None, practice_types: 'tuple[str, ...] | None' = None) -> list[Candidate]

Lazy single-node candidate generation (either a memory id or a ZK note id).

Used by the on-demand connection path: given one node, find its cross-store neighbours in the other store via semantic NN + shared provenance. canon_types/practice_types optionally restrict which note types are accepted on the canon (ZK) and practice (memory) endpoints — None (the default) accepts all, preserving the original behaviour.

Source code in zettelkasten/synapse/candidates.py

def candidates_for(
    node_id: str,
    zk_get_graph: GetGraph,
    projects: list[str] | None = None,
    graphs_dir: "Path | None" = None,
    per_node_k: int = 8,
    sim_threshold: float = 0.45,
    canon_types: "tuple[str, ...] | None" = None,
    practice_types: "tuple[str, ...] | None" = None,
) -> list[Candidate]:
    """Lazy single-node candidate generation (either a memory id or a ZK note id).

    Used by the on-demand connection path: given one node, find its cross-store
    neighbours in the other store via semantic NN + shared provenance.
    ``canon_types``/``practice_types`` optionally restrict which note types are
    accepted on the canon (ZK) and practice (memory) endpoints — ``None`` (the
    default) accepts all, preserving the original behaviour.
    """
    sources, get_graph = resolve_scope(zk_get_graph, projects=projects, graphs_dir=graphs_dir)
    mem = get_graph(MEMORY_SOURCE)
    zk_sources = [s for s in sources if s != MEMORY_SOURCE]
    out: dict[tuple[str, str], Candidate] = {}

    if node_id in mem.notes:
        # Memory node → find ZK neighbours.
        mnote = mem.notes[node_id]
        if not _type_in(mnote, practice_types):
            return []
        text = _node_text(mnote)
        mprov = _memory_provenance(mnote)
        for source in zk_sources:
            try:
                zg = get_graph(source)
            except Exception:
                continue
            emb = getattr(zg, "embeddings", None)
            sem = emb.search(text, top_k=per_node_k) if emb is not None else []
            for zid, sim in sem:
                if sim < sim_threshold or zid not in getattr(zg, "notes", {}):
                    continue
                znote = zg.notes[zid]
                if not _type_in(znote, canon_types):
                    continue
                out[(node_id, zid)] = Candidate(node_id, zid, source, semantic_sim=sim,
                                                prefer_bonus=_prefer_bonus(source, znote))
            if mprov:
                for zid, znote in getattr(zg, "notes", {}).items():
                    if not _type_in(znote, canon_types):
                        continue
                    shared = sorted(mprov & _zk_provenance(znote))
                    if shared:
                        c = out.get((node_id, zid)) or Candidate(
                            node_id, zid, source, prefer_bonus=_prefer_bonus(source, znote))
                        c.shared_provenance = shared
                        out[(node_id, zid)] = c
        return sorted(out.values(), key=lambda c: c.score(), reverse=True)

    # Otherwise treat node_id as a ZK note id: find memory neighbours.
    prov_index = _memory_provenance_index(mem)
    for source in zk_sources:
        try:
            zg = get_graph(source)
        except Exception:
            continue
        znote = getattr(zg, "notes", {}).get(node_id)
        if znote is None:
            continue
        if not _type_in(znote, canon_types):
            continue
        bonus = _prefer_bonus(source, znote)
        if mem.embeddings is not None:
            for mid, sim in mem.embeddings.search(_node_text(znote), top_k=per_node_k):
                if sim < sim_threshold or mid not in mem.notes:
                    continue
                if not _type_in(mem.notes[mid], practice_types):
                    continue
                out[(mid, node_id)] = Candidate(mid, node_id, source, semantic_sim=sim,
                                                prefer_bonus=bonus)
        for tok in _zk_provenance(znote):
            for mid in prov_index.get(tok, ()):  # type: ignore[union-attr]
                if not _type_in(mem.notes.get(mid), practice_types):
                    continue
                c = out.get((mid, node_id)) or Candidate(mid, node_id, source, prefer_bonus=bonus)
                if tok not in c.shared_provenance:
                    c.shared_provenance.append(tok)
                out[(mid, node_id)] = c
    return sorted(out.values(), key=lambda c: c.score(), reverse=True)

generate_candidates

generate_candidates(zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None, per_node_k: int = 5, sim_threshold: float = 0.45, limit: int | None = None, canon_types: 'tuple[str, ...] | None' = None, practice_types: 'tuple[str, ...] | None' = None, overfetch: int = 1) -> list[Candidate]

Generate cross-store candidate pairs across the intersection scope.

ZK-anchored: each ZK note is used as a query into memory's vector space (bounded by per_node_k neighbours per note, sim_threshold floor), and the shared-provenance index is intersected. limit caps the number of canon-endpoint (candidate) notes scanned for lazy/partial builds — with a canon_types filter only canon-passing notes count against it, so claims beyond the first limit non-canon notes stay reachable.

The claim-aligned build restricts the two registers with canon_types (only these ZK note types become canon endpoints — e.g. claim/finding) and practice_types (only these memory entry types become practice endpoints — e.g. decision/experiment/checkpoint, which keeps a large annotation/note-heavy tree from exploding the candidate pool). Because that type filter is applied AFTER semantic recall, cross-register recall would otherwise bleed — so overfetch (>=1) widens the per-note neighbour fetch to per_node_k * overfetch and then keeps the first per_node_k that pass the practice filter. _cross synthesis-claim endpoints get a small score prior so they are preferred over single-source findings covering a pair. The shared-provenance channel is register-independent and always kept. canon_types/practice_types None (the default) preserves the original, unfiltered note-note behaviour exactly.

Source code in zettelkasten/synapse/candidates.py

def generate_candidates(
    zk_get_graph: GetGraph,
    projects: list[str] | None = None,
    graphs_dir: "Path | None" = None,
    per_node_k: int = 5,
    sim_threshold: float = 0.45,
    limit: int | None = None,
    canon_types: "tuple[str, ...] | None" = None,
    practice_types: "tuple[str, ...] | None" = None,
    overfetch: int = 1,
) -> list[Candidate]:
    """Generate cross-store candidate pairs across the intersection scope.

    ZK-anchored: each ZK note is used as a query into memory's vector space
    (bounded by ``per_node_k`` neighbours per note, ``sim_threshold`` floor), and
    the shared-provenance index is intersected. ``limit`` caps the number of
    canon-endpoint (candidate) notes scanned for lazy/partial builds — with a
    ``canon_types`` filter only canon-passing notes count against it, so claims
    beyond the first ``limit`` non-canon notes stay reachable.

    The claim-aligned build restricts the two registers with ``canon_types``
    (only these ZK note types become canon endpoints — e.g. ``claim``/``finding``)
    and ``practice_types`` (only these memory entry types become practice
    endpoints — e.g. ``decision``/``experiment``/``checkpoint``, which keeps a
    large annotation/note-heavy tree from exploding the candidate pool). Because
    that type filter is applied AFTER semantic recall, cross-register recall would
    otherwise bleed — so ``overfetch`` (>=1) widens the per-note neighbour fetch
    to ``per_node_k * overfetch`` and then keeps the first ``per_node_k`` that pass
    the practice filter. ``_cross`` synthesis-claim endpoints get a small score
    prior so they are preferred over single-source findings covering a pair. The
    shared-provenance channel is register-independent and always kept.
    ``canon_types``/``practice_types`` ``None`` (the default) preserves the
    original, unfiltered note-note behaviour exactly.
    """
    sources, get_graph = resolve_scope(zk_get_graph, projects=projects, graphs_dir=graphs_dir)
    mem = get_graph(MEMORY_SOURCE)
    prov_index = _memory_provenance_index(mem)
    zk_sources = [s for s in sources if s != MEMORY_SOURCE]
    fetch_k = per_node_k * max(1, overfetch)

    cands: dict[tuple[str, str], Candidate] = {}

    def _get(mid: str, zid: str, source: str, znote: Any) -> Candidate:
        key = (mid, zid)
        c = cands.get(key)
        if c is None:
            c = Candidate(memory_id=mid, zk_id=zid, zk_source=source,
                          prefer_bonus=_prefer_bonus(source, znote))
            cands[key] = c
        return c

    scanned = 0
    for source in zk_sources:
        try:
            zg = get_graph(source)
        except Exception:
            continue
        for zid, znote in getattr(zg, "notes", {}).items():
            if limit is not None and scanned >= limit:
                break
            # Canon-type filter: a ZK note that is not an in-scope canon type is
            # never a candidate endpoint. It is NOT counted toward ``limit`` — only
            # canon-passing notes consume the scan budget, so a partial/lazy build
            # can still reach claim/finding notes that sit beyond the first ``limit``
            # non-canon notes (which would otherwise exhaust the budget and yield an
            # empty candidate set even though in-scope claims exist).
            if not _type_in(znote, canon_types):
                continue
            scanned += 1

            # Semantic channel (overfetch, then keep the first per_node_k that
            # pass the practice-type filter).
            if mem.embeddings is not None:
                kept = 0
                for mid, sim in mem.embeddings.search(_node_text(znote), top_k=fetch_k):
                    if sim < sim_threshold or mid not in mem.notes:
                        continue
                    if not _type_in(mem.notes[mid], practice_types):
                        continue
                    c = _get(mid, zid, source, znote)
                    c.semantic_sim = max(c.semantic_sim, sim)
                    kept += 1
                    if kept >= per_node_k:
                        break

            # Provenance channel (register-independent; still practice-filtered).
            if prov_index:
                for tok in _zk_provenance(znote):
                    for mid in prov_index.get(tok, ()):  # type: ignore[union-attr]
                        if not _type_in(mem.notes.get(mid), practice_types):
                            continue
                        c = _get(mid, zid, source, znote)
                        if tok not in c.shared_provenance:
                            c.shared_provenance.append(tok)
        if limit is not None and scanned >= limit:
            break

    return sorted(cands.values(), key=lambda c: c.score(), reverse=True)

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

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

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

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

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)

build_matrix_lens

build_matrix_lens(zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None, overlay: dict[str, Any] | None = None, min_confidence: float = 0.0, max_rows: int = _MAX_AXIS, max_columns: int = _MAX_AXIS) -> dict[str, Any]

Project the connection overlay into a practice x canon matrix (read-only).

Rows = memory nodes, columns = ZK nodes, cells = typed edges. Titles are resolved best-effort via the stores. Axes are capped (busiest nodes first) with a truncation flag.

Source code in zettelkasten/synapse/matrix.py

def build_matrix_lens(
    zk_get_graph: GetGraph,
    projects: list[str] | None = None,
    graphs_dir: "Path | None" = None,
    overlay: dict[str, Any] | None = None,
    min_confidence: float = 0.0,
    max_rows: int = _MAX_AXIS,
    max_columns: int = _MAX_AXIS,
) -> dict[str, Any]:
    """Project the connection overlay into a practice x canon matrix (read-only).

    Rows = memory nodes, columns = ZK nodes, cells = typed edges. Titles are
    resolved best-effort via the stores. Axes are capped (busiest nodes first)
    with a truncation flag.
    """
    from zettelkasten.synapse import overlay as _overlay

    if overlay is None:
        overlay = _overlay.load_overlay()
    edges = [e for e in overlay.get("edges", []) if float(e.get("confidence", 0.0)) >= min_confidence]

    _sources, get_graph = resolve_scope(zk_get_graph, projects=projects, graphs_dir=graphs_dir)

    def _memory_title(mid: str) -> tuple[str, str]:
        try:
            note = get_graph(MEMORY_SOURCE).notes.get(mid)
        except Exception:
            note = None
        return (getattr(note, "title", "") if note else ""), (getattr(note, "type", "") if note else "")

    def _zk_title(zid: str, source: str) -> str:
        try:
            note = get_graph(source).notes.get(zid)
        except Exception:
            note = None
        return getattr(note, "title", "") if note else ""

    # Degree-order the axes so a cap keeps the most-connected nodes.
    row_deg: dict[str, int] = {}
    col_deg: dict[str, int] = {}
    zk_source_of: dict[str, str] = {}
    for e in edges:
        row_deg[e["memory_id"]] = row_deg.get(e["memory_id"], 0) + 1
        col_deg[e["zk_id"]] = col_deg.get(e["zk_id"], 0) + 1
        zk_source_of.setdefault(e["zk_id"], e.get("zk_source", ""))

    row_ids = [rid for rid, _ in sorted(row_deg.items(), key=lambda kv: (-kv[1], kv[0]))]
    col_ids = [cid for cid, _ in sorted(col_deg.items(), key=lambda kv: (-kv[1], kv[0]))]
    rows_truncated = len(row_ids) > max_rows
    cols_truncated = len(col_ids) > max_columns
    row_ids = row_ids[:max_rows]
    col_ids = col_ids[:max_columns]
    row_set, col_set = set(row_ids), set(col_ids)

    cells = [
        {
            "memory_id": e["memory_id"],
            "zk_id": e["zk_id"],
            # Carry the canon source so a downstream consumer (e.g. the grounded
            # synthesis provider) can resolve the claim by its EXACT (zk_source,
            # zk_id) key — the same note id can exist in multiple in-scope graphs.
            "zk_source": e.get("zk_source", ""),
            "relation": e.get("relation"),
            "confidence": e.get("confidence"),
            "rationale": e.get("rationale", ""),
            # Present only on the claim-aligned overlay (additive; ``None`` on the
            # generic note-note overlay).
            "claim_strength": e.get("claim_strength"),
            "priority": e.get("priority"),
        }
        for e in edges
        if e["memory_id"] in row_set and e["zk_id"] in col_set
    ]

    relation_totals: dict[str, int] = {}
    for c in cells:
        relation_totals[c["relation"]] = relation_totals.get(c["relation"], 0) + 1

    rows = []
    for mid in row_ids:
        title, mtype = _memory_title(mid)
        rows.append({"id": mid, "store": "memory", "tier": "practice", "title": title, "type": mtype})
    columns = []
    for zid in col_ids:
        source = zk_source_of.get(zid, "")
        columns.append({"id": zid, "store": "zk", "tier": "canon",
                        "title": _zk_title(zid, source), "source": source})

    return {
        "rows": rows,
        "columns": columns,
        "cells": cells,
        "relation_totals": relation_totals,
        "row_count": len(rows),
        "column_count": len(columns),
        "edge_count": len(cells),
        "truncated": {"rows": rows_truncated, "columns": cols_truncated},
        "scope": {"projects": overlay.get("manifest", {}).get("projects", [])},
    }

synthesize_axes

synthesize_axes(lens: dict[str, Any], synth: AxisSynthesizer | None = None) -> dict[str, Any]

Add row/column summaries and a rolled-up apex synthesis to a matrix lens.

synth(kind, header, items) returns summary text; defaults to a tool-free LLM pass. Purely additive and read-only — the returned lens gains row_synthesis / column_synthesis / apex fields.

Source code in zettelkasten/synapse/matrix.py

def synthesize_axes(lens: dict[str, Any], synth: AxisSynthesizer | None = None) -> dict[str, Any]:
    """Add row/column summaries and a rolled-up apex synthesis to a matrix lens.

    ``synth(kind, header, items)`` returns summary text; defaults to a tool-free
    LLM pass. Purely additive and read-only — the returned lens gains
    ``row_synthesis`` / ``column_synthesis`` / ``apex`` fields.
    """
    synth = synth or _llm_synth

    row_summaries: dict[str, str] = {}
    for r in lens["rows"]:
        items = _row_items(lens, r["id"])
        if items:
            row_summaries[r["id"]] = synth("row", r["title"] or r["id"], items)

    column_summaries: dict[str, str] = {}
    for c in lens["columns"]:
        items = _col_items(lens, c["id"])
        if items:
            column_summaries[c["id"]] = synth("column", c["title"] or c["id"], items)

    apex_items = (
        [{"counterpart": (r["title"] or r["id"]), "relation": "practice", "rationale": row_summaries[r["id"]]}
         for r in lens["rows"] if r["id"] in row_summaries]
        + [{"counterpart": (c["title"] or c["id"]), "relation": "canon", "rationale": column_summaries[c["id"]]}
           for c in lens["columns"] if c["id"] in column_summaries]
    )
    apex = synth("apex", "cross-store synthesis", apex_items) if apex_items else ""

    lens = dict(lens)
    lens["row_synthesis"] = row_summaries
    lens["column_synthesis"] = column_summaries
    lens["apex"] = apex
    return lens

adapter_expander

adapter_expander(adapters: 'list[SourceAdapter]', budget: 'Budget | None' = None) -> 'Expander'

Build the navigator's :data:Expander seam over adapters.

The returned callable matches the navigation contract exactly — it is invoked as expander(neighborhood, target) where target is a node token, and it returns (neighbors, edges) for the navigator to merge. It locates the target node in the current neighborhood, dispatches to the owning adapter by the node's store + source, and returns that adapter's within-store one-hop expansion. An unknown target (not in the neighborhood, or no owning adapter) or a failing adapter yields ([], []) — a best-effort expansion never sinks a run.

budget is the DECLARED assembly budget (default a fresh :class:~zettelkasten.synapse.substrate.Budget): it is passed to adapter.expand and, crucially, bounds each expand's fan-out — a single call may reveal no more neighbors than the remaining room under budget.max_nodes when set, truncating deterministically in adapter order. None leaves fan-out count-uncapped; hop and navigator token bounds still terminate the run. The navigator's _merge enforces the SAME optional ceiling globally; this keeps the two growth paths consistent.

Source code in zettelkasten/synapse/assemble.py

def adapter_expander(
    adapters: "list[SourceAdapter]", budget: "Budget | None" = None
) -> "Expander":
    """Build the navigator's :data:`Expander` seam over ``adapters``.

    The returned callable matches the navigation contract exactly — it is invoked
    as ``expander(neighborhood, target)`` where ``target`` is a node token, and it
    returns ``(neighbors, edges)`` for the navigator to merge. It locates the
    target node in the current neighborhood, dispatches to the owning adapter by
    the node's ``store`` + ``source``, and returns that adapter's within-store
    one-hop expansion. An unknown target (not in the neighborhood, or no owning
    adapter) or a failing adapter yields ``([], [])`` — a best-effort expansion
    never sinks a run.

    ``budget`` is the DECLARED assembly budget (default a fresh
    :class:`~zettelkasten.synapse.substrate.Budget`): it is passed to
    ``adapter.expand`` and, crucially, bounds each expand's fan-out — a single
    call may reveal no more neighbors than the remaining room under
    ``budget.max_nodes`` when set, truncating deterministically in adapter order.
    ``None`` leaves fan-out count-uncapped; hop and navigator token bounds still
    terminate the run. The navigator's ``_merge`` enforces the SAME optional
    ceiling globally; this keeps the two growth paths consistent.
    """
    from zettelkasten.synapse.substrate import Budget, _safe_token

    budget = budget or Budget()

    def _expand(neighborhood, target):
        node = neighborhood.nodes.get(target)
        if node is None:
            return [], []
        adapter = _dispatch_adapter(adapters, node.store, node.source)
        if adapter is None:
            return [], []
        try:
            neighbors, edges = adapter.expand(node, budget)
        except Exception:  # noqa: BLE001 - best-effort: never crash the run
            return [], []
        # Bound the fan-out by the declared budget so one expand can never grow
        # the neighborhood past ``max_nodes``, then close the edge-loss class with
        # a single uniform rule (no per-branch early return). ``room`` is the
        # remaining node headroom: admit at most that many NEW neighbors in
        # deterministic adapter/link order, and admit NONE when saturated
        # (``room <= 0``) — but never early-return, because edges among
        # already-present nodes must survive even when no new neighbor fits.
        # Then filter edges ONCE against the retained set: keep an edge when its
        # ``dst`` is either a retained new neighbor OR a node already present in
        # the neighborhood (both endpoints exist ⇒ not dangling), and drop the
        # rest (edges to truncated new neighbors). ``src`` is the expanded node,
        # always present, so only ``dst`` is checked. On the non-truncation path
        # every ``dst`` is present-or-kept, so this filter is a harmless no-op.
        # ``_safe_token`` keeps the kept-set best-effort — an unaddressable
        # neighbor simply matches no edge and adds no raise path.
        room = (
            None
            if budget.max_nodes is None
            else budget.max_nodes - len(neighborhood.nodes)
        )
        if room is not None:
            neighbors = neighbors[:room] if room > 0 else []
        kept = {t for n in neighbors if (t := _safe_token(n)) is not None}
        edges = [e for e in edges if e.dst in kept or e.dst in neighborhood.nodes]
        return neighbors, edges

    return _expand

adapter_grounder

adapter_grounder(adapters: 'list[SourceAdapter]', registry: 'object | None' = None) -> 'Grounder'

Build the navigator's :data:Grounder seam over adapters.

The returned callable matches the navigation contract exactly — it is invoked as grounder(token) and returns the re-fetched :class:~zettelkasten. synapse.substrate.Node (with its full, un-elided body) or None when the token does not resolve. It parses the token to a MemDSL :class:~zettelkasten.synapse.memdsl.schema.Address, dispatches to the owning adapter by store + source, and returns adapter.get(address).

registry is accepted for API symmetry with the router's verifier registry (so a future grounder can consult it) but is unused by the v1 seam, whose sole job per the navigation contract is to re-fetch a node by token.

Source code in zettelkasten/synapse/assemble.py

def adapter_grounder(
    adapters: "list[SourceAdapter]", registry: "object | None" = None
) -> "Grounder":
    """Build the navigator's :data:`Grounder` seam over ``adapters``.

    The returned callable matches the navigation contract exactly — it is invoked
    as ``grounder(token)`` and returns the re-fetched :class:`~zettelkasten.
    synapse.substrate.Node` (with its full, un-elided body) or ``None`` when the
    token does not resolve. It parses the token to a MemDSL
    :class:`~zettelkasten.synapse.memdsl.schema.Address`, dispatches to the owning
    adapter by ``store`` + ``source``, and returns ``adapter.get(address)``.

    ``registry`` is accepted for API symmetry with the router's verifier registry
    (so a future grounder can consult it) but is unused by the v1 seam, whose sole
    job per the navigation contract is to re-fetch a node by token.
    """
    from zettelkasten.synapse.memdsl.schema import Address, MemDSLParseError

    def _ground(token):
        try:
            address = Address.parse(token)
        except MemDSLParseError:
            return None
        adapter = _dispatch_adapter(adapters, address.store, address.source)
        if adapter is None:
            return None
        try:
            return adapter.get(address)
        except Exception:  # noqa: BLE001 - grounding fetch-back is best-effort
            return None

    return _ground

build_adapters

build_adapters(projects: 'list[str] | None' = None, *, graphs_dir: 'object | None' = None, include_code: bool = False) -> 'list[SourceAdapter]'

Compose the store adapters for a cross-store scope (read-only).

Resolves the scope with :func:zettelkasten.synapse.retrieval.resolve_scope — the SAME entry point the synapse() read paths use — and builds one adapter per resolved source:

• a single :class:~zettelkasten.synapse.substrate.MemoryAdapter for the .memory/ research tree (always included; the _memory source that resolve_scope appends is skipped in the loop since this adapter already covers it);
• one :class:~zettelkasten.synapse.substrate.ZettelAdapter per resolved ZK box, whose lazy graph_provider routes through the combined grapher that resolve_scope returns (so a cached/rebuilt graph is picked up fresh);
• optionally a :class:~zettelkasten.synapse.substrate.CodeAdapter when include_code is set.

projects=None uses the intersection config; an explicit list overrides it ([] = memory only). graphs_dir optionally points the ZK grapher at a non-default store base (tests). A federated <repo_id>:<box> source IS included: its identity is encoded into an address-safe source segment (:func:_federated_source) so it is representable by the store:source:id address scheme, while its graph_provider still resolves the REAL federated graph through the combined grapher's <repo_id>:<box> routing.

Two soundness invariants are ENFORCED (not merely asserted) so no adapter can ever be built with an address-invalid or colliding source:

• Validate-and-skip. The FINAL resolved source (local or federated) is checked against the Address-forbidden character set (:func:_is_address_safe_source) right before the adapter is constructed; a failing source is omitted (continue) rather than emitted. This backstops federated identities carrying the reserved ~ delimiter or a forbidden char AND boundary-colon names (":box", "repo:") that :func:~zettelkasten.federation.split_namespace declines and that would otherwise reach Address verbatim and crash the read path.
• No collision with the reserved federation namespace. A LOCAL box whose name begins with the reserved ~fed~ marker is skipped, so a local source can never equal a federated encoded source (which always carries that marker). This keeps _dispatch_adapter from silently shadowing a federated node with a same-named local box.

Returns the adapters in deterministic scope order (memory first).

Source code in zettelkasten/synapse/assemble.py

def build_adapters(
    projects: "list[str] | None" = None,
    *,
    graphs_dir: "object | None" = None,
    include_code: bool = False,
) -> "list[SourceAdapter]":
    """Compose the store adapters for a cross-store scope (read-only).

    Resolves the scope with :func:`zettelkasten.synapse.retrieval.resolve_scope`
    — the SAME entry point the ``synapse()`` read paths use — and builds one
    adapter per resolved source:

    * a single :class:`~zettelkasten.synapse.substrate.MemoryAdapter` for the
      ``.memory/`` research tree (always included; the ``_memory`` source that
      ``resolve_scope`` appends is skipped in the loop since this adapter already
      covers it);
    * one :class:`~zettelkasten.synapse.substrate.ZettelAdapter` per resolved ZK
      box, whose lazy ``graph_provider`` routes through the combined grapher that
      ``resolve_scope`` returns (so a cached/rebuilt graph is picked up fresh);
    * optionally a :class:`~zettelkasten.synapse.substrate.CodeAdapter` when
      ``include_code`` is set.

    ``projects=None`` uses the intersection config; an explicit list overrides it
    (``[]`` = memory only). ``graphs_dir`` optionally points the ZK grapher at a
    non-default store base (tests). A federated ``<repo_id>:<box>`` source IS
    included: its identity is encoded into an address-safe ``source`` segment
    (:func:`_federated_source`) so it is representable by the ``store:source:id``
    address scheme, while its ``graph_provider`` still resolves the REAL federated
    graph through the combined grapher's ``<repo_id>:<box>`` routing.

    Two soundness invariants are ENFORCED (not merely asserted) so no adapter can
    ever be built with an address-invalid or colliding ``source``:

    * **Validate-and-skip.** The FINAL resolved ``source`` (local or federated) is
      checked against the Address-forbidden character set
      (:func:`_is_address_safe_source`) right before the adapter is constructed; a
      failing source is omitted (``continue``) rather than emitted. This backstops
      federated identities carrying the reserved ``~`` delimiter or a forbidden
      char AND boundary-colon names (``":box"``, ``"repo:"``) that
      :func:`~zettelkasten.federation.split_namespace` declines and that would
      otherwise reach ``Address`` verbatim and crash the read path.
    * **No collision with the reserved federation namespace.** A LOCAL box whose
      name begins with the reserved ``~fed~`` marker is skipped, so a local source
      can never equal a federated encoded source (which always carries that
      marker). This keeps ``_dispatch_adapter`` from silently shadowing a federated
      node with a same-named local box.

    Returns the adapters in deterministic scope order (memory first).
    """
    from zettelkasten import federation
    from zettelkasten.graph import ZettelGraph
    from zettelkasten.synapse import retrieval
    from zettelkasten.synapse.memory_source import MEMORY_SOURCE
    from zettelkasten.synapse.substrate import CodeAdapter, MemoryAdapter, ZettelAdapter

    # A lightweight, per-call cached local ZK grapher — one ZettelGraph per box,
    # loaded on first access. resolve_scope wraps it in combined_grapher (which
    # also routes ``_memory`` and federated names), returning the get_graph the
    # ZettelAdapters read through.
    cache: dict = {}

    def zk_get_graph(name: str):
        graph = cache.get(name)
        if graph is None:
            graph = ZettelGraph(name, graphs_dir=graphs_dir)
            graph.load()
            cache[name] = graph
        return graph

    sources, get_graph = retrieval.resolve_scope(
        zk_get_graph, projects=projects, graphs_dir=graphs_dir
    )

    adapters: "list[SourceAdapter]" = [MemoryAdapter()]
    for name in sources:
        # ``_memory`` is already covered by MemoryAdapter.
        if name == MEMORY_SOURCE:
            continue
        # A namespaced ``<repo_id>:<box>`` source is federated: encode its identity
        # into an address-safe ``source`` (see _federated_source), but resolve the
        # real graph through the ``<repo_id>:<box>`` name the combined grapher
        # routes. An unsound identity is validate-and-skipped, not emitted.
        # ``link_graph`` is the graph name the backing notes stamp on an
        # intra-graph ``link.graph``. It stays ``None`` (== ``source``) for a local
        # box; for a federated box it is the REAL local box name, since a peer's
        # ZettelGraph is loaded under that local name (retrieval._federated_graph)
        # and its notes carry it — NOT the encoded ``source`` — on ``link.graph``.
        link_graph: "str | None" = None
        parts = federation.split_namespace(name)
        if parts is not None:
            source = _federated_source(*parts)
            if source is None:
                logger.debug(
                    "build_adapters: skipping unsound federated identity %r "
                    "(_federated_source returned None)", name,
                )
                continue
            link_graph = parts[1]
        else:
            # A local box occupying the reserved ``~fed~`` namespace would collide
            # with a federated encoding and silently shadow it in dispatch, so it
            # is skipped rather than admitted.
            if name.startswith(_FED_PREFIX):
                logger.debug(
                    "build_adapters: skipping local box %r occupying the reserved "
                    "%r federation namespace", name, _FED_PREFIX,
                )
                continue
            source = name
        # Load-bearing backstop: never construct an adapter whose ``source`` would
        # crash the read path when stamped into an Address (e.g. a boundary-colon
        # name split_namespace declines and that falls through to the local branch).
        if not _is_address_safe_source(source):
            continue
        adapters.append(
            ZettelAdapter(source, lambda n=name: get_graph(n), link_graph=link_graph)
        )
    if include_code:
        adapters.append(CodeAdapter())
    return adapters

navigate_scope

navigate_scope(question: str, projects: 'list[str] | None' = None, *, llm: 'object | None' = None, budget: 'Budget | None' = None) -> 'NavResult'

Run the full grounded-navigation loop over a resolved scope.

A thin convenience that ties the pieces together: compose the store adapters for projects (:func:build_adapters), assemble the in-memory neighborhood for question (:func:~zettelkasten.synapse.substrate.assemble_neighborhood), and run a :class:~zettelkasten.synapse.navigation.Navigator wired with the adapter-backed :func:adapter_expander / :func:adapter_grounder seams.

llm defaults to :func:zettelkasten.synapse.navigation.default_llm (the production model seam); a test injects a scripted stub. budget bounds the neighborhood assembly (:class:~zettelkasten.synapse.substrate.Budget): its hops>0 turns on pre-loop breadth-first expansion and its max_nodes is threaded through as the navigator's hard in-loop ceiling.

Degrades GRACEFULLY: any assembly/model/navigation failure (a scope-resolve, kglite, embedding-build, or LLM error) is caught and returned as a labeled abstain :class:~zettelkasten.synapse.navigation.NavResult — never a raise — mirroring the synapse(action="navigate") handler. Returns the navigator's :class:~zettelkasten.synapse.navigation.NavResult.

Source code in zettelkasten/synapse/assemble.py

def navigate_scope(
    question: str,
    projects: "list[str] | None" = None,
    *,
    llm: "object | None" = None,
    budget: "Budget | None" = None,
) -> "NavResult":
    """Run the full grounded-navigation loop over a resolved scope.

    A thin convenience that ties the pieces together: compose the store adapters
    for ``projects`` (:func:`build_adapters`), assemble the in-memory neighborhood
    for ``question`` (:func:`~zettelkasten.synapse.substrate.assemble_neighborhood`),
    and run a :class:`~zettelkasten.synapse.navigation.Navigator` wired with the
    adapter-backed :func:`adapter_expander` / :func:`adapter_grounder` seams.

    ``llm`` defaults to :func:`zettelkasten.synapse.navigation.default_llm` (the
    production model seam); a test injects a scripted stub. ``budget`` bounds the
    neighborhood assembly (:class:`~zettelkasten.synapse.substrate.Budget`): its
    ``hops>0`` turns on pre-loop breadth-first expansion and its ``max_nodes`` is
    threaded through as the navigator's hard in-loop ceiling.

    Degrades GRACEFULLY: any assembly/model/navigation failure (a scope-resolve,
    kglite, embedding-build, or LLM error) is caught and returned as a labeled
    abstain :class:`~zettelkasten.synapse.navigation.NavResult` — never a raise —
    mirroring the ``synapse(action="navigate")`` handler.
    Returns the navigator's :class:`~zettelkasten.synapse.navigation.NavResult`.
    """
    from zettelkasten.synapse import navigation
    from zettelkasten.synapse.substrate import assemble_neighborhood

    try:
        adapters = build_adapters(projects)
        neighborhood = assemble_neighborhood(
            question, adapters, budget, expand=bool(budget and budget.hops > 0)
        )
        navigator = navigation.Navigator(
            llm or navigation.default_llm(),
            expander=adapter_expander(adapters, budget=budget),
            grounder=adapter_grounder(adapters),
            relevance=deterministic_relevance_map(neighborhood, adapters),
            config=(
                navigation.NavConfig(max_nodes=budget.max_nodes)
                if budget is not None
                else None
            ),
        )
        return navigator.navigate(question, neighborhood)
    except Exception as exc:  # noqa: BLE001 - degrade to a labeled abstain, never raise
        return _abstain_result(f"navigation could not run: {exc}")

render_neighborhood

render_neighborhood(neighborhood: Neighborhood, node_status: Mapping[str, StatusVerdict], *, focus: str | None = None, expanded: Iterable[str] = (), max_body_chars: int | None = None) -> Envelope

Render an assembled substrate neighborhood into a MemDSL :class:Envelope.

neighborhood is the assembled nodes + edges (see :func:zettelkasten.synapse.substrate.assemble_neighborhood); node_status maps each :attr:Node.token to the :class:StatusVerdict epistemics derived (see :func:zettelkasten.synapse.epistemics.apply_node_statuses). The wire status of every rendered node is CONSUMED from that map, never re-derived.

focus optionally names the node token the render is centered on (resolved to its assigned envelope-local ref). expanded names the node tokens whose neighborhoods have already been fetched, so an expand affordance is only offered toward not-yet-expanded neighbors. max_body_chars optionally elides long bodies to a lossy view (each elided node gains an elided gap and a ground fetch-back handle).

Determinism: nodes are ordered by their address token before refs are assigned, and gaps are emitted node-by-node in that order (each node's own gaps in a fixed stale → unresolved → outdated → missing → elided order), so the same neighborhood always yields the same envelope. Returns the :class:Envelope.

Source code in zettelkasten/synapse/memdsl/render.py

def render_neighborhood(
    neighborhood: Neighborhood,
    node_status: Mapping[str, StatusVerdict],
    *,
    focus: str | None = None,
    expanded: Iterable[str] = (),
    max_body_chars: int | None = None,
) -> Envelope:
    """Render an assembled substrate neighborhood into a MemDSL :class:`Envelope`.

    ``neighborhood`` is the assembled nodes + edges (see
    :func:`zettelkasten.synapse.substrate.assemble_neighborhood`); ``node_status``
    maps each :attr:`Node.token` to the :class:`StatusVerdict` epistemics derived
    (see :func:`zettelkasten.synapse.epistemics.apply_node_statuses`). The wire
    status of every rendered node is CONSUMED from that map, never re-derived.

    ``focus`` optionally names the node token the render is centered on (resolved
    to its assigned envelope-local ref). ``expanded`` names the node tokens whose
    neighborhoods have already been fetched, so an ``expand`` affordance is only
    offered toward not-yet-expanded neighbors. ``max_body_chars`` optionally
    elides long bodies to a lossy view (each elided node gains an ``elided`` gap
    and a ``ground`` fetch-back handle).

    Determinism: nodes are ordered by their address token before refs are
    assigned, and gaps are emitted node-by-node in that order (each node's own
    gaps in a fixed stale → unresolved → outdated → missing → elided order), so
    the same neighborhood always yields the same envelope. Returns the
    :class:`Envelope`.
    """
    expanded_set = set(expanded)

    # Deterministic node order: sort by the stable address token, never dict
    # insertion order, so identical input yields byte-identical output. A bad-id
    # node whose ``token``/:class:`Address` cannot be constructed would raise
    # :class:`MemDSLParseError` in the sort key, so filter such nodes out FIRST
    # (via ``_safe_token``, a debug log per drop) rather than letting one
    # unaddressable node sink the whole render. Surviving nodes keep their order.
    addressable = [n for n in neighborhood.node_list() if _safe_token(n) is not None]
    if len(addressable) != len(neighborhood):
        logger.debug(
            "render: dropped %d unaddressable node(s) before render",
            len(neighborhood) - len(addressable),
        )
    ordered = sorted(addressable, key=lambda n: n.token)
    present = {n.token for n in ordered}

    # Index outgoing edges by source token (preserving edge order per source) so
    # each node's un-expanded neighbors resolve to ``expand`` handles.
    edges_by_src: dict[str, list[str]] = {}
    for edge in neighborhood.edges:
        edges_by_src.setdefault(edge.src, []).append(edge.dst)

    rendered_nodes: list[RenderedNode] = []
    gaps: list[GapToken] = []
    ref_by_token: dict[str, str] = {}

    for offset, node in enumerate(ordered):
        ref = f"n{offset + 1}"
        ref_by_token[node.token] = ref
        rendered, node_gaps = _render_node(
            node,
            ref,
            node_status.get(node.token),
            expand_targets=_outgoing_targets(edges_by_src, node, present, expanded_set),
            max_body_chars=max_body_chars,
        )
        rendered_nodes.append(rendered)
        gaps.extend(node_gaps)

    # ``assign_refs`` is the contract's canonical ref assignment; re-apply it over
    # the final ordered list so the refs are the single source of truth (the
    # per-node ``ref`` above and this must agree — this call makes it authoritative).
    assign_refs(rendered_nodes)

    focus_ref = ref_by_token.get(focus) if focus is not None else None
    return Envelope(nodes=rendered_nodes, gaps=gaps, focus=focus_ref)