scripts/odysseus-memory

#!/usr/bin/env python3
"""odysseus-memory — shell wrapper for the AI memory store.

The chat assistant accumulates durable facts about the user (name,
projects, preferences, contacts) into `data/memory.json`. This CLI
lets you inspect, search, and prune them from the shell.

    odysseus-memory list                          # all entries (paged)
    odysseus-memory list --category preference
    odysseus-memory search "tokyo"                # substring match
    odysseus-memory show MEMORY_ID
    odysseus-memory add "User lives in Tokyo" --category fact
    odysseus-memory delete MEMORY_ID
    odysseus-memory categories                    # counts per category

Reads/writes the same `data/memory.json` the web UI uses via the
`MemoryManager` helper. Atomic — the manager handles the temp-file
swap so partial writes can't corrupt the file.
"""

from __future__ import annotations
import sys
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "_lib"))
from cli import quiet_logs, emit, fail, common_parser, run, REPO_ROOT as _REPO_ROOT
quiet_logs()

import argparse, json, logging, os, sys
from pathlib import Path

try:
    from services.memory.memory import MemoryManager
    quiet_logs()
except ModuleNotFoundError as e:
    sys.stderr.write(f"error: {e}\nhint: run from repo root with venv active.\n")
    sys.exit(2)


_DATA_DIR = str(_REPO_ROOT / "data")
_mgr: MemoryManager | None = None


def _manager() -> MemoryManager:
    global _mgr
    if _mgr is None:
        _mgr = MemoryManager(_DATA_DIR)
    return _mgr


def _memory_entries(entries):
    return [e for e in entries or [] if isinstance(e, dict)]


def cmd_list(args):
    entries = _memory_entries(_manager().load_all())
    if args.category:
        entries = [e for e in entries if (e.get("category") or "fact") == args.category]
    if args.source:
        entries = [e for e in entries if (e.get("source") or "user") == args.source]
    if args.owner:
        entries = [e for e in entries if (e.get("owner") or "") == args.owner]
    # Newest first
    entries = sorted(entries, key=lambda e: e.get("timestamp", 0), reverse=True)
    emit(entries[: args.limit], args)


def cmd_search(args):
    q = args.query.lower()
    entries = _memory_entries(_manager().load_all())
    matches = [e for e in entries if q in (e.get("text") or "").lower()]
    matches = sorted(matches, key=lambda e: e.get("timestamp", 0), reverse=True)
    emit(matches[: args.limit], args)


def cmd_show(args):
    for e in _memory_entries(_manager().load_all()):
        if e.get("id") == args.id:
            emit(e, args)
            return
    fail(f"no memory with id {args.id!r}")


def cmd_add(args):
    entry = _manager().add_entry(
        args.text,
        source="cli",
        category=args.category,
        owner=args.owner,
    )
    # add_entry doesn't save by default — the call in chat does it
    # after dedup checks. Persist here so a one-shot CLI add sticks.
    all_entries = _manager().load_all()
    if not any(e.get("id") == entry.get("id") for e in all_entries):
        all_entries.append(entry)
        _manager().save(all_entries)
    emit(entry, args)


def cmd_delete(args):
    entries = _memory_entries(_manager().load_all())
    target = next((e for e in entries if e.get("id") == args.id), None)
    if not target:
        fail(f"no memory with id {args.id!r}")
    remaining = [e for e in entries if e.get("id") != args.id]
    _manager().save(remaining)
    emit({"ok": True, "deleted": target}, args)


def cmd_categories(args):
    counts: dict[str, int] = {}
    for e in _memory_entries(_manager().load_all()):
        cat = e.get("category") or "fact"
        counts[cat] = counts.get(cat, 0) + 1
    rows = sorted(counts.items(), key=lambda kv: -kv[1])
    emit([{"category": c, "count": n} for c, n in rows], args)


def _build_parser():
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--pretty", action="store_true")
    p = argparse.ArgumentParser(prog="odysseus-memory", parents=[common])
    sub = p.add_subparsers(dest="cmd", required=True)

    pl = sub.add_parser("list", parents=[common])
    pl.add_argument("--category")
    pl.add_argument("--source")
    pl.add_argument("--owner")
    pl.add_argument("--limit", type=int, default=50)
    pl.set_defaults(func=cmd_list)

    ps = sub.add_parser("search", parents=[common])
    ps.add_argument("query")
    ps.add_argument("--limit", type=int, default=50)
    ps.set_defaults(func=cmd_search)

    psh = sub.add_parser("show", parents=[common])
    psh.add_argument("id")
    psh.set_defaults(func=cmd_show)

    pa = sub.add_parser("add", parents=[common])
    pa.add_argument("text")
    pa.add_argument("--category", default="fact")
    pa.add_argument("--owner")
    pa.set_defaults(func=cmd_add)

    pd = sub.add_parser("delete", parents=[common])
    pd.add_argument("id")
    pd.set_defaults(func=cmd_delete)

    pc = sub.add_parser("categories", parents=[common])
    pc.set_defaults(func=cmd_categories)

    return p


if __name__ == "__main__":
    sys.exit(run(_build_parser()))