Present the memory tree as a read-only, ZK-retrievable source graph.
zettelkasten.framing.project_query retrieves over a list of "source graphs",
each of which must expose:
.notes — a dict[id, note] where each note has .id/.title/.type/
.tags/.aliases/.body/.links (the keyword channel iterates these)..embeddings — optional; an object with .search(query, top_k) ->
list[(id, score)] (the semantic channel; skipped when None)..get_backlinks(id) — list of {relation: ...} dicts.
:class:MemorySourceGraph wraps the .memory/ tree in exactly that shape so a
memory entry can surface side-by-side with ZK notes in unified retrieval. It is
strictly READ-ONLY: it never writes .memory/ and never rebuilds/overwrites
memory's own memory.kgl cache.
Provenance of a memory hit is its entry id (e.g. deci-198cef35); the entry's
git-pinned files are carried along for downstream connection typing.
Embedding reuse (no corpus re-embed): both stores embed with the same 512-d
model2vec model, so the semantic channel simply LOADS memory's existing
memory.kgl (whose vectors were computed by the memory MCP server) and embeds
only the query at search time. If that cache is missing, a minimal entry-only
index is built in-process as a fallback (never persisted, so memory keeps sole
ownership of its cache).
kglite has process-global native state, so every native call here serializes on
the SAME lock the zettelkasten embedding index uses (_NATIVE_LOCK) — mixing
handles without a shared lock can SIGSEGV.
MemoryNote
dataclass
A memory entry adapted to the note surface project_query reads.
links/backlinks are intentionally empty: cross-store relationships live
in synapse's own overlay, and within-memory edges (related_to) aren't
part of retrieval. entry keeps the raw record so downstream (connection
typing, provenance) can read files/parent_id/timestamps.
Source code in zettelkasten/synapse/memory_source.py
| @dataclass
class MemoryNote:
"""A memory entry adapted to the note surface ``project_query`` reads.
``links``/backlinks are intentionally empty: cross-store relationships live
in synapse's own overlay, and within-memory edges (``related_to``) aren't
part of retrieval. ``entry`` keeps the raw record so downstream (connection
typing, provenance) can read ``files``/``parent_id``/timestamps.
"""
id: str
title: str
type: str
tags: list[str]
body: str
aliases: list[str] = field(default_factory=list)
links: list[Any] = field(default_factory=list)
entry: dict[str, Any] = field(default_factory=dict)
|
MemorySemanticIndex
Semantic search over memory entries via memory's existing vectors.
Loads memory.kgl (query-time embedding only). On a missing cache, builds
a minimal entry-only index in-process as a fallback. Mirrors
:meth:zettelkasten.embeddings.EmbeddingIndex.search.
Source code in zettelkasten/synapse/memory_source.py
| class MemorySemanticIndex:
"""Semantic search over memory entries via memory's existing vectors.
Loads ``memory.kgl`` (query-time embedding only). On a missing cache, builds
a minimal entry-only index in-process as a fallback. Mirrors
:meth:`zettelkasten.embeddings.EmbeddingIndex.search`.
"""
def __init__(self, entries: list[dict[str, Any]], embedder: Any) -> None:
self._entries = entries
self._embedder = embedder
self._graph: Any = None
self._loaded = False
self._lock = _native_lock()
def _ensure_graph(self) -> None:
if self._loaded:
return
# Caller holds self._lock.
if self._embedder is None:
self._loaded = True
return
import kglite
kgl_path = _memory_kgl_path()
graph: Any = None
if kgl_path.exists():
try:
graph = kglite.load(str(kgl_path))
sig_path = storage.sig_path_for(kgl_path)
fresh = (
sig_path.exists()
and sig_path.read_text(encoding="utf-8") == storage.source_fileset_signature()
)
if not fresh:
logger.info(
"synapse: memory.kgl is stale vs .memory/ sources; using it "
"anyway for retrieval (vectors may lag recent edits)."
)
except Exception as exc:
logger.warning("synapse: could not load memory.kgl (%s); rebuilding minimal index", exc)
graph = None
if graph is None:
graph = self._build_fallback_graph(kglite)
if graph is not None:
try:
graph.set_embedder(self._embedder)
except Exception as exc:
logger.warning("synapse: could not register embedder on memory index: %s", exc)
graph = None
self._graph = graph
self._loaded = True
def _build_fallback_graph(self, kglite: Any) -> Any:
"""Build an entry-only kglite index from ``.memory/`` files (not persisted)."""
try:
import pandas as pd
rows: list[dict[str, Any]] = []
for e in self._entries:
eid = e.get("id")
if not eid:
continue
title = str(e.get("title", ""))
tags = _coerce_tags(e.get("tags"))
body = str(e.get("body", ""))
plan = e.get("plan_content") or ""
if plan:
body = f"{body}\n{plan}"
rows.append(
{"id": str(eid), "title": title, "search_text": _build_search_text(title, tags, body)}
)
if not rows:
return None
g = kglite.KnowledgeGraph()
g.add_nodes(
pd.DataFrame(rows),
node_type="entry",
unique_id_field="id",
node_title_field="title",
nullable_int_downcast=True,
)
g.set_embedder(self._embedder)
g.embed_texts("entry", "search_text", replace=False, show_progress=False)
logger.info("synapse: built fallback memory index (%d entries)", len(rows))
return g
except Exception as exc:
logger.warning("synapse: fallback memory index build failed: %s", exc)
return None
def search(
self, query: str, top_k: int = 10, exclude_ids: set[str] | None = None
) -> list[tuple[str, float]]:
"""Semantic search: ``list[(entry_id, similarity_score)]`` (empty on any failure)."""
exclude = set(exclude_ids or set())
with self._lock:
self._ensure_graph()
if self._graph is None:
return []
want = top_k + len(exclude)
try:
hits = self._graph.select("entry").search_text("search_text", query, top_k=want)
except Exception as exc:
logger.warning("synapse: memory semantic search failed: %s", exc)
return []
results: list[tuple[str, float]] = []
for hit in hits or []:
nid = hit.get("id")
if not nid or nid in exclude:
continue
results.append((str(nid), float(hit.get("score", 0.0))))
if len(results) >= top_k:
break
return results
|
search
search(query: str, top_k: int = 10, exclude_ids: set[str] | None = None) -> list[tuple[str, float]]
Semantic search: list[(entry_id, similarity_score)] (empty on any failure).
Source code in zettelkasten/synapse/memory_source.py
| def search(
self, query: str, top_k: int = 10, exclude_ids: set[str] | None = None
) -> list[tuple[str, float]]:
"""Semantic search: ``list[(entry_id, similarity_score)]`` (empty on any failure)."""
exclude = set(exclude_ids or set())
with self._lock:
self._ensure_graph()
if self._graph is None:
return []
want = top_k + len(exclude)
try:
hits = self._graph.select("entry").search_text("search_text", query, top_k=want)
except Exception as exc:
logger.warning("synapse: memory semantic search failed: %s", exc)
return []
results: list[tuple[str, float]] = []
for hit in hits or []:
nid = hit.get("id")
if not nid or nid in exclude:
continue
results.append((str(nid), float(hit.get("score", 0.0))))
if len(results) >= top_k:
break
return results
|
MemorySourceGraph
The .memory/ tree adapted to the ZK source-graph surface (read-only).
Source code in zettelkasten/synapse/memory_source.py
| class MemorySourceGraph:
"""The ``.memory/`` tree adapted to the ZK source-graph surface (read-only)."""
def __init__(self, entries: list[dict[str, Any]] | None = None, embedder: Any = None) -> None:
if entries is None:
entries = _read_entries()
self.name = MEMORY_SOURCE
self.notes: dict[str, MemoryNote] = {}
for e in entries:
eid = e.get("id")
if not eid:
continue
self.notes[str(eid)] = MemoryNote(
id=str(eid),
title=str(e.get("title", "")),
type=str(e.get("type", "note")),
tags=_coerce_tags(e.get("tags")),
body=str(e.get("body", "")),
entry=e,
)
self.embeddings: MemorySemanticIndex | None = (
MemorySemanticIndex(entries, embedder) if embedder is not None else None
)
def get_backlinks(self, note_id: str) -> list[dict[str, Any]]:
# Memory has no supports/replicates edges; corroboration is a ZK concept.
return []
|
get_memory_source
Return a cached :class:MemorySourceGraph, rebuilt when .memory/ changes.
The cache is keyed on storage.source_fileset_signature() so an edit,
add, delete, or rename of any memory source file transparently rebuilds the
.notes map and drops the stale semantic index.
Source code in zettelkasten/synapse/memory_source.py
| def get_memory_source(force_refresh: bool = False) -> MemorySourceGraph:
"""Return a cached :class:`MemorySourceGraph`, rebuilt when ``.memory/`` changes.
The cache is keyed on ``storage.source_fileset_signature()`` so an edit,
add, delete, or rename of any memory source file transparently rebuilds the
``.notes`` map and drops the stale semantic index.
"""
global _CACHED_SOURCE, _CACHED_SIG
sig = storage.source_fileset_signature()
with _CACHE_LOCK:
if not force_refresh and _CACHED_SOURCE is not None and _CACHED_SIG == sig:
return _CACHED_SOURCE
entries = _read_entries()
_CACHED_SOURCE = MemorySourceGraph(entries, embedder=_shared_embedder())
_CACHED_SIG = sig
return _CACHED_SOURCE
|