NexusOS/bin/sync.py

#!/usr/bin/env python3
"""NexusOS backup and restore - one entry point for Linux and native Windows.

Same command on both boxes. Everything portable lives here: the git sync with
Gitea, the memory-DB dump / compare / rebuild, and the venv + web-UI rebuild.
The parts that only mean something on Linux - apt packages, the bundled Ollama
binary, the XFCE desktop wiring, the desktop snapshot - stay in bash and get
called from here, skipped outright on Windows.

    python bin/sync.py restore [--check]
    python bin/sync.py backup  [--check] [--full] [--force-db]
    python bin/sync.py compare                 # print the direction verdict only

Standard library only, so it runs before the Promethean venv exists and needs no
sqlite3 binary on PATH - Windows has none, which is why the old
bin/db-compare.sh could never work there.

A fresh machine clones first (git clone <repo> nexus-core), then runs this.
"""
import argparse
import os
import re
import shutil
import sqlite3
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
DB = ROOT / "synapse" / "memory" / "memory.db"
DB_SQL = ROOT / "synapse" / "memory" / "memory.db.sql"
DB_SQL_REL = "synapse/memory/memory.db.sql"   # git paths are always posix-style


def run(*args, capture=False, check=False):
    """Run a command from the repo root. Returns CompletedProcess."""
    exe = shutil.which(args[0])   # resolves npm -> npm.cmd on Windows
    if exe is None:
        raise SystemExit(f"'{args[0]}' not found on PATH")
    return subprocess.run(
        [exe, *args[1:]], cwd=ROOT, check=check,
        capture_output=capture, text=True, encoding="utf-8",
    )


def ensure_exec_bits() -> None:
    """Force +x on every path git tracks as executable (mode 100755), straight
    from `git ls-files` rather than a hardcoded list - self-maintaining as
    scripts are added. Guards against core.fileMode=false (repo-local or a
    machine-wide /etc/gitconfig) silently dropping exec bits on checkout, which
    otherwise surfaces later as a confusing "Permission denied" on whichever
    script happens to run next (fetch-ollama.sh, the ncp symlink target, ...)
    rather than as an obvious failure right here."""
    if os.name == "nt":
        return
    result = run("git", "ls-files", "-s", capture=True)
    for line in result.stdout.splitlines():
        mode, _, rest = line.partition(" ")
        if mode != "100755":
            continue
        path = ROOT / rest.split("\t", 1)[1]
        if path.exists():
            path.chmod(path.stat().st_mode | 0o111)


def linux_stage(script: str, *args) -> None:
    """Run one of the Linux-only bash stages. A no-op on Windows, where apt,
    xfconf, plank and the rest have nothing to act on."""
    if os.name == "nt":
        return
    path = ROOT / "bin" / script
    bash = shutil.which("bash")
    if path.exists() and bash:
        # Pin the stage to THIS repo. It defaults to ~/nexus-core otherwise, so a
        # clone in a scratch dir would restore over the real machine instead.
        subprocess.run([bash, str(path), *args], cwd=ROOT, check=False,
                       env={**os.environ, "NEXUS_ROOT": str(ROOT)})


def venv_python() -> Path:
    """Path to the Promethean interpreter, creating the venv if it's missing."""
    venv = ROOT / "Promethean"
    py = venv / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
    if py.exists() and subprocess.run(
            [str(py), "-m", "pip", "--version"], capture_output=True).returncode:
        # venv creation can partially succeed: the interpreter gets built but
        # ensurepip's bootstrap fails (e.g. the matching pythonX.Y-venv package
        # wasn't installed yet), leaving pip missing. Existence of `py` alone
        # can't tell a venv like that apart from a good one, so a prior failed
        # run would otherwise be reused forever instead of getting rebuilt now
        # that whatever broke ensurepip is fixed.
        print("Existing venv has no pip - recreating it...")
        shutil.rmtree(venv)
    if not py.exists():
        result = subprocess.run([sys.executable, "-m", "venv", str(venv)])
        if result.returncode:
            raise SystemExit(
                "\nvenv creation failed - the Python 'venv' module isn't usable here.\n"
                "On Debian/Ubuntu this is usually a missing pythonX.Y-venv package; "
                "check the 'Installing system packages' warning further up, install "
                "it by hand, then re-run."
            )
    return py


def requirements() -> str:
    """Pick the PyTorch overlay for this host."""
    if os.name == "nt":
        return "requirements-windows.txt"        # CPU / pure-Python, right for native Windows
    if shutil.which("nvidia-smi"):
        return "requirements-nvidia.txt"
    lspci = shutil.which("lspci")
    if lspci and "nvidia" in subprocess.run(
            [lspci], capture_output=True, text=True).stdout.lower():
        return "requirements-nvidia.txt"
    return "requirements-amd.txt"


