memory-sync.py

#!/usr/bin/env python3
"""memory-sync — bidirectional Codex <-> Claude Code long-term memory sync.

Companion to agent-session-sync.py. Where that tool mirrors conversation
transcripts (inert until you resume one), this one moves the notes each agent
keeps *about you* — files that get injected into every future session. That is
a higher-trust write, so it is opt-in, it lives behind its own lock and state,
and it only ever writes two files it owns:

  Codex  -> Claude : $CODEX_HOME/memories/memory_summary.md
                     -> <claude projects>/<project>/memory/codex_memory_sync.md
  Claude -> Codex  : <claude projects>/*/memory/*.md
                     -> $CODEX_HOME/memories/claude_code_sync.md

Plus one idempotent pointer line in each side's index so the agent knows to
read it: Claude's MEMORY.md, and Codex's global AGENTS.md.

Everything else is read-only. Codex's own MEMORY.md, memory_summary.md and
raw_memories.md are never written. A file that exists at an owned path but
lacks our generated-by marker is left alone and reported, never clobbered.

The Claude -> Codex bundle is full memory text while that fits under
--warn-bytes, and otherwise an index: one line per memory file with its
description and absolute path, for Codex to read on demand. The pointer only
ever costs a line of AGENTS.md; pointing it at a megabyte would just spend a
turn on a truncated slice of it.

Codex -> Claude no-ops unless memory_summary.md exists; recent Codex keeps
memory in memories_*.sqlite instead, and this script does not read it.

  memory-sync.py --dry-run -v    # show what would change, write nothing
  memory-sync.py                 # sync both directions
"""

import argparse
import hashlib
import json
import os
import re
import sys
from datetime import datetime, timezone

try:
    import fcntl
    msvcrt = None
except ImportError:
    fcntl = None
    import msvcrt

CODEX_HOME = os.path.expanduser(os.environ.get("CODEX_HOME") or "~/.codex")
CLAUDE_HOME = os.path.expanduser(os.environ.get("CLAUDE_CONFIG_DIR") or "~/.claude")
CLAUDE_PROJECTS = os.path.join(CLAUDE_HOME, "projects")

CODEX_MEM = os.path.join(CODEX_HOME, "memories")
CODEX_SUMMARY = os.path.join(CODEX_MEM, "memory_summary.md")
CODEX_OWNED = os.path.join(CODEX_MEM, "claude_code_sync.md")
CODEX_AGENTS = os.path.join(CODEX_HOME, "AGENTS.md")
CLAUDE_OWNED_NAME = "codex_memory_sync.md"

MARKER = "<!-- generated by memory-sync.py — edits will be overwritten -->"
POINTER = (f"At session start, also read {CODEX_OWNED} for Claude Code memory "
           f"about this host's projects and user preferences.")
INDEX_LINE = (f"- [{CLAUDE_OWNED_NAME}]({CLAUDE_OWNED_NAME}) — Auto-synced Codex memory: "
              f"session knowledge, reusable patterns, user preferences")

# never read into the Claude -> Codex bundle
SKIP_NAMES = {CLAUDE_OWNED_NAME, "MEMORY.md"}
DESC_CAP = 100          # chars of description kept per file in index mode

VERBOSE = False


def default_state_dir():
    override = os.environ.get("AGENT_SESSION_SYNC_STATE_DIR")
    if override:
        return os.path.expanduser(override)
    if os.name == "nt":
        base = os.environ.get("LOCALAPPDATA") or os.path.expanduser("~\\AppData\\Local")
        return os.path.join(base, "agent-session-sync")
    if getattr(os, "geteuid", lambda: 1)() == 0:
        return "/var/lib/agent-session-sync"
    xdg = os.environ.get("XDG_STATE_HOME") or os.path.expanduser("~/.local/state")
    return os.path.join(xdg, "agent-session-sync")


STATE_DIR = default_state_dir()
STATE_FILE = os.path.join(STATE_DIR, "memory-state.json")   # distinct from the session sync
LOCK_FILE = os.path.join(STATE_DIR, "memory.lock")


def log(msg):
    if VERBOSE:
        print(msg, file=sys.stderr)


def sha(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def read(path, default=""):
    try:
        with open(path, encoding="utf-8", errors="replace") as fh:
            return fh.read()
    except OSError:
        return default


def acquire_lock(path):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    fh = open(path, "w")
    try:
        if fcntl is not None:
            fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
        else:
            msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1)
    except OSError:
        fh.close()
        return None
    return fh


def load_state():
    try:
        with open(STATE_FILE) as fh:
            return json.load(fh)
    except (OSError, json.JSONDecodeError):
        return {}


def save_state(state):
    os.makedirs(STATE_DIR, exist_ok=True)
    tmp = STATE_FILE + ".tmp"
    with open(tmp, "w", encoding="utf-8", newline="\n") as fh:
        json.dump(state, fh, indent=1)
    os.replace(tmp, STATE_FILE)


