memory.storage

memory.storage

File-based storage for the memory tree.

Entries, skills, sessions, and documents are stored as individual markdown files with YAML frontmatter inside .memory/. The KGLite .memory.kgl file becomes a disposable cache rebuilt from these files.

Directory layout::

.memory/
    config.yaml
    project.md
    entries/<id>.md
    skills/<name>.md
    sessions/<id>.md
    documents/<id>.md
    archive/entries/<id>.md

Derived caches (git-ignored, under .angelo/)::

.angelo/memory.kgl
.angelo/memory.index.json
.angelo/memory.embeddings.json

pid_alive

pid_alive(pid: int) -> bool

Best-effort check whether a process with this PID is still running.

Used to detect live-session sidecars left behind by dead MCP processes. PID reuse can cause false positives; heartbeat staleness covers that case.

Source code in memory/storage.py

def pid_alive(pid: int) -> bool:
    """Best-effort check whether a process with this PID is still running.

    Used to detect live-session sidecars left behind by dead MCP processes.
    PID reuse can cause false positives; heartbeat staleness covers that case.
    """
    if not isinstance(pid, int) or pid <= 0:
        return False
    if os.name == "nt":
        import ctypes
        import ctypes.wintypes

        PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
        STILL_ACTIVE = 259
        handle = ctypes.windll.kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
        if not handle:
            return False
        try:
            exit_code = ctypes.wintypes.DWORD()
            ok = ctypes.windll.kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code))
            return bool(ok) and exit_code.value == STILL_ACTIVE
        finally:
            ctypes.windll.kernel32.CloseHandle(handle)
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except OSError:
        return False
    return True

source_fileset_signature

source_fileset_signature() -> str

Exact signature of the local .memory/ SOURCE files the graph is built from.

Covers project.md plus every *.md under the entries, skills, sessions, and documents directories — the authoritative source of truth. Each 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. The parts are \x00-joined; the directory globals are read at call time so tests that monkeypatch them (or a re-anchored workspace) are honoured.

Derived caches (the .kgl snapshot) are DELIBERATELY excluded. They are rebuilt from these files, so folding a cache's own mtime into the freshness key would make it self-referential: merely re-saving the cache would invalidate it. Keeping the signature source-only also lets it double as a cross-process cache-validity token — any process that rebuilds from these files can persist the resulting .kgl beside this signature (see :func:write_cache_signature), and any other process may adopt that cache iff the signature still matches its own view of the sources.

Source code in memory/storage.py

def source_fileset_signature() -> str:
    """Exact signature of the local ``.memory/`` SOURCE files the graph is built from.

    Covers ``project.md`` plus every ``*.md`` under the entries, skills, sessions,
    and documents directories — the authoritative source of truth. Each 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. The parts
    are ``\\x00``-joined; the directory globals are read at call time so tests that
    monkeypatch them (or a re-anchored workspace) are honoured.

    Derived caches (the ``.kgl`` snapshot) are DELIBERATELY excluded. They are
    rebuilt *from* these files, so folding a cache's own mtime into the freshness
    key would make it self-referential: merely re-saving the cache would
    invalidate it. Keeping the signature source-only also lets it double as a
    *cross-process cache-validity token* — any process that rebuilds from these
    files can persist the resulting ``.kgl`` beside this signature (see
    :func:`write_cache_signature`), and any other process may adopt that cache iff
    the signature still matches its own view of the sources.
    """
    parts: list[str] = []
    # project.md holds the project name/metadata rendered by /tree. A local
    # project rename edits only this file — no entry/skill/session/document file
    # changes — so its (path, mtime) must be folded in explicitly or a rename
    # would leave the key (and any adopted cache) stale.
    project_file = MEMORY_DIR / "project.md"
    try:
        parts.append(f"{project_file}|{project_file.stat().st_mtime_ns}")
    except OSError:
        pass
    for subdir in (ENTRIES_DIR, SKILLS_DIR, SESSIONS_DIR, DOCUMENTS_DIR):
        try:
            files = sorted(subdir.glob("*.md"))
        except OSError:
            continue
        for f in files:
            try:
                parts.append(f"{f}|{f.stat().st_mtime_ns}")
            except OSError:
                continue
    return "\x00".join(parts)

sig_path_for

sig_path_for(kgl_path: PathLike | str) -> Path

Path of the .sig sidecar that records a .kgl cache's source signature.

