class ZettelGraph:
"""In-memory index over a single concept box (graph)."""
def __init__(self, name: str, *, graphs_dir: Path | None = None, repo_id: str | None = None):
# The graph name becomes a directory under the graphs root, so it must be
# a single safe path segment. This guards every caller (MCP server and
# dashboard routes alike) against traversal via crafted names.
validate_id(name, kind="graph name", for_filename=True)
self.name = name
# ``graphs_dir`` lets a caller point a graph at a different .zettelkasten/
# root than the local one — used by federation to read another repo's
# graphs read-only. Defaults to this repo's GRAPHS_DIR.
base = graphs_dir if graphs_dir is not None else GRAPHS_DIR
self.path = safe_join(base, name)
# A non-None ``repo_id`` marks this graph as a read-only projection of a
# federated repo. It also namespaces the disposable embedding cache so two
# federated repos (or a federated repo and the local store) that share a
# box name never collide under .angelo/zettel/.
self.repo_id = repo_id
self.read_only = repo_id is not None
self.notes: dict[str, Note] = {}
self._title_index: dict[str, str] = {} # lowercase title -> id
self._alias_index: dict[str, str] = {} # lowercase alias -> id
self._tag_index: dict[str, set[str]] = {} # tag -> set of ids
# Reverse-backlink adjacency (mirrors claims.py's ``by_target``), so
# ``get_backlinks`` is an O(backlinks) lookup instead of an O(all notes ×
# all links) full scan. ``_backlink_index`` maps a target id to the
# sources that link to it: ``{target_id: {source_id: [Link, ...]}}`` —
# the inner list preserves a source's per-link order and multiplicity so
# a source with several edges to the same target yields several rows,
# exactly as the full scan did. ``_backlink_out`` records, for each
# source, the set of targets currently registered for it in the reverse
# index; it is the authoritative record used to retract a source's OLD
# edges on re-index (callers such as spine ``ensure_link``/``remove_link``
# mutate ``note.links`` IN PLACE before calling ``save_note``, so the old
# edge set cannot be recovered from ``self.notes`` — it must be tracked
# here). ``_note_order`` assigns each note a monotonic sequence at first
# index so ``get_backlinks`` can emit sources in ``self.notes`` insertion
# order, matching the previous ``for note in self.notes.values()`` scan.
self._backlink_index: dict[str, dict[str, list[Link]]] = {}
self._backlink_out: dict[str, set[str]] = {}
self._note_order: dict[str, int] = {}
self._note_seq: int = 0
# Staleness token for the reverse-backlink index. The incremental index
# above is only maintained via ``_index_note``/``load``; code that assigns
# ``self.notes`` DIRECTLY (``zg.notes = {...}`` in discover_spines and many
# fixtures) or deletes straight from it (``del self.notes[id]`` in
# server/spine) bypasses that maintenance, leaving the index empty or
# stale. ``get_backlinks`` compares this token — the identity of the
# ``self.notes`` object plus its length — against the live dict and lazily
# rebuilds when they diverge. Identity catches whole-dict reassignment
# (a new object) regardless of size; length catches direct add/delete on
# the same object (mooting the stale-entry leak from direct deletes). The
# common save_note/load path keeps the index in sync and re-stamps the
# token, so it stays O(backlinks) with no rebuild.
self._backlink_notes_ref: dict | None = self.notes
self._backlink_notes_len: int = 0
self._embeddings: "EmbeddingIndex | None" = None
# Serializes *construction* of the lazy embedding index and access to
# self.notes across the MCP threadpool. kglite is not thread-safe and a
# single ZettelGraph instance is shared per box, so two concurrent tool
# calls must not each build a separate EmbeddingIndex (each carries its
# own internal lock, giving zero mutual exclusion) nor iterate
# self.notes while another thread mutates it. Re-entrant so a holder can
# call back into other guarded methods on the same thread.
self._embeddings_lock = threading.RLock()
def _embedding_cache_path(self) -> "Path | None":
"""Cache path for this graph's kglite embedding index.
Local graphs use the default location (``None`` → ``.angelo/zettel/<box>.kgl``).
Federated graphs cache under ``.angelo/zettel/federated/<repo_id>/<box>.kgl``
so a federated repo's index (rebuilt locally from its committed notes,
since the remote ``.kgl`` is gitignored and may not exist) never clobbers
the local box of the same name.
"""
if self.repo_id is None:
return None
from zettelkasten.embeddings import _cache_dir
return _cache_dir() / "federated" / self.repo_id / f"{self.name}.kgl"
@property
def embeddings(self):
"""Lazy-build the kglite-backed embedding index.
The index is a disposable per-box cache rebuilt from the live notes:
``load_cache`` recovers prior vectors so unchanged notes are not
re-embedded, then ``index_notes`` (re)populates the kglite graph from
the notes currently in memory and saves the cache. Because the graph is
rebuilt from live notes only, vectors for deleted notes can never
resurface as ghost search hits.
Construction uses double-checked locking on ``self._embeddings_lock``:
the fast path returns an already-built index without locking, while a
cold box builds exactly one ``EmbeddingIndex`` under the lock even when
several threadpool calls race in together. This mirrors the memory
server's ``_rebuild_with_embeddings`` (one lock spanning rebuild+embed).
The notes are snapshotted while holding the lock so a concurrent
``save_note``/``_index_note`` can't change ``self.notes`` mid-iteration
(CPython would raise "dictionary changed size during iteration").
"""
if self._embeddings is None:
from zettelkasten.embeddings import EmbeddingIndex
with self._embeddings_lock:
if self._embeddings is None:
index = EmbeddingIndex(self.path, cache_path=self._embedding_cache_path())
index.load_cache()
notes_snapshot = list(self.notes.values())
# Notes rebuild from the live .md sources; citations rebuild
# from the live ``_citations/*.yaml`` source the same way, so a
# .kgl wipe rebuilds both and a citation deleted while cold
# cannot resurface as a ghost (see _reconcile_citations).
citations_changed = self._reconcile_citations(index, notes_snapshot)
if notes_snapshot:
index.index_notes(notes_snapshot)
# Persist when notes were (re)indexed or citation state moved;
# a truly empty, citation-free box writes no cache (unchanged).
if notes_snapshot or citations_changed:
index.save_cache()
# Publish only after the index is fully built so other
# threads either block on the lock or see a complete index.
self._embeddings = index
# Warm/backfill the global ANN index from this freshly built
# box (reusing its vectors, skipping unchanged notes, and
# evicting ghosts). Best-effort — never blocks the box build.
if notes_snapshot:
self._mirror_box_to_global(index, notes_snapshot)
return self._embeddings
def _reconcile_citations(self, index, notes_snapshot: list["Note"]) -> bool:
"""Reconcile the index's citation nodes against the live ``_citations``.
The ``.kgl`` cache is disposable but ``load_cache`` re-hydrates citation
nodes straight from it with NO check against the authoritative live
``_citations/*.yaml`` source. Notes get this for free (they are rebuilt
from the live ``.md`` files by ``index_notes``); citations have no such
live-reload step, so a citation whose yaml was deleted while the index
was cold would otherwise linger as a ghost in ``citation_count`` /
set-cover / search. We close that gap here, in the long-lived process:
1. Prune any indexed citation id NOT in this box's PER-BOX relevant set
(the live citations its notes currently link to) — the only prune
path (:meth:`remove_citation`). This subsumes two drifts: a citation
whose yaml was deleted while cold (absent from the live source, hence
never relevant) AND a citation UNLINKED from THIS box but still
globally alive in another box. Without the latter, an unlinked-but-
live citation would linger in this box's ``citation_count`` /
``CITATION_NODE_TYPE`` search forever (per-box drift). A citation that
IS still linked to the box stays in ``relevant`` and is never pruned.
2. (Re)index that same per-box relevant subset. ``index_citations`` is
additive/refresh and never prunes ids it wasn't passed, so passing a
per-box subset cannot delete another box's citations; step 1 is the
only deletion. A ``.kgl`` wipe therefore rebuilds this box's citations
from yaml exactly as ``index_notes`` rebuilds notes from ``.md``.
"""
try:
live = load_citations(graphs_dir=self.path.parent)
except Exception as exc:
# A malformed/unreadable ``_citations`` dir leaves the index
# unreconciled (it keeps serving whatever was rehydrated/indexed).
# Surface it explicitly via a warning rather than failing silently so
# the stale-citation condition is observable, then bail without
# mutating the index.
logger.warning("Could not load citations for reconcile of box '%s': %s", self.name, exc)
return False
# Per-box relevant subset: the live citations this box's notes link to.
relevant: set[str] = set()
for note in notes_snapshot:
for link in note.links:
if link.graph == "_citations" and link.target in live:
relevant.add(link.target)
changed = False
# Prune every indexed id outside the per-box relevant set: a globally
# deleted citation (absent from ``live``) AND one merely unlinked from
# THIS box (still in ``live`` but no longer referenced here) both drop
# out, so this box's index/count/search reflect only its own live links.
for cid in index.citation_ids:
if cid not in relevant:
index.remove_citation(cid)
changed = True
if relevant:
index.index_citations({cid: live[cid] for cid in relevant})
changed = True
return changed
def load(self, eager_embeddings: bool = True) -> None:
"""Load all notes from disk into memory.
Args:
eager_embeddings: If True, initialize and sync the embedding index
immediately rather than waiting for first search.
Held entirely under ``_embeddings_lock`` so the clear-then-rebuild is
atomic with respect to the ``embeddings`` property and ``_index_note``
(both re-entrant on the same lock): a concurrent reader can never
observe a half-cleared index or an index built from a partially
repopulated ``self.notes``. The lock is re-entrant, so the nested
``self.embeddings`` access below does not deadlock.
"""
with self._embeddings_lock:
self.notes.clear()
self._title_index.clear()
self._alias_index.clear()
self._tag_index.clear()
self._backlink_index.clear()
self._backlink_out.clear()
self._note_order.clear()
self._note_seq = 0
self._embeddings = None
# Empty state is in sync with the (cleared) index; each _index_note
# below re-stamps as notes are added, and this covers the no-notes
# early-return path too.
self._mark_backlinks_fresh()
if not self.path.exists():
return
for filepath in self.path.glob("*.md"):
if filepath.name.startswith("_"):
continue
note = parse_note(filepath)
if note:
self._index_note(note)
if eager_embeddings and self.notes:
# Force embedding index initialization so search is instant
_ = self.embeddings
def _index_note(self, note: Note) -> None:
# Quarantine notes whose id is not a safe filename. The id is used to
# build on-disk paths (save_note/delete), so a hostile or corrupt note
# file with a traversal id (e.g. "../../etc/x") must never enter the
# index where it could later be written or deleted by path.
try:
validate_id(note.id, kind="note id", for_filename=True)
except (ValueError, TypeError) as exc:
logger.warning(
"Skipping note with unsafe id %r in graph '%s': %s",
note.id,
self.name,
exc,
)
return
# Hold the embeddings lock for the dict writes so they are serialized
# against the notes snapshot taken while (re)building the index.
with self._embeddings_lock:
# A note already present is an in-place update: CPython keeps its
# position in ``self.notes`` on reassignment, so its ``_note_order``
# sequence (hence its backlink ordering) must be preserved. A note
# NOT present is either brand new or a re-add after an external
# delete (server ``delete_note`` / spine ``delete_node`` remove it
# straight from ``self.notes``); in both cases it lands at the END of
# ``self.notes`` and must get a fresh, higher sequence to match.
if note.id not in self._note_order or note.id not in self.notes:
self._note_order[note.id] = self._note_seq
self._note_seq += 1
self.notes[note.id] = note
self._title_index[note.title.lower()] = note.id
for alias in note.aliases:
self._alias_index[alias.lower()] = note.id
for tag in note.tags:
self._tag_index.setdefault(tag, set()).add(note.id)
self._reindex_backlinks(note)
# The index now reflects self.notes (this note added/updated in
# place); re-stamp the staleness token so get_backlinks serves from
# the index without a rebuild on the common save_note/load path.
self._mark_backlinks_fresh()
def _reindex_backlinks(self, note: Note) -> None:
"""Refresh ``note``'s outgoing edges in the reverse-backlink index.
Must be called holding ``_embeddings_lock``. Retracts the source's OLD
edges (looked up from ``_backlink_out`` — NOT from ``self.notes``, whose
note may already carry the new links because callers mutate ``note.links``
in place before ``save_note``) before registering the new ones, so a
link edit never leaves a stale reverse edge. Mirrors the ``by_target``
adjacency built in ``claims.py``.
"""
source_id = note.id
old_targets = self._backlink_out.get(source_id)
if old_targets:
for tgt in old_targets:
bucket = self._backlink_index.get(tgt)
if bucket is not None:
bucket.pop(source_id, None)
if not bucket:
del self._backlink_index[tgt]
new_targets: set[str] = set()
for link in note.links:
self._backlink_index.setdefault(link.target, {}).setdefault(
source_id, []
).append(link)
new_targets.add(link.target)
if new_targets:
self._backlink_out[source_id] = new_targets
else:
self._backlink_out.pop(source_id, None)
def _mark_backlinks_fresh(self) -> None:
"""Record that the reverse-backlink index reflects the current ``self.notes``.
Must be called holding ``_embeddings_lock``. Stores the identity and size
of the live ``self.notes`` so ``get_backlinks`` can cheaply tell whether a
direct reassignment/mutation has since bypassed index maintenance.
"""
self._backlink_notes_ref = self.notes
self._backlink_notes_len = len(self.notes)
def _backlink_index_is_stale(self) -> bool:
"""Whether the reverse-backlink index has diverged from ``self.notes``.
Must be called holding ``_embeddings_lock``. True when the notes dict was
reassigned wholesale (a different object) or changed size out-of-band
(a direct ``del``/insert that bypassed ``_index_note``).
"""
return (
self.notes is not self._backlink_notes_ref
or len(self.notes) != self._backlink_notes_len
)
def _rebuild_backlink_index(self) -> None:
"""Rebuild the reverse-backlink index and ordering from ``self.notes``.
Must be called holding ``_embeddings_lock``. Recovers from a direct
``self.notes`` reassignment (``zg.notes = {...}``) or direct delete
(``del self.notes[id]`` / ``.pop``) that bypassed ``_index_note`` and left
the incrementally-maintained index empty or carrying stale entries. Notes
are (re)ordered by ``self.notes`` iteration order so the rebuilt index
matches what a full scan over ``self.notes.values()`` would produce.
"""
self._backlink_index.clear()
self._backlink_out.clear()
self._note_order.clear()
self._note_seq = 0
for note in self.notes.values():
self._note_order[note.id] = self._note_seq
self._note_seq += 1
self._reindex_backlinks(note)
self._mark_backlinks_fresh()
def generate_unique_id(self, title: str) -> str:
"""Return a note id derived from ``title`` that does not collide.
``generate_id`` is minute-resolution + a title slug, so two notes created
in the same minute (same title, or titles with the same slug) would
otherwise produce identical ids and silently overwrite each other on
``save_note``. This appends ``-2``, ``-3``, ... until the id is free both
in the in-memory index and on disk.
"""
base = generate_id(title)
candidate = base
suffix = 2
while candidate in self.notes or (self.path / f"{candidate}.md").exists():
candidate = f"{base}-{suffix}"
suffix += 1
return candidate
def save_note(
self,
note: Note,
*,
lock_key: str | None = None,
graphs_dir: Path | str | None = None,
) -> Path:
"""Write a note to disk and index it (including embeddings).
The ``.md`` source of truth is written CRASH-SAFELY (temp + fsync +
rename via :func:`atomic_write_text`) and back-to-back with the in-memory
``_index_note`` mutation so a concurrent reader never observes a written
file whose index entry is missing/stale (or vice versa).
``lock_key`` (default ``None``) is the CROSS-PROCESS backstop for shared
spine-node writers: when provided, the write WINDOW (the atomic ``.md``
write + index mutation + incremental embedding index) is serialized under
:func:`zettelkasten.commit.review_write_lock` keyed by ``lock_key``, so a
separate OS process (the MCP server) writing the SAME node on the same key
cannot interleave. The lock is held ONLY around the brief write window —
never across a caller's wider multi-second build — so it never starves
another process holding the file lock (fixes the P1 head-of-line stall).
``lock_key=None`` preserves the historical unlocked behaviour, so existing
callers are unaffected. ``graphs_dir`` names the store base the lock file
derives from (defaults to this graph's own store root) so both processes
derive one shared lock; it is ignored when ``lock_key`` is ``None``.
"""
# The note id becomes a filename, so reject path separators, '..', and
# Windows filename hazards before touching the filesystem; safe_join is
# a defense-in-depth containment guard (also catches symlinked dirs).
validate_id(note.id, kind="note id", for_filename=True)
safe_join(self.path, f"{note.id}.md")
self.path.mkdir(parents=True, exist_ok=True)
filepath = self.path / f"{note.id}.md"
if lock_key is not None:
# Import lazily: ``commit`` imports ``graph`` (lazily), so a top-level
# import here would risk a circular import at module load.
from zettelkasten.commit import review_write_lock
base = graphs_dir if graphs_dir is not None else self.path.parent
with review_write_lock(lock_key, graphs_dir=base):
self._write_note_indexed(note, filepath)
else:
self._write_note_indexed(note, filepath)
_signal_note_written(self.name)
return filepath
def _write_note_indexed(self, note: Note, filepath: Path) -> None:
"""Atomically write ``note``'s ``.md`` and update every in-memory index.
The ``.md`` write and ``_index_note`` happen together so a reader never
sees file/index skew. Callers that need cross-process exclusion wrap this
in :func:`review_write_lock` via ``save_note``'s ``lock_key``.
"""
atomic_write_text(filepath, note_to_markdown(note))
self._index_note(note)
# Update the embedding index under the shared lock so this incremental
# embed cannot race a concurrent lazy build / reload / delete_note that
# also touches the same index — they all serialize on
# ``_embeddings_lock`` (mirrors delete_note and _invalidate_warm_citation).
# The lock is an RLock, so re-taking it after ``_index_note`` released it
# is safe.
#
# ``index_note`` embeds ONLY this one changed note (never the whole set)
# and stores its vector in the live graph synchronously, so a subsequent
# search/get_vector sees it immediately. The ``.kgl`` PERSIST, however, is
# debounced rather than run synchronously on every save: a burst of note
# writes coalesces into a single cache rewrite once the box goes quiet.
# This is safe because the ``.md`` source of truth was already written
# above (``atomic_write_text``) and the ``.kgl`` is a disposable cache
# rebuilt from the live notes on load — so a coalesced or crash-dropped
# persist can never diverge from a full rebuild (delete_note /
# remove_citation still persist synchronously, and a fired debounced save
# snapshots the current graph atomically). Scheduling the timer only takes
# the index's own tiny timer lock, so it does not extend time under
# ``_embeddings_lock`` or the native lock.
with self._embeddings_lock:
if self._embeddings is not None:
self._embeddings.index_note(note)
self._embeddings.save_cache_debounced()
# Mirror into the global ANN index, reusing the vector the per-box
# index just computed (no second embed). Best-effort: a global
# index failure must never break the .md write.
self._mirror_note_to_global(note)
def _mirror_note_to_global(self, note: "Note") -> None:
"""Upsert one just-written note into the global ANN index (best-effort).
Reuses the vector the per-box index computed for this exact text, so no
second embed happens on the write path. Skips federated (namespaced)
graphs — the global store is the LOCAL corpus only.
"""
if getattr(self, "repo_id", None):
return
try:
from zettelkasten.global_index import get_global_index
from zettelkasten.graph_io import source_date_int
gi = get_global_index()
if gi is None:
return
vec = None
if self._embeddings is not None:
vec = self._embeddings.get_vector(note.id)
date = source_date_int(self.name, graphs_dir=self.path.parent)
gi.upsert_note(self.name, note, date, vector=vec)
except Exception as exc: # pragma: no cover - best-effort
logger.debug("Global index: could not mirror note '%s': %s", note.id, exc)
def _mirror_box_to_global(self, index, notes: list["Note"]) -> None:
"""Mirror a freshly (re)built box into the global ANN index (best-effort).
Reuses the per-box index's vectors and only pushes notes whose text
changed since the global store last saw them (``skip_unchanged``), then
reconciles so a note deleted while the process was down cannot linger as a
global ghost. Skips federated (namespaced) graphs.
"""
if getattr(self, "repo_id", None):
return
try:
from zettelkasten.global_index import get_global_index
from zettelkasten.graph_io import source_date_int
gi = get_global_index()
if gi is None:
return
date = source_date_int(self.name, graphs_dir=self.path.parent)
vectors: dict[str, list[float]] = {}
for n in notes:
v = index.get_vector(n.id)
if v:
vectors[n.id] = v
gi.upsert_notes(self.name, notes, date, vectors=vectors, skip_unchanged=True)
gi.reconcile_box(self.name, [n.id for n in notes])
except Exception as exc: # pragma: no cover - best-effort
logger.debug("Global index: could not mirror box '%s': %s", self.name, exc)
def find_by_title(self, title: str) -> Note | None:
"""Look up a note by title or alias (case-insensitive)."""
lower = title.lower()
note_id = self._title_index.get(lower) or self._alias_index.get(lower)
return self.notes.get(note_id) if note_id else None
def search(self, query: str, limit: int | None = None) -> list[Note]:
"""Simple substring search across titles, aliases, tags, and body.
When ``limit`` is a positive int, scanning stops as soon as that many
matches are collected, so the cost is bounded on large graphs instead of
always scanning every note.
"""
q = query.lower()
results = []
for note in self.notes.values():
if (q in note.title.lower()
or q in note.body.lower()
or any(q in t for t in note.tags)
or any(q in a.lower() for a in note.aliases)):
results.append(note)
if limit is not None and limit > 0 and len(results) >= limit:
break
return results
def get_linked(self, note_id: str, relation: str | None = None) -> list[dict]:
"""Get all notes linked from the given note, optionally filtered by relation."""
note = self.notes.get(note_id)
if not note:
return []
results = []
for link in note.links:
if relation and link.relation != relation:
continue
target_note = self.notes.get(link.target)
entry: dict = {
"target_id": link.target,
"relation": link.relation,
"direction": link.direction,
"target_title": target_note.title if target_note else "(unresolved)",
}
if link.graph:
entry["graph"] = link.graph
results.append(entry)
return results
def get_backlinks(self, note_id: str) -> list[dict]:
"""Get all notes that link TO this note.
Served from the incrementally-maintained ``_backlink_index`` (O(number
of backlinks) rather than a scan of every note × every link). The result
is byte-for-byte identical to a full scan: sources are ordered by their
``self.notes`` insertion order (via ``_note_order``), a source's multiple
edges to this note appear once each in link order, the current
``source_title`` is used, and only sources still present in ``self.notes``
are emitted.
Correctness under direct ``self.notes`` mutation: the index is only
maintained via ``save_note``/``load``/``_index_note``, so code that
assigns or edits ``self.notes`` directly (``zg.notes = {...}`` in
discover_spines and fixtures; ``del self.notes[id]`` in server/spine)
would otherwise see an empty or stale index. We detect that via a cheap
staleness token and lazily rebuild the index from ``self.notes`` before
serving; the common indexed path never rebuilds.
This reflects INDEXED/PERSISTED state: the backlink index is maintained
from what has been saved (``save_note``/``load``/``_index_note``), so an
in-place edit to ``note.links`` is only visible here once it is followed
by ``save_note`` — the real link helpers (``server.link_notes``,
``spine.ensure_link``/``remove_link``) already do this. (This is why we
rely on the staleness token + lazy rebuild rather than an O(n)-per-call
scan of every note.)
Concurrency: the whole read runs under ``_embeddings_lock`` and the
buckets/notes are snapshotted into local lists INSIDE the lock, so the
row-building loop below cannot observe a torn structure. This does NOT
guarantee unconditional crash-freedom: it serializes ONLY against
mutators that ALSO hold ``_embeddings_lock`` (``save_note``,
``_reindex_backlinks``, ``load``, and ``spine.delete_node``). Any code
that mutates ``self.notes`` directly (assign/``pop``/``del``) MUST hold
``_embeddings_lock`` too, or a concurrent lazy rebuild iterating
``self.notes`` here can still raise "dictionary changed size during
iteration"/``KeyError``. The lock is re-entrant, so the lazy rebuild
nested here is safe.
"""
with self._embeddings_lock:
if self._backlink_index_is_stale():
self._rebuild_backlink_index()
by_source = self._backlink_index.get(note_id)
if not by_source:
return []
# Snapshot to live sources under the lock: copy each source's link
# list (it is mutated in place by _reindex_backlinks) and capture the
# ordering key + current title, so the result can be built after the
# lock is released.
snapshot = [
(
sid,
self._note_order.get(sid, 0),
self.notes[sid].title,
list(links),
)
for sid, links in by_source.items()
if sid in self.notes
]
snapshot.sort(key=lambda row: row[1])
results = []
for sid, _order, title, links in snapshot:
for link in links:
results.append({
"source_id": sid,
"source_title": title,
"relation": link.relation,
})
return results
def get_prerequisites_chain(self, note_id: str, visited: set | None = None) -> list[str]:
"""Recursively resolve the prerequisite chain for a note."""
if visited is None:
visited = set()
note = self.notes.get(note_id)
if not note or note_id in visited:
return []
visited.add(note_id)
chain = []
for prereq_id in note.prerequisites:
chain.extend(self.get_prerequisites_chain(prereq_id, visited))
chain.append(prereq_id)
return chain
def list_notes(self, type_filter: str | None = None, tag: str | None = None) -> list[dict]:
"""List notes with optional filters, returning summaries."""
results = []
for note in self.notes.values():
if type_filter and note.type != type_filter:
continue
if tag and tag not in note.tags:
continue
results.append({
"id": note.id,
"title": note.title,
"type": note.type,
"tags": note.tags,
"link_count": len(note.links),
"aliases": note.aliases,
"excerpt": lead_gloss(note.body),
# A curated glossary entry (user-added via the Inspect "Add" bar)
# has an empty source and no grounding; a paper-anchored evidence
# note (ground_paper) carries one or both. The Terms/Concepts
# views use this to keep evidence out of the working glossary.
"grounded": bool(note.grounding) or bool(note.source),
})
return sorted(results, key=lambda x: x["id"])
def missing_definitions(self) -> list[dict]:
"""Find terms this graph references but never defines.
A term is "referenced" when a note links to it or lists it as a
prerequisite; it's "missing" when that reference doesn't resolve to any
note in this graph (by id, title, or alias). Cross-graph links (those
carrying an explicit foreign `graph`) are ignored — only same-graph gaps
count. Results are ranked by how many notes reference the term, so the
backbone concepts the graph leans on most surface first. This powers the
glossary's "missing definitions" surface.
"""
ids = set(self.notes.keys())
titles = {n.title.strip().lower() for n in self.notes.values()}
aliases = {a.strip().lower() for n in self.notes.values() for a in n.aliases}
def resolved(ref: str) -> bool:
r = ref.strip()
return bool(r) and (r in ids or r.lower() in titles or r.lower() in aliases)
candidates: dict[str, dict] = {}
for note in self.notes.values():
refs: list[str] = []
for link in note.links:
if link.graph and link.graph != self.name:
continue # cross-graph reference — not this graph's gap
refs.append(link.target)
refs.extend(note.prerequisites or [])
for ref in refs:
if resolved(ref):
continue
term = ref.strip()
if not term:
continue
entry = candidates.setdefault(
term.lower(), {"term": term, "count": 0, "referenced_by": []}
)
entry["count"] += 1
if note.title not in entry["referenced_by"]:
entry["referenced_by"].append(note.title)
return sorted(candidates.values(), key=lambda c: (-c["count"], c["term"].lower()))