zettelkasten.synapse.assemble

zettelkasten.synapse.assemble

Wiring that makes the synapse navigation loop runnable against real stores.

This is the connective tissue between the three lower synapse workstreams and the navigation engine:

• :mod:zettelkasten.synapse.substrate supplies the store adapters and the in-memory neighborhood assembler, but nothing composes those adapters from a scope.
• :mod:zettelkasten.synapse.navigation supplies the deterministic executor, but its production expander / grounder seams default to NO-OPs — so expand reveals no real neighbors and ground re-fetches no real body.

This module closes both gaps WITHOUT touching either lower module's signatures:

1. :func:build_adapters composes a :class:~zettelkasten.synapse.substrate. MemoryAdapter plus one :class:~zettelkasten.synapse.substrate.ZettelAdapter per resolved source, REUSING :func:zettelkasten.synapse.retrieval.resolve_scope (and, through it, :func:~zettelkasten.synapse.retrieval.combined_grapher) exactly as the synapse() read paths do — it never re-implements scope resolution.
2. :func:adapter_expander / :func:adapter_grounder turn a list of adapters into the injectable :data:~zettelkasten.synapse.navigation.Expander / :data:~zettelkasten.synapse.navigation.Grounder seams the navigator calls, dispatching each request to the owning adapter by the node's store + source.
3. :func:navigate_scope ties it all together: build adapters, assemble a neighborhood, and run the navigator to a :class:~zettelkasten.synapse. navigation.NavResult.

Design invariants (mirrors the rest of synapse):

• Strictly read-only. Nothing here writes .memory/, .zettelkasten/, the link overlay, or .synapse/. The only side effects are store reads and the injected LLM call.
• Pure + deterministic. Ordering follows the deterministic source order that :func:~zettelkasten.synapse.retrieval.resolve_scope returns; adapter dispatch returns the first matching adapter.
• No heavy top-level imports. Every heavy dependency (framing / retrieval / navigation / the graph loader) is imported function-locally, matching the substrate module's convention and avoiding import cycles.

deterministic_relevance_map

deterministic_relevance_map(neighborhood: 'object', adapters: 'list[SourceAdapter]') -> dict[str, float]

Score assembled handles from stable adapter priority and retrieval rank.

assemble_neighborhood preserves adapter order and each adapter's returned rank in the insertion order of neighborhood.nodes. This projection reuses that deterministic order: a node's score is 1 / ((adapter_priority + 1) * (rank_within_adapter + 1)). Expanded nodes arrive after retrieved seeds and therefore receive lower per-adapter rank. No model-generated relevance enters the map.

Source code in zettelkasten/synapse/assemble.py

def deterministic_relevance_map(
    neighborhood: "object",
    adapters: "list[SourceAdapter]",
) -> dict[str, float]:
    """Score assembled handles from stable adapter priority and retrieval rank.

    ``assemble_neighborhood`` preserves adapter order and each adapter's returned
    rank in the insertion order of ``neighborhood.nodes``. This projection reuses
    that deterministic order: a node's score is
    ``1 / ((adapter_priority + 1) * (rank_within_adapter + 1))``. Expanded nodes
    arrive after retrieved seeds and therefore receive lower per-adapter rank.
    No model-generated relevance enters the map.
    """

    nodes = getattr(neighborhood, "nodes", {})
    if not isinstance(nodes, dict):
        return {}
    priority_by_identity = {id(adapter): index for index, adapter in enumerate(adapters)}
    ranks: dict[int, int] = {}
    relevance: dict[str, float] = {}
    fallback_priority = len(adapters)
    for token, node in nodes.items():
        adapter = _dispatch_adapter(
            adapters,
            str(getattr(node, "store", "")),
            str(getattr(node, "source", "")),
        )
        priority = (
            priority_by_identity.get(id(adapter), fallback_priority)
            if adapter is not None
            else fallback_priority
        )
        rank = ranks.get(priority, 0)
        ranks[priority] = rank + 1
        relevance[str(token)] = 1.0 / ((priority + 1) * (rank + 1))
    return relevance

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

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

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