class GraphLoader:
"""Rebuilds KGLite from .memory/ markdown files, with mtime-based reload."""
def __init__(self, path: Path):
self._path = path
self._graph: kglite.KnowledgeGraph | None = None
self._mtime: float = 0.0
# Exact LOCAL source-fileset signature the live graph was built from — the
# sorted (path, st_mtime_ns) set over project.md + every *.md under the
# entries/skills/sessions/documents dirs (the derived .kgl is excluded;
# see storage.source_fileset_signature). Rebuilds gate on THIS (see
# _resolve), not merely on the newest mtime, so a delete of a non-newest
# file or a same-count/older-mtime swap — neither of which advances the
# newest mtime — still triggers a rebuild. ``None`` until the first build.
# Peer/federation files are deliberately excluded.
self._fileset_sig: str | None = None
# Monotonic timestamp of the last filesystem mtime scan, used to coalesce
# the frequent polling requests into at most one scan (and so at most one
# rebuild) per _SCAN_MIN_INTERVAL window.
self._last_scan: float = 0.0
# Re-entrant so the _LockedGraph facade can hold it across a native call
# without deadlocking helpers that re-lock. Guards every touch of the
# Rust-backed graph (build, read, write, save) — see _LockedGraph.
self._lock = threading.RLock()
self._memory_dir = storage.MEMORY_DIR
# Persistent cold-start caches. Rebuilding the graph from a large .memory/
# tree costs ~30s (50k entries); when the source fileset is UNCHANGED we
# instead kglite.load() a snapshot in sub-second. Loading is gated by a
# .sig sidecar holding the EXACT source-fileset signature the snapshot was
# built from (the same signature that drives _resolve's rebuild gate), so
# any local change — edit, add, delete, or same-count/older-mtime swap —
# misses the cache and rebuilds. Two caches are consulted (see
# _load_snapshot), both gitignored/disposable under .angelo/:
# • the PRIVATE snapshot — written only by THIS loader when it rebuilds,
# so there is never a cross-process write race on it; makes a dashboard
# RESTART fast.
# • the SHARED memory.kgl (self._path) the memory server persists with a
# matching .sig sidecar — we only ever READ it (the server owns writes)
# to make the FIRST dashboard open fast when an agent session already
# built the graph (cross-process cold start).
# Disable both via DASHBOARD_GRAPH_SNAPSHOT=0.
self._snapshot_enabled = os.environ.get("DASHBOARD_GRAPH_SNAPSHOT", "1") != "0"
self._snapshot_path = path.parent / "dashboard-graph.kgl"
self._snapshot_sig_path = storage.sig_path_for(self._snapshot_path)
self._shared_sig_path = storage.sig_path_for(self._path)
def _load_snapshot(self, sig: str) -> "kglite.KnowledgeGraph | None":
"""Load a persisted graph cache IFF it was built from ``sig``.
Tries two signature-gated caches in order and returns the first whose
recorded signature matches the CURRENT source fileset:
1. the PRIVATE dashboard snapshot — fast path for a dashboard *restart*;
2. the SHARED ``memory.kgl`` the memory server persists — fast path for the
*first* dashboard open after an agent session already built the graph
(cross-process cold start). This is the same kglite.load ``_resolve``
already performs when ``.memory/`` is absent, so it is schema-safe; the
signature gate just proves the server's cache matches the live sources.
Returns ``None`` on any miss, signature mismatch, or corruption so the
caller falls back to a full rebuild — a cache is a pure accelerator, never
a source of truth. Each ``.sig`` sidecar is written AFTER its ``.kgl`` (see
:meth:`_save_snapshot` and :func:`memory.storage.write_cache_signature`),
so a signature match implies a complete cache; a torn write leaves the sig
absent or stale → miss → rebuild.
"""
if not self._snapshot_enabled:
return None
private = self._try_load_cache(self._snapshot_path, self._snapshot_sig_path, sig)
if private is not None:
logger.info("Dashboard graph loaded from private snapshot (fileset unchanged)")
return private
shared = self._try_load_cache(self._path, self._shared_sig_path, sig)
if shared is not None:
logger.info(
"Dashboard graph adopted from memory-server cache %s (fileset matches)",
self._path.name,
)
return shared
return None
def _try_load_cache(
self, kgl_path: Path, sig_path: Path, sig: str
) -> "kglite.KnowledgeGraph | None":
"""``kglite.load`` ``kgl_path`` IFF ``sig_path`` records exactly ``sig``.
Any miss (either file absent), signature mismatch, or load error returns
``None`` — the cache is an accelerator, so a failure just means "rebuild".
"""
try:
if not kgl_path.exists() or not sig_path.exists():
return None
if sig_path.read_text(encoding="utf-8") != sig:
return None
return kglite.load(str(kgl_path))
except Exception:
logger.warning(
"Graph cache load failed for %s; ignoring", kgl_path, exc_info=True
)
return None
def _save_snapshot(self, sig: str) -> None:
"""Persist the live graph + its fileset signature (best-effort).
Ordering is load-bearing: the snapshot is written first (atomically, via
``atomic_graph_save``), the ``.sig`` sidecar second (atomic temp+rename).
If the process dies between the two, the sig is missing or stale, so the
next start rebuilds rather than loading a snapshot that doesn't match its
recorded fileset. A save failure is non-fatal — we just forgo the cache.
Called under ``self._lock`` (from _resolve) so ``self._graph`` is stable.
"""
if not self._snapshot_enabled or self._graph is None:
return
try:
atomic_graph_save(self._graph, self._snapshot_path)
tmp = self._snapshot_sig_path.parent / (self._snapshot_sig_path.name + ".tmp")
tmp.write_text(sig, encoding="utf-8")
tmp.replace(self._snapshot_sig_path)
except Exception:
logger.warning("Dashboard graph snapshot save failed (non-fatal)", exc_info=True)
def _latest_mtime(self) -> float:
"""Get the newest mtime across all .memory/ source files."""
latest = 0.0
if self._path.exists():
latest = max(latest, self._path.stat().st_mtime)
for subdir in [storage.ENTRIES_DIR, storage.SKILLS_DIR, storage.SESSIONS_DIR]:
if subdir.exists():
for f in subdir.glob("*.md"):
latest = max(latest, f.stat().st_mtime)
return latest
def _local_fileset_signature(self) -> str:
"""Exact signature of the LOCAL ``.memory/`` source files the graph is built from.
Delegates to :func:`memory.storage.source_fileset_signature` — the single
canonical implementation shared with ``routes/tree._local_fileset_signature``
(a guard test asserts parity) and the memory server. Each source file
contributes its path AND ``st_mtime_ns``, so an add/delete/rename (a changed
path *set*) or an in-place edit (a changed mtime) both move the value —
unlike :meth:`_latest_mtime`, which a delete of a non-newest file or a
same-count/older-mtime swap leaves unchanged.
The derived ``.kgl`` snapshot is DELIBERATELY excluded from the key so it
can double as a cross-process cache-validity token (see the helper's
docstring). Peer/federation files are also excluded: the federated
projection reads peers fresh on every build, so a peer-only change must
never force a local KGLite rebuild (see ``ensure_local_fresh`` / trap 2).
"""
return storage.source_fileset_signature()
def _resolve(self) -> kglite.KnowledgeGraph:
"""Return the live raw graph, rebuilding it if the LOCAL fileset changed.
A reload builds a fresh graph and atomically rebinds ``self._graph``, so
a facade still holding the previous object keeps reading a consistent,
fully-built graph rather than one mutated in place. Callers must not run
native operations on the returned object except through a _LockedGraph.
The rebuild gate is the EXACT local fileset signature, not merely a bump
in the newest mtime: an mtime-only gate misses a delete of a non-newest
file and a same-count/older-mtime swap (neither advances the max mtime),
so it would keep serving a stale graph for those. ``_latest_mtime`` is
still used only to set :attr:`version` on a rebuild, preserving the
``version == newest mtime`` contract other caches rely on.
"""
if not self._memory_dir.exists() and not self._path.exists():
raise FileNotFoundError(f"Neither .memory/ nor {self._path} found")
# Coalesce the rapid polling requests: once a graph is built, only
# re-scan the filesystem (an O(files) stat walk) at most once per
# _SCAN_MIN_INTERVAL. Within that window a poll serves the cached graph
# without stat'ing or rebuilding, so a busy dashboard can't peg the CPU
# rebuilding on every request. invalidate() and a drifted
# ensure_local_fresh() both reset _last_scan, so a delete/swap/explicit
# refresh still takes effect immediately (the conditional-GET body path
# threads its exact signature through ensure_local_fresh, so it never
# depends on this window elapsing — see the class of bugs in trap 1).
if self._graph is not None:
now = time.monotonic()
if now - self._last_scan < _SCAN_MIN_INTERVAL:
return self._graph
self._last_scan = now
# With a live graph we must re-walk to compare signatures; with no graph
# (first access, invalidate(), or an ensure_local_fresh() drop) a rebuild
# is unconditional, so trust a signature ensure_local_fresh() already
# threaded rather than repeat its O(files) walk.
if self._graph is not None:
current_sig = self._local_fileset_signature()
if current_sig == self._fileset_sig:
return self._graph
else:
current_sig = (
self._fileset_sig
if self._fileset_sig is not None
else self._local_fileset_signature()
)
with self._lock:
# Double-checked: a concurrent call may have already rebuilt to a
# graph matching current_sig while we waited for the lock.
if self._graph is not None and current_sig == self._fileset_sig:
return self._graph
if self._memory_dir.exists():
# Fast path: an unchanged fileset since our last build means the
# persisted snapshot is still valid — load it (sub-second) instead
# of rebuilding from every markdown file (~30s at 50k). On any
# miss/corruption _load_snapshot returns None and we rebuild, then
# refresh the snapshot for the next cold start.
snapshot = self._load_snapshot(current_sig)
if snapshot is not None:
self._graph = snapshot
else:
self._graph = _rebuild_from_files()
logger.info("Dashboard graph rebuilt from .memory/ files")
self._save_snapshot(current_sig)
else:
self._graph = kglite.load(str(self._path))
logger.info("Dashboard graph loaded from %s", self._path)
self._mtime = self._latest_mtime()
self._fileset_sig = current_sig
self._last_scan = time.monotonic()
return self._graph
@property
def graph(self) -> _LockedGraph:
"""Thread-safe handle to the graph.
Returns a :class:`_LockedGraph` facade bound to the currently-live graph,
so every native call a handler makes is serialized behind ``self._lock``.
KGLite is not safe for concurrent access; without this, two threadpool
requests touching the graph at once crash the process (PyBorrowError).
"""
return _LockedGraph(self._lock, self._resolve())
@property
def version(self) -> float:
"""Freshness token that advances whenever the graph is rebuilt.
Equals the newest source mtime the live graph was built from, so callers
can cache derived structures (e.g. a search corpus) and rebuild only when
this changes. ``invalidate()`` resets it so forced refreshes are seen.
"""
return self._mtime
def ensure_local_fresh(self, local_sig: str) -> None:
"""Guarantee the next graph access reflects the current LOCAL fileset.
Called by a route BEFORE it reads :attr:`graph` to assemble a 200 body,
passing the exact local signature it ALREADY computed for its cache key /
ETag (so no second O(files) walk is done). This closes the last
anti-stale hole: the route's cache key is exact/uncoalesced, so a delete
of a non-newest file or a same-count/older-mtime swap is a cache MISS —
but ``_resolve``'s own scan is coalesced behind ``_SCAN_MIN_INTERVAL``, so
without this the body could still be assembled over the stale cached
graph within that window. When the passed signature differs from the one
the live graph was built against we drop the graph and reset the scan
clock so the very next :attr:`graph` access rebuilds for THIS change,
bypassing the coalescing window; when it matches we refresh the scan clock
so the immediately-following ``_resolve`` serves the cached graph without
repeating the walk.
``local_sig`` is the LOCAL signature only — peer/federation files are
excluded — so a peer-only change never forces a local KGLite rebuild
(the federated projection reads peers fresh on every build; trap 2).
"""
with self._lock:
if self._graph is not None and local_sig == self._fileset_sig:
self._last_scan = time.monotonic()
return
# Drift (or nothing built yet): drop the cached graph and record the
# signature to rebuild against. Storing it here lets _resolve trust it
# (it rebuilds unconditionally when _graph is None) instead of
# repeating the O(files) walk the caller already did.
self._graph = None
self._fileset_sig = local_sig
self._last_scan = 0.0
def invalidate(self) -> None:
"""Force a rebuild from .memory/ files on the next graph access.
Used after mutations (e.g. deletes) that the mtime-based reload can't
detect — removing a file doesn't advance the newest source mtime, so the
cached graph would otherwise keep serving the deleted node.
"""
with self._lock:
self._graph = None
self._mtime = 0.0
self._fileset_sig = None
self._last_scan = 0.0
def save(self) -> None:
"""Persist the in-memory graph back to disk atomically."""
with self._lock:
if self._graph is not None:
atomic_graph_save(self._graph, self._path)
self._mtime = self._path.stat().st_mtime
# Saving rewrites the .kgl (advancing its mtime), which is part of
# the local fileset signature. Refresh it so the just-persisted
# in-memory graph isn't needlessly rebuilt on the next access.
self._fileset_sig = self._local_fileset_signature()