def write_owned(path, body, dry, state):
    """Write a file we own. Refuse if something without our marker sits there.

    The refusal is a standing condition, and this runs on a timer — so complain
    once per distinct obstruction rather than on every tick, and complain again
    if the thing in the way changes.
    """
    if os.path.exists(path):
        blocking = read(path)
        if MARKER not in blocking:
            key = "blocked:" + path
            digest = sha(blocking)
            if state.get(key) != digest:
                print(f"refusing to overwrite {path}: not generated by memory-sync",
                      file=sys.stderr)
                state[key] = digest
            return False
    state.pop("blocked:" + path, None)
    if dry:
        return True
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8", newline="\n") as fh:
        fh.write(body)
    os.replace(tmp, path)
    return True


def append_once(path, needle, line, dry, create=True):
    """Append `line` to `path` unless `needle` already appears in it."""
    existing = read(path, default=None if not create else "")
    if existing is None:
        return False
    if needle in existing:
        return False
    if dry:
        return True
    os.makedirs(os.path.dirname(path), exist_ok=True)
    sep = "" if (not existing or existing.endswith("\n")) else "\n"
    with open(path, "a", encoding="utf-8", newline="\n") as fh:
        fh.write(f"{sep}\n{line}\n")
    return True


def strip_frontmatter(text):
    if not text.startswith("---"):
        return text
    parts = text.split("\n")
    if parts[0].strip() != "---":
        return text
    for i in range(1, len(parts)):
        if parts[i].strip() == "---":
            return "\n".join(parts[i + 1:]).lstrip("\n")
    return text


def munge_cwd(cwd):
    return re.sub(r"[^A-Za-z0-9-]", "-", cwd)


def memory_dirs():
    """Every <project>/memory directory, home project first (global context)."""
    if not os.path.isdir(CLAUDE_PROJECTS):
        return []
    home = munge_cwd(os.path.expanduser("~"))
    found = []
    for proj in sorted(os.listdir(CLAUDE_PROJECTS)):
        d = os.path.join(CLAUDE_PROJECTS, proj, "memory")
        if os.path.isdir(d):
            found.append((proj, d))
    found.sort(key=lambda p: (p[0] != home, p[0]))
    return found


# ---------------------------------------------------------------------------

def codex_to_claude(state, args):
    summary = read(CODEX_SUMMARY, default=None)
    if summary is None:
        log(f"codex->claude: no {CODEX_SUMMARY}, nothing to do")
        return 0
    digest = sha(summary)
    if state.get("codex_summary_sha") == digest:
        log("codex->claude: unchanged")
        return 0

    project = args.project or munge_cwd(os.path.expanduser("~"))
    memdir = os.path.join(CLAUDE_PROJECTS, project, "memory")
    target = os.path.join(memdir, CLAUDE_OWNED_NAME)
    body = (
        "---\n"
        "name: codex-memory-sync\n"
        "description: Auto-synced Codex memory — regenerated by memory-sync.py\n"
        "metadata:\n"
        "  type: project\n"
        "---\n\n"
        f"{MARKER}\n\n"
        f"{summary.strip()}\n"
    )
    if not write_owned(target, body, args.dry_run, state):
        return 0
    added = append_once(os.path.join(memdir, "MEMORY.md"),
                        CLAUDE_OWNED_NAME, INDEX_LINE, args.dry_run)
    if not args.dry_run:
        state["codex_summary_sha"] = digest
    print(f"{'would sync' if args.dry_run else 'synced'} codex->claude: {target}"
          + (" (+ MEMORY.md index)" if added else ""))
    return 1


def _tidy(s):
    s = " ".join(s.split()).strip().strip('"').strip("'").strip()
    return s if len(s) <= DESC_CAP else s[:DESC_CAP].rsplit(" ", 1)[0] + "…"


def describe(text):
    """One line about a memory file: its frontmatter description, else its first prose line."""
    if text.startswith("---"):
        m = re.search(r"^description:\s*(.+)$", text, re.M)
        if m:
            return _tidy(m.group(1))
        text = strip_frontmatter(text)   # else `name:` etc. would masquerade as prose
    for line in text.split("\n"):
        s = line.strip()
        if s and not s.startswith(("---", "#")):
            return _tidy(s)
    return ""


def memory_groups():
    out = []
    for project, memdir in memory_dirs():
        files = [f for f in sorted(os.listdir(memdir))
                 if f.endswith(".md") and f not in SKIP_NAMES]
        if files:
            out.append((project, memdir, files))
    return out


def build_full(groups):
    chunks = []
    for project, memdir, files in groups:
        chunks.append(f"## project: {project}")
        for name in files:
            chunks.append(f"### {name}")
            chunks.append(strip_frontmatter(read(os.path.join(memdir, name))).strip())
            chunks.append("")
    return "\n".join(chunks).rstrip() + "\n"


