zettelkasten.synapse.candidates

zettelkasten.synapse.candidates

Cross-store connection candidates: which memory<->ZK pairs MIGHT be related.

Candidates are the cheap, deterministic pre-filter before the (expensive) LLM typing pass. Two signals, unioned:

• Semantic NN — using each ZK note's text as a query into memory's shared vector space (and the reverse for a single node), pairs whose cosine clears a floor are candidates. Both stores embed with the same model, so this is a true cross-store nearest-neighbour.
• Shared provenance — a memory entry's git-pinned files overlapping a ZK note's referenced provenance (cited-work ids, or file-path tokens in its body). Rare but high-signal: it means both nodes are anchored to the same artifact.

Every candidate is BIPARTITE by construction: one memory endpoint, one ZK endpoint. Read-only; no store is touched.

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

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)

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)