zettelkasten.embeddings

zettelkasten.embeddings

Embedding-based semantic search for the Zettelkasten.

Backed by kglite — the same strategy the memory subsystem uses. The .md files under the graphs directory remain the source of truth; each concept box gets its own disposable kglite KnowledgeGraph cached at <ANGELO_DIR>/zettel/<box>.kgl and rebuilt from the live notes on load. Notes become nodes, intra-box links become edges, and embedding + similarity search go through kglite's native embed_texts / search_text / vector_search (model2vec, see :mod:memory.embedders).

Because the cache is rebuilt from the notes currently on disk, vectors for notes deleted while the index was cold can never resurface as ghost search hits — the stale-vector bug in the old JSON/numpy index is gone by design.

EmbeddingIndex

kglite-backed semantic index for a single concept box.

Public surface is unchanged from the legacy numpy implementation: search, find_similar, index_note, index_notes, remove_note, needs_indexing, indexed_count, load_cache, save_cache — plus get_vector for callers that need a raw vector.

Source code in zettelkasten/embeddings.py

class EmbeddingIndex:
    """kglite-backed semantic index for a single concept box.

    Public surface is unchanged from the legacy numpy implementation:
    ``search``, ``find_similar``, ``index_note``, ``index_notes``,
    ``remove_note``, ``needs_indexing``, ``indexed_count``, ``load_cache``,
    ``save_cache`` — plus ``get_vector`` for callers that need a raw vector.
    """

    def __init__(
        self,
        graph_path: Path,
        *,
        embedder: Any = _UNSET,
        cache_path: Path | None = None,
    ):
        self.graph_path = graph_path
        self.box_name = graph_path.name
        self.cache_path = cache_path or (_cache_dir() / f"{self.box_name}.kgl")
        # Legacy float-list JSON blob; read once for back-compat, never written.
        self.legacy_cache_path = graph_path / "_embeddings.json"
        self._embedder = _build_embedder() if embedder is _UNSET else embedder
        self._graph: Any = None
        self._node_ids: set[str] = set()
        # Vectors recovered from a prior cache, used to skip re-embedding
        # unchanged notes after a rebuild.
        self._cached_vectors: dict[str, list[float]] = {}
        # Content hash each cached vector was computed against (note id -> hash).
        # A missing entry means the cache predates hash tracking (legacy JSON or
        # an older .kgl); such vectors can't be verified and are re-embedded
        # once on upgrade rather than trusted (see _embed_pending).
        self._cached_hashes: dict[str, str] = {}
        # Content hash of the text each currently-indexed node was built from.
        self._note_hashes: dict[str, str] = {}
        # Citation (cited-but-unowned work) state, mirroring the note fields
        # above but keyed by citation id and stored under CITATION_NODE_TYPE.
        self._citation_ids: set[str] = set()
        self._cited_vectors: dict[str, list[float]] = {}
        self._cited_hashes: dict[str, str] = {}
        # Negative cache: content hash of citations whose text embedded to a
        # zero-norm (OOV/unstorable) vector, so an UNCHANGED such work
        # short-circuits on the next index_citations call instead of being
        # re-embedded-and-dropped every pass. In-memory only (never persisted):
        # a dropped citation has no vector in the .kgl, so a reload cannot
        # resurrect it, and the first post-reload pass simply repopulates this.
        self._citation_skipped_hashes: dict[str, str] = {}
        # kglite.KnowledgeGraph is not thread-safe and one graph instance is
        # shared across the MCP threadpool for a box, so every method touching
        # self._graph serializes on a re-entrant lock. This is the SHARED
        # process-wide native lock (not a per-instance one): kglite has global
        # native state, so distinct boxes/handles must still mutually exclude or a
        # rebuild-then-swap on one thread races a query on another and SIGSEGVs.
        # See _NATIVE_LOCK for why kglite and the sklearn compute share it.
        self._lock = _NATIVE_LOCK
        # Debounced-save state (see ``save_cache_debounced``). A dedicated,
        # NON-native lock guards only the timer handle — it is held briefly and
        # never nested inside ``_lock`` — so scheduling a save from the write
        # path is cheap and cannot contend with native compute. The timer's
        # callback takes ``_lock`` (via ``save_cache``) alone, so the global
        # native-lock acquisition order is unchanged.
        self._save_timer: "threading.Timer | None" = None
        self._save_timer_lock = threading.Lock()
        # Monotonic token identifying the most recently scheduled debounce
        # timer. Each ``save_cache_debounced`` bumps it and stamps the new timer
        # with the current value; a firing timer compares its stamp against this
        # to tell whether it is STILL the scheduled timer (vs. one that was
        # cancelled-and-replaced by a later save). Lets the callback clear the
        # handle / prune the pending registry ONLY when it truly owns them.
        self._save_timer_gen = 0

    # ── cache lifecycle ─────────────────────────────────────────────────────

    def load_cache(self) -> None:
        """Start a fresh in-memory graph and recover prior vectors.

        The graph itself is never treated as authoritative: it is rebuilt from
        the live notes by :meth:`index_notes`. We only salvage the embedding
        vectors (keyed by note id) so unchanged notes don't have to be
        re-embedded.
        """
        import kglite

        with self._lock:
            self._graph = kglite.KnowledgeGraph()
            self._node_ids = set()
            self._cached_vectors = {}
            self._cached_hashes = {}
            self._note_hashes = {}
            self._citation_ids = set()
            self._cited_vectors = {}
            self._cited_hashes = {}
            self._citation_skipped_hashes = {}
            # Full citation node rows (id/title/text/hash) recovered from the
            # prior .kgl, kept aside so we can re-hydrate citation NODES into
            # the fresh graph after the dimension check below.
            cited_rows: list[dict] = []

            if self._embedder is not None:
                try:
                    self._graph.set_embedder(self._embedder)
                except Exception as exc:
                    logger.warning("Could not register embedder for box '%s': %s", self.box_name, exc)

            # Recover vectors (and the hashes they were computed against) from
            # the prior .kgl cache, if any.
            if self.cache_path.exists():
                try:
                    prior = kglite.load(str(self.cache_path))
                    stored = prior.embeddings(NODE_TYPE, TEXT_FIELD) or {}
                    self._cached_vectors = {str(k): list(v) for k, v in stored.items()}
                    try:
                        rows = prior.cypher(
                            f"MATCH (n:{NODE_TYPE}) RETURN n.id AS id, n.{HASH_FIELD} AS h"
                        )
                        for row in rows or []:
                            nid = row.get("id")
                            h = row.get("h")
                            if nid is not None and h:
                                self._cached_hashes[str(nid)] = str(h)
                    except Exception as exc:
                        logger.debug("Could not read cached hashes for box '%s': %s", self.box_name, exc)
                    # Recover citation vectors + hashes the same way (separate
                    # node type, same .kgl). Absent when no citations were ever
                    # embedded into this box.
                    try:
                        stored_cit = prior.embeddings(CITATION_NODE_TYPE, TEXT_FIELD) or {}
                        self._cited_vectors = {str(k): list(v) for k, v in stored_cit.items()}
                        rows = prior.cypher(
                            f"MATCH (n:{CITATION_NODE_TYPE}) RETURN n.id AS id, "
                            f"n.title AS title, n.{TEXT_FIELD} AS text, n.{HASH_FIELD} AS h"
                        )
                        for row in rows or []:
                            cid = row.get("id")
                            h = row.get("h")
                            if cid is None:
                                continue
                            cid = str(cid)
                            if h:
                                self._cited_hashes[cid] = str(h)
                            cited_rows.append(
                                {
                                    "id": cid,
                                    "title": row.get("title"),
                                    "text": row.get("text"),
                                    "h": str(h) if h else None,
                                }
                            )
                    except Exception as exc:
                        logger.debug("Could not read cached citations for box '%s': %s", self.box_name, exc)
                except Exception as exc:
                    logger.warning("Could not load zettel cache '%s': %s", self.cache_path, exc)

            # One-time back-compat import of the legacy JSON blob when there is
            # no .kgl yet. These vectors carry no hash, so they can't be verified
            # against the current note text; _embed_pending re-embeds them once
            # on upgrade so every vector ends up hash-tracked.
            if not self._cached_vectors and self.legacy_cache_path.exists():
                try:
                    data = json.loads(self.legacy_cache_path.read_text(encoding="utf-8"))
                    self._cached_vectors = {
                        str(k): list(v) for k, v in (data.get("vectors") or {}).items()
                    }
                except (OSError, ValueError, TypeError) as exc:
                    logger.warning("Could not read legacy embeddings '%s': %s", self.legacy_cache_path, exc)

            # Drop cached vectors whose dimension no longer matches the active
            # embedder (a model/dimension upgrade). Restoring them would mix
            # dimensions in one kglite store and corrupt search; clearing the
            # cache makes index_notes re-embed every note fresh at the new dim.
            self._discard_mismatched_cached_vectors()

            # Re-hydrate citation NODES into the fresh graph so graph-native
            # Wave B queries (``select(CITATION_NODE_TYPE).vector_search``) and
            # ``citation_count`` reflect cached citations WITHOUT requiring
            # ``index_citations`` to be re-run first. Notes are rebuilt from the
            # live .md sources by ``index_notes``; citations have no equivalent
            # live-reload step here, so we restore them straight from the .kgl.
            # Runs after the dimension check so cleared (mismatched) vectors are
            # not re-hydrated.
            self._rehydrate_citations(cited_rows)

    def _discard_mismatched_cached_vectors(self) -> None:
        """Clear recovered vectors if their dimension != the embedder's."""
        if self._embedder is None:
            return
        want = getattr(self._embedder, "dimension", None)
        if not want:
            return
        for vec in self._cached_vectors.values():
            if vec and len(vec) != want:
                logger.info(
                    "Box '%s': cached vectors are %d-d but embedder is %d-d; "
                    "rebuilding embeddings at the new dimension.",
                    self.box_name, len(vec), want,
                )
                self._cached_vectors = {}
                self._cached_hashes = {}
                break
        for vec in self._cited_vectors.values():
            if vec and len(vec) != want:
                logger.info(
                    "Box '%s': cached citation vectors are %d-d but embedder is "
                    "%d-d; rebuilding citation embeddings at the new dimension.",
                    self.box_name, len(vec), want,
                )
                self._cited_vectors = {}
                self._cited_hashes = {}
                break

    def _rehydrate_citations(self, cited_rows: list[dict]) -> None:
        """Re-add cached citation NODES (and their vectors) to the live graph.

        Called at the end of :meth:`load_cache`. Each row is restored only when
        it still has a usable (present, right-dimension, non-zero-norm) cached
        vector, so a reload alone leaves ``citation_count`` and graph-native
        ``CITATION_NODE_TYPE`` search reflecting the cache — no ``index_citations``
        round-trip required first. A later ``index_citations`` with the same
        works is then idempotent (hashes match → restore, no re-embed).
        """
        if self._graph is None or not cited_rows:
            return
        import pandas as pd

        rows: list[dict] = []
        payload: dict[str, list[float]] = {}
        for r in cited_rows:
            cid = r.get("id")
            if cid is None:
                continue
            cid = str(cid)
            vec = self._cited_vectors.get(cid)
            # Skip vectors dropped by the dimension check or any stored garbage.
            if vec is None or _is_zero_norm(vec):
                continue
            text = r.get("text") or ""
            h = r.get("h") or _text_hash(text)
            title = r.get("title") or (text.split("\n", 1)[0] if text else cid)
            rows.append({"id": cid, "title": title, TEXT_FIELD: text, HASH_FIELD: h})
            payload[cid] = vec
        if not rows:
            return
        try:
            self._graph.add_nodes(
                pd.DataFrame(rows),
                node_type=CITATION_NODE_TYPE,
                unique_id_field="id",
                node_title_field="title",
                conflict_handling="update",
            )
            self._graph.add_embeddings(CITATION_NODE_TYPE, TEXT_FIELD, payload)
        except Exception as exc:
            logger.warning("Could not re-hydrate citations for box '%s': %s", self.box_name, exc)
            return
        for row in rows:
            self._citation_ids.add(row["id"])

    def save_cache(self) -> None:
        """Persist the kglite graph atomically (temp file + rename).

        The tmp filename is made unique per write (pid + uuid) because the MCP
        server and the dashboard backend share one ``<box>.kgl``; a fixed tmp
        path would let two concurrent saves clobber each other into a torn file.
        The cache is disposable, so a lost write is fine — a corrupt one is not.
        """
        with self._lock:
            if self._graph is None:
                return
            tmp = self.cache_path.with_suffix(f".kgl.{os.getpid()}.{uuid.uuid4().hex}.tmp")
            try:
                self.cache_path.parent.mkdir(parents=True, exist_ok=True)
                self._graph.save(str(tmp))
                os.replace(tmp, self.cache_path)
            except Exception as exc:
                logger.warning("Could not save zettel cache '%s': %s", self.cache_path, exc)
                try:
                    tmp.unlink(missing_ok=True)
                except OSError:
                    pass

    def save_cache_debounced(self) -> None:
        """Schedule a debounced :meth:`save_cache`, coalescing rapid saves.

        Called on the ``save_note`` write path instead of persisting the whole
        ``.kgl`` synchronously on every note write. Rapid successive saves reset
        a single per-index timer, so a burst of writes collapses into ONE cache
        rewrite once the box goes quiet (see :data:`_SAVE_DEBOUNCE_SEC`).

        Safety: the incremental embed already ran synchronously in
        :meth:`index_note`, so the new vector is live in-memory the instant this
        returns (an immediate ``get_vector``/``search`` sees it) — only the
        on-disk persist is deferred. The ``.kgl`` is a disposable cache rebuilt
        from the live ``.md`` sources on load, and the caller writes that source
        of truth synchronously BEFORE scheduling here, so a coalesced or even
        crash-dropped persist can never diverge from a full rebuild. The fired
        save serializes the CURRENT live graph under ``_lock`` and writes it via
        the same atomic temp+rename as :meth:`save_cache`, so an interleaved
        synchronous saver (``delete_note``/``remove_citation``) and this timer
        never tear the file. The daemon timer never blocks interpreter exit; a
        best-effort ``atexit`` flush persists anything still pending.
        """
        with self._save_timer_lock:
            if self._save_timer is not None:
                self._save_timer.cancel()
            self._save_timer_gen += 1
            gen = self._save_timer_gen
            timer = threading.Timer(
                _SAVE_DEBOUNCE_SEC, self._fire_debounced_save, args=(gen,)
            )
            timer.daemon = True
            self._save_timer = timer
            _PENDING_SAVE_INDEXES.add(self)
            timer.start()

    def _fire_debounced_save(self, gen: int) -> None:
        """Timer callback: clear the pending handle, then persist synchronously.

        ``gen`` is the token this timer was stamped with when scheduled. We only
        touch the shared timer/registry state if we are STILL the scheduled
        timer (``gen == self._save_timer_gen``): a save scheduled after us
        bumped the token and installed a new timer, so it owns the handle now
        and pruning here would orphan it (its timer would fire an extra,
        uncancellable write). All of this is decided under ``_save_timer_lock``,
        so a save being scheduled concurrently either wins the lock first (bumps
        the token, we see the mismatch and leave everything for it) or waits
        until we finish (then re-adds ``self`` to the registry), and no genuinely
        pending save is ever dropped.
        """
        with self._save_timer_lock:
            if gen == self._save_timer_gen:
                self._save_timer = None
                # No newer save was scheduled after us, so once we persist below
                # this index has no pending write: drop it from the atexit
                # registry to keep the shutdown flush O(genuinely-pending)
                # rather than O(every index ever touched).
                _PENDING_SAVE_INDEXES.discard(self)
        self.save_cache()

    def flush_cache(self) -> None:
        """Cancel any pending debounced save and persist immediately.

        Idempotent and safe to call whether or not a save is pending: with no
        timer it is just a plain :meth:`save_cache`. Used by the ``atexit`` flush
        so a scheduled-but-unfired persist is not wasted on a clean shutdown.
        """
        with self._save_timer_lock:
            if self._save_timer is not None:
                self._save_timer.cancel()
                self._save_timer = None
            # Invalidate any timer already mid-fire (its cancel() was a no-op) so
            # it sees the token mismatch and does not re-null the handle or
            # re-prune after we persist. We flush the pending write here, so this
            # index is no longer pending: drop it from the atexit registry.
            self._save_timer_gen += 1
            _PENDING_SAVE_INDEXES.discard(self)
        self.save_cache()

    # ── indexing ────────────────────────────────────────────────────────────

    def _node_row(self, note: "Note") -> dict:
        text = note_to_embed_text(note)
        return {
            "id": note.id,
            "title": note.title,
            "type": note.type,
            TEXT_FIELD: text,
            HASH_FIELD: _text_hash(text),
        }

    def _add_nodes(self, notes: list["Note"]) -> bool:
        import pandas as pd

        if self._graph is None:
            self.load_cache()
        if not notes:
            return False
        rows = [self._node_row(n) for n in notes]
        df = pd.DataFrame(rows)
        try:
            self._graph.add_nodes(
                df,
                node_type=NODE_TYPE,
                unique_id_field="id",
                node_title_field="title",
                conflict_handling="update",
            )
        except Exception as exc:
            logger.warning("Could not add notes to box '%s': %s", self.box_name, exc)
            return False
        for row in rows:
            self._node_ids.add(row["id"])
            self._note_hashes[row["id"]] = row[HASH_FIELD]
        return True

    def _add_edges(self, notes: list["Note"]) -> None:
        """Mirror intra-box links as kglite edges (best-effort).

        Only links whose target is a node already in this box and which carry
        no cross-graph reference are added; the relation is stored as an edge
        property under a single ``links`` connection type so relation names with
        hyphens never clash with kglite's edge-type identifier rules.
        """
        import pandas as pd

        if self._graph is None:
            return
        src_ids = [n.id for n in notes]
        if not src_ids:
            return
        # Build the replacement payload BEFORE deleting anything, so a failure
        # while assembling it can never leave a note edgeless. Only links whose
        # target is already a node in this box and which carry no cross-graph
        # reference are mirrored as edges.
        edges = []
        for n in notes:
            for link in n.links:
                if link.graph:
                    continue
                if link.target not in self._node_ids:
                    continue
                edges.append({"src": n.id, "tgt": link.target, "relation": link.relation})
        # Clear these notes' existing OUTGOING edges so edits or removals of a
        # note's links don't leave stale edges behind (add-only would never
        # prune them). If the clear itself fails, leave the prior edges intact
        # rather than risk a half-applied state.
        try:
            self._graph.cypher(
                f"MATCH (n:{NODE_TYPE})-[r:{LINK_TYPE}]->() WHERE n.id IN $ids DELETE r",
                params={"ids": src_ids},
            )
        except Exception as exc:
            logger.warning("Could not clear edges for box '%s': %s", self.box_name, exc)
            return
        if not edges:
            return
        try:
            self._graph.add_connections(
                pd.DataFrame(edges),
                connection_type=LINK_TYPE,
                source_type=NODE_TYPE,
                source_id_field="src",
                target_type=NODE_TYPE,
                target_id_field="tgt",
            )
        except Exception as exc:
            # Log loudly: the old edges were just cleared, so swallowing this
            # silently would leave the note with zero outgoing edges unnoticed.
            logger.warning("Could not add edges for box '%s': %s", self.box_name, exc)

    def _embed_pending(self) -> None:
        """Restore known vectors, then embed only the notes still missing one."""
        if self._embedder is None or self._graph is None:
            return
        if self._cached_vectors:
            payload = {}
            for nid, vec in self._cached_vectors.items():
                if nid not in self._node_ids:
                    continue
                cached_hash = self._cached_hashes.get(nid)
                # A missing cached hash means a legacy/older cache (legacy JSON
                # blob or a pre-hash .kgl) with no hash to verify against. Do
                # NOT restore it: there is no way to know it still matches the
                # current note text, and restoring it would let _add_nodes stamp
                # the *current* text's hash onto a possibly-stale vector,
                # permanently masking the staleness on every later reload. Skip
                # it so embed_texts recomputes a fresh vector (stamped with the
                # correct hash) — a one-time re-embed per legacy note on upgrade
                # that restores full staleness tracking from then on.
                if cached_hash is None:
                    continue
                # Otherwise a mismatch means the .md changed out-of-band while
                # cold: drop it so embed_texts recomputes.
                if cached_hash != self._note_hashes.get(nid):
                    continue
                payload[nid] = vec
            if payload:
                try:
                    self._graph.add_embeddings(NODE_TYPE, TEXT_FIELD, payload)
                except Exception as exc:
                    logger.warning("Could not restore vectors for box '%s': %s", self.box_name, exc)
        try:
            self._graph.embed_texts(NODE_TYPE, TEXT_FIELD, show_progress=False, replace=False)
        except Exception as exc:
            logger.warning("Embedding failed for box '%s': %s", self.box_name, exc)

    def index_notes(self, notes: list["Note"]) -> None:
        """(Re)build the graph from ``notes`` and embed any not yet embedded.

        Notes whose type is in :data:`EMBED_SKIP_TYPES` (quotes) are dropped up
        front: they are supporting evidence reached structurally, so embedding
        them is pure wasted compute (~half the corpus). They never enter this
        semantic index — not as nodes, vectors, edges, or cluster members.
        """
        notes = [n for n in notes if n.type not in EMBED_SKIP_TYPES]
        if not notes:
            return
        with self._lock:
            if not self._add_nodes(notes):
                return
            self._add_edges(notes)
            self._embed_pending()

    def index_note(self, note: "Note") -> None:
        """Add/update a single note and (re)embed it.

        Unlike the bulk path, this always recomputes the note's vector so an
        edit to an existing note refreshes its embedding instead of keeping a
        stale cached one. Types in :data:`EMBED_SKIP_TYPES` (quotes) are skipped
        entirely — see :meth:`index_notes`.
        """
        if note.type in EMBED_SKIP_TYPES:
            return
        with self._lock:
            if not self._add_nodes([note]):
                return
            self._add_edges([note])
            if self._embedder is None or self._graph is None:
                return
            try:
                vecs = self._embedder.embed([note_to_embed_text(note)])
            except Exception as exc:
                logger.warning("Could not embed note '%s': %s", note.id, exc)
                return
            if not vecs:
                return
            vec = list(vecs[0])
            try:
                self._graph.add_embeddings(NODE_TYPE, TEXT_FIELD, {note.id: vec})
                self._cached_vectors[note.id] = vec
                self._cached_hashes[note.id] = self._note_hashes.get(
                    note.id, _text_hash(note_to_embed_text(note))
                )
            except Exception as exc:
                logger.warning("Could not store vector for note '%s': %s", note.id, exc)

    def _citation_row(self, cid: str, text: str) -> dict:
        return {
            "id": cid,
            "title": text.split("\n", 1)[0],
            TEXT_FIELD: text,
            HASH_FIELD: _text_hash(text),
        }

    def index_citations(self, citations: dict[str, dict]) -> None:
        """Embed cited-but-unowned works (``title + abstract``) into this index.

        ``citations`` maps citation id -> ``_citations/*.yaml`` payload (the
        shape returned by :func:`zettelkasten.graph.load_citations`). Each work
        is embedded from :func:`citation_to_embed_text`; works with no
        embeddable text (no title and no abstract) are skipped rather than
        embedded as empty vectors. Citation nodes share the box's embedder and
        ``.kgl`` cache with notes but live under :data:`CITATION_NODE_TYPE`, so
        their vectors are comparable to note vectors yet never surface in note
        search.

        Re-embedding is driven by a content hash so the same long-lived index
        stays correct without a reload. A work is re-embedded whenever its
        freshly-computed hash differs from the hash its currently-tracked vector
        was computed against — this catches BOTH out-of-band edits seen across a
        reload AND in-session edits (re-calling with an edited title/abstract on
        the same index), since citations have no singular force-recompute path
        like :meth:`index_note`. Unchanged works are restored from their tracked
        vector and never re-embedded. Out-of-vocabulary-but-nonempty text that
        embeds to a zero-norm vector is dropped (never stored or counted) so it
        cannot pollute downstream cosine similarity.

        A work whose text changes to OOV/unstorable evicts any prior node and
        vector for that id (so :meth:`get_citation_vector` returns ``None`` and
        :attr:`citation_count` drops) and is negative-cached by content hash, so
        an unchanged OOV work short-circuits on later calls rather than
        re-embedding-and-dropping every pass. Changing such text back to
        embeddable re-embeds it and makes it retrievable again.

        This call is purely additive/refresh — it NEVER prunes ids that are
        absent from ``citations``. Wave B may pass a SUBSET of the relevant
        works for a box per call, so omitting an id must not delete it; explicit
        removal goes through :meth:`remove_citation`.
        """
        if not citations:
            return
        import pandas as pd

        with self._lock:
            if self._graph is None:
                self.load_cache()

            # Build the embed text + content hash for every work with something
            # embeddable; works with neither title nor abstract are skipped.
            pending: dict[str, tuple[str, str]] = {}
            for cid, data in citations.items():
                if not isinstance(data, dict):
                    continue
                text = citation_to_embed_text(data)
                if not text:
                    continue
                pending[str(cid)] = (text, _text_hash(text))
            if not pending:
                return

            # Partition into unchanged (restore the tracked vector, no re-embed)
            # and changed/new (must be embedded now). "Unchanged" requires a
            # tracked, non-garbage vector whose hash still matches the current
            # text; anything else — an edit (in-session or cross-reload) or a
            # first sighting — is re-embedded.
            restore: dict[str, list[float]] = {}
            to_embed: dict[str, str] = {}
            for cid, (text, h) in pending.items():
                vec = self._cited_vectors.get(cid)
                if vec is not None and self._cited_hashes.get(cid) == h and not _is_zero_norm(vec):
                    restore[cid] = vec
                elif self._citation_skipped_hashes.get(cid) == h:
                    # Negative-cache hit: this exact text already embedded to a
                    # zero-norm vector and was dropped. Skip it — don't re-embed
                    # (and don't store) the same garbage every pass.
                    continue
                else:
                    to_embed[cid] = text

            # Embed the changed/new works in one batch, dropping any that embed
            # to a zero-norm vector (OOV-but-nonempty text under model2vec).
            embedded: dict[str, list[float]] = {}
            dropped: dict[str, str] = {}
            if to_embed and self._embedder is not None:
                cids = list(to_embed)
                try:
                    vecs = self._embedder.embed([to_embed[c] for c in cids])
                except Exception as exc:
                    logger.warning("Could not embed citations for box '%s': %s", self.box_name, exc)
                    vecs = []
                for cid, vec in zip(cids, vecs):
                    vec = list(vec)
                    if _is_zero_norm(vec):
                        logger.debug(
                            "Box '%s': citation '%s' has out-of-vocabulary text that embeds "
                            "to a zero-norm vector; skipping it.", self.box_name, cid,
                        )
                        dropped[cid] = pending[cid][1]
                        continue
                    embedded[cid] = vec

            # An id that was targeted for re-embed but produced no storable
            # vector must not strand stale per-id state: evict any prior node +
            # vector (so get_citation_vector returns None and citation_count
            # drops) and negative-cache its hash so an unchanged OOV work
            # short-circuits next time. This runs BEFORE the empty-final return
            # so a good->OOV edit can't keep serving the stale pre-edit vector.
            for cid, h in dropped.items():
                self._evict_citation_node(cid)
                self._citation_skipped_hashes[cid] = h

            # Final set with a usable vector: restored + freshly embedded.
            final: dict[str, list[float]] = {**restore, **embedded}
            if not final:
                return

            rows = [self._citation_row(cid, pending[cid][0]) for cid in final]
            try:
                self._graph.add_nodes(
                    pd.DataFrame(rows),
                    node_type=CITATION_NODE_TYPE,
                    unique_id_field="id",
                    node_title_field="title",
                    conflict_handling="update",
                )
                self._graph.add_embeddings(CITATION_NODE_TYPE, TEXT_FIELD, final)
            except Exception as exc:
                logger.warning("Could not add citations to box '%s': %s", self.box_name, exc)
                return
            for cid, vec in final.items():
                h = pending[cid][1]
                self._citation_ids.add(cid)
                self._cited_vectors[cid] = vec
                self._cited_hashes[cid] = h
                # Text that was previously OOV is now embeddable again: clear
                # its negative-cache entry so it is retrievable from here on.
                self._citation_skipped_hashes.pop(cid, None)

    def _evict_citation_node(self, cid: str) -> None:
        """Delete a citation node + its tracked vector/hash state.

        Clears the per-citation bookkeeping structures and DETACH DELETEs the
        node from the live graph (so a later ``save_cache`` persists the
        removal). Does NOT touch the negative cache — the caller decides whether
        the id should be remembered as OOV (eviction-on-drop) or forgotten
        entirely (:meth:`remove_citation`). Acquires the re-entrant ``_lock``
        itself, so held-lock callers re-enter safely and off-lock callers are
        protected too.
        """
        with self._lock:
            self._cited_vectors.pop(cid, None)
            self._cited_hashes.pop(cid, None)
            self._citation_ids.discard(cid)
            if self._graph is None:
                return
            try:
                self._graph.cypher(
                    f"MATCH (n:{CITATION_NODE_TYPE}) WHERE n.id = $id DETACH DELETE n",
                    params={"id": cid},
                )
            except Exception as exc:
                logger.debug("Could not delete citation '%s' from box '%s': %s", cid, self.box_name, exc)

    def remove_citation(self, cid: str) -> None:
        """Drop a cited work's node + all tracked state from graph and caches.

        Mirrors :meth:`remove_note` for the citation node type. This is the ONLY
        deletion path for citations: :meth:`index_citations` never prunes ids it
        wasn't passed, so a citation that should disappear from
        :attr:`citation_count` and graph-native ``CITATION_NODE_TYPE`` search
        must be removed here. The node is DETACH DELETEd from the live graph so
        a subsequent ``save_cache`` persists the removal and a later
        ``load_cache`` will not re-hydrate it.
        """
        with self._lock:
            self._citation_skipped_hashes.pop(cid, None)
            self._evict_citation_node(cid)

    def remove_note(self, note_id: str) -> None:
        """Drop a note's node (and edges) from the live graph and caches."""
        with self._lock:
            self._cached_vectors.pop(note_id, None)
            self._cached_hashes.pop(note_id, None)
            self._note_hashes.pop(note_id, None)
            self._node_ids.discard(note_id)
            if self._graph is None:
                return
            try:
                self._graph.cypher(
                    f"MATCH (n:{NODE_TYPE}) WHERE n.id = $id DETACH DELETE n",
                    params={"id": note_id},
                )
            except Exception as exc:
                logger.debug("Could not delete node '%s' from box '%s': %s", note_id, self.box_name, exc)

    # ── queries ─────────────────────────────────────────────────────────────

    def _collect_hits(self, hits: Any, top_k: int, exclude: set[str]) -> list[tuple[str, float]]:
        results: list[tuple[str, float]] = []
        for hit in hits or []:
            nid = hit.get("id")
            if nid is None:
                continue
            nid = str(nid)
            if nid in exclude:
                continue
            results.append((nid, float(hit.get("score", 0.0))))
            if len(results) >= top_k:
                break
        return results

    def search(
        self, query: str, top_k: int = 10, exclude_ids: set[str] | None = None
    ) -> list[tuple[str, float]]:
        """Semantic search: returns list of ``(note_id, similarity_score)``.

        Encodes ``query`` with the SHARED, already-loaded model2vec adapter and
        runs a pure :meth:`vector_search`, instead of kglite's native
        ``search_text``. ``search_text`` re-instantiates the ~120MB model2vec model
        on EVERY call — so a cross-store query that fans out over N boxes paid N
        cold model loads per query (the dominant warm-query cost at scale). The
        query is embedded once here with the cached adapter (idempotent load) into
        the exact vector space the note vectors were built in (both go through
        ``self._embedder.embed``), so results are equivalent without the reload.
        A query that embeds to nothing / a zero-norm vector (OOV, e.g. symbol-only)
        yields no semantic hits (the keyword channel drives), matching the prior
        contract that a directionless vector carries no similarity signal.
        """
        with self._lock:
            if self._graph is None or self._embedder is None or not self._node_ids:
                return []
            qvec = self._embed_query(query)
            if qvec is None:
                return []
            return self._search_vector_locked(qvec, top_k, exclude_ids)

    def _embed_query(self, query: str) -> "list[float] | None":
        """Encode a query string to a vector via the cached adapter (no reload).

        Returns ``None`` when embeddings are disabled or the query maps to a
        missing / zero-norm (directionless) vector. Assumes ``self._lock`` is held.
        """
        if self._embedder is None:
            return None
        try:
            vecs = self._embedder.embed([query])
        except Exception as exc:
            logger.warning("Query embed failed for box '%s': %s", self.box_name, exc)
            return None
        qvec = list(vecs[0]) if vecs else None
        if not qvec or _is_zero_norm(qvec):
            return None
        return qvec

    def _search_vector_locked(
        self, vec: list[float], top_k: int, exclude_ids: set[str] | None
    ) -> list[tuple[str, float]]:
        """Nearest-note vector search for a pre-encoded query. Assumes lock held."""
        if self._graph is None or not self._node_ids:
            return []
        exclude = set(exclude_ids or set())
        want = top_k + len(exclude)
        try:
            hits = self._graph.select(NODE_TYPE).vector_search(TEXT_FIELD, vec, top_k=want)
        except Exception as exc:
            logger.warning("Semantic search failed for box '%s': %s", self.box_name, exc)
            return []
        return self._collect_hits(hits, top_k, exclude)

    def search_vector(
        self, vec: list[float], top_k: int = 10, exclude_ids: set[str] | None = None
    ) -> list[tuple[str, float]]:
        """Semantic search for a PRE-ENCODED query vector (skip re-embedding).

        Lets a caller fanning out over many boxes embed the query ONCE (see
        :meth:`embed_query`) and reuse the vector across every box, so the model is
        touched a single time per query rather than once per box. The vector must
        come from the same model2vec space the index was built in.
        """
        with self._lock:
            if self._embedder is None:
                return []
            return self._search_vector_locked(vec, top_k, exclude_ids)

    def embed_query(self, query: str) -> "list[float] | None":
        """Public one-shot query encoder (shared model2vec space); ``None`` if empty.

        Pair with :meth:`search_vector` to embed a query once and fan it out over
        many boxes without re-encoding per box.
        """
        with self._lock:
            return self._embed_query(query)

    def find_similar(
        self, note_id: str, top_k: int = 10, exclude_ids: set[str] | None = None
    ) -> list[tuple[str, float]]:
        """Find notes most similar to a given note."""
        with self._lock:
            if self._graph is None or self._embedder is None:
                return []
            vec = self.get_vector(note_id)
            if vec is None:
                return []
            exclude = set(exclude_ids or set()) | {note_id}
            want = top_k + len(exclude)
            try:
                hits = self._graph.select(NODE_TYPE).vector_search(TEXT_FIELD, vec, top_k=want)
            except Exception as exc:
                logger.warning("Similarity search failed for box '%s': %s", self.box_name, exc)
                return []
            return self._collect_hits(hits, top_k, exclude)

    def get_vector(self, note_id: str) -> list[float] | None:
        """Return the stored embedding vector for a note, or ``None``.

        Stable accessor for callers (e.g. cross-graph similarity) that need raw
        vectors without reaching into private state.
        """
        with self._lock:
            if self._graph is not None:
                try:
                    vec = self._graph.embedding(NODE_TYPE, TEXT_FIELD, note_id)
                    if vec is not None:
                        return list(vec)
                except Exception:
                    pass
            cached = self._cached_vectors.get(note_id)
            return list(cached) if cached is not None else None

    def get_vectors(self, note_ids: "Iterable[str]") -> dict[str, list[float]]:
        """Batch :meth:`get_vector`: ``{note_id: vector}`` for the ids that have one.

        Amortizes the lock acquisition across the whole set (the reranker's MMR
        pass fetches a vector per candidate; taking the lock once per box instead
        of once per candidate is the point). Missing / unembedded notes are simply
        absent from the result — never an error.
        """
        out: dict[str, list[float]] = {}
        with self._lock:
            for note_id in note_ids:
                if self._graph is not None:
                    try:
                        vec = self._graph.embedding(NODE_TYPE, TEXT_FIELD, note_id)
                        if vec is not None:
                            out[note_id] = list(vec)
                            continue
                    except Exception:
                        pass
                cached = self._cached_vectors.get(note_id)
                if cached is not None:
                    out[note_id] = list(cached)
        return out

    def get_citation_vector(self, citation_id: str) -> list[float] | None:
        """Return the stored embedding vector for a cited work, or ``None``.

        Mirrors :meth:`get_vector` for the citation node type so callers can
        fetch a citation's raw vector for comparison against note vectors. A
        zero-norm vector is never returned (it would be useless for cosine
        comparison and signals OOV garbage that should never have been stored),
        so Wave B can treat a non-``None`` result as directly comparable.
        """
        with self._lock:
            if self._graph is not None:
                try:
                    vec = self._graph.embedding(CITATION_NODE_TYPE, TEXT_FIELD, citation_id)
                    if vec is not None:
                        vec = list(vec)
                        return None if _is_zero_norm(vec) else vec
                except Exception:
                    pass
            cached = self._cited_vectors.get(citation_id)
            if cached is None or _is_zero_norm(cached):
                return None
            return list(cached)

    @property
    def indexed_count(self) -> int:
        return len(self._node_ids)

    @property
    def citation_count(self) -> int:
        return len(self._citation_ids)

    @property
    def citation_ids(self) -> set[str]:
        """The citation ids currently live in this index (a defensive copy).

        Exposes the rehydrated/indexed citation set so a caller can reconcile it
        against the authoritative live ``_citations`` source — diffing this
        against the current citation ids to find ghosts whose yaml was deleted
        while the index was cold (see :meth:`remove_citation`). A copy so the
        caller can iterate while mutating the index.
        """
        with self._lock:
            return set(self._citation_ids)

    def needs_indexing(self, note_ids: set[str]) -> set[str]:
        """Return note IDs that aren't in the index yet."""
        return set(note_ids) - self._node_ids