# -- memory DB -----------------------------------------------------------------

def _vec0_extension() -> "Path | None":
    """Path to the sqlite_vec native extension inside the venv, if built. Found
    by glob rather than importing sqlite_vec - this script runs stdlib-only,
    before the venv necessarily exists."""
    matches = list((ROOT / "Promethean").glob("**/sqlite_vec/vec0.*"))
    return matches[0] if matches else None


_DERIVED = re.compile(
    r'^\s*(?:CREATE\s+(?:VIRTUAL\s+)?TABLE|INSERT\s+INTO)\s+"?vec_(?:messages|documents)',
    re.IGNORECASE,
)


def _is_derived_stmt(stmt: str) -> bool:
    """True for a dump statement that builds or fills a sqlite_vec table.

    Matched on the statement's TARGET, never as a substring: a chat message
    whose text happens to mention vec_messages is an INSERT INTO "messages" and
    has to survive. (It didn't, the first time - the test caught it.)
    """
    if _DERIVED.match(stmt):
        return True
    # iterdump writes a virtual table straight into the schema table, and parks
    # its shadow tables' AUTOINCREMENT counters in sqlite_sequence. Both name the
    # vec table as a quoted VALUE rather than as the statement's target.
    head = stmt.lstrip().upper()
    if (head.startswith("INSERT INTO SQLITE_MASTER")
            or head.startswith('INSERT INTO "SQLITE_SEQUENCE"')) and "'vec_" in stmt:
        return True
    # Only meaningful when replaying over an existing DB, and sqlite_sequence
    # exists only if some AUTOINCREMENT table survives the filter - today true
    # by luck alone. Restore deletes the DB file first, so drop it.
    return stmt.lstrip().upper().startswith('DELETE FROM "SQLITE_SEQUENCE"')


def dump_db() -> bool:
    """Dump the (gitignored, binary, WAL) memory DB to a diff-friendly SQL file
    so git backs up the assistant's memory + conversation history. sqlite3 reads
    through the WAL, so the dump has the latest committed data uncheckpointed."""
    if not DB.exists():
        return False
    try:
        with sqlite3.connect(f"file:{DB}?mode=ro", uri=True) as conn:
            ext = _vec0_extension()
            if ext:
                try:
                    conn.enable_load_extension(True)
                    conn.load_extension(str(ext))
                    conn.enable_load_extension(False)
                except (sqlite3.OperationalError, AttributeError):
                    pass  # iterdump below will fail with a clear "no such module" if this was needed
            # Drop the sqlite_vec virtual tables and their shadow tables.
            # iterdump() serializes a virtual table as a raw
            # INSERT INTO sqlite_master(...) followed by inserts into a table
            # the connection can't see yet, so replaying the dump dies with
            # "no such table: vec_messages" and leaves ZERO tables behind -
            # the whole backup was unrestorable. They're derived data anyway:
            # _backfill_vec_msgs()/_backfill_vec() rebuild both indexes from
            # message_vectors and documents on the next search.
            # Filter the generator, not the joined text - each yield is one
            # complete statement, while splitting the text on ";\n" tears
            # apart INSERTs whose content contains newlines.
            stmts = [s for s in conn.iterdump() if not _is_derived_stmt(s)]
            DB_SQL.write_text("\n".join(stmts) + "\n", encoding="utf-8")
    except sqlite3.Error as exc:
        print(f"Warning: could not dump {DB} ({exc}) - DB not captured")
        return False
    print(f"Dumped memory DB -> {DB_SQL}")
    return True


def _state(conn):
    """What a machine holds: conversations as {id: updated_at} and memory facts
    as a set of ids. NOT messages.id - it's INTEGER AUTOINCREMENT, so both
    machines hand out the same ids independently and comparing them is
    meaningless. A new message bumps its conversation's updated_at, which is
    what actually gets caught here."""
    def query(sql):
        try:
            return conn.execute(sql).fetchall()
        except sqlite3.Error:
            return []   # table absent in an old dump - treat as empty
    return (
        dict(query("select id, updated_at from conversations")),
        {row[0] for row in query("select id from memory")},
    )


def _extra(a, b) -> int:
    """How much `a` holds that `b` lacks: conversations `b` is missing or has an
    older copy of, plus memory facts `b` doesn't have at all. An out-of-date
    copy is NOT extra content on b's side - that asymmetry is what separates
    'one box is simply ahead' from a real divergence."""
    a_conv, a_mem = a
    b_conv, b_mem = b
    # Membership test first, never a placeholder default: updated_at is a REAL
    # column, so ANY typed default is a cross-type comparison against the one
    # case this function exists to count - a conversation the other side lacks.
    # ("" < 1778553309.83 raises TypeError; 0 only worked by accident.)
    newer = sum(1 for cid, ts in a_conv.items() if cid not in b_conv or b_conv[cid] < ts)
    return newer + len(a_mem - b_mem)


