zettelkasten.global_index

zettelkasten.global_index

Global note-level ANN index for the Zettelkasten.

A single combined kglite store over EVERY box's note vectors, so a large-scope query (e.g. "all earnings calls this quarter across my universe") is ONE approximate-nearest-neighbour query instead of an O(#boxes) per-box fan-out. The per-box .kgl indexes in :mod:zettelkasten.embeddings remain the authoritative rebuild source and the small-scope / federated path; this store is a second, disposable derived cache that mirrors them.

Design (mirrors the single memory.kgl precedent):

• One graph at <ANGELO_DIR>/zettel/_global.kgl. Node id is the COMPOSITE f"{box}::{note_id}" because zettel ids are not unique across boxes. Each node carries box, note_id, date (int YYYYMMDD or 0), title, type, search_text and text_hash properties, so a pre-filtered vector search can restrict recall to a project's boxes and/or a date range BEFORE ranking (kglite select().where(...).vector_search(...)).
• The embedder, native lock, embed-text builder and quote-skip set are shared with :mod:zettelkasten.embeddings so a note's global vector is byte-identical to its per-box vector (same model, same text).
• Disposable: the .md files are the source of truth. A corrupt or stale _global.kgl is rebuilt from the live notes by :meth:GlobalIndex.rebuild; text_hash per node lets a rebuild reuse unchanged vectors instead of re-embedding the whole corpus.

GlobalIndex

A single combined kglite store over all boxes' note vectors.

Source code in zettelkasten/global_index.py