citation_ids property

citation_ids: set[str]

The citation ids currently live in this index (a defensive copy).

Exposes the rehydrated/indexed citation set so a caller can reconcile it against the authoritative live _citations source — diffing this against the current citation ids to find ghosts whose yaml was deleted while the index was cold (see :meth:remove_citation). A copy so the caller can iterate while mutating the index.

load_cache

load_cache() -> None

Start a fresh in-memory graph and recover prior vectors.

The graph itself is never treated as authoritative: it is rebuilt from the live notes by :meth:index_notes. We only salvage the embedding vectors (keyed by note id) so unchanged notes don't have to be re-embedded.

Source code in zettelkasten/embeddings.py

def load_cache(self) -> None:
    """Start a fresh in-memory graph and recover prior vectors.

    The graph itself is never treated as authoritative: it is rebuilt from
    the live notes by :meth:`index_notes`. We only salvage the embedding
    vectors (keyed by note id) so unchanged notes don't have to be
    re-embedded.
    """
    import kglite

    with self._lock:
        self._graph = kglite.KnowledgeGraph()
        self._node_ids = set()
        self._cached_vectors = {}
        self._cached_hashes = {}
        self._note_hashes = {}
        self._citation_ids = set()
        self._cited_vectors = {}
        self._cited_hashes = {}
        self._citation_skipped_hashes = {}
        # Full citation node rows (id/title/text/hash) recovered from the
        # prior .kgl, kept aside so we can re-hydrate citation NODES into
        # the fresh graph after the dimension check below.
        cited_rows: list[dict] = []

        if self._embedder is not None:
            try:
                self._graph.set_embedder(self._embedder)
            except Exception as exc:
                logger.warning("Could not register embedder for box '%s': %s", self.box_name, exc)

        # Recover vectors (and the hashes they were computed against) from
        # the prior .kgl cache, if any.
        if self.cache_path.exists():
            try:
                prior = kglite.load(str(self.cache_path))
                stored = prior.embeddings(NODE_TYPE, TEXT_FIELD) or {}
                self._cached_vectors = {str(k): list(v) for k, v in stored.items()}
                try:
                    rows = prior.cypher(
                        f"MATCH (n:{NODE_TYPE}) RETURN n.id AS id, n.{HASH_FIELD} AS h"
                    )
                    for row in rows or []:
                        nid = row.get("id")
                        h = row.get("h")
                        if nid is not None and h:
                            self._cached_hashes[str(nid)] = str(h)
                except Exception as exc:
                    logger.debug("Could not read cached hashes for box '%s': %s", self.box_name, exc)
                # Recover citation vectors + hashes the same way (separate
                # node type, same .kgl). Absent when no citations were ever
                # embedded into this box.
                try:
                    stored_cit = prior.embeddings(CITATION_NODE_TYPE, TEXT_FIELD) or {}
                    self._cited_vectors = {str(k): list(v) for k, v in stored_cit.items()}
                    rows = prior.cypher(
                        f"MATCH (n:{CITATION_NODE_TYPE}) RETURN n.id AS id, "
                        f"n.title AS title, n.{TEXT_FIELD} AS text, n.{HASH_FIELD} AS h"
                    )
                    for row in rows or []:
                        cid = row.get("id")
                        h = row.get("h")
                        if cid is None:
                            continue
                        cid = str(cid)
                        if h:
                            self._cited_hashes[cid] = str(h)
                        cited_rows.append(
                            {
                                "id": cid,
                                "title": row.get("title"),
                                "text": row.get("text"),
                                "h": str(h) if h else None,
                            }
                        )
                except Exception as exc:
                    logger.debug("Could not read cached citations for box '%s': %s", self.box_name, exc)
            except Exception as exc:
                logger.warning("Could not load zettel cache '%s': %s", self.cache_path, exc)

        # One-time back-compat import of the legacy JSON blob when there is
        # no .kgl yet. These vectors carry no hash, so they can't be verified
        # against the current note text; _embed_pending re-embeds them once
        # on upgrade so every vector ends up hash-tracked.
        if not self._cached_vectors and self.legacy_cache_path.exists():
            try:
                data = json.loads(self.legacy_cache_path.read_text(encoding="utf-8"))
                self._cached_vectors = {
                    str(k): list(v) for k, v in (data.get("vectors") or {}).items()
                }
            except (OSError, ValueError, TypeError) as exc:
                logger.warning("Could not read legacy embeddings '%s': %s", self.legacy_cache_path, exc)

        # Drop cached vectors whose dimension no longer matches the active
        # embedder (a model/dimension upgrade). Restoring them would mix
        # dimensions in one kglite store and corrupt search; clearing the
        # cache makes index_notes re-embed every note fresh at the new dim.
        self._discard_mismatched_cached_vectors()

        # Re-hydrate citation NODES into the fresh graph so graph-native
        # Wave B queries (``select(CITATION_NODE_TYPE).vector_search``) and
        # ``citation_count`` reflect cached citations WITHOUT requiring
        # ``index_citations`` to be re-run first. Notes are rebuilt from the
        # live .md sources by ``index_notes``; citations have no equivalent
        # live-reload step here, so we restore them straight from the .kgl.
        # Runs after the dimension check so cleared (mismatched) vectors are
        # not re-hydrated.
        self._rehydrate_citations(cited_rows)