A single convention (<name>.kgl → <name>.kgl.sig) shared by every writer (the memory server) and reader (the dashboard loader) so they agree on where the cross-process validity token lives.

Source code in memory/storage.py

def sig_path_for(kgl_path: os.PathLike | str) -> Path:
    """Path of the ``.sig`` sidecar that records a ``.kgl`` cache's source signature.

    A single convention (``<name>.kgl`` → ``<name>.kgl.sig``) shared by every
    writer (the memory server) and reader (the dashboard loader) so they agree on
    where the cross-process validity token lives.
    """
    p = Path(kgl_path)
    return p.parent / (p.name + ".sig")

write_cache_signature

write_cache_signature(kgl_path: PathLike | str) -> Path

Persist the current source-fileset signature beside a .kgl cache.

Call this immediately after saving the .kgl so a separate process can validate and adopt that cache on a cold start (skipping a full rebuild) iff the signature still matches its own view of the source files. Writing the sig second is load-bearing: a torn write leaves the sig absent or stale, which a reader treats as a miss and falls back to rebuilding — never a false adoption. The write is atomic; callers may ignore failures (a missing sig is a safe skip).

Source code in memory/storage.py

def write_cache_signature(kgl_path: os.PathLike | str) -> Path:
    """Persist the current source-fileset signature beside a ``.kgl`` cache.

    Call this immediately *after* saving the ``.kgl`` so a separate process can
    validate and adopt that cache on a cold start (skipping a full rebuild) iff
    the signature still matches its own view of the source files. Writing the sig
    second is load-bearing: a torn write leaves the sig absent or stale, which a
    reader treats as a miss and falls back to rebuilding — never a false adoption.
    The write is atomic; callers may ignore failures (a missing sig is a safe skip).
    """
    return atomic_write(sig_path_for(kgl_path), source_fileset_signature())

migrate_legacy_caches

migrate_legacy_caches(kgl_target: Path | None = None) -> None

Move legacy cache locations into ANGELO_DIR.

Handles both pre-.cassius root-level files and the pre-.angelo .cassius/ directory. Safe to call on every startup: only moves a legacy location when it exists and the new location doesn't.

Parameters:

Name	Type	Description	Default
kgl_target	Path | None	the configured MEMORY_FILE path. The legacy .memory.kgl is only moved there when the target differs from the legacy location (i.e. the user hasn't pinned the old path via MEMORY_FILE_PATH).	None

Source code in memory/storage.py

def migrate_legacy_caches(kgl_target: Path | None = None) -> None:
    """Move legacy cache locations into ``ANGELO_DIR``.

    Handles both pre-.cassius root-level files and the pre-.angelo
    ``.cassius/`` directory. Safe to call on every startup: only moves a
    legacy location when it exists and the new location doesn't.

    Args:
        kgl_target: the configured ``MEMORY_FILE`` path. The legacy
            ``.memory.kgl`` is only moved there when the target differs from
            the legacy location (i.e. the user hasn't pinned the old path via
            ``MEMORY_FILE_PATH``).
    """
    legacy_dir = Path(".cassius")
    if legacy_dir.exists() and not ANGELO_DIR.exists():
        try:
            legacy_dir.replace(ANGELO_DIR)
            logger.info("Migrated legacy cache dir %s -> %s", legacy_dir, ANGELO_DIR)
        except OSError as e:
            logger.warning("Could not migrate %s: %s", legacy_dir, e)
    _move_legacy(Path(".memory.index.json"), INDEX_FILE)
    _move_legacy(Path(".memory.embeddings.json"), EMBEDDINGS_FILE)
    legacy_kgl = Path(".memory.kgl")
    if kgl_target is not None and kgl_target.resolve() != legacy_kgl.resolve():
        _move_legacy(legacy_kgl, kgl_target)

write_entry

write_entry(entry: dict[str, Any]) -> Path

Write an entry to .memory/entries/<id>.md.

Source code in memory/storage.py

def write_entry(entry: dict[str, Any]) -> Path:
    """Write an entry to ``.memory/entries/<id>.md``."""
    _ensure_dirs()
    meta, body = _split_meta_body(entry, _ENTRY_META_FIELDS)
    entry_id = meta["id"]
    return _write_md(_id_filename(ENTRIES_DIR, entry_id, kind="entry id"), meta, body)

write_skill

write_skill(skill: dict[str, Any]) -> Path

Write a skill to .memory/skills/<name>.md.

Source code in memory/storage.py

def write_skill(skill: dict[str, Any]) -> Path:
    """Write a skill to ``.memory/skills/<name>.md``."""
    _ensure_dirs()
    meta, body = _split_meta_body(skill, _SKILL_META_FIELDS, body_key="content")
    name = _safe_filename(meta.get("name") or meta["id"])
    return _write_md(SKILLS_DIR / f"{name}.md", meta, body)

write_session

write_session(session: dict[str, Any]) -> Path

Write a session to .memory/sessions/<id>.md.

Source code in memory/storage.py

def write_session(session: dict[str, Any]) -> Path:
    """Write a session to ``.memory/sessions/<id>.md``."""
    _ensure_dirs()
    meta, body = _split_meta_body(session, _SESSION_META_FIELDS, body_key="search_text")
    session_id = meta["id"]
    return _write_md(_id_filename(SESSIONS_DIR, session_id, kind="session id"), meta, body)

write_document

write_document(doc: dict[str, Any]) -> Path

Write a document to .memory/documents/<id>.md.

Source code in memory/storage.py

def write_document(doc: dict[str, Any]) -> Path:
    """Write a document to ``.memory/documents/<id>.md``."""
    _ensure_dirs()
    meta, body = _split_meta_body(doc, _DOCUMENT_META_FIELDS, body_key="description")
    doc_id = meta["id"]
    return _write_md(_id_filename(DOCUMENTS_DIR, doc_id, kind="document id"), meta, body)

write_project

write_project(project: dict[str, Any]) -> Path

Write project metadata to .memory/project.md.

Source code in memory/storage.py

def write_project(project: dict[str, Any]) -> Path:
    """Write project metadata to ``.memory/project.md``."""
    _ensure_dirs()
    meta, body = _split_meta_body(project, _PROJECT_META_FIELDS, body_key="description")
    return _write_md(MEMORY_DIR / "project.md", meta, body)

archive_entry

archive_entry(entry_id: str) -> Path | None

Move an entry file to archive/entries/. Returns the new path or None.

Source code in memory/storage.py

def archive_entry(entry_id: str) -> Path | None:
    """Move an entry file to ``archive/entries/``.  Returns the new path or None."""
    src = _id_filename(ENTRIES_DIR, entry_id, kind="entry id")
    if not src.exists():
        return None
    _ensure_dirs()
    dst = _id_filename(ARCHIVE_ENTRIES_DIR, entry_id, kind="entry id")
    shutil.move(str(src), str(dst))
    return dst

update_entry_field

update_entry_field(entry_id: str, updates: dict[str, Any]) -> Path | None

Read an existing entry file, apply field updates, write it back.

The read-modify-write is guarded by a per-id cross-process lock so two MCP processes updating the same entry cannot lose each other's field changes. write_entry itself does not take the lock, so there is no nested-lock deadlock when it is called from inside the critical section.

Source code in memory/storage.py

def update_entry_field(entry_id: str, updates: dict[str, Any]) -> Path | None:
    """Read an existing entry file, apply field updates, write it back.

    The read-modify-write is guarded by a per-id cross-process lock so two MCP
    processes updating the same entry cannot lose each other's field changes.
    ``write_entry`` itself does not take the lock, so there is no nested-lock
    deadlock when it is called from inside the critical section.
    """
    path = _id_filename(ENTRIES_DIR, entry_id, kind="entry id")
    with file_lock(_update_lock_path("entry", entry_id), timeout=_LOCK_TIMEOUT):
        if not path.exists():
            return None
        data = read_file(path)
        data.update(updates)
        return write_entry(data)

update_skill_field

update_skill_field(name: str, updates: dict[str, Any]) -> Path | None

Read an existing skill file, apply field updates, write it back.

Source code in memory/storage.py

def update_skill_field(name: str, updates: dict[str, Any]) -> Path | None:
    """Read an existing skill file, apply field updates, write it back."""
    safe = _safe_filename(name)
    path = SKILLS_DIR / f"{safe}.md"
    with file_lock(_update_lock_path("skill", safe), timeout=_LOCK_TIMEOUT):
        if not path.exists():
            return None
        data = read_file(path)
        if "body" in data and "content" not in data:
            data["content"] = data.pop("body")
        data.update(updates)
        return write_skill(data)

update_document_field

update_document_field(doc_id: str, updates: dict[str, Any]) -> Path | None

Read an existing document file, apply field updates, write it back.

Source code in memory/storage.py

def update_document_field(doc_id: str, updates: dict[str, Any]) -> Path | None:
    """Read an existing document file, apply field updates, write it back."""
    path = _id_filename(DOCUMENTS_DIR, doc_id, kind="document id")
    with file_lock(_update_lock_path("document", doc_id), timeout=_LOCK_TIMEOUT):
        if not path.exists():
            return None
        data = read_file(path)
        if "body" in data and "description" not in data:
            data["description"] = data.pop("body")
        data.update(updates)
        return write_document(data)

read_file

read_file(path: Path) -> dict[str, Any]

Parse a single markdown file back into a flat dict.

Source code in memory/storage.py

def read_file(path: Path) -> dict[str, Any]:
    """Parse a single markdown file back into a flat dict."""
    text = path.read_text(encoding="utf-8")
    meta, body = _parse_frontmatter(text)
    meta["body"] = body.rstrip("\n") if body else ""
    return meta

read_all

read_all() -> dict[str, Any]

Read the entire .memory/ tree into a dict.

Source code in memory/storage.py

def read_all() -> dict[str, Any]:
    """Read the entire ``.memory/`` tree into a dict."""
    result: dict[str, Any] = {
        "entries": _read_dir(ENTRIES_DIR),
        "skills": _read_dir(SKILLS_DIR, body_key="content"),
        "sessions": _read_dir(SESSIONS_DIR, body_key="search_text"),
        "documents": _read_dir(DOCUMENTS_DIR, body_key="description"),
        "project": None,
    }
    project_path = MEMORY_DIR / "project.md"
    if project_path.exists():
        proj = read_file(project_path)
        proj["description"] = proj.pop("body", "")
        result["project"] = proj
    return result

write_config

write_config(config: dict[str, Any]) -> Path

Write .memory/config.yaml.

Source code in memory/storage.py

def write_config(config: dict[str, Any]) -> Path:
    """Write ``.memory/config.yaml``."""
    _ensure_dirs()
    path = MEMORY_DIR / "config.yaml"
    atomic_write(
        path,
        yaml.dump(config, default_flow_style=False, allow_unicode=True, sort_keys=False),
    )
    return path

read_config

read_config() -> dict[str, Any]

Read .memory/config.yaml, or return defaults if missing.

Source code in memory/storage.py

def read_config() -> dict[str, Any]:
    """Read ``.memory/config.yaml``, or return defaults if missing."""
    path = MEMORY_DIR / "config.yaml"
    if not path.exists():
        return {"format_version": FORMAT_VERSION}
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except yaml.YAMLError:
        data = {}
    return data

store_is_established

store_is_established() -> bool

Whether .memory/ is an authoritative file-based store.

A store counts as established once it carries the format stamp (config.yaml) or any entry files. A merely-present .memory/ that is empty or holds only runtime sidecars (e.g. a stray active-sessions/ directory, or a folder a user created by hand) does NOT count — so a legacy .kgl is still allowed to migrate into it.

This is the startup gate's source of truth: True means rebuild the cache from files (and never overwrite them); False with a .kgl present means migrate. Checking for entry files (not just config.yaml) protects a real but unstamped store from being clobbered by a stale .kgl.

Source code in memory/storage.py

def store_is_established() -> bool:
    """Whether ``.memory/`` is an authoritative file-based store.

    A store counts as established once it carries the format stamp
    (``config.yaml``) **or** any entry files. A merely-present ``.memory/`` that
    is empty or holds only runtime sidecars (e.g. a stray ``active-sessions/``
    directory, or a folder a user created by hand) does NOT count — so a legacy
    ``.kgl`` is still allowed to migrate into it.

    This is the startup gate's source of truth: ``True`` means rebuild the cache
    from files (and never overwrite them); ``False`` with a ``.kgl`` present
    means migrate. Checking for entry files (not just ``config.yaml``) protects
    a real but unstamped store from being clobbered by a stale ``.kgl``.
    """
    if not MEMORY_DIR.exists():
        return False
    if (MEMORY_DIR / "config.yaml").exists():
        return True
    try:
        return any(ENTRIES_DIR.glob("*.md"))
    except OSError:
        return False

build_index

build_index(prev_index: dict[str, dict[str, Any]] | None = None) -> dict[str, dict[str, Any]]

Walk .memory/ and build {rel_path: {mtime, mtime_ns, size, content_hash}}.

Change detection is two-pass: every file is stat-ed for a (mtime_ns, size) fingerprint and its content SHA-256 is recomputed only when that fingerprint differs from the cached entry — otherwise the prior hash is reused. This avoids re-hashing the entire corpus (O(total bytes)) on every auto-sync poll while keeping :func:get_changed_files exact: a file is reported changed iff its content hash changed (mtime/size are only a fast pre-filter; a bare mtime bump forces a re-hash and comparison, never a false positive). When prev_index is omitted the persisted index cache is used as the reuse source.

Source code in memory/storage.py

def build_index(
    prev_index: dict[str, dict[str, Any]] | None = None,
) -> dict[str, dict[str, Any]]:
    """Walk ``.memory/`` and build ``{rel_path: {mtime, mtime_ns, size, content_hash}}``.

    Change detection is two-pass: every file is ``stat``-ed for a
    ``(mtime_ns, size)`` fingerprint and its content SHA-256 is recomputed only
    when that fingerprint differs from the cached entry — otherwise the prior
    hash is reused. This avoids re-hashing the entire corpus (O(total bytes)) on
    every auto-sync poll while keeping :func:`get_changed_files` exact: a file is
    reported changed iff its content hash changed (mtime/size are only a fast
    pre-filter; a bare mtime bump forces a re-hash and comparison, never a false
    positive). When ``prev_index`` is omitted the persisted index cache is used
    as the reuse source.
    """
    if prev_index is None:
        prev_index = load_index()
    index: dict[str, dict[str, Any]] = {}
    if not MEMORY_DIR.exists():
        return index
    for f in MEMORY_DIR.rglob("*.md"):
        rel = _index_key(f)
        entry = _index_entry(f, rel, prev_index)
        if entry is not None:
            index[rel] = entry
    config_path = MEMORY_DIR / "config.yaml"
    if config_path.exists():
        rel = _index_key(config_path)
        entry = _index_entry(config_path, rel, prev_index)
        if entry is not None:
            index[rel] = entry
    return index

load_index

load_index() -> dict[str, dict[str, Any]]

Read the index cache from ANGELO_DIR.

Source code in memory/storage.py

def load_index() -> dict[str, dict[str, Any]]:
    """Read the index cache from ``ANGELO_DIR``."""
    if not INDEX_FILE.exists():
        return {}
    try:
        return json.loads(INDEX_FILE.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return {}

save_index

save_index(index: dict[str, dict[str, Any]]) -> None

Write the index cache under ANGELO_DIR (crash-safe).

Source code in memory/storage.py

def save_index(index: dict[str, dict[str, Any]]) -> None:
    """Write the index cache under ``ANGELO_DIR`` (crash-safe)."""
    atomic_write(INDEX_FILE, json.dumps(index, indent=2))

get_changed_files

get_changed_files(old_index: dict[str, dict[str, Any]], new_index: dict[str, dict[str, Any]]) -> list[str]

Return file paths that were added, modified, or removed.

Source code in memory/storage.py

def get_changed_files(
    old_index: dict[str, dict[str, Any]],
    new_index: dict[str, dict[str, Any]],
) -> list[str]:
    """Return file paths that were added, modified, or removed."""
    changed: list[str] = []
    all_keys = set(old_index) | set(new_index)
    for key in sorted(all_keys):
        old = old_index.get(key)
        new = new_index.get(key)
        if old is None or new is None:
            changed.append(key)
        elif old.get("content_hash") != new.get("content_hash"):
            changed.append(key)
    return changed

classify_memory_file

classify_memory_file(rel_key: str) -> tuple[str, str] | None

Map a changed .memory/ index key to (node_type, file_stem).

Used by the incremental delta path to decide which graph node a changed file backs. Returns None for anything that is not a graph-backed node file — config.yaml, archived entries, runtime sidecars, or any file that is not a direct child of one of the node directories. The returned file_stem equals the node id for entries/sessions/documents; for skills (filename is a sanitized name) and the project it is only a locator and the caller must read the file for the authoritative id.

Keys are workspace-relative POSIX paths (as produced by :func:_index_key), so the comparison works whether the storage dirs are the historical relative .memory or an absolute path anchored to ANGELO_WORKSPACE.

Source code in memory/storage.py

def classify_memory_file(rel_key: str) -> tuple[str, str] | None:
    """Map a changed ``.memory/`` index key to ``(node_type, file_stem)``.

    Used by the incremental delta path to decide which graph node a changed
    file backs. Returns ``None`` for anything that is *not* a graph-backed node
    file — ``config.yaml``, archived entries, runtime sidecars, or any file that
    is not a direct child of one of the node directories. The returned
    ``file_stem`` equals the node id for entries/sessions/documents; for skills
    (filename is a sanitized name) and the project it is only a locator and the
    caller must read the file for the authoritative id.

    Keys are workspace-relative POSIX paths (as produced by :func:`_index_key`),
    so the comparison works whether the storage dirs are the historical
    relative ``.memory`` or an absolute path anchored to ``ANGELO_WORKSPACE``.
    """
    key = rel_key.replace("\\", "/")
    if not key.endswith(".md"):
        return None
    parent, _, name = key.rpartition("/")
    stem = name[:-3]  # strip ".md"
    for node_type, directory in (
        ("entry", ENTRIES_DIR),
        ("skill", SKILLS_DIR),
        ("session", SESSIONS_DIR),
        ("document", DOCUMENTS_DIR),
    ):
        if parent == _index_key(directory).replace("\\", "/"):
            return (node_type, stem)
    if name == "project.md" and parent == _index_key(MEMORY_DIR).replace("\\", "/"):
        return ("project", stem)
    return None

load_embeddings_sidecar

load_embeddings_sidecar() -> dict[str, list[float]]

Read the embeddings sidecar: {entry_id: [float, ...], ...}.

Source code in memory/storage.py

def load_embeddings_sidecar() -> dict[str, list[float]]:
    """Read the embeddings sidecar: ``{entry_id: [float, ...], ...}``."""
    if not EMBEDDINGS_FILE.exists():
        return {}
    try:
        return json.loads(EMBEDDINGS_FILE.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return {}

save_embeddings_sidecar

save_embeddings_sidecar(data: dict[str, list[float]]) -> None

Write the embeddings sidecar under ANGELO_DIR (crash-safe).

Source code in memory/storage.py

def save_embeddings_sidecar(data: dict[str, list[float]]) -> None:
    """Write the embeddings sidecar under ``ANGELO_DIR`` (crash-safe)."""
    atomic_write(EMBEDDINGS_FILE, json.dumps(data))

load_embedding_meta

load_embedding_meta() -> dict[str, Any]

Read the embedding metadata: {"model": str, "dimension": int}.

Returns {} when absent (legacy repos embedded before meta existed) or unreadable. Callers must treat a missing/empty meta as "unknown" and fall back to inspecting the actual stored vector dimension.

Source code in memory/storage.py

def load_embedding_meta() -> dict[str, Any]:
    """Read the embedding metadata: ``{"model": str, "dimension": int}``.

    Returns ``{}`` when absent (legacy repos embedded before meta existed) or
    unreadable. Callers must treat a missing/empty meta as "unknown" and fall
    back to inspecting the actual stored vector dimension.
    """
    if not EMBEDDINGS_META_FILE.exists():
        return {}
    try:
        data = json.loads(EMBEDDINGS_META_FILE.read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except (json.JSONDecodeError, OSError):
        return {}

save_embedding_meta

save_embedding_meta(model: str, dimension: int) -> None

Write the embedding metadata under ANGELO_DIR (crash-safe).

Write this only after a successful (re-)embed so a crashed/aborted migration never advertises a dimension the stored vectors don't match.

Source code in memory/storage.py

def save_embedding_meta(model: str, dimension: int) -> None:
    """Write the embedding metadata under ``ANGELO_DIR`` (crash-safe).

    Write this only *after* a successful (re-)embed so a crashed/aborted
    migration never advertises a dimension the stored vectors don't match.
    """
    atomic_write(
        EMBEDDINGS_META_FILE,
        json.dumps({"model": model, "dimension": int(dimension)}),
    )

migrate_from_kgl

migrate_from_kgl(kgl_path: Path) -> dict[str, Any]

Read a KGLite .kgl file and export all nodes to .memory/ files.

Captures every node property present and reconstructs relationships that legacy (pre-file-storage) graphs stored only as edges, so the subsequent rebuild-from-files preserves the hierarchy and cross-references. A timestamped backup of the source .kgl is written before any files are created (the rewrite is one-way and an upgrading user may have no other copy).

Returns a stats dict with counts of exported nodes plus the backup path.

Source code in memory/storage.py

def migrate_from_kgl(kgl_path: Path) -> dict[str, Any]:
    """Read a KGLite ``.kgl`` file and export all nodes to ``.memory/`` files.

    Captures every node property present *and* reconstructs relationships that
    legacy (pre-file-storage) graphs stored only as edges, so the subsequent
    rebuild-from-files preserves the hierarchy and cross-references. A
    timestamped backup of the source ``.kgl`` is written before any files are
    created (the rewrite is one-way and an upgrading user may have no other
    copy).

    Returns a stats dict with counts of exported nodes plus the backup path.
    """
    import kglite as kgl

    if not kgl_path.exists():
        return {"error": "kgl file not found", "migrated": 0}

    # Safety net: back up the source before writing anything.
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    backup_path: Path | None = kgl_path.with_suffix(kgl_path.suffix + f".bak-{stamp}")
    try:
        shutil.copy2(kgl_path, backup_path)
        logger.info("Backed up %s -> %s before migration", kgl_path, backup_path)
    except OSError as e:
        logger.warning("Could not back up %s before migration: %s", kgl_path, e)
        backup_path = None

    g = kgl.load(str(kgl_path))
    _ensure_dirs()

    # ---- Collect nodes (defer writes; edge reconstruction may fill fields) ----
    project_row: dict[str, Any] | None = None
    for row in g.cypher(
        "MATCH (p:project) RETURN p.id AS id, p.name AS name, "
        "p.description AS description, p.body AS body, p.status AS status, "
        "p.created_at AS created_at"
    ):
        project_row = _drop_none(dict(row))
        # Legacy projects may carry the text under ``body``; storage expects it
        # under ``description``.
        if not project_row.get("description") and project_row.get("body"):
            project_row["description"] = project_row["body"]
        project_row.pop("body", None)
        break  # at most one project node

    entries: dict[str, dict[str, Any]] = {}
    for row in g.cypher(
        "MATCH (e:entry) RETURN e.id AS id, e.type AS type, e.project AS project, "
        "e.parent_id AS parent_id, e.title AS title, e.node_label AS node_label, "
        "e.tags AS tags, e.status AS status, e.success AS success, e.files AS files, "
        "e.open_threads AS open_threads, e.body AS body, "
        "e.created_at AS created_at, e.updated_at AS updated_at, "
        "e.invalidates AS invalidates, e.invalidated_by AS invalidated_by, "
        "e.related_to AS related_to, e.has_child_rationale AS has_child_rationale, "
        "e.external_plan AS external_plan, e.plan_content AS plan_content, "
        "e.session_id AS session_id, e.notes AS notes"
    ):
        d = _drop_none(dict(row))
        if d.get("id"):
            entries[d["id"]] = d

    skills: dict[str, dict[str, Any]] = {}
    for row in g.cypher(
        "MATCH (s:skill) RETURN s.id AS id, s.name AS name, s.category AS category, "
        "s.content AS content, s.tags AS tags, s.status AS status, "
        "s.workflow AS workflow, "
        "s.created_at AS created_at, s.updated_at AS updated_at"
    ):
        d = _drop_none(dict(row))
        if d.get("id"):
            skills[d["id"]] = d

    sessions: dict[str, dict[str, Any]] = {}
    for row in g.cypher(
        "MATCH (s:session) RETURN s.id AS id, s.project AS project, "
        "s.started_at AS started_at, s.ended_at AS ended_at, s.title AS title, "
        "s.summary AS summary, s.kind AS kind, s.notes AS notes, "
        "s.entries_created AS entries_created, "
        "s.entries_accessed AS entries_accessed, s.entries_modified AS entries_modified, "
        "s.search_text AS search_text"
    ):
        d = _drop_none(dict(row))
        if d.get("id"):
            sessions[d["id"]] = d

    documents: dict[str, dict[str, Any]] = {}
    for row in g.cypher(
        "MATCH (d:document) RETURN d.id AS id, d.path AS path, d.title AS title, "
        "d.description AS description, d.project AS project, d.status AS status, "
        "d.links_to AS links_to, d.created_at AS created_at"
    ):
        dd = _drop_none(dict(row))
        if dd.get("id"):
            documents[dd["id"]] = dd

    # ---- Reconstruct relationships stored only as edges (legacy graphs) ----
    _reconstruct_edges(g, entries, documents)

    # ---- Write everything ----
    stats: dict[str, Any] = {
        "entries": 0, "skills": 0, "sessions": 0, "documents": 0, "project": 0,
    }
    if project_row:
        write_project(project_row)
        stats["project"] = 1
    for d in entries.values():
        write_entry(d)
        stats["entries"] += 1
    for d in skills.values():
        write_skill(d)
        stats["skills"] += 1
    for d in sessions.values():
        write_session(d)
        stats["sessions"] += 1
    for d in documents.values():
        write_document(d)
        stats["documents"] += 1

    config: dict[str, Any] = {
        "format_version": FORMAT_VERSION, "migrated_from": str(kgl_path),
    }
    if backup_path is not None:
        config["migration_backup"] = str(backup_path)
        stats["backup"] = str(backup_path)
    write_config(config)

    new_index = build_index()
    save_index(new_index)

    logger.info("Migration complete: %s", stats)
    return stats

verify_migration

verify_migration(kgl_path: Path) -> bool

Verify migrated files against the source .kgl.

Compares node id sets (not just counts) for entries/skills/sessions/ documents — a mismatch fails the check. Also logs a relationship-preservation summary comparing source edge counts to the relationship fields written to files; relationship differences are warning-only because legacy edge direction and rationale cannot always round-trip exactly.

Source code in memory/storage.py

def verify_migration(kgl_path: Path) -> bool:
    """Verify migrated files against the source ``.kgl``.

    Compares node **id sets** (not just counts) for entries/skills/sessions/
    documents — a mismatch fails the check. Also logs a relationship-preservation
    summary comparing source edge counts to the relationship fields written to
    files; relationship differences are warning-only because legacy edge
    direction and rationale cannot always round-trip exactly.
    """
    import kglite as kgl

    if not kgl_path.exists():
        return False

    g = kgl.load(str(kgl_path))
    data = read_all()

    def _kgl_ids(label: str, var: str) -> set[str]:
        return {
            r["id"]
            for r in g.cypher(f"MATCH ({var}:{label}) RETURN {var}.id AS id")
            if r.get("id")
        }

    def _file_ids(key: str) -> set[str]:
        return {x.get("id") for x in data.get(key, []) if x.get("id")}

    ok = True
    for key, (label, var) in {
        "entries": ("entry", "e"),
        "skills": ("skill", "s"),
        "sessions": ("session", "s"),
        "documents": ("document", "d"),
    }.items():
        kgl_ids = _kgl_ids(label, var)
        file_ids = _file_ids(key)
        if kgl_ids != file_ids:
            ok = False
            logger.warning(
                "Migration %s id mismatch (kgl=%d files=%d) missing=%s extra=%s",
                key, len(kgl_ids), len(file_ids),
                sorted(kgl_ids - file_ids)[:10], sorted(file_ids - kgl_ids)[:10],
            )

    # Relationship preservation summary (informational only).
    def _edge_count(query: str) -> int:
        try:
            return len(list(g.cypher(query)))
        except Exception:  # pragma: no cover - defensive
            return -1

    kgl_has_child = _edge_count("MATCH (s)-[:has_child]->(c:entry) RETURN c.id AS id")
    kgl_related = _edge_count("MATCH (a:entry)-[:related]->(b:entry) RETURN a.id AS id")
    kgl_invalidated = _edge_count(
        "MATCH (o:entry)-[:invalidated_by]->(n:entry) RETURN n.id AS id"
    )
    kgl_doc_links = _edge_count(
        "MATCH (d:document)-[:related]->(e:entry) RETURN d.id AS id"
    )

    file_entries = data.get("entries", [])
    file_parents = sum(1 for e in file_entries if e.get("parent_id"))
    file_related = sum(
        len([t for t in str(e.get("related_to", "") or "").split(",") if t.strip()])
        for e in file_entries
    )
    file_invalidates = sum(1 for e in file_entries if e.get("invalidates"))
    file_links = sum(1 for d in data.get("documents", []) if d.get("links_to"))

    logger.info(
        "Relationship preservation — has_child: kgl=%d files(parent_id)=%d | "
        "related: kgl=%d files=%d | invalidated_by: kgl=%d files(invalidates)=%d | "
        "doc links: kgl=%d files=%d",
        kgl_has_child, file_parents, kgl_related, file_related,
        kgl_invalidated, file_invalidates, kgl_doc_links, file_links,
    )

    return ok