class GlobalIndex:
    """A single combined kglite store over all boxes' note vectors."""

    def __init__(self, *, embedder: Any = _UNSET, cache_path: Path | None = None):
        self.cache_path = cache_path or _global_cache_path()
        # Lazy embedder: build only when actually needed (a query, or embedding a
        # vector-less note). ``embedder=_UNSET`` defers; an explicit value (tests)
        # or ``None`` (disabled) is honored immediately.
        self._embedder_pending = embedder is _UNSET
        self._embedder = None if embedder is _UNSET else embedder
        self._graph: Any = None
        # Composite id -> content hash the currently-indexed vector was built
        # from, so a rebuild/mirror reuses unchanged vectors instead of
        # re-embedding (and can skip re-adding unchanged nodes).
        self._note_hashes: dict[str, str] = {}
        self._lock = _NATIVE_LOCK
        self._save_timer: "threading.Timer | None" = None
        self._save_timer_lock = threading.Lock()
        self._save_timer_gen = 0

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

    def load_cache(self) -> None:
        """Load the persisted store as the LIVE, queryable graph.

        Unlike the per-box index (which starts empty and rebuilds every box from
        ``.md`` on load), the global store loads its prior ``.kgl`` directly so it
        is immediately queryable across process restarts without walking the whole
        corpus. ``text_hash`` tracking is recovered so later upserts can skip
        unchanged notes; :meth:`rebuild` is the disposable full-refresh that
        re-derives everything from the live ``.md`` sources (and evicts any node
        for a note deleted while the process was down).
        """
        import kglite

        with self._lock:
            self._note_hashes = {}
            loaded = None
            if self.cache_path.exists():
                try:
                    loaded = kglite.load(str(self.cache_path))
                except Exception as exc:
                    logger.warning("Global index: could not load cache '%s': %s", self.cache_path, exc)
                    loaded = None

            if loaded is not None:
                self._graph = loaded
                self._note_hashes = self._read_hashes(self._graph)
            else:
                self._graph = kglite.KnowledgeGraph()
            # NOTE: search uses ``vector_search`` with an explicit query vector, so
            # the embedder is NOT registered on the graph here — and is not built
            # at load time (see ``_ensure_embedder``), keeping the write mirror
            # embedder-free. A stale vector dimension (model change) is handled the
            # first time the embedder is actually built.

    def _ensure_embedder(self) -> Any:
        """Build the embedder on first real need; drop a stale-dim store once.

        Deferred so the vector-supplying write mirror never builds an embedder.
        On the first build, if the loaded store's vectors don't match the active
        embedder's dimension (a model change), the store is reset to empty so a
        rebuild refills it at the new dimension rather than serving garbage.
        """
        if self._embedder_pending:
            self._embedder = _build_embedder()
            self._embedder_pending = False
            if (
                self._embedder is not None
                and self._graph is not None
                and self._dimension_mismatch(self._graph)
            ):
                import kglite

                logger.info("Global index: stale vector dimension; starting empty (rebuild to refill).")
                self._graph = kglite.KnowledgeGraph()
                self._note_hashes = {}
        return self._embedder

    @staticmethod
    def _read_hashes(graph: Any) -> dict[str, str]:
        out: dict[str, str] = {}
        try:
            rows = graph.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:
                    out[str(nid)] = str(h)
        except Exception as exc:  # pragma: no cover - defensive
            logger.debug("Global index: could not read cached hashes: %s", exc)
        return out

    def _dimension_mismatch(self, graph: Any) -> bool:
        """True when a loaded store's vectors don't match the active embedder dim."""
        want = getattr(self._embedder, "dimension", None)
        if not want:
            return False
        try:
            stored = graph.embeddings(NODE_TYPE, TEXT_FIELD) or {}
        except Exception:  # pragma: no cover - defensive
            return False
        for vec in stored.values():
            return len(list(vec)) != want
        return False

    def _ensure_graph(self) -> None:
        if self._graph is None:
            self.load_cache()

    def save_cache(self) -> None:
        """Persist the kglite graph atomically (temp + rename); disposable."""
        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("Global index: could not save 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 persists."""
        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:
        with self._save_timer_lock:
            if gen == self._save_timer_gen:
                self._save_timer = None
                _PENDING_SAVE_INDEXES.discard(self)
        self.save_cache()

    def flush_cache(self) -> None:
        """Cancel any pending debounced save and persist immediately."""
        with self._save_timer_lock:
            if self._save_timer is not None:
                self._save_timer.cancel()
                self._save_timer = None
            self._save_timer_gen += 1
            _PENDING_SAVE_INDEXES.discard(self)
        self.save_cache()

    # ── maintenance ──────────────────────────────────────────────────────────

    def _row(self, box: str, note: "Note", date: int) -> dict:
        text = note_to_embed_text(note)
        return {
            "id": composite_id(box, note.id),
            "title": note.title,
            "type": note.type,
            "box": box,
            "note_id": note.id,
            "date": int(date or 0),
            TEXT_FIELD: text,
            HASH_FIELD: _text_hash(text),
        }

    def upsert_note(
        self, box: str, note: "Note", date: int = 0, *, vector: list[float] | None = None
    ) -> None:
        """Add/update a single note's global node + vector.

        ``vector`` may be supplied by the caller (the per-box index already
        embedded this exact text) to avoid a second embed on the write path; when
        ``None`` the note is embedded here. Quote-type notes (``EMBED_SKIP_TYPES``)
        are skipped, matching the per-box index.
        """
        if note.type in EMBED_SKIP_TYPES:
            return
        self.upsert_notes(box, [note], date, vectors={note.id: vector} if vector else None)

    def upsert_notes(
        self,
        box: str,
        notes: list["Note"],
        date: int = 0,
        *,
        vectors: dict[str, list[float]] | None = None,
        skip_unchanged: bool = False,
    ) -> None:
        """Add/update many notes for one box in a single batch.

        ``vectors`` optionally maps note id -> precomputed vector (reused from the
        per-box index); any note missing from it is embedded here.
        ``skip_unchanged`` (used by the warm-mirror path) drops notes whose
        ``text_hash`` already matches the indexed one, so re-mirroring a box on a
        cold access is a near no-op instead of re-adding every node.
        """
        import pandas as pd

        notes = [n for n in notes if n.type not in EMBED_SKIP_TYPES]
        if not notes:
            return
        with self._lock:
            self._ensure_graph()
            rows = [self._row(box, n, date) for n in notes]
            if skip_unchanged:
                rows = [r for r in rows if self._note_hashes.get(r["id"]) != r[HASH_FIELD]]
                if not rows:
                    return
                kept = {r["id"] for r in rows}
                notes = [n for n in notes if composite_id(box, n.id) in kept]
            try:
                self._graph.add_nodes(
                    pd.DataFrame(rows),
                    node_type=NODE_TYPE,
                    unique_id_field="id",
                    node_title_field="title",
                    conflict_handling="update",
                )
            except Exception as exc:
                logger.warning("Global index: could not add notes for box '%s': %s", box, exc)
                return

            payload: dict[str, list[float]] = {}
            to_embed: list[tuple[str, str]] = []  # (composite_id, text)
            supplied = vectors or {}
            for n, row in zip(notes, rows):
                cid = row["id"]
                pre = supplied.get(n.id)
                if pre and not _is_zero_norm(list(pre)):
                    payload[cid] = list(pre)
                else:
                    to_embed.append((cid, row[TEXT_FIELD]))
            # Only build the embedder when a note actually needs embedding — the
            # all-vectors-supplied mirror path stays embedder-free.
            if to_embed and self._ensure_embedder() is not None:
                try:
                    vecs = self._embedder.embed([t for _, t in to_embed])
                except Exception as exc:
                    logger.warning("Global index: embed failed for box '%s': %s", box, exc)
                    vecs = []
                for (cid, _text), vec in zip(to_embed, vecs):
                    vec = list(vec)
                    if not _is_zero_norm(vec):
                        payload[cid] = vec
            if payload:
                try:
                    self._graph.add_embeddings(NODE_TYPE, TEXT_FIELD, payload)
                except Exception as exc:
                    logger.warning("Global index: could not store vectors for box '%s': %s", box, exc)
                    return
            for row in rows:
                self._note_hashes[row["id"]] = row[HASH_FIELD]
        self.save_cache_debounced()

    def remove_note(self, box: str, note_id: str) -> None:
        """Delete one note's global node (synchronous persist, like delete_note)."""
        cid = composite_id(box, note_id)
        with self._lock:
            self._note_hashes.pop(cid, None)
            if self._graph is None:
                return
            try:
                self._graph.cypher(
                    f"MATCH (n:{NODE_TYPE}) WHERE n.id = $id DETACH DELETE n",
                    params={"id": cid},
                )
            except Exception as exc:
                logger.debug("Global index: could not delete '%s': %s", cid, exc)
                return
        self.save_cache()

    def remove_box(self, box: str) -> None:
        """Delete every node belonging to a box (e.g. when the box is removed)."""
        with self._lock:
            if self._graph is None:
                self.load_cache()
            try:
                self._graph.cypher(
                    f"MATCH (n:{NODE_TYPE}) WHERE n.box = $box DETACH DELETE n",
                    params={"box": box},
                )
            except Exception as exc:
                logger.debug("Global index: could not delete box '%s': %s", box, exc)
                return
            prefix = f"{box}::"
            for cid in [c for c in self._note_hashes if c.startswith(prefix)]:
                self._note_hashes.pop(cid, None)
        self.save_cache()

    def reconcile_box(self, box: str, current_note_ids: Iterable[str]) -> int:
        """Evict global nodes for a box whose note no longer exists on disk.

        Called when a box is (re)built from its ``.md`` sources so a note deleted
        while the process was down (e.g. an external ``git pull``) cannot linger
        as a ghost in the global index — mirroring the per-box index's rebuild
        guarantee. Returns the number of evicted nodes.
        """
        keep = {composite_id(box, nid) for nid in current_note_ids}
        with self._lock:
            if self._graph is None:
                return 0
            try:
                rows = self._graph.cypher(
                    f"MATCH (n:{NODE_TYPE}) WHERE n.box = $box RETURN n.id AS id",
                    params={"box": box},
                )
            except Exception as exc:  # pragma: no cover - defensive
                logger.debug("Global index: reconcile query failed for box '%s': %s", box, exc)
                return 0
            stale = [str(r.get("id")) for r in (rows or []) if str(r.get("id")) not in keep]
            for cid in stale:
                try:
                    self._graph.cypher(
                        f"MATCH (n:{NODE_TYPE}) WHERE n.id = $id DETACH DELETE n",
                        params={"id": cid},
                    )
                except Exception:  # pragma: no cover - defensive
                    continue
                self._note_hashes.pop(cid, None)
        if stale:
            self.save_cache_debounced()
        return len(stale)

    # ── query ────────────────────────────────────────────────────────────────

    def search(
        self,
        query: str,
        top_k: int = 10,
        *,
        boxes: Iterable[str] | None = None,
        date_range: tuple[int, int] | None = None,
    ) -> list[tuple[str, str, float]]:
        """Pre-filtered ANN search; returns ``[(box, note_id, score)]``.

        ``boxes`` restricts recall to those boxes (an EMPTY iterable yields no
        results — a project with no sources matches nothing). ``date_range`` is an
        inclusive ``(lo, hi)`` on the ``YYYYMMDD`` ``date`` property (notes with an
        unknown date ``0`` are excluded when a range is given). Both filters are
        applied BEFORE ranking so recall for a small project inside a huge corpus
        stays correct.
        """
        with self._lock:
            if self._graph is None or self._ensure_embedder() is None:
                return []
            try:
                qvecs = self._embedder.embed([query])
            except Exception as exc:
                logger.warning("Global index: could not embed query: %s", exc)
                return []
            if not qvecs:
                return []
            qv = list(qvecs[0])

            cond: dict[str, Any] = {}
            if boxes is not None:
                box_list = list(dict.fromkeys(boxes))
                if not box_list:
                    return []
                cond["box"] = {"in": box_list}
            if date_range is not None:
                lo, hi = date_range
                cond["date"] = {"between": [int(lo), int(hi)]}

            try:
                sel = self._graph.select(NODE_TYPE)
                if cond:
                    sel = sel.where(cond)
                hits = sel.vector_search(TEXT_FIELD, qv, top_k=top_k)
            except Exception as exc:
                logger.warning("Global index: search failed: %s", exc)
                return []

        out: list[tuple[str, str, float]] = []
        for hit in hits or []:
            box = hit.get("box")
            note_id = hit.get("note_id")
            if box is None or note_id is None:
                # Fall back to parsing the composite id.
                cid = str(hit.get("id") or "")
                if "::" not in cid:
                    continue
                box, note_id = cid.split("::", 1)
            out.append((str(box), str(note_id), float(hit.get("score", 0.0))))
        return out

    # ── rebuild ──────────────────────────────────────────────────────────────

    def rebuild(self, graphs_dir: Path | None = None) -> dict:
        """Rebuild the whole store from the live ``.md`` notes across all boxes.

        Walks the graphs directory, indexing every real source box (and
        ``_cross``), reusing prior vectors whose ``text_hash`` still matches so a
        rebuild does not re-embed the entire corpus. Returns a small summary.
        """
        import kglite
        import pandas as pd

        from zettelkasten import graph as zk_graph
        from zettelkasten.graph_io import source_date_int

        base = graphs_dir or zk_graph.GRAPHS_DIR
        with self._lock:
            if self._ensure_embedder() is None:
                return {"boxes": 0, "notes": 0, "embedded": 0, "reused": 0}

            # Snapshot prior vectors + hashes from the CURRENT store so an
            # unchanged note is not re-embedded, then build a fresh graph so any
            # node for a since-deleted note is dropped (disposable full refresh).
            prior_vectors: dict[str, list[float]] = {}
            prior_hashes: dict[str, str] = {}
            if self._graph is not None:
                try:
                    prior_vectors = {
                        str(k): list(v)
                        for k, v in (self._graph.embeddings(NODE_TYPE, TEXT_FIELD) or {}).items()
                    }
                except Exception:  # pragma: no cover - defensive
                    prior_vectors = {}
                prior_hashes = dict(self._note_hashes) or self._read_hashes(self._graph)

            self._graph = kglite.KnowledgeGraph()
            try:
                self._graph.set_embedder(self._embedder)
            except Exception:  # pragma: no cover - defensive
                pass
            self._note_hashes = {}

            boxes = 0
            total = 0
            reused = 0
            embedded = 0
            rows_all: list[dict] = []
            payload: dict[str, list[float]] = {}
            to_embed: list[tuple[str, str]] = []

            if base.exists():
                for d in sorted(base.iterdir()):
                    if not d.is_dir():
                        continue
                    if d.name.startswith("_") and d.name != "_cross":
                        continue
                    try:
                        zg = zk_graph.ZettelGraph(d.name, graphs_dir=base)
                        zg.load(eager_embeddings=False)
                    except Exception as exc:  # pragma: no cover - defensive
                        logger.warning("Global index rebuild: could not load box '%s': %s", d.name, exc)
                        continue
                    boxes += 1
                    date = source_date_int(d.name, graphs_dir=base)
                    for note in zg.notes.values():
                        if note.type in EMBED_SKIP_TYPES:
                            continue
                        row = self._row(d.name, note, date)
                        rows_all.append(row)
                        total += 1
                        cid = row["id"]
                        cached = prior_vectors.get(cid)
                        if (
                            cached is not None
                            and prior_hashes.get(cid) == row[HASH_FIELD]
                            and not _is_zero_norm(cached)
                        ):
                            payload[cid] = cached
                            reused += 1
                        else:
                            to_embed.append((cid, row[TEXT_FIELD]))

            if rows_all:
                try:
                    self._graph.add_nodes(
                        pd.DataFrame(rows_all),
                        node_type=NODE_TYPE,
                        unique_id_field="id",
                        node_title_field="title",
                        conflict_handling="update",
                    )
                except Exception as exc:
                    logger.warning("Global index rebuild: add_nodes failed: %s", exc)
                    return {"boxes": boxes, "notes": total, "embedded": 0, "reused": 0}

            if to_embed:
                try:
                    vecs = self._embedder.embed([t for _, t in to_embed])
                except Exception as exc:
                    logger.warning("Global index rebuild: embed failed: %s", exc)
                    vecs = []
                for (cid, _t), vec in zip(to_embed, vecs):
                    vec = list(vec)
                    if not _is_zero_norm(vec):
                        payload[cid] = vec
                        embedded += 1

            if payload:
                try:
                    self._graph.add_embeddings(NODE_TYPE, TEXT_FIELD, payload)
                except Exception as exc:
                    logger.warning("Global index rebuild: add_embeddings failed: %s", exc)

            self._note_hashes = {row["id"]: row[HASH_FIELD] for row in rows_all}

        self.flush_cache()
        return {"boxes": boxes, "notes": total, "embedded": embedded, "reused": reused}

    def covered_boxes(self, candidates: Iterable[str]) -> set[str]:
        """Subset of ``candidates`` that currently hold >=1 indexed note.

        Lets a caller treat the ANN as the SOLE semantic source for a covered box
        — an empty ANN result for it means "no match", not "not indexed" — while
        still falling back to a per-box search for uncovered boxes (the memory
        tree, federated peers, or a box not yet mirrored in). Derived from the
        in-memory composite-id hash map (``box::note_id``), so it reflects the live
        store with no embedder build and no KGLite dialect assumptions. Short-
        circuits once every candidate is accounted for (the common fully-covered
        scope).
        """
        names = set(candidates)
        if not names:
            return set()
        found: set[str] = set()
        with self._lock:
            for cid in self._note_hashes:
                box = cid.split("::", 1)[0]
                if box in names and box not in found:
                    found.add(box)
                    if len(found) == len(names):
                        break
        return found

    def build_if_empty(self, graphs_dir: Path | None = None) -> "dict | None":
        """One-time backfill: rebuild ONLY when the store holds no notes yet.

        Backward-compat path for repos indexed by an older angelo (no
        ``_global.kgl`` on disk) — the first warmup after an update finds an empty
        store and populates it from the live ``.md`` corpus. A store that already
        holds notes (loaded from a persisted cache, or warmed incrementally) is
        left untouched, so an already-indexed repo is NEVER re-indexed. Returns the
        :meth:`rebuild` summary when a build ran, else ``None`` (nothing to do).
        """
        from zettelkasten import graph as zk_graph

        with self._lock:
            if self.indexed_count > 0:
                return None
        base = graphs_dir or zk_graph.GRAPHS_DIR
        # Skip the (embedding) rebuild for an empty corpus — nothing to index, and
        # we don't want to spin up the embedder needlessly.
        try:
            has_source = base.exists() and any(
                d.is_dir() and (not d.name.startswith("_") or d.name == "_cross")
                for d in base.iterdir()
            )
        except OSError:
            has_source = False
        if not has_source:
            return None
        return self.rebuild(graphs_dir=graphs_dir)

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

load_cache

load_cache() -> None

Load the persisted store as the LIVE, queryable graph.

Unlike the per-box index (which starts empty and rebuilds every box from .md on load), the global store loads its prior .kgl directly so it is immediately queryable across process restarts without walking the whole corpus. text_hash tracking is recovered so later upserts can skip unchanged notes; :meth:rebuild is the disposable full-refresh that re-derives everything from the live .md sources (and evicts any node for a note deleted while the process was down).

Source code in zettelkasten/global_index.py

def load_cache(self) -> None:
    """Load the persisted store as the LIVE, queryable graph.

    Unlike the per-box index (which starts empty and rebuilds every box from
    ``.md`` on load), the global store loads its prior ``.kgl`` directly so it
    is immediately queryable across process restarts without walking the whole
    corpus. ``text_hash`` tracking is recovered so later upserts can skip
    unchanged notes; :meth:`rebuild` is the disposable full-refresh that
    re-derives everything from the live ``.md`` sources (and evicts any node
    for a note deleted while the process was down).
    """
    import kglite

    with self._lock:
        self._note_hashes = {}
        loaded = None
        if self.cache_path.exists():
            try:
                loaded = kglite.load(str(self.cache_path))
            except Exception as exc:
                logger.warning("Global index: could not load cache '%s': %s", self.cache_path, exc)
                loaded = None

        if loaded is not None:
            self._graph = loaded
            self._note_hashes = self._read_hashes(self._graph)
        else:
            self._graph = kglite.KnowledgeGraph()

save_cache

save_cache() -> None

Persist the kglite graph atomically (temp + rename); disposable.

Source code in zettelkasten/global_index.py

def save_cache(self) -> None:
    """Persist the kglite graph atomically (temp + rename); disposable."""
    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("Global index: could not save 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 persists.

Source code in zettelkasten/global_index.py

def save_cache_debounced(self) -> None:
    """Schedule a debounced :meth:`save_cache`, coalescing rapid persists."""
    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.

Source code in zettelkasten/global_index.py

def flush_cache(self) -> None:
    """Cancel any pending debounced save and persist immediately."""
    with self._save_timer_lock:
        if self._save_timer is not None:
            self._save_timer.cancel()
            self._save_timer = None
        self._save_timer_gen += 1
        _PENDING_SAVE_INDEXES.discard(self)
    self.save_cache()

upsert_note

upsert_note(box: str, note: 'Note', date: int = 0, *, vector: list[float] | None = None) -> None

Add/update a single note's global node + vector.

vector may be supplied by the caller (the per-box index already embedded this exact text) to avoid a second embed on the write path; when None the note is embedded here. Quote-type notes (EMBED_SKIP_TYPES) are skipped, matching the per-box index.

Source code in zettelkasten/global_index.py

def upsert_note(
    self, box: str, note: "Note", date: int = 0, *, vector: list[float] | None = None
) -> None:
    """Add/update a single note's global node + vector.

    ``vector`` may be supplied by the caller (the per-box index already
    embedded this exact text) to avoid a second embed on the write path; when
    ``None`` the note is embedded here. Quote-type notes (``EMBED_SKIP_TYPES``)
    are skipped, matching the per-box index.
    """
    if note.type in EMBED_SKIP_TYPES:
        return
    self.upsert_notes(box, [note], date, vectors={note.id: vector} if vector else None)

upsert_notes

upsert_notes(box: str, notes: list['Note'], date: int = 0, *, vectors: dict[str, list[float]] | None = None, skip_unchanged: bool = False) -> None

Add/update many notes for one box in a single batch.

vectors optionally maps note id -> precomputed vector (reused from the per-box index); any note missing from it is embedded here. skip_unchanged (used by the warm-mirror path) drops notes whose text_hash already matches the indexed one, so re-mirroring a box on a cold access is a near no-op instead of re-adding every node.

Source code in zettelkasten/global_index.py

def upsert_notes(
    self,
    box: str,
    notes: list["Note"],
    date: int = 0,
    *,
    vectors: dict[str, list[float]] | None = None,
    skip_unchanged: bool = False,
) -> None:
    """Add/update many notes for one box in a single batch.

    ``vectors`` optionally maps note id -> precomputed vector (reused from the
    per-box index); any note missing from it is embedded here.
    ``skip_unchanged`` (used by the warm-mirror path) drops notes whose
    ``text_hash`` already matches the indexed one, so re-mirroring a box on a
    cold access is a near no-op instead of re-adding every node.
    """
    import pandas as pd

    notes = [n for n in notes if n.type not in EMBED_SKIP_TYPES]
    if not notes:
        return
    with self._lock:
        self._ensure_graph()
        rows = [self._row(box, n, date) for n in notes]
        if skip_unchanged:
            rows = [r for r in rows if self._note_hashes.get(r["id"]) != r[HASH_FIELD]]
            if not rows:
                return
            kept = {r["id"] for r in rows}
            notes = [n for n in notes if composite_id(box, n.id) in kept]
        try:
            self._graph.add_nodes(
                pd.DataFrame(rows),
                node_type=NODE_TYPE,
                unique_id_field="id",
                node_title_field="title",
                conflict_handling="update",
            )
        except Exception as exc:
            logger.warning("Global index: could not add notes for box '%s': %s", box, exc)
            return

        payload: dict[str, list[float]] = {}
        to_embed: list[tuple[str, str]] = []  # (composite_id, text)
        supplied = vectors or {}
        for n, row in zip(notes, rows):
            cid = row["id"]
            pre = supplied.get(n.id)
            if pre and not _is_zero_norm(list(pre)):
                payload[cid] = list(pre)
            else:
                to_embed.append((cid, row[TEXT_FIELD]))
        # Only build the embedder when a note actually needs embedding — the
        # all-vectors-supplied mirror path stays embedder-free.
        if to_embed and self._ensure_embedder() is not None:
            try:
                vecs = self._embedder.embed([t for _, t in to_embed])
            except Exception as exc:
                logger.warning("Global index: embed failed for box '%s': %s", box, exc)
                vecs = []
            for (cid, _text), vec in zip(to_embed, vecs):
                vec = list(vec)
                if not _is_zero_norm(vec):
                    payload[cid] = vec
        if payload:
            try:
                self._graph.add_embeddings(NODE_TYPE, TEXT_FIELD, payload)
            except Exception as exc:
                logger.warning("Global index: could not store vectors for box '%s': %s", box, exc)
                return
        for row in rows:
            self._note_hashes[row["id"]] = row[HASH_FIELD]
    self.save_cache_debounced()

remove_note

remove_note(box: str, note_id: str) -> None

Delete one note's global node (synchronous persist, like delete_note).

Source code in zettelkasten/global_index.py

def remove_note(self, box: str, note_id: str) -> None:
    """Delete one note's global node (synchronous persist, like delete_note)."""
    cid = composite_id(box, note_id)
    with self._lock:
        self._note_hashes.pop(cid, None)
        if self._graph is None:
            return
        try:
            self._graph.cypher(
                f"MATCH (n:{NODE_TYPE}) WHERE n.id = $id DETACH DELETE n",
                params={"id": cid},
            )
        except Exception as exc:
            logger.debug("Global index: could not delete '%s': %s", cid, exc)
            return
    self.save_cache()

remove_box

remove_box(box: str) -> None

Delete every node belonging to a box (e.g. when the box is removed).

Source code in zettelkasten/global_index.py

def remove_box(self, box: str) -> None:
    """Delete every node belonging to a box (e.g. when the box is removed)."""
    with self._lock:
        if self._graph is None:
            self.load_cache()
        try:
            self._graph.cypher(
                f"MATCH (n:{NODE_TYPE}) WHERE n.box = $box DETACH DELETE n",
                params={"box": box},
            )
        except Exception as exc:
            logger.debug("Global index: could not delete box '%s': %s", box, exc)
            return
        prefix = f"{box}::"
        for cid in [c for c in self._note_hashes if c.startswith(prefix)]:
            self._note_hashes.pop(cid, None)
    self.save_cache()

reconcile_box

reconcile_box(box: str, current_note_ids: Iterable[str]) -> int

Evict global nodes for a box whose note no longer exists on disk.

Called when a box is (re)built from its .md sources so a note deleted while the process was down (e.g. an external git pull) cannot linger as a ghost in the global index — mirroring the per-box index's rebuild guarantee. Returns the number of evicted nodes.

Source code in zettelkasten/global_index.py

def reconcile_box(self, box: str, current_note_ids: Iterable[str]) -> int:
    """Evict global nodes for a box whose note no longer exists on disk.

    Called when a box is (re)built from its ``.md`` sources so a note deleted
    while the process was down (e.g. an external ``git pull``) cannot linger
    as a ghost in the global index — mirroring the per-box index's rebuild
    guarantee. Returns the number of evicted nodes.
    """
    keep = {composite_id(box, nid) for nid in current_note_ids}
    with self._lock:
        if self._graph is None:
            return 0
        try:
            rows = self._graph.cypher(
                f"MATCH (n:{NODE_TYPE}) WHERE n.box = $box RETURN n.id AS id",
                params={"box": box},
            )
        except Exception as exc:  # pragma: no cover - defensive
            logger.debug("Global index: reconcile query failed for box '%s': %s", box, exc)
            return 0
        stale = [str(r.get("id")) for r in (rows or []) if str(r.get("id")) not in keep]
        for cid in stale:
            try:
                self._graph.cypher(
                    f"MATCH (n:{NODE_TYPE}) WHERE n.id = $id DETACH DELETE n",
                    params={"id": cid},
                )
            except Exception:  # pragma: no cover - defensive
                continue
            self._note_hashes.pop(cid, None)
    if stale:
        self.save_cache_debounced()
    return len(stale)

search

search(query: str, top_k: int = 10, *, boxes: Iterable[str] | None = None, date_range: tuple[int, int] | None = None) -> list[tuple[str, str, float]]

Pre-filtered ANN search; returns [(box, note_id, score)].

boxes restricts recall to those boxes (an EMPTY iterable yields no results — a project with no sources matches nothing). date_range is an inclusive (lo, hi) on the YYYYMMDD date property (notes with an unknown date 0 are excluded when a range is given). Both filters are applied BEFORE ranking so recall for a small project inside a huge corpus stays correct.

Source code in zettelkasten/global_index.py

def search(
    self,
    query: str,
    top_k: int = 10,
    *,
    boxes: Iterable[str] | None = None,
    date_range: tuple[int, int] | None = None,
) -> list[tuple[str, str, float]]:
    """Pre-filtered ANN search; returns ``[(box, note_id, score)]``.

    ``boxes`` restricts recall to those boxes (an EMPTY iterable yields no
    results — a project with no sources matches nothing). ``date_range`` is an
    inclusive ``(lo, hi)`` on the ``YYYYMMDD`` ``date`` property (notes with an
    unknown date ``0`` are excluded when a range is given). Both filters are
    applied BEFORE ranking so recall for a small project inside a huge corpus
    stays correct.
    """
    with self._lock:
        if self._graph is None or self._ensure_embedder() is None:
            return []
        try:
            qvecs = self._embedder.embed([query])
        except Exception as exc:
            logger.warning("Global index: could not embed query: %s", exc)
            return []
        if not qvecs:
            return []
        qv = list(qvecs[0])

        cond: dict[str, Any] = {}
        if boxes is not None:
            box_list = list(dict.fromkeys(boxes))
            if not box_list:
                return []
            cond["box"] = {"in": box_list}
        if date_range is not None:
            lo, hi = date_range
            cond["date"] = {"between": [int(lo), int(hi)]}

        try:
            sel = self._graph.select(NODE_TYPE)
            if cond:
                sel = sel.where(cond)
            hits = sel.vector_search(TEXT_FIELD, qv, top_k=top_k)
        except Exception as exc:
            logger.warning("Global index: search failed: %s", exc)
            return []

    out: list[tuple[str, str, float]] = []
    for hit in hits or []:
        box = hit.get("box")
        note_id = hit.get("note_id")
        if box is None or note_id is None:
            # Fall back to parsing the composite id.
            cid = str(hit.get("id") or "")
            if "::" not in cid:
                continue
            box, note_id = cid.split("::", 1)
        out.append((str(box), str(note_id), float(hit.get("score", 0.0))))
    return out

rebuild

rebuild(graphs_dir: Path | None = None) -> dict

Rebuild the whole store from the live .md notes across all boxes.

Walks the graphs directory, indexing every real source box (and _cross), reusing prior vectors whose text_hash still matches so a rebuild does not re-embed the entire corpus. Returns a small summary.

Source code in zettelkasten/global_index.py

def rebuild(self, graphs_dir: Path | None = None) -> dict:
    """Rebuild the whole store from the live ``.md`` notes across all boxes.

    Walks the graphs directory, indexing every real source box (and
    ``_cross``), reusing prior vectors whose ``text_hash`` still matches so a
    rebuild does not re-embed the entire corpus. Returns a small summary.
    """
    import kglite
    import pandas as pd

    from zettelkasten import graph as zk_graph
    from zettelkasten.graph_io import source_date_int

    base = graphs_dir or zk_graph.GRAPHS_DIR
    with self._lock:
        if self._ensure_embedder() is None:
            return {"boxes": 0, "notes": 0, "embedded": 0, "reused": 0}

        # Snapshot prior vectors + hashes from the CURRENT store so an
        # unchanged note is not re-embedded, then build a fresh graph so any
        # node for a since-deleted note is dropped (disposable full refresh).
        prior_vectors: dict[str, list[float]] = {}
        prior_hashes: dict[str, str] = {}
        if self._graph is not None:
            try:
                prior_vectors = {
                    str(k): list(v)
                    for k, v in (self._graph.embeddings(NODE_TYPE, TEXT_FIELD) or {}).items()
                }
            except Exception:  # pragma: no cover - defensive
                prior_vectors = {}
            prior_hashes = dict(self._note_hashes) or self._read_hashes(self._graph)

        self._graph = kglite.KnowledgeGraph()
        try:
            self._graph.set_embedder(self._embedder)
        except Exception:  # pragma: no cover - defensive
            pass
        self._note_hashes = {}

        boxes = 0
        total = 0
        reused = 0
        embedded = 0
        rows_all: list[dict] = []
        payload: dict[str, list[float]] = {}
        to_embed: list[tuple[str, str]] = []

        if base.exists():
            for d in sorted(base.iterdir()):
                if not d.is_dir():
                    continue
                if d.name.startswith("_") and d.name != "_cross":
                    continue
                try:
                    zg = zk_graph.ZettelGraph(d.name, graphs_dir=base)
                    zg.load(eager_embeddings=False)
                except Exception as exc:  # pragma: no cover - defensive
                    logger.warning("Global index rebuild: could not load box '%s': %s", d.name, exc)
                    continue
                boxes += 1
                date = source_date_int(d.name, graphs_dir=base)
                for note in zg.notes.values():
                    if note.type in EMBED_SKIP_TYPES:
                        continue
                    row = self._row(d.name, note, date)
                    rows_all.append(row)
                    total += 1
                    cid = row["id"]
                    cached = prior_vectors.get(cid)
                    if (
                        cached is not None
                        and prior_hashes.get(cid) == row[HASH_FIELD]
                        and not _is_zero_norm(cached)
                    ):
                        payload[cid] = cached
                        reused += 1
                    else:
                        to_embed.append((cid, row[TEXT_FIELD]))

        if rows_all:
            try:
                self._graph.add_nodes(
                    pd.DataFrame(rows_all),
                    node_type=NODE_TYPE,
                    unique_id_field="id",
                    node_title_field="title",
                    conflict_handling="update",
                )
            except Exception as exc:
                logger.warning("Global index rebuild: add_nodes failed: %s", exc)
                return {"boxes": boxes, "notes": total, "embedded": 0, "reused": 0}

        if to_embed:
            try:
                vecs = self._embedder.embed([t for _, t in to_embed])
            except Exception as exc:
                logger.warning("Global index rebuild: embed failed: %s", exc)
                vecs = []
            for (cid, _t), vec in zip(to_embed, vecs):
                vec = list(vec)
                if not _is_zero_norm(vec):
                    payload[cid] = vec
                    embedded += 1

        if payload:
            try:
                self._graph.add_embeddings(NODE_TYPE, TEXT_FIELD, payload)
            except Exception as exc:
                logger.warning("Global index rebuild: add_embeddings failed: %s", exc)

        self._note_hashes = {row["id"]: row[HASH_FIELD] for row in rows_all}

    self.flush_cache()
    return {"boxes": boxes, "notes": total, "embedded": embedded, "reused": reused}

covered_boxes

covered_boxes(candidates: Iterable[str]) -> set[str]

Subset of candidates that currently hold >=1 indexed note.

Lets a caller treat the ANN as the SOLE semantic source for a covered box — an empty ANN result for it means "no match", not "not indexed" — while still falling back to a per-box search for uncovered boxes (the memory tree, federated peers, or a box not yet mirrored in). Derived from the in-memory composite-id hash map (box::note_id), so it reflects the live store with no embedder build and no KGLite dialect assumptions. Short- circuits once every candidate is accounted for (the common fully-covered scope).

Source code in zettelkasten/global_index.py

def covered_boxes(self, candidates: Iterable[str]) -> set[str]:
    """Subset of ``candidates`` that currently hold >=1 indexed note.

    Lets a caller treat the ANN as the SOLE semantic source for a covered box
    — an empty ANN result for it means "no match", not "not indexed" — while
    still falling back to a per-box search for uncovered boxes (the memory
    tree, federated peers, or a box not yet mirrored in). Derived from the
    in-memory composite-id hash map (``box::note_id``), so it reflects the live
    store with no embedder build and no KGLite dialect assumptions. Short-
    circuits once every candidate is accounted for (the common fully-covered
    scope).
    """
    names = set(candidates)
    if not names:
        return set()
    found: set[str] = set()
    with self._lock:
        for cid in self._note_hashes:
            box = cid.split("::", 1)[0]
            if box in names and box not in found:
                found.add(box)
                if len(found) == len(names):
                    break
    return found

build_if_empty

build_if_empty(graphs_dir: Path | None = None) -> 'dict | None'

One-time backfill: rebuild ONLY when the store holds no notes yet.

Backward-compat path for repos indexed by an older angelo (no _global.kgl on disk) — the first warmup after an update finds an empty store and populates it from the live .md corpus. A store that already holds notes (loaded from a persisted cache, or warmed incrementally) is left untouched, so an already-indexed repo is NEVER re-indexed. Returns the :meth:rebuild summary when a build ran, else None (nothing to do).

Source code in zettelkasten/global_index.py

def build_if_empty(self, graphs_dir: Path | None = None) -> "dict | None":
    """One-time backfill: rebuild ONLY when the store holds no notes yet.

    Backward-compat path for repos indexed by an older angelo (no
    ``_global.kgl`` on disk) — the first warmup after an update finds an empty
    store and populates it from the live ``.md`` corpus. A store that already
    holds notes (loaded from a persisted cache, or warmed incrementally) is
    left untouched, so an already-indexed repo is NEVER re-indexed. Returns the
    :meth:`rebuild` summary when a build ran, else ``None`` (nothing to do).
    """
    from zettelkasten import graph as zk_graph

    with self._lock:
        if self.indexed_count > 0:
            return None
    base = graphs_dir or zk_graph.GRAPHS_DIR
    # Skip the (embedding) rebuild for an empty corpus — nothing to index, and
    # we don't want to spin up the embedder needlessly.
    try:
        has_source = base.exists() and any(
            d.is_dir() and (not d.name.startswith("_") or d.name == "_cross")
            for d in base.iterdir()
        )
    except OSError:
        has_source = False
    if not has_source:
        return None
    return self.rebuild(graphs_dir=graphs_dir)

composite_id

composite_id(box: str, note_id: str) -> str

Stable global node id for a (box, note_id) pair.

Source code in zettelkasten/global_index.py

def composite_id(box: str, note_id: str) -> str:
    """Stable global node id for a ``(box, note_id)`` pair."""
    return f"{box}::{note_id}"

get_global_index

get_global_index() -> 'GlobalIndex | None'

Return the process-wide global index, or None when embeddings are off.

Lazily built and cache-loaded on first use. Returns None when semantic search is disabled (ZETTEL_SKIP_EMBEDDINGS=1 or no embedder), so callers transparently fall back to the per-box path.

Source code in zettelkasten/global_index.py

def get_global_index() -> "GlobalIndex | None":
    """Return the process-wide global index, or ``None`` when embeddings are off.

    Lazily built and cache-loaded on first use. Returns ``None`` when semantic
    search is disabled (``ZETTEL_SKIP_EMBEDDINGS=1`` or no embedder), so callers
    transparently fall back to the per-box path.
    """
    global _singleton
    if _skip_embeddings():
        return None
    if _singleton is not None:
        return _singleton
    with _singleton_lock:
        if _singleton is not None:
            return _singleton
        idx = GlobalIndex()
        idx.load_cache()
        _singleton = idx
        return _singleton

reset_global_index

reset_global_index() -> None

Drop the process-wide singleton (tests / cache-path repoint).

Source code in zettelkasten/global_index.py

def reset_global_index() -> None:
    """Drop the process-wide singleton (tests / cache-path repoint)."""
    global _singleton
    with _singleton_lock:
        _singleton = None

warm_global_index

warm_global_index(*, auto_build: bool = False) -> 'GlobalIndex | None'

Eagerly load the global store so the first query doesn't pay the cold cost.

Loads the persisted .kgl (via :func:get_global_index). With auto_build set, an EMPTY store (old repo with no persisted index) is then populated once from the live .md corpus via :meth:GlobalIndex.build_if_empty — this is the backward-compat backfill so a repo indexed by an older angelo gets an ANN on the next server start. An already-populated store is never rebuilt. Returns the index, or None when embeddings are disabled. Safe to call from a background thread — the load and the (idempotent) build are lock-guarded.

Source code in zettelkasten/global_index.py

def warm_global_index(*, auto_build: bool = False) -> "GlobalIndex | None":
    """Eagerly load the global store so the first query doesn't pay the cold cost.

    Loads the persisted ``.kgl`` (via :func:`get_global_index`). With
    ``auto_build`` set, an EMPTY store (old repo with no persisted index) is then
    populated once from the live ``.md`` corpus via
    :meth:`GlobalIndex.build_if_empty` — this is the backward-compat backfill so a
    repo indexed by an older angelo gets an ANN on the next server start. An
    already-populated store is never rebuilt. Returns the index, or ``None`` when
    embeddings are disabled. Safe to call from a background thread — the load and
    the (idempotent) build are lock-guarded.
    """
    idx = get_global_index()
    if idx is not None and auto_build:
        try:
            summary = idx.build_if_empty()
            if summary and summary.get("notes"):
                logger.info(
                    "Global index: backfilled %d note(s) across %d box(es) "
                    "(embedded %d, reused %d).",
                    summary.get("notes", 0), summary.get("boxes", 0),
                    summary.get("embedded", 0), summary.get("reused", 0),
                )
        except Exception as exc:  # pragma: no cover - backfill is best-effort
            logger.warning("Global index: auto-build failed: %s", exc)
    return idx

global_index_status

global_index_status() -> dict

Cheap health snapshot of the global store — never builds the embedder.

Reports the cache file's presence/size/age and, when the singleton is already built THIS process, its in-memory indexed-note count. Deliberately avoids constructing the index (which would load the embedding model) so health is fast even when semantic search has not been touched yet.

Source code in zettelkasten/global_index.py

def global_index_status() -> dict:
    """Cheap health snapshot of the global store — never builds the embedder.

    Reports the cache file's presence/size/age and, when the singleton is already
    built THIS process, its in-memory indexed-note count. Deliberately avoids
    constructing the index (which would load the embedding model) so ``health`` is
    fast even when semantic search has not been touched yet.
    """
    import time

    path = _global_cache_path()
    exists = path.exists()
    status: dict = {
        "cache_path": str(path),
        "cache_exists": exists,
        "enabled": not _skip_embeddings(),
    }
    if exists:
        try:
            st = path.stat()
            status["cache_bytes"] = st.st_size
            status["cache_age_seconds"] = round(max(0.0, time.time() - st.st_mtime), 1)
        except OSError:
            pass
    if _singleton is not None:
        status["indexed_notes"] = _singleton.indexed_count
        status["loaded"] = True
    else:
        status["loaded"] = False
    return status