save_cache

save_cache() -> None

Persist the kglite graph atomically (temp file + rename).

The tmp filename is made unique per write (pid + uuid) because the MCP server and the dashboard backend share one <box>.kgl; a fixed tmp path would let two concurrent saves clobber each other into a torn file. The cache is disposable, so a lost write is fine — a corrupt one is not.

Source code in zettelkasten/embeddings.py

def save_cache(self) -> None:
    """Persist the kglite graph atomically (temp file + rename).

    The tmp filename is made unique per write (pid + uuid) because the MCP
    server and the dashboard backend share one ``<box>.kgl``; a fixed tmp
    path would let two concurrent saves clobber each other into a torn file.
    The cache is disposable, so a lost write is fine — a corrupt one is not.
    """
    with self._lock:
        if self._graph is None:
            return
        tmp = self.cache_path.with_suffix(f".kgl.{os.getpid()}.{uuid.uuid4().hex}.tmp")
        try:
            self.cache_path.parent.mkdir(parents=True, exist_ok=True)
            self._graph.save(str(tmp))
            os.replace(tmp, self.cache_path)
        except Exception as exc:
            logger.warning("Could not save zettel cache '%s': %s", self.cache_path, exc)
            try:
                tmp.unlink(missing_ok=True)
            except OSError:
                pass