def build_index(groups):
    """A table of contents with absolute paths. Codex reads only what it needs."""
    chunks = []
    for project, memdir, files in groups:
        chunks.append(f"## project: {project}  ({memdir})")
        for name in files:
            chunks.append(f"- {name} — {describe(read(os.path.join(memdir, name)))}")
        chunks.append("")
    return "\n".join(chunks).rstrip() + "\n"


def claude_to_codex(state, args):
    groups = memory_groups()
    if not groups:
        log("claude->codex: no memory files, nothing to do")
        return 0
    n_files = sum(len(f) for _, _, f in groups)

    # Full text when it fits; otherwise an index, because telling Codex to read a
    # megabyte just burns a turn on a truncated slice of it.
    mode = args.mode
    full = build_full(groups)
    if mode == "auto":
        mode = "full" if len(full.encode("utf-8")) <= args.warn_bytes else "index"
    if mode == "full":
        body_core = full
        note = f"{n_files} files, full text"
    else:
        body_core = build_index(groups)
        note = (f"{n_files} files across {len(groups)} projects, index only — "
                f"read a path above for its full text")

    digest = sha(mode + body_core)   # timestamp excluded, so an idle run is a no-op
    if state.get("claude_memory_sha") == digest:
        log("claude->codex: unchanged")
        return 0

    stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    body = (f"# Claude Code Memory\n{MARKER}\nupdated: {stamp}\n"
            f"scope: Auto-synced from Claude Code memory files on this host\n"
            f"mode: {note}\n\n{body_core}")
    if not write_owned(CODEX_OWNED, body, args.dry_run, state):
        return 0

    # The pointer costs one line of AGENTS.md; the bundle only reaches the model if
    # it acts on that line and reads the file. Pointing at something huge therefore
    # doesn't blow the context window — it wastes a turn on a slice of the file,
    # since Codex truncates tool output. Only reachable via --mode full: auto
    # already downgrades to an index. An existing pointer is left alone.
    size = len(body.encode("utf-8"))
    oversized = size > args.warn_bytes
    already = "claude_code_sync" in read(CODEX_AGENTS)
    if oversized and not already:
        print(f"not pointing {os.path.basename(CODEX_AGENTS)} at a {size // 1024}K bundle "
              f"(~{size // 4000}k tokens): Codex would be told to read it, and truncate it. "
              f"Use --mode index, or accept it with --warn-bytes {size}.", file=sys.stderr)
        added = False
    else:
        if oversized:
            print(f"warning: {CODEX_OWNED} is {size // 1024}K (~{size // 4000}k tokens); "
                  f"a Codex session told to read it will get a truncated slice.",
                  file=sys.stderr)
        added = append_once(CODEX_AGENTS, "claude_code_sync", POINTER, args.dry_run)
    if not args.dry_run:
        state["claude_memory_sha"] = digest
    print(f"{'would sync' if args.dry_run else 'synced'} claude->codex: {CODEX_OWNED} "
          f"[{mode}, {size // 1024}K]"
          + (f" (+ pointer in {os.path.basename(CODEX_AGENTS)})" if added else ""))
    return 1


def main():
    global VERBOSE
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--to-claude", action="store_true", help="only sync codex -> claude")
    ap.add_argument("--to-codex", action="store_true", help="only sync claude -> codex")
    ap.add_argument("--project", help="Claude project dir to receive Codex memory "
                                      "(default: the one matching your home dir)")
    ap.add_argument("--mode", choices=("auto", "full", "index"), default="auto",
                    help="what to hand Codex: full memory text, or an index of names, "
                         "descriptions and paths. auto (default) picks full when it fits "
                         "under --warn-bytes, else index")
    ap.add_argument("--warn-bytes", type=int, default=65536,
                    help="warn when the bundle Codex must read exceeds this size "
                         "(default 65536; it costs context on every session)")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("-v", "--verbose", action="store_true")
    ap.add_argument("--quiet", action="store_true", help="no output when idle (for cron)")
    args = ap.parse_args()
    VERBOSE = args.verbose

    if not os.path.isdir(CODEX_HOME) or not os.path.isdir(CLAUDE_PROJECTS):
        log("codex or claude home missing, nothing to do")
        return 0

    lock = acquire_lock(LOCK_FILE)
    if lock is None:
        return 0

    state = load_state()
    n = 0
    if not args.to_codex:
        n += codex_to_claude(state, args)
    if not args.to_claude:
        n += claude_to_codex(state, args)
    if not args.dry_run:
        save_state(state)
    if n == 0 and not args.quiet:
        print("memory-sync: nothing to do")
    return 0


if __name__ == "__main__":
    sys.exit(main())