def compare(db: Path = DB, dump: Path = DB_SQL) -> str:
    """Which way the sync should go. One of: same, local-ahead, local-behind,
    diverged, no-live, no-dump, unreadable.
    ponytail: detects direction, does not merge. Diverged is reported, not resolved."""
    if not dump.exists() or not dump.stat().st_size:
        return "no-dump"
    if not db.exists():
        return "no-live"
    try:
        with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as live_conn:
            live = _state(live_conn)
        with sqlite3.connect(":memory:") as dump_conn:
            dump_conn.executescript(dump.read_text(encoding="utf-8"))
            backup = _state(dump_conn)
    except (sqlite3.Error, OSError):
        # A dump that won't replay is NOT a divergence. Reporting it as one
        # blocked backup and restore alike with a verdict that looked like a
        # real answer - dumps written before the vec-table filter above land
        # here every time. Still refuses both directions, but says why.
        return "unreadable"
    ahead, behind = _extra(live, backup), _extra(backup, live)
    if ahead and behind:
        return "diverged"
    if ahead:
        return "local-ahead"
    if behind:
        return "local-behind"
    return "same"


def restore_db() -> None:
    """Rebuild the memory DB from the pulled dump, keeping a rollback copy."""
    print()
    state = compare()
    if state == "same":
        print("Memory DB already matches the backup.")
        return
    if state == "local-ahead":
        print("WARNING: this machine has conversations the backup doesn't.")
        print("         Not applying the dump - run `backup` HERE first.")
        print(f"         To discard local history instead: delete {DB} and restore")
        return
    if state == "diverged":
        print("WARNING: this machine and the backup each have conversations the other lacks.")
        print("         Not applying the dump - nothing is merging these automatically.")
        print(f"         Keep local: run `backup` here. Keep remote: delete {DB} and restore")
        return
    if state == "no-dump":
        print("No memory dump in the backup - skipping DB restore.")
        return
    if state == "unreadable":
        print("WARNING: the backup's memory dump will not replay (written before the")
        print("         vec-table fix). Not applying it - it would leave an empty DB.")
        print("         Run `backup --force-db` on the machine with the good history")
        print("         to publish a clean dump, then restore here.")
        return

    print("Restoring memory DB from backup...")
    rollback = DB.with_suffix(".db.pre-restore")
    if DB.exists():
        shutil.copyfile(DB, rollback)
    for suffix in ("", "-wal", "-shm"):
        Path(str(DB) + suffix).unlink(missing_ok=True)
    try:
        with sqlite3.connect(DB) as conn:
            conn.executescript(DB_SQL.read_text(encoding="utf-8"))
        print("Memory DB restored (conversations + history + facts).")
    except sqlite3.Error as exc:
        print(f"Warning: memory DB rebuild failed ({exc}).")
        if rollback.exists():
            shutil.move(rollback, DB)
            print("Rolled back to previous DB.")


# -- rebuild -------------------------------------------------------------------

def rebuild_env() -> None:
    print("\nRebuilding Python environment...")
    py, req = venv_python(), requirements()
    if (ROOT / req).exists():
        subprocess.run([str(py), "-m", "pip", "install", "--upgrade", "pip", "-q"], check=False)
        subprocess.run([str(py), "-m", "pip", "install", "-r", req], cwd=ROOT, check=False)
    else:
        print(f"Warning: {req} not found - skipping pip install.")

    print("\nRebuilding frontend dependencies...")
    web = ROOT / "interface" / "web"
    npm = shutil.which("npm")
    if npm is None:
        print("Warning: npm not found - the web UI will not be built.")
        return
    subprocess.run([npm, "install"], cwd=web, check=False)
    # Build the UI so the backend can serve it single-process (it mounts
    # interface/web/dist at :8000). Without this the app has no UI to show.
    print("Building frontend...")
    subprocess.run([npm, "run", "build"], cwd=web, check=False)
    if not (web / "dist" / "index.html").exists():
        print("Warning: interface/web/dist/index.html missing - the app will serve no UI.")


# -- commands ------------------------------------------------------------------

def cmd_restore(args) -> int:
    if args.check:
        print("Fetching to preview restore (no changes)...")
        run("git", "fetch", "origin")
        print("\nCommits a restore would apply:")
        print(run("git", "log", "--oneline", "..origin/main", capture=True).stdout or "(up to date)")
        print(run("git", "diff", "--stat", "..origin/main", capture=True).stdout)
        print(f"Memory DB vs backup: {compare()}")
        print("\n(dry-run only - nothing changed. Apply with: restore)")
        return 0

    # System packages first - the venv and npm build below need them present.
    linux_stage("restore-linux.sh", "prep")

    print("\nPulling latest from Gitea...")
    if run("git", "pull", "--ff-only", "origin", "main").returncode:
        print("Pull failed (diverged? stash/commit local changes).")
        return 1
    ensure_exec_bits()
    restore_db()
    rebuild_env()
    linux_stage("restore-linux.sh", "runtime")
    if not args.no_desktop:
        linux_stage("restore-linux.sh", "desktop")
    print("\nRestore complete. Nexus is ready to start.")
    print("Note: Ollama models are not in the backup. Open the Models tab -> Required")
    print("and pull the two models NexusOS depends on (memory curator + embeddings)")
    print("before anything else, then pick a hardware-fit chat model.")
    return 0


