zettelkasten.synapse.navigation

zettelkasten.synapse.navigation

The navigation engine of enforced epistemic honesty.

This is the STEERING layer of the synapse stack. It closes the loop between an LLM that reasons freely and a deterministic engine that owns every ground truth: the LLM chooses navigation actions and emits a cited reasoning-DAG, but a DETERMINISTIC executor here feeds it retrieval, enforces the token budget, and delegates the final ANSWER-vs-ABSTAIN verdict to :mod:zettelkasten.synapse.grounding_router. Because the navigator never decides grounding itself, fabrication is structurally impossible and abstention is automatic whenever coverage or grounding breaks.

It integrates the four upstream synapse workstreams (read in full before editing this module): the MemDSL contract (:mod:zettelkasten.synapse.memdsl.schema / .parser), the substrate assembler (:mod:zettelkasten.synapse.substrate), the deterministic epistemics engine (:mod:zettelkasten.synapse.epistemics), and the grounding router (:mod:zettelkasten.synapse.grounding_router). It EMITS and CONSUMES the contract vocabulary rather than inventing a parallel one.

The pieces, in the order the executor uses them:

• Deterministic action executor + action grammar. The v1 read actions — expand / join / ground / prune / abstain / answer — are exactly the ones registered in :data:zettelkasten.synapse.memdsl.schema.ACTION_REGISTRY. The executor runs the LLM's non-terminal actions in a FIXED, deterministic order (topologic expand → join, then parametric ground, then prune), then handles the single terminal decision. All I/O (the LLM, tether verification, verifier grounding, neighborhood expansion, the decision-tree walk) is injected, so the executor is a pure function of (inputs, seams) and runs in tests with no model or store.

• Injectable, advisory VOI. The production default remains :func:minimal_voi (relevance × status_gap); the deterministic :func:resolvability_cost_voi policy additionally divides by bounded token cost. VOI only ranks bounded prompt annotations and labels fallback diagnostics. It never executes an action or weakens the hard token/hop guardrails.

• Bounded self-correction. A verifier failure (a router false-premise abstain) triggers a RE-NAVIGATE — the executor feeds the failure back and lets the LLM pick a different path rather than doubling down — capped at :attr:NavConfig.max_renavigations.

• Sequential multi-hop state. Cross-type join moves record :class:BridgeBindings carrying the WEAKEST-LINK status across the joined endpoints, and a later hop resting on a binding is capped by it — a hop can be no more grounded than the binding it rests on.

• Model-agnostic LLM seam. The LLM is reached through :func:zettelkasten.llm_adapter.run_toolless_agent (wrapped by :func:default_llm), but the executor only ever calls the injected :data:LLM callable, so a test injects a scripted stub with no real model call. The LLM's structured output IS the MemDSL OUTPUT contract, parsed by :func:zettelkasten.synapse.memdsl.parser.parse_output; a parse failure gets ONE repair retry (the validator error is fed back) before the turn abstains.

VOICandidate dataclass

One handle the VOI model scores: a rendered node plus its derived status.

relevance is the retrieval relevance of the node (a [0, 1]-ish score; the v1 default is a flat 1.0 for every present node — a richer relevance channel is an injected map). status is the node's deterministically derived :class:~zettelkasten.synapse.epistemics.StatusVerdict. cost is an estimated token cost of grounding/expanding the handle. A Fang-2026 resolvability/cost VOI reads the SAME fields, so it drops in without a signature change.

Source code in zettelkasten/synapse/navigation.py

@dataclass(frozen=True)
class VOICandidate:
    """One handle the VOI model scores: a rendered node plus its derived status.

    ``relevance`` is the retrieval relevance of the node (a ``[0, 1]``-ish score;
    the v1 default is a flat ``1.0`` for every present node — a richer relevance
    channel is an injected map). ``status`` is the node's deterministically
    derived :class:`~zettelkasten.synapse.epistemics.StatusVerdict`. ``cost`` is
    an estimated token cost of grounding/expanding the handle. A Fang-2026
    resolvability/cost VOI reads the SAME fields, so it drops in without a
    signature change.
    """

    ref: str
    token: str
    relevance: float
    status: StatusVerdict
    cost: int

VOIAnnotation dataclass

The advisory VOI annotation attached to a candidate handle.

voi is the (advisory) value-of-information estimate; cost the estimated token cost. The executor uses these only for bounded prompt hints and to label a structural fallback; the hard budget guardrail is separate.

