ai-agent/backend/memories.py

"""Claude *memories* — the persistent file-based memory the assistant keeps for
this repo (one fact per file, with YAML frontmatter), surfaced in the dashboard.

Memories live in ``MEMORIES_DIR`` (the repo's
``.claude/projects/<encoded>/memory`` dir, mounted read-only). Each file is::

    ---
    name: <slug>
    description: <one-line summary>
    metadata:
      node_type: memory
      type: user | feedback | project | reference
      originSessionId: <session that first wrote it>
    ---

    <markdown body, may link related memories with [[other-slug]]>

We parse the frontmatter (no PyYAML dependency — the shape is fixed and simple),
derive each memory's related **projects** (``projects/<slug>`` mentions in the
body, same convention the conversation parser uses) and its ``[[links]]`` to
other memories, and expose them to the frontend. The ``MEMORY.md`` index file is
skipped (it only mirrors these records).
"""
from __future__ import annotations

import datetime
import os
import pathlib
import re

MEMORIES_DIR = pathlib.Path(os.environ.get("MEMORIES_DIR", "/memories")).resolve()

# `projects/<slug>` mentions in a body → the projects a memory relates to (the
# leading-dash filter drops Claude's own encoded transcript dir names, mirroring
# conversations._PROJECT_RE).
_PROJECT_RE = re.compile(r"(?:^|[/\s(`\"'])projects/([A-Za-z0-9._][\w.-]*)")
# `services/<slug>` mentions in a body → the homelab services a memory relates to.
_SERVICE_RE = re.compile(r"(?:^|[/\s(`\"'])services/([A-Za-z0-9._][\w.-]*)")
# `[[other-slug]]` wiki-links between memories.
_LINK_RE = re.compile(r"\[\[([A-Za-z0-9][\w.-]*)\]\]")

MAX_BODY = 40_000  # cap a memory body so detail responses stay bounded


def _iso(epoch: float) -> str:
    return (datetime.datetime.fromtimestamp(epoch, datetime.timezone.utc)
            .isoformat().replace("+00:00", "Z"))


def _parse_frontmatter(text: str) -> tuple[dict, str]:
    """Split a ``---`` frontmatter block from the body. Returns (data, body).

    Supports one level of nesting (the ``metadata:`` map), which is all these
    files use. Values are unquoted; missing frontmatter yields ({}, text)."""
    if not text.startswith("---"):
        return {}, text
    end = text.find("\n---", 3)
    if end == -1:
        return {}, text
    block = text[3:end].strip("\n")
    body = text[end + 4:].lstrip("\n")
    data: dict = {}
    cur = data
    for line in block.splitlines():
        if not line.strip():
            continue
        indent = len(line) - len(line.lstrip())
        key, sep, val = line.strip().partition(":")
        if not sep:
            continue
        key = key.strip()
        val = val.strip().strip('"').strip("'")
        if indent == 0:
            if val == "":
                cur = data.setdefault(key, {})  # start a nested map
            else:
                data[key] = val
                cur = data
        elif isinstance(cur, dict):
            cur[key] = val
    return data, body


def _record(path: pathlib.Path, with_body: bool = False) -> dict:
    slug = path.stem
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
        mtime = path.stat().st_mtime
    except OSError:
        text, mtime = "", 0.0
    fm, body = _parse_frontmatter(text)
    meta = fm.get("metadata") if isinstance(fm.get("metadata"), dict) else {}
    projects = sorted({m for m in _PROJECT_RE.findall(body)
                       if not m.startswith("-")})
    services = sorted({m for m in _SERVICE_RE.findall(body)
                       if not m.startswith("-")})
    links = sorted({m for m in _LINK_RE.findall(body) if m != slug})
    rec = {
        "slug": slug,
        "name": fm.get("name") or slug,
        "description": fm.get("description") or "",
        "type": meta.get("type"),
        "originSessionId": meta.get("originSessionId"),
        "projects": projects,
        "services": services,
        "links": links,
        "words": len(body.split()),
        "bytes": len(text.encode("utf-8")),
        "updatedAt": _iso(mtime) if mtime else None,
    }
    if with_body:
        rec["content"] = body[:MAX_BODY] + (
            f"\n\n… [{len(body) - MAX_BODY} more chars truncated]"
            if len(body) > MAX_BODY else "")
    return rec


def _iter_files():
    if not MEMORIES_DIR.is_dir():
        return
    for p in sorted(MEMORIES_DIR.glob("*.md")):
        if p.name == "MEMORY.md":  # the index, not a memory itself
            continue
        yield p


def list_memories() -> list[dict]:
    """All memory summaries, newest-updated first."""
    items = [_record(p) for p in _iter_files()]
    items.sort(key=lambda m: m.get("updatedAt") or "", reverse=True)
    return items


def _safe_path(slug: str) -> pathlib.Path | None:
    name = pathlib.PurePosixPath(slug).name
    if not name or name == "MEMORY.md":
        return None
    p = (MEMORIES_DIR / f"{name}.md").resolve()
    try:
        p.relative_to(MEMORIES_DIR)
    except ValueError:
        return None
    return p if p.is_file() else None


def memory_detail(slug: str) -> dict | None:
    p = _safe_path(slug)
    if not p:
        return None
    return _record(p, with_body=True)


def slugs() -> set[str]:
    return {p.stem for p in _iter_files()}


def by_origin(session_id: str | None) -> list[dict]:
    """Memory summaries first written by a given conversation (origin session)."""
    if not session_id:
        return []
    return [m for m in list_memories() if m.get("originSessionId") == session_id]