def _remote_dump(tmp: Path) -> bool:
    """Write origin/main's dump to tmp. False if it can't be fetched."""
    if run("git", "fetch", "-q", "origin").returncode:
        print("Note: could not reach Gitea - skipping backup safety check.")
        return False
    shown = run("git", "show", f"origin/main:{DB_SQL_REL}", capture=True)
    if shown.returncode:
        return False
    tmp.write_text(shown.stdout, encoding="utf-8")
    return True


def check_db_direction(tmp_dir: Path) -> bool:
    """Mirror of restore's guard: refuse to dump a stale live DB over a backup
    that already holds newer conversations from the other machine. Compares
    against the REMOTE dump, not the working-tree copy - that copy is from this
    box's last backup and is exactly what goes stale when the other box pushes."""
    if not DB.exists():
        return True
    tmp = tmp_dir / "remote.sql"
    if not _remote_dump(tmp):
        return True
    state = compare(DB, tmp)
    if state == "local-behind":
        print("REFUSING TO BACK UP: the backup has conversations this machine doesn't.")
        print("  Backing up now would overwrite them with this box's older history.")
        print("  Run `restore` here first, then back up.")
        return False
    if state == "diverged":
        print("REFUSING TO BACK UP: this machine and the backup each have conversations")
        print("  the other lacks. Nothing merges these automatically.")
        print("  Force this box's history to win: python bin/sync.py backup --force-db")
        return False
    if state == "unreadable":
        print("REFUSING TO BACK UP: the remote memory dump will not replay, so there is")
        print("  no way to tell whether it holds history this machine lacks.")
        print("  If this box has the good history: python bin/sync.py backup --force-db")
        print("  (that publishes a clean dump and clears this for good).")
        return False
    return True


def cmd_backup(args) -> int:
    import tempfile
    if args.full:
        linux_stage("backup-linux.sh")
    dump_db()
    with tempfile.TemporaryDirectory() as tmp_dir:
        safe = check_db_direction(Path(tmp_dir))

    if args.check:
        print(f"\nMemory DB vs backup: {'safe to back up' if safe else 'STALE - restore first'}")
        print("\nFiles a backup would commit:")
        print(run("git", "status", "--short", capture=True).stdout)
        print("Local commits not yet pushed:")
        print(run("git", "log", "--oneline", "@{u}..", capture=True).stdout or "(none / no upstream)")
        print("(dry-run only - nothing committed or pushed. Apply with: backup)")
        return 0
    if not safe and not args.force_db:
        return 1

    run("git", "add", "-A")
    if run("git", "diff", "--cached", "--quiet").returncode:
        stamp = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
        if run("git", "commit", "-q", "-m", f"backup: {stamp}").returncode:
            print("Commit failed - see error above (e.g. git identity not set).")
            return 1
        print("Committed backup snapshot.")
    else:
        print("No changes to commit.")

    print("Pushing to Gitea...")
    if run("git", "push", "origin", "main").returncode:
        print("Push failed. Set up credentials once with:")
        print("  git config credential.helper store   # then push once and enter your Gitea token")
        return 1
    print("Backup complete.")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="cmd", required=True)

    restore = sub.add_parser("restore", help="pull from Gitea, rebuild DB + venv + web UI")
    restore.add_argument("-c", "--check", action="store_true", help="dry run, change nothing")
    restore.add_argument("--no-desktop", action="store_true",
                         help="skip the XFCE desktop wiring (panel, theme, os-release). "
                              "Use for a test clone - that stage writes to $HOME, not the repo.")
    restore.set_defaults(func=cmd_restore)

    backup = sub.add_parser("backup", help="dump DB, commit and push to Gitea")
    backup.add_argument("-c", "--check", action="store_true", help="dry run, change nothing")
    backup.add_argument("-f", "--full", action="store_true",
                        help="also snapshot the live desktop wiring + Claude notes (Linux)")
    backup.add_argument("--force-db", action="store_true", help="skip the staleness guard")
    backup.set_defaults(func=cmd_backup)

    compare_cmd = sub.add_parser("compare", help="print the sync direction verdict")
    compare_cmd.set_defaults(func=lambda a: (print(compare()), 0)[1])

    args = parser.parse_args()
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())