Source code in zettelkasten/synapse/navigation.py

@dataclass(frozen=True)
class VOIAnnotation:
    """The advisory VOI annotation attached to a candidate handle.

    ``voi`` is the (advisory) value-of-information estimate; ``cost`` the
    estimated token cost. The executor uses these only for bounded prompt hints
    and to label a structural fallback; the hard budget guardrail is separate.
    """

    voi: float
    cost: int

BridgeBinding dataclass

A cross-hop binding: an entity that bridges two hops, with weakest status.

A join across type records the bridge entity (the shared node), the hop_from / hop_to node tokens it links, and the WEAKEST-LINK :class:~zettelkasten.synapse.epistemics.StatusVerdict over the two endpoints (composed with :func:zettelkasten.synapse.epistemics.propagate, so tier is capped by the weaker endpoint and taint is unioned). The executor then caps the hop_to node's status by this binding when routing — a later hop can be no more grounded than the binding it rests on.

Source code in zettelkasten/synapse/navigation.py

@dataclass(frozen=True)
class BridgeBinding:
    """A cross-hop binding: an entity that bridges two hops, with weakest status.

    A ``join`` across type records the bridge ``entity`` (the shared node), the
    ``hop_from`` / ``hop_to`` node tokens it links, and the WEAKEST-LINK
    :class:`~zettelkasten.synapse.epistemics.StatusVerdict` over the two endpoints
    (composed with :func:`zettelkasten.synapse.epistemics.propagate`, so tier is
    capped by the weaker endpoint and taint is unioned). The executor then caps
    the ``hop_to`` node's status by this binding when routing — a later hop can be
    no more grounded than the binding it rests on.
    """

    entity: str
    hop_from: str
    hop_to: str
    status: StatusVerdict

NavConfig dataclass

Deterministic bounds on a navigation run.

token_budget is the HARD guardrail: the executor halts the loop (with a VOI-max fallback) once it can no longer afford the next turn. The default of 8000 is deliberately generous: NAV_SYSTEM now carries TWO @memdsl.out/1 exemplars (answer + abstain), so the old 4000 floor was too tight to reliably reach the model on a non-trivial neighborhood. max_hops caps the number of LLM turns; max_renavigations caps verifier-failure re-navigations; max_parse_repairs caps structured-output repair retries. trace_floor is forwarded to the router's prose gate. max_nodes is the hard ceiling on the neighborhood size — LLM-driven expand/ground merges never grow it past this — mirroring the substrate :class:~zettelkasten.synapse.substrate.Budget.max_nodes; None (the default) leaves in-loop growth unbounded (only the token budget bounds it). render_max_body_chars caps how many characters of each node body are SHOWN in the render: None shows the full body (no elision), a positive int truncates a long body to a lossy view (the render then emits an elided gap and a ground fetch-back handle, so the agent can re-fetch the FULL body). It bounds per-node body DEPTH, NOT neighborhood WIDTH: a wide hub (a node with many neighbors) is bounded by token_budget / max_nodes, never by this knob — trimming each body shorter does not reduce the NUMBER of nodes shown. Elision only bounds the PROMPT — the grounder re-fetches and the grounding gate always verifies against the full body — so it never weakens grounding; it only keeps a body-heavy neighborhood from blowing the token budget BEFORE the model is ever called. The default of 600 keeps the budget guardrail from halting on a body-heavy neighborhood. voi_policy selects the deterministic advisory scorer. The shipped default remains "minimal" until benchmark evidence justifies promotion; "resolvability_cost" activates :func:resolvability_cost_voi. focus_handle_limit bounds the number of top-scoring handles shown as prompt hints. Neither setting executes or suppresses an action.

Source code in zettelkasten/synapse/navigation.py