save_cache_debounced

save_cache_debounced() -> None

Schedule a debounced :meth:save_cache, coalescing rapid saves.

Called on the save_note write path instead of persisting the whole .kgl synchronously on every note write. Rapid successive saves reset a single per-index timer, so a burst of writes collapses into ONE cache rewrite once the box goes quiet (see :data:_SAVE_DEBOUNCE_SEC).

Safety: the incremental embed already ran synchronously in :meth:index_note, so the new vector is live in-memory the instant this returns (an immediate get_vector/search sees it) — only the on-disk persist is deferred. The .kgl is a disposable cache rebuilt from the live .md sources on load, and the caller writes that source of truth synchronously BEFORE scheduling here, so a coalesced or even crash-dropped persist can never diverge from a full rebuild. The fired save serializes the CURRENT live graph under _lock and writes it via the same atomic temp+rename as :meth:save_cache, so an interleaved synchronous saver (delete_note/remove_citation) and this timer never tear the file. The daemon timer never blocks interpreter exit; a best-effort atexit flush persists anything still pending.

Source code in zettelkasten/embeddings.py

def save_cache_debounced(self) -> None:
    """Schedule a debounced :meth:`save_cache`, coalescing rapid saves.

    Called on the ``save_note`` write path instead of persisting the whole
    ``.kgl`` synchronously on every note write. Rapid successive saves reset
    a single per-index timer, so a burst of writes collapses into ONE cache
    rewrite once the box goes quiet (see :data:`_SAVE_DEBOUNCE_SEC`).

    Safety: the incremental embed already ran synchronously in
    :meth:`index_note`, so the new vector is live in-memory the instant this
    returns (an immediate ``get_vector``/``search`` sees it) — only the
    on-disk persist is deferred. The ``.kgl`` is a disposable cache rebuilt
    from the live ``.md`` sources on load, and the caller writes that source
    of truth synchronously BEFORE scheduling here, so a coalesced or even
    crash-dropped persist can never diverge from a full rebuild. The fired
    save serializes the CURRENT live graph under ``_lock`` and writes it via
    the same atomic temp+rename as :meth:`save_cache`, so an interleaved
    synchronous saver (``delete_note``/``remove_citation``) and this timer
    never tear the file. The daemon timer never blocks interpreter exit; a
    best-effort ``atexit`` flush persists anything still pending.
    """
    with self._save_timer_lock:
        if self._save_timer is not None:
            self._save_timer.cancel()
        self._save_timer_gen += 1
        gen = self._save_timer_gen
        timer = threading.Timer(
            _SAVE_DEBOUNCE_SEC, self._fire_debounced_save, args=(gen,)
        )
        timer.daemon = True
        self._save_timer = timer
        _PENDING_SAVE_INDEXES.add(self)
        timer.start()

