VoiceAgent/brain.py

"""Read from and write to the Metamate personal brain.

The brain is the long-term memory: `briefing.md` carries active work and pinned
reminders, `preferences.md` how Adolfo likes to be worked with, `profile.md` who
he is, and `projects/` a directory per workstream. It lives remotely and is
reached through `meta agents.memory`, so everything here shells out.

Two directions:

- **In.** The three context files go into the system prompt at startup, so the
  agent knows what's active without being told. Project directory names also
  become vocabulary, since "CIP-Unified-Cooldown" and "pSMSL" are exactly the
  words a recogniser mangles.
- **Out.** Durable things learned in conversation are appended back, following
  the discipline the brain's own README sets out: lasting preferences to
  `preferences.md`, techniques and lessons to `notes.md`.

`briefing.md` is deliberately *not* written to automatically. A daily cron owns
that file, and an agent appending to it unprompted would fight the cron and
corrupt the one file everything else reads first.
"""

import json
import subprocess
from pathlib import Path

from loguru import logger

META = "meta"
CONTEXT_FILES = ("briefing.md", "preferences.md", "profile.md")

# Writes go only to files a human owns, never to cron-managed ones.
WRITABLE = {
    "preference": "preferences.md",
    "note": "notes.md",
}

# Learned entries get their own section so they never land inside a hand-written
# one, and so it stays obvious which lines the agent added.
LEARNED_HEADING = "## Learned in voice sessions"

_READ_TIMEOUT = 25.0
_WRITE_TIMEOUT = 25.0

# Used only when the brain can't be reached, so a flight or a VPN drop doesn't
# cost the agent all of its context.
CACHE = Path.home() / ".cache" / "voice-agent" / "brain-context.json"


def _run(args: list[str], timeout: float) -> tuple[bool, str]:
    try:
        result = subprocess.run(
            [META, "agents.memory", *args],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except (OSError, subprocess.SubprocessError) as e:
        return False, str(e)
    if result.returncode != 0:
        return False, (result.stderr.strip() or result.stdout.strip())[:200]
    return True, result.stdout


class Brain:
    """The personal brain, or a graceful no-op when it can't be reached."""

    def __init__(self):
        self._context: dict[str, str] = {}
        self._projects: list[str] = []
        self.reachable = False

    def load(self) -> bool:
        """Fetch context and project names. Falls back to cache when offline."""
        ok, out = _run(
            ["read-batch", f"--paths={','.join(CONTEXT_FILES)}", "--output=json"],
            _READ_TIMEOUT,
        )
        if ok:
            try:
                payload = json.loads(out)
            except json.JSONDecodeError:
                ok = False
            else:
                self._context = {
                    name: entry.get("content", "")
                    for name, entry in payload.items()
                    if isinstance(entry, dict) and entry.get("content")
                }
                self.reachable = bool(self._context)

        if self.reachable:
            self._projects = self._list_projects()
            self._save_cache()
            logger.info(
                f"Brain: loaded {len(self._context)} context files, "
                f"{len(self._projects)} projects"
            )
            return True

        if self._load_cache():
            logger.warning(f"Brain unreachable ({out[:80]}); using cached context.")
            return True
        logger.warning(f"Brain unavailable: {out[:120]}")
        return False

    def _list_projects(self) -> list[str]:
        ok, out = _run(["list", "--path=projects", "-l", "200"], _READ_TIMEOUT)
        if not ok:
            return []
        names = []
        for line in out.splitlines():
            parts = line.split()
            # Rows look like "NAME  dir  -"; skip the header and rules.
            if len(parts) >= 2 and parts[1] == "dir":
                names.append(parts[0])
        return names

    def _save_cache(self):
        try:
            CACHE.parent.mkdir(parents=True, exist_ok=True)
            CACHE.write_text(
                json.dumps({"context": self._context, "projects": self._projects})
            )
        except OSError:
            pass

    def _load_cache(self) -> bool:
        try:
            payload = json.loads(CACHE.read_text())
        except (OSError, json.JSONDecodeError):
            return False
        self._context = payload.get("context", {})
        self._projects = payload.get("projects", [])
        return bool(self._context)

    @property
    def projects(self) -> list[str]:
        return self._projects

    def prompt_block(self) -> str:
        """The context files, framed as reference rather than as instructions.

        Framing matters: briefing.md is dense markdown with links and bold, and
        without a clear label the agent starts answering in the same register —
        which is wrong out loud.
        """
        if not self._context:
            return ""
        sections = [
            f"### {name}\n{text.strip()}"
            for name, text in self._context.items()
            if text.strip()
        ]
        if not sections:
            return ""
        return (
            "Below is your memory of Adolfo's work, from his personal brain. "
            "Treat it as reference you already know, not as something to read "
            "back. It is written notes — never mirror their formatting when you "
            "speak.\n\n" + "\n\n".join(sections)
        )

    def append(self, kind: str, content: str) -> tuple[bool, str]:
        """Append a line to one of the writable brain files.

        Args:
            kind: A key of ``WRITABLE`` — "preference" or "note".
            content: One line, already phrased as a durable statement.
        """
        path = WRITABLE.get(kind)
        if not path:
            return False, f"nothing writable for {kind!r}; expected {sorted(WRITABLE)}"
        if not content.strip():
            return False, "refusing to write empty content"

        # Append lands at the end of the file, which would tuck the new line
        # inside whatever the last section happens to be. These are curated
        # files, so learned entries get their own heading the first time.
        body = content.strip()
        ok, existing = _run(["read", f"--path=/{path}"], _READ_TIMEOUT)
        if ok and LEARNED_HEADING not in existing:
            body = f"\n{LEARNED_HEADING}\n{body}"

        ok, out = _run(
            ["append", f"--path={path}", f"--content={body}"],
            _WRITE_TIMEOUT,
        )
        if ok:
            logger.info(f"Brain: appended a {kind} to {path}")
            return True, path
        return False, out