@dataclass(frozen=True)
class NavConfig:
    """Deterministic bounds on a navigation run.

    ``token_budget`` is the HARD guardrail: the executor halts the loop (with a
    VOI-max fallback) once it can no longer afford the next turn. The default of
    ``8000`` is deliberately generous: ``NAV_SYSTEM`` now carries TWO
    ``@memdsl.out/1`` exemplars (answer + abstain), so the old ``4000`` floor was
    too tight to reliably reach the model on a non-trivial neighborhood.
    ``max_hops`` caps the number of LLM turns; ``max_renavigations`` caps
    verifier-failure re-navigations; ``max_parse_repairs`` caps structured-output
    repair retries. ``trace_floor`` is forwarded to the router's prose gate.
    ``max_nodes`` is the hard ceiling on the neighborhood size — LLM-driven
    ``expand``/``ground`` merges never grow it past this — mirroring the substrate
    :class:`~zettelkasten.synapse.substrate.Budget.max_nodes`; ``None`` (the
    default) leaves in-loop growth unbounded (only the token budget bounds it).
    ``render_max_body_chars`` caps how many characters of each node body are
    SHOWN in the render: ``None`` shows the full body (no elision), a positive int
    truncates a long body to a lossy view (the render then emits an ``elided`` gap
    and a ``ground`` fetch-back handle, so the agent can re-fetch the FULL body).
    It bounds per-node body DEPTH, NOT neighborhood WIDTH: a wide hub (a node with
    many neighbors) is bounded by ``token_budget`` / ``max_nodes``, never by this
    knob — trimming each body shorter does not reduce the NUMBER of nodes shown.
    Elision only bounds the PROMPT — the grounder re-fetches and the grounding
    gate always verifies against the full body — so it never weakens grounding; it
    only keeps a body-heavy neighborhood from blowing the token budget BEFORE the
    model is ever called. The default of ``600`` keeps the budget guardrail from
    halting on a body-heavy neighborhood.
    ``voi_policy`` selects the deterministic advisory scorer. The shipped default
    remains ``"minimal"`` until benchmark evidence justifies promotion;
    ``"resolvability_cost"`` activates :func:`resolvability_cost_voi`.
    ``focus_handle_limit`` bounds the number of top-scoring handles shown as
    prompt hints. Neither setting executes or suppresses an action.
    """

    # Generous by design: NAV_SYSTEM embeds two @memdsl.out/1 exemplars, so a
    # 4000 floor was too tight to reliably reach the model on a non-trivial
    # neighborhood. Body DEPTH is trimmed by render_max_body_chars; neighborhood
    # WIDTH is bounded here (and by max_nodes).
    token_budget: int = 8000
    max_hops: int = 8
    max_renavigations: int = 1
    max_parse_repairs: int = 2
    trace_floor: float = gr.DEFAULT_TRACE_FLOOR
    max_nodes: int | None = None
    render_max_body_chars: int | None = 600
    voi_policy: str = "minimal"
    focus_handle_limit: int = 3

    def __post_init__(self) -> None:
        """Validate the bounded advisory policy controls."""

        if self.voi_policy not in {"minimal", "resolvability_cost"}:
            raise ValueError(f"unknown advisory VOI policy {self.voi_policy!r}")
        if (
            isinstance(self.focus_handle_limit, bool)
            or not isinstance(self.focus_handle_limit, int)
            or not 0 <= self.focus_handle_limit <= 10
        ):
            raise ValueError("focus_handle_limit must be an integer in [0, 10]")

NavResult dataclass

The outcome of a navigation run.

decision is the router's (or a structurally-forced) ANSWER/ABSTAIN verdict; halted_reason names why the loop ended (answer / abstain / budget / hops / parse-fail); loop_break is "voi-max" when a budget/hop guardrail broke a loop, else empty. The remaining fields expose the run's telemetry for inspection and testing.

Source code in zettelkasten/synapse/navigation.py

@dataclass
class NavResult:
    """The outcome of a navigation run.

    ``decision`` is the router's (or a structurally-forced) ANSWER/ABSTAIN
    verdict; ``halted_reason`` names why the loop ended (``answer`` / ``abstain``
    / ``budget`` / ``hops`` / ``parse-fail``); ``loop_break`` is ``"voi-max"``
    when a budget/hop guardrail broke a loop, else empty. The remaining fields
    expose the run's telemetry for inspection and testing.
    """

    decision: RouterDecision
    halted_reason: str
    output: AgentOutput | None = None
    envelope: Envelope | None = None
    hops: int = 0
    tokens_spent: int = 0
    renavigations: int = 0
    parse_repairs: int = 0
    loop_break: str = ""
    voi: dict[str, VOIAnnotation] = field(default_factory=dict)
    bindings: list[BridgeBinding] = field(default_factory=list)
    trace: list[str] = field(default_factory=list)

    @property
    def answered(self) -> bool:
        return self.decision.verdict == ANSWER

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

default_llm

default_llm(model: str = '', *, timeout: float | None = None) -> LLM

Build the production :data:LLM seam over :func:run_toolless_agent.

model keeps the navigator model-agnostic at its own API boundary and is forwarded to :func:zettelkasten.llm_adapter.run_toolless_agent, which passes it to DashboardAgent's supported model option. It also remains in the agent name for traceability. timeout bounds the call.