flush_cache

flush_cache() -> None

Cancel any pending debounced save and persist immediately.

Idempotent and safe to call whether or not a save is pending: with no timer it is just a plain :meth:save_cache. Used by the atexit flush so a scheduled-but-unfired persist is not wasted on a clean shutdown.

Source code in zettelkasten/embeddings.py

def flush_cache(self) -> None:
    """Cancel any pending debounced save and persist immediately.

    Idempotent and safe to call whether or not a save is pending: with no
    timer it is just a plain :meth:`save_cache`. Used by the ``atexit`` flush
    so a scheduled-but-unfired persist is not wasted on a clean shutdown.
    """
    with self._save_timer_lock:
        if self._save_timer is not None:
            self._save_timer.cancel()
            self._save_timer = None
        # Invalidate any timer already mid-fire (its cancel() was a no-op) so
        # it sees the token mismatch and does not re-null the handle or
        # re-prune after we persist. We flush the pending write here, so this
        # index is no longer pending: drop it from the atexit registry.
        self._save_timer_gen += 1
        _PENDING_SAVE_INDEXES.discard(self)
    self.save_cache()

index_notes

index_notes(notes: list['Note']) -> None

(Re)build the graph from notes and embed any not yet embedded.

Notes whose type is in :data:EMBED_SKIP_TYPES (quotes) are dropped up front: they are supporting evidence reached structurally, so embedding them is pure wasted compute (~half the corpus). They never enter this semantic index — not as nodes, vectors, edges, or cluster members.

Source code in zettelkasten/embeddings.py

def index_notes(self, notes: list["Note"]) -> None:
    """(Re)build the graph from ``notes`` and embed any not yet embedded.

    Notes whose type is in :data:`EMBED_SKIP_TYPES` (quotes) are dropped up
    front: they are supporting evidence reached structurally, so embedding
    them is pure wasted compute (~half the corpus). They never enter this
    semantic index — not as nodes, vectors, edges, or cluster members.
    """
    notes = [n for n in notes if n.type not in EMBED_SKIP_TYPES]
    if not notes:
        return
    with self._lock:
        if not self._add_nodes(notes):
            return
        self._add_edges(notes)
        self._embed_pending()

index_note

index_note(note: 'Note') -> None

Add/update a single note and (re)embed it.

Unlike the bulk path, this always recomputes the note's vector so an edit to an existing note refreshes its embedding instead of keeping a stale cached one. Types in :data:EMBED_SKIP_TYPES (quotes) are skipped entirely — see :meth:index_notes.

Source code in zettelkasten/embeddings.py

def index_note(self, note: "Note") -> None:
    """Add/update a single note and (re)embed it.

    Unlike the bulk path, this always recomputes the note's vector so an
    edit to an existing note refreshes its embedding instead of keeping a
    stale cached one. Types in :data:`EMBED_SKIP_TYPES` (quotes) are skipped
    entirely — see :meth:`index_notes`.
    """
    if note.type in EMBED_SKIP_TYPES:
        return
    with self._lock:
        if not self._add_nodes([note]):
            return
        self._add_edges([note])
        if self._embedder is None or self._graph is None:
            return
        try:
            vecs = self._embedder.embed([note_to_embed_text(note)])
        except Exception as exc:
            logger.warning("Could not embed note '%s': %s", note.id, exc)
            return
        if not vecs:
            return
        vec = list(vecs[0])
        try:
            self._graph.add_embeddings(NODE_TYPE, TEXT_FIELD, {note.id: vec})
            self._cached_vectors[note.id] = vec
            self._cached_hashes[note.id] = self._note_hashes.get(
                note.id, _text_hash(note_to_embed_text(note))
            )
        except Exception as exc:
            logger.warning("Could not store vector for note '%s': %s", note.id, exc)

index_citations

index_citations(citations: dict[str, dict]) -> None

Embed cited-but-unowned works (title + abstract) into this index.

citations maps citation id -> _citations/*.yaml payload (the shape returned by :func:zettelkasten.graph.load_citations). Each work is embedded from :func:citation_to_embed_text; works with no embeddable text (no title and no abstract) are skipped rather than embedded as empty vectors. Citation nodes share the box's embedder and .kgl cache with notes but live under :data:CITATION_NODE_TYPE, so their vectors are comparable to note vectors yet never surface in note search.

Re-embedding is driven by a content hash so the same long-lived index stays correct without a reload. A work is re-embedded whenever its freshly-computed hash differs from the hash its currently-tracked vector was computed against — this catches BOTH out-of-band edits seen across a reload AND in-session edits (re-calling with an edited title/abstract on the same index), since citations have no singular force-recompute path like :meth:index_note. Unchanged works are restored from their tracked vector and never re-embedded. Out-of-vocabulary-but-nonempty text that embeds to a zero-norm vector is dropped (never stored or counted) so it cannot pollute downstream cosine similarity.

A work whose text changes to OOV/unstorable evicts any prior node and vector for that id (so :meth:get_citation_vector returns None and :attr:citation_count drops) and is negative-cached by content hash, so an unchanged OOV work short-circuits on later calls rather than re-embedding-and-dropping every pass. Changing such text back to embeddable re-embeds it and makes it retrievable again.

This call is purely additive/refresh — it NEVER prunes ids that are absent from citations. Wave B may pass a SUBSET of the relevant works for a box per call, so omitting an id must not delete it; explicit removal goes through :meth:remove_citation.

Source code in zettelkasten/embeddings.py

def index_citations(self, citations: dict[str, dict]) -> None:
    """Embed cited-but-unowned works (``title + abstract``) into this index.

    ``citations`` maps citation id -> ``_citations/*.yaml`` payload (the
    shape returned by :func:`zettelkasten.graph.load_citations`). Each work
    is embedded from :func:`citation_to_embed_text`; works with no
    embeddable text (no title and no abstract) are skipped rather than
    embedded as empty vectors. Citation nodes share the box's embedder and
    ``.kgl`` cache with notes but live under :data:`CITATION_NODE_TYPE`, so
    their vectors are comparable to note vectors yet never surface in note
    search.

    Re-embedding is driven by a content hash so the same long-lived index
    stays correct without a reload. A work is re-embedded whenever its
    freshly-computed hash differs from the hash its currently-tracked vector
    was computed against — this catches BOTH out-of-band edits seen across a
    reload AND in-session edits (re-calling with an edited title/abstract on
    the same index), since citations have no singular force-recompute path
    like :meth:`index_note`. Unchanged works are restored from their tracked
    vector and never re-embedded. Out-of-vocabulary-but-nonempty text that
    embeds to a zero-norm vector is dropped (never stored or counted) so it
    cannot pollute downstream cosine similarity.

    A work whose text changes to OOV/unstorable evicts any prior node and
    vector for that id (so :meth:`get_citation_vector` returns ``None`` and
    :attr:`citation_count` drops) and is negative-cached by content hash, so
    an unchanged OOV work short-circuits on later calls rather than
    re-embedding-and-dropping every pass. Changing such text back to
    embeddable re-embeds it and makes it retrievable again.

    This call is purely additive/refresh — it NEVER prunes ids that are
    absent from ``citations``. Wave B may pass a SUBSET of the relevant
    works for a box per call, so omitting an id must not delete it; explicit
    removal goes through :meth:`remove_citation`.
    """
    if not citations:
        return
    import pandas as pd

    with self._lock:
        if self._graph is None:
            self.load_cache()

        # Build the embed text + content hash for every work with something
        # embeddable; works with neither title nor abstract are skipped.
        pending: dict[str, tuple[str, str]] = {}
        for cid, data in citations.items():
            if not isinstance(data, dict):
                continue
            text = citation_to_embed_text(data)
            if not text:
                continue
            pending[str(cid)] = (text, _text_hash(text))
        if not pending:
            return

        # Partition into unchanged (restore the tracked vector, no re-embed)
        # and changed/new (must be embedded now). "Unchanged" requires a
        # tracked, non-garbage vector whose hash still matches the current
        # text; anything else — an edit (in-session or cross-reload) or a
        # first sighting — is re-embedded.
        restore: dict[str, list[float]] = {}
        to_embed: dict[str, str] = {}
        for cid, (text, h) in pending.items():
            vec = self._cited_vectors.get(cid)
            if vec is not None and self._cited_hashes.get(cid) == h and not _is_zero_norm(vec):
                restore[cid] = vec
            elif self._citation_skipped_hashes.get(cid) == h:
                # Negative-cache hit: this exact text already embedded to a
                # zero-norm vector and was dropped. Skip it — don't re-embed
                # (and don't store) the same garbage every pass.
                continue
            else:
                to_embed[cid] = text

        # Embed the changed/new works in one batch, dropping any that embed
        # to a zero-norm vector (OOV-but-nonempty text under model2vec).
        embedded: dict[str, list[float]] = {}
        dropped: dict[str, str] = {}
        if to_embed and self._embedder is not None:
            cids = list(to_embed)
            try:
                vecs = self._embedder.embed([to_embed[c] for c in cids])
            except Exception as exc:
                logger.warning("Could not embed citations for box '%s': %s", self.box_name, exc)
                vecs = []
            for cid, vec in zip(cids, vecs):
                vec = list(vec)
                if _is_zero_norm(vec):
                    logger.debug(
                        "Box '%s': citation '%s' has out-of-vocabulary text that embeds "
                        "to a zero-norm vector; skipping it.", self.box_name, cid,
                    )
                    dropped[cid] = pending[cid][1]
                    continue
                embedded[cid] = vec

        # An id that was targeted for re-embed but produced no storable
        # vector must not strand stale per-id state: evict any prior node +
        # vector (so get_citation_vector returns None and citation_count
        # drops) and negative-cache its hash so an unchanged OOV work
        # short-circuits next time. This runs BEFORE the empty-final return
        # so a good->OOV edit can't keep serving the stale pre-edit vector.
        for cid, h in dropped.items():
            self._evict_citation_node(cid)
            self._citation_skipped_hashes[cid] = h

        # Final set with a usable vector: restored + freshly embedded.
        final: dict[str, list[float]] = {**restore, **embedded}
        if not final:
            return

        rows = [self._citation_row(cid, pending[cid][0]) for cid in final]
        try:
            self._graph.add_nodes(
                pd.DataFrame(rows),
                node_type=CITATION_NODE_TYPE,
                unique_id_field="id",
                node_title_field="title",
                conflict_handling="update",
            )
            self._graph.add_embeddings(CITATION_NODE_TYPE, TEXT_FIELD, final)
        except Exception as exc:
            logger.warning("Could not add citations to box '%s': %s", self.box_name, exc)
            return
        for cid, vec in final.items():
            h = pending[cid][1]
            self._citation_ids.add(cid)
            self._cited_vectors[cid] = vec
            self._cited_hashes[cid] = h
            # Text that was previously OOV is now embeddable again: clear
            # its negative-cache entry so it is retrievable from here on.
            self._citation_skipped_hashes.pop(cid, None)

remove_citation

remove_citation(cid: str) -> None

Drop a cited work's node + all tracked state from graph and caches.

Mirrors :meth:remove_note for the citation node type. This is the ONLY deletion path for citations: :meth:index_citations never prunes ids it wasn't passed, so a citation that should disappear from :attr:citation_count and graph-native CITATION_NODE_TYPE search must be removed here. The node is DETACH DELETEd from the live graph so a subsequent save_cache persists the removal and a later load_cache will not re-hydrate it.

Source code in zettelkasten/embeddings.py

def remove_citation(self, cid: str) -> None:
    """Drop a cited work's node + all tracked state from graph and caches.

    Mirrors :meth:`remove_note` for the citation node type. This is the ONLY
    deletion path for citations: :meth:`index_citations` never prunes ids it
    wasn't passed, so a citation that should disappear from
    :attr:`citation_count` and graph-native ``CITATION_NODE_TYPE`` search
    must be removed here. The node is DETACH DELETEd from the live graph so
    a subsequent ``save_cache`` persists the removal and a later
    ``load_cache`` will not re-hydrate it.
    """
    with self._lock:
        self._citation_skipped_hashes.pop(cid, None)
        self._evict_citation_node(cid)

remove_note

remove_note(note_id: str) -> None

Drop a note's node (and edges) from the live graph and caches.

Source code in zettelkasten/embeddings.py

def remove_note(self, note_id: str) -> None:
    """Drop a note's node (and edges) from the live graph and caches."""
    with self._lock:
        self._cached_vectors.pop(note_id, None)
        self._cached_hashes.pop(note_id, None)
        self._note_hashes.pop(note_id, None)
        self._node_ids.discard(note_id)
        if self._graph is None:
            return
        try:
            self._graph.cypher(
                f"MATCH (n:{NODE_TYPE}) WHERE n.id = $id DETACH DELETE n",
                params={"id": note_id},
            )
        except Exception as exc:
            logger.debug("Could not delete node '%s' from box '%s': %s", note_id, self.box_name, exc)

search

search(query: str, top_k: int = 10, exclude_ids: set[str] | None = None) -> list[tuple[str, float]]

Semantic search: returns list of (note_id, similarity_score).

Encodes query with the SHARED, already-loaded model2vec adapter and runs a pure :meth:vector_search, instead of kglite's native search_text. search_text re-instantiates the ~120MB model2vec model on EVERY call — so a cross-store query that fans out over N boxes paid N cold model loads per query (the dominant warm-query cost at scale). The query is embedded once here with the cached adapter (idempotent load) into the exact vector space the note vectors were built in (both go through self._embedder.embed), so results are equivalent without the reload. A query that embeds to nothing / a zero-norm vector (OOV, e.g. symbol-only) yields no semantic hits (the keyword channel drives), matching the prior contract that a directionless vector carries no similarity signal.

Source code in zettelkasten/embeddings.py

def search(
    self, query: str, top_k: int = 10, exclude_ids: set[str] | None = None
) -> list[tuple[str, float]]:
    """Semantic search: returns list of ``(note_id, similarity_score)``.

    Encodes ``query`` with the SHARED, already-loaded model2vec adapter and
    runs a pure :meth:`vector_search`, instead of kglite's native
    ``search_text``. ``search_text`` re-instantiates the ~120MB model2vec model
    on EVERY call — so a cross-store query that fans out over N boxes paid N
    cold model loads per query (the dominant warm-query cost at scale). The
    query is embedded once here with the cached adapter (idempotent load) into
    the exact vector space the note vectors were built in (both go through
    ``self._embedder.embed``), so results are equivalent without the reload.
    A query that embeds to nothing / a zero-norm vector (OOV, e.g. symbol-only)
    yields no semantic hits (the keyword channel drives), matching the prior
    contract that a directionless vector carries no similarity signal.
    """
    with self._lock:
        if self._graph is None or self._embedder is None or not self._node_ids:
            return []
        qvec = self._embed_query(query)
        if qvec is None:
            return []
        return self._search_vector_locked(qvec, top_k, exclude_ids)

search_vector

search_vector(vec: list[float], top_k: int = 10, exclude_ids: set[str] | None = None) -> list[tuple[str, float]]

Semantic search for a PRE-ENCODED query vector (skip re-embedding).

Lets a caller fanning out over many boxes embed the query ONCE (see :meth:embed_query) and reuse the vector across every box, so the model is touched a single time per query rather than once per box. The vector must come from the same model2vec space the index was built in.

Source code in zettelkasten/embeddings.py

def search_vector(
    self, vec: list[float], top_k: int = 10, exclude_ids: set[str] | None = None
) -> list[tuple[str, float]]:
    """Semantic search for a PRE-ENCODED query vector (skip re-embedding).

    Lets a caller fanning out over many boxes embed the query ONCE (see
    :meth:`embed_query`) and reuse the vector across every box, so the model is
    touched a single time per query rather than once per box. The vector must
    come from the same model2vec space the index was built in.
    """
    with self._lock:
        if self._embedder is None:
            return []
        return self._search_vector_locked(vec, top_k, exclude_ids)

embed_query

embed_query(query: str) -> 'list[float] | None'

Public one-shot query encoder (shared model2vec space); None if empty.

Pair with :meth:search_vector to embed a query once and fan it out over many boxes without re-encoding per box.

Source code in zettelkasten/embeddings.py

def embed_query(self, query: str) -> "list[float] | None":
    """Public one-shot query encoder (shared model2vec space); ``None`` if empty.

    Pair with :meth:`search_vector` to embed a query once and fan it out over
    many boxes without re-encoding per box.
    """
    with self._lock:
        return self._embed_query(query)

find_similar

find_similar(note_id: str, top_k: int = 10, exclude_ids: set[str] | None = None) -> list[tuple[str, float]]

Find notes most similar to a given note.

Source code in zettelkasten/embeddings.py

def find_similar(
    self, note_id: str, top_k: int = 10, exclude_ids: set[str] | None = None
) -> list[tuple[str, float]]:
    """Find notes most similar to a given note."""
    with self._lock:
        if self._graph is None or self._embedder is None:
            return []
        vec = self.get_vector(note_id)
        if vec is None:
            return []
        exclude = set(exclude_ids or set()) | {note_id}
        want = top_k + len(exclude)
        try:
            hits = self._graph.select(NODE_TYPE).vector_search(TEXT_FIELD, vec, top_k=want)
        except Exception as exc:
            logger.warning("Similarity search failed for box '%s': %s", self.box_name, exc)
            return []
        return self._collect_hits(hits, top_k, exclude)

get_vector

get_vector(note_id: str) -> list[float] | None

Return the stored embedding vector for a note, or None.

Stable accessor for callers (e.g. cross-graph similarity) that need raw vectors without reaching into private state.

Source code in zettelkasten/embeddings.py

def get_vector(self, note_id: str) -> list[float] | None:
    """Return the stored embedding vector for a note, or ``None``.

    Stable accessor for callers (e.g. cross-graph similarity) that need raw
    vectors without reaching into private state.
    """
    with self._lock:
        if self._graph is not None:
            try:
                vec = self._graph.embedding(NODE_TYPE, TEXT_FIELD, note_id)
                if vec is not None:
                    return list(vec)
            except Exception:
                pass
        cached = self._cached_vectors.get(note_id)
        return list(cached) if cached is not None else None

get_vectors

get_vectors(note_ids: 'Iterable[str]') -> dict[str, list[float]]

Batch :meth:get_vector: {note_id: vector} for the ids that have one.

Amortizes the lock acquisition across the whole set (the reranker's MMR pass fetches a vector per candidate; taking the lock once per box instead of once per candidate is the point). Missing / unembedded notes are simply absent from the result — never an error.

Source code in zettelkasten/embeddings.py

def get_vectors(self, note_ids: "Iterable[str]") -> dict[str, list[float]]:
    """Batch :meth:`get_vector`: ``{note_id: vector}`` for the ids that have one.

    Amortizes the lock acquisition across the whole set (the reranker's MMR
    pass fetches a vector per candidate; taking the lock once per box instead
    of once per candidate is the point). Missing / unembedded notes are simply
    absent from the result — never an error.
    """
    out: dict[str, list[float]] = {}
    with self._lock:
        for note_id in note_ids:
            if self._graph is not None:
                try:
                    vec = self._graph.embedding(NODE_TYPE, TEXT_FIELD, note_id)
                    if vec is not None:
                        out[note_id] = list(vec)
                        continue
                except Exception:
                    pass
            cached = self._cached_vectors.get(note_id)
            if cached is not None:
                out[note_id] = list(cached)
    return out

get_citation_vector

get_citation_vector(citation_id: str) -> list[float] | None

Return the stored embedding vector for a cited work, or None.

Mirrors :meth:get_vector for the citation node type so callers can fetch a citation's raw vector for comparison against note vectors. A zero-norm vector is never returned (it would be useless for cosine comparison and signals OOV garbage that should never have been stored), so Wave B can treat a non-None result as directly comparable.

Source code in zettelkasten/embeddings.py

def get_citation_vector(self, citation_id: str) -> list[float] | None:
    """Return the stored embedding vector for a cited work, or ``None``.

    Mirrors :meth:`get_vector` for the citation node type so callers can
    fetch a citation's raw vector for comparison against note vectors. A
    zero-norm vector is never returned (it would be useless for cosine
    comparison and signals OOV garbage that should never have been stored),
    so Wave B can treat a non-``None`` result as directly comparable.
    """
    with self._lock:
        if self._graph is not None:
            try:
                vec = self._graph.embedding(CITATION_NODE_TYPE, TEXT_FIELD, citation_id)
                if vec is not None:
                    vec = list(vec)
                    return None if _is_zero_norm(vec) else vec
            except Exception:
                pass
        cached = self._cited_vectors.get(citation_id)
        if cached is None or _is_zero_norm(cached):
            return None
        return list(cached)

needs_indexing

needs_indexing(note_ids: set[str]) -> set[str]

Return note IDs that aren't in the index yet.

Source code in zettelkasten/embeddings.py

def needs_indexing(self, note_ids: set[str]) -> set[str]:
    """Return note IDs that aren't in the index yet."""
    return set(note_ids) - self._node_ids

set_embedder_factory

set_embedder_factory(factory: Callable[[], Any] | None) -> None

Register (or clear) the embedder factory used by new EmbeddingIndexes.

Pass a zero-arg callable returning an embedder or None to disable; pass None to restore the production default (model2vec).

Source code in zettelkasten/embeddings.py

def set_embedder_factory(factory: Callable[[], Any] | None) -> None:
    """Register (or clear) the embedder factory used by new ``EmbeddingIndex``es.

    Pass a zero-arg callable returning an embedder or ``None`` to disable;
    pass ``None`` to restore the production default (model2vec).
    """
    global _embedder_factory, _shared_embedder
    _embedder_factory = factory
    # A factory change (set or clear) must not be shadowed by a previously cached
    # default embedder — drop it so the next build honors the new configuration.
    with _shared_embedder_lock:
        _shared_embedder = None

propose_clusters

propose_clusters(vectors: 'dict[str, list[float] | None]', *, threshold: float = 0.55, max_k: int = 12) -> 'dict[str, list]'

Propose candidate clusters from note vectors — a pure, dependency-light step.

Given {note_id -> vector} this greedily agglomerates notes by cosine similarity. A note joins the existing cluster whose CENTROID it is most similar to, provided that centroid similarity meets threshold; otherwise it seeds a new cluster. Comparing against the centroid (the running mean of the members) rather than against any single member is a cohesion guard: it stops a chain of pairwise-close-but-globally-spread notes (e.g. unit vectors fanned out along an arc, each adjacent pair similar but the endpoints anti-correlated) from collapsing into one sprawling cluster the way naive single linkage would. max_k caps the number of clusters — once the cap is reached a note that would have seeded a new cluster folds into its nearest existing cluster instead, so nothing is ever dropped for want of a slot.

Notes whose vector is missing, empty, or zero-norm (no usable direction for cosine) are NEVER dropped: they are returned in unclustered.

Deterministic: ids are processed in sorted order and both the cluster members and the cluster list itself are returned sorted, so identical input always yields identical output — which keeps any downstream row id derived from a cluster's member set stable across rebuilds.

Returns {"clusters": [[id, ...], ...], "unclustered": [id, ...]}.

Source code in zettelkasten/embeddings.py

def propose_clusters(
    vectors: "dict[str, list[float] | None]",
    *,
    threshold: float = 0.55,
    max_k: int = 12,
) -> "dict[str, list]":
    """Propose candidate clusters from note vectors — a pure, dependency-light step.

    Given ``{note_id -> vector}`` this greedily agglomerates notes by cosine
    similarity. A note joins the existing cluster whose CENTROID it is most
    similar to, provided that centroid similarity meets ``threshold``; otherwise
    it seeds a new cluster. Comparing against the centroid (the running mean of
    the members) rather than against any single member is a cohesion guard: it
    stops a chain of pairwise-close-but-globally-spread notes (e.g. unit vectors
    fanned out along an arc, each adjacent pair similar but the endpoints
    anti-correlated) from collapsing into one sprawling cluster the way naive
    single linkage would. ``max_k`` caps the number of clusters — once the cap is
    reached a note that would have seeded a new cluster folds into its nearest
    existing cluster instead, so nothing is ever dropped for want of a slot.

    Notes whose vector is missing, empty, or zero-norm (no usable direction for
    cosine) are NEVER dropped: they are returned in ``unclustered``.

    Deterministic: ids are processed in sorted order and both the cluster members
    and the cluster list itself are returned sorted, so identical input always
    yields identical output — which keeps any downstream row id derived from a
    cluster's member set stable across rebuilds.

    Returns ``{"clusters": [[id, ...], ...], "unclustered": [id, ...]}``.
    """
    usable: dict[str, list[float]] = {}
    unclustered: list[str] = []
    for nid in sorted(vectors):
        vec = vectors.get(nid)
        if vec is None or _is_zero_norm(vec):
            unclustered.append(nid)
        else:
            usable[nid] = [float(x) for x in vec]

    cap = max(1, int(max_k))
    clusters: list[list[str]] = []
    # Running per-cluster centroids, kept parallel to ``clusters``. Cosine is
    # scale-invariant, so a centroid is represented by the component-wise SUM of
    # its members' vectors (cheaper to maintain incrementally than the mean, and
    # ``_cosine_sim(vec, sum)`` equals ``_cosine_sim(vec, mean)``).
    centroids: list[list[float]] = []
    for nid in sorted(usable):
        vec = usable[nid]
        best_idx = -1
        best_sim = -1.0
        for idx, centroid in enumerate(centroids):
            sim = _cosine_sim(vec, centroid)
            if sim > best_sim:
                best_sim = sim
                best_idx = idx
        if best_idx >= 0 and best_sim >= threshold:
            clusters[best_idx].append(nid)
            _accumulate(centroids[best_idx], vec)
        elif len(clusters) < cap:
            clusters.append([nid])
            centroids.append(list(vec))
        elif best_idx >= 0:
            # max-k reached: fold into the nearest cluster rather than drop it.
            clusters[best_idx].append(nid)
            _accumulate(centroids[best_idx], vec)
        else:  # defensive: cap>=1 guarantees a cluster exists once we're full
            clusters.append([nid])
            centroids.append(list(vec))

    out_clusters = sorted((sorted(c) for c in clusters), key=lambda c: c[0])
    return {"clusters": out_clusters, "unclustered": sorted(unclustered)}

note_to_embed_text

note_to_embed_text(note: 'Note') -> str

Convert a note to the text representation used for embedding.

Source code in zettelkasten/embeddings.py

def note_to_embed_text(note: "Note") -> str:
    """Convert a note to the text representation used for embedding."""
    parts = [note.title]
    if note.aliases:
        parts.append(f"Also known as: {', '.join(note.aliases)}")
    parts.append(note.body)
    if note.tags:
        parts.append(f"Tags: {', '.join(note.tags)}")
    return "\n".join(parts)

citation_to_embed_text

citation_to_embed_text(citation: dict) -> str | None

Text used to embed a cited-but-unowned work: its title + abstract.

citation is a _citations/*.yaml payload as returned by :func:zettelkasten.graph.load_citations; the relevant fields are title and the OpenAlex-reconstructed plain-text abstract.

Returns None only when there is literally nothing to embed (both title and abstract missing/empty after .strip()), so the caller skips the work rather than embedding the empty string. This guards the empty-text case ONLY — non-empty but out-of-vocabulary text (emoji/CJK/symbol-only) still returns a string here and can embed to a zero-norm vector; that garbage-vector case is defended separately at store time in :meth:EmbeddingIndex.index_citations (see :func:_is_zero_norm). Falls back to title-only when the abstract is missing/empty (and to abstract-only in the unusual case of a titleless entry).

Source code in zettelkasten/embeddings.py

def citation_to_embed_text(citation: dict) -> str | None:
    """Text used to embed a cited-but-unowned work: its title + abstract.

    ``citation`` is a ``_citations/*.yaml`` payload as returned by
    :func:`zettelkasten.graph.load_citations`; the relevant fields are ``title``
    and the OpenAlex-reconstructed plain-text ``abstract``.

    Returns ``None`` only when there is *literally nothing* to embed (both
    title and abstract missing/empty after ``.strip()``), so the caller skips
    the work rather than embedding the empty string. This guards the empty-text
    case ONLY — non-empty but out-of-vocabulary text (emoji/CJK/symbol-only)
    still returns a string here and can embed to a zero-norm vector; that
    garbage-vector case is defended separately at store time in
    :meth:`EmbeddingIndex.index_citations` (see :func:`_is_zero_norm`). Falls
    back to title-only when the abstract is missing/empty (and to abstract-only
    in the unusual case of a titleless entry).
    """
    title = str(citation.get("title") or "").strip()
    abstract = str(citation.get("abstract") or "").strip()
    if not title and not abstract:
        return None
    if not abstract:
        return title
    if not title:
        return abstract
    return f"{title}\n{abstract}"