Source code in zettelkasten/synapse/navigation.py

def default_llm(model: str = "", *, timeout: float | None = None) -> LLM:
    """Build the production :data:`LLM` seam over :func:`run_toolless_agent`.

    ``model`` keeps the navigator model-agnostic at its own API boundary and is
    forwarded to :func:`zettelkasten.llm_adapter.run_toolless_agent`, which passes
    it to ``DashboardAgent``'s supported ``model`` option. It also remains in the
    agent name for traceability. ``timeout`` bounds the call.
    """
    agent_name = f"synapse-navigator[{model}]" if model else "synapse-navigator"

    def _run(system: str, prompt: str) -> str:
        return run_toolless_agent(
            agent_name,
            system,
            prompt,
            timeout=timeout,
            model=model,
        )

    return _run

status_gap

status_gap(verdict: StatusVerdict) -> float

How much a node's status LIMITS the answer, in [0, 1].

The gap is what grounding a node could still buy: a clean VERIFIED node (grounded / measured with no taint) is already maximally useful, so its gap is 0.0 (grounding it further has no value); an inferred node, or any node carrying stale / unresolved / contested taint, has a full gap of 1.0 (resolving it is where the value is).

Source code in zettelkasten/synapse/navigation.py

def status_gap(verdict: StatusVerdict) -> float:
    """How much a node's status LIMITS the answer, in ``[0, 1]``.

    The gap is what grounding a node could still buy: a clean VERIFIED node
    (``grounded`` / ``measured`` with no taint) is already maximally useful, so
    its gap is ``0.0`` (grounding it further has no value); an ``inferred`` node,
    or any node carrying ``stale`` / ``unresolved`` / ``contested`` taint, has a
    full gap of ``1.0`` (resolving it is where the value is).
    """
    if verdict.tainted:
        return 1.0
    return 0.0 if verdict.wire() in _VERIFIED_WIRE else 1.0

minimal_voi

minimal_voi(candidate: VOICandidate) -> VOIAnnotation

The v1-minimal, advisory VOI: voi ≈ relevance × status_gap.

Deliberately simple — relevance weights how on-topic a handle is and the status gap weights how much its inferred/tainted status still limits the answer, so the highest-VOI handle is the most relevant not-yet-grounded node. This is the placeholder behind :data:VOIModel: the full Fang-2026 resolvability/cost-ratio VOI slots in as a different callable of the same shape with no change here or in the executor.

Source code in zettelkasten/synapse/navigation.py

def minimal_voi(candidate: VOICandidate) -> VOIAnnotation:
    """The v1-minimal, advisory VOI: ``voi ≈ relevance × status_gap``.

    Deliberately simple — relevance weights how on-topic a handle is and the
    status gap weights how much its inferred/tainted status still limits the
    answer, so the highest-VOI handle is the most relevant not-yet-grounded node.
    This is the placeholder behind :data:`VOIModel`: the full Fang-2026
    resolvability/cost-ratio VOI slots in as a different callable of the same
    shape with no change here or in the executor.
    """
    voi = max(0.0, candidate.relevance) * status_gap(candidate.status)
    return VOIAnnotation(voi=voi, cost=candidate.cost)

resolvability_cost_voi

resolvability_cost_voi(candidate: VOICandidate) -> VOIAnnotation

Cost-aware advisory VOI over deterministic evidence.

Relevance is clamped to [0, 1]; status gap is router-owned and monotonic; token cost is bounded to [1, 100_000] before division. A clean verified node therefore remains exactly zero, while increasing relevance or reducing cost can never lower a non-zero score. The annotation is advisory only.

Source code in zettelkasten/synapse/navigation.py

def resolvability_cost_voi(candidate: VOICandidate) -> VOIAnnotation:
    """Cost-aware advisory VOI over deterministic evidence.

    Relevance is clamped to ``[0, 1]``; status gap is router-owned and monotonic;
    token cost is bounded to ``[1, 100_000]`` before division. A clean verified
    node therefore remains exactly zero, while increasing relevance or reducing
    cost can never lower a non-zero score. The annotation is advisory only.
    """

    relevance = min(1.0, max(0.0, candidate.relevance))
    cost = min(100_000, max(1, candidate.cost))
    voi = relevance * status_gap(candidate.status) / cost
    return VOIAnnotation(voi=voi, cost=candidate.cost)