replica-omnisciente/scripts/sync-agents-to-brain.py

#!/usr/bin/env python3
"""
Unified agent → Réplica Omnisciente brain importer.

Imports session history, plans and artifacts from every coding agent in the
fleet into the unified brain at <replica>/.aurelio/brain/, mirroring the Kimi
bridge (`scripts/sync-kimi-to-brain.py`) but for all sources:

  kimi          ~/.kimi/sessions/**/context.jsonl            (+ plans)
  hermes        ~/.hermes/sessions/{saved,request_dump_*}.json
  claude-code   ~/.claude/projects/<hash>/<session>.jsonl    (+ plans)
  antigravity   ~/.gemini/antigravity-{cli,ide}/conversations/*.db (sqlite)
  qwen-code     ~/.qwen[-code]/**/{context.jsonl,*.jsonl}    (no-op if absent)
  mimocode      ./.mimocode/plans/*.md  (+ future *.jsonl)
  pi            ~/.pi/agent/sessions/**/*.jsonl
  opencode      ~/.local/share/opencode/opencode.db (sqlite)

Output contract per session (see agent_importers/engine.py):
  .aurelio/brain/session-<source>-<id>/{session.jsonl, session_memory.md,
    .system_generated/chat_history.json, .system_generated/logs/overview.txt}

Idempotent: re-running writes only when content changes; timestamps are derived
from the source artifact so identical inputs produce byte-identical outputs.

Usage:
  python3 scripts/sync-agents-to-brain.py [--source NAME]... [--dry-run]
                                          [--skip-active] [--replica-root PATH]
                                          [--summary]
  # --source may be repeated; default = all known sources.
"""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Any

# Make `agent_importers` importable whether run as a module or a script.
sys.path.insert(0, str(Path(__file__).resolve().parent))

from agent_importers import (  # noqa: E402
    ADAPTERS,
    REPLICA_ROOT,
    write_session,
    derive_title,
)
from agent_importers.adapters import (  # noqa: E402
    discover_kimi,
    discover_hermes,
    discover_claude_code,
    discover_antigravity_cli,
    discover_qwen_code,
    discover_mimocode,
    discover_pi,
    discover_opencode,
)


def _is_recent(path_str: str, seconds: int = 60) -> bool:
    try:
        return (datetime.now().timestamp() - Path(path_str).stat().st_mtime) < seconds
    except OSError:
        return False


def _title_for(sess) -> str:
    return sess.title or derive_title(sess.messages, sess.workspace_path, f"{sess.source} session")


def run(args: argparse.Namespace) -> dict[str, Any]:
    replica_root = Path(args.replica_root).resolve() if args.replica_root else REPLICA_ROOT
    # Patch the module-level roots used by the engine/writer.
    import agent_importers.engine as eng
    eng.REPLICA_ROOT = replica_root
    eng.BRAIN_DIR = replica_root / ".aurelio" / "brain"

    requested = args.source or list(ADAPTERS.keys())
    summary: dict[str, Any] = {
        "replica_root": str(replica_root),
        "brain_dir": str(eng.BRAIN_DIR),
        "dry_run": args.dry_run,
        "sources": {},
        "totals": {"sessions": 0, "written": 0, "skipped_empty": 0, "skipped_active": 0},
    }

    dispatch = {
        "kimi": lambda: discover_kimi(Path(args.kimi_dir) if args.kimi_dir else None),
        "hermes": lambda: discover_hermes(Path(args.hermes_dir) if args.hermes_dir else None),
        "claude-code": lambda: discover_claude_code(Path(args.claude_dir) if args.claude_dir else None),
        "antigravity": lambda: discover_antigravity_cli(Path(args.gemini_dir) if args.gemini_dir else None),
        "qwen-code": lambda: discover_qwen_code(Path(args.qwen_dir) if args.qwen_dir else None),
        "mimocode": lambda: discover_mimocode(repo_root=replica_root),
        "pi": lambda: discover_pi(Path(args.pi_dir) if args.pi_dir else None),
        "opencode": lambda: discover_opencode(Path(args.opencode_db) if args.opencode_db else None),
    }

    for name in requested:
        if name not in ADAPTERS:
            print(f"unknown source: {name}", file=sys.stderr)
            continue
        src_stats = {"sessions": 0, "written": 0, "skipped_empty": 0, "skipped_active": 0, "files": []}
        try:
            sessions = list(dispatch[name]())
        except Exception as exc:  # defensive: one bad source must not abort the run
            print(f"[{name}] discovery failed: {exc}", file=sys.stderr)
            sessions = []

        for sess in sessions:
            if not sess.title:
                sess.title = _title_for(sess)
            if args.skip_active and _is_recent(sess.source_locator):
                src_stats["skipped_active"] += 1
                continue
            res = write_session(sess, dry_run=args.dry_run)
            src_stats["sessions"] += 1
            if res.get("skipped_empty"):
                src_stats["skipped_empty"] += 1
            wrote = res.get("written") or res.get("would_write") or []
            if wrote:
                src_stats["written"] += len(wrote)
                if args.verbose:
                    src_stats["files"].append({res["brain_id"]: wrote})

        summary["sources"][name] = src_stats
        summary["totals"]["sessions"] += src_stats["sessions"]
        summary["totals"]["written"] += src_stats["written"]
        summary["totals"]["skipped_empty"] += src_stats["skipped_empty"]
        summary["totals"]["skipped_active"] += src_stats["skipped_active"]

    return summary


def main() -> int:
    ap = argparse.ArgumentParser(description="Import all coding-agent artifacts into the Réplica brain.")
    ap.add_argument("--source", action="append", choices=sorted(ADAPTERS.keys()),
                    help="Limit to one or more sources (repeatable). Default: all.")
    ap.add_argument("--dry-run", action="store_true", help="Report what would be written without touching disk.")
    ap.add_argument("--skip-active", action="store_true", help="Skip artifacts modified in the last 60s.")
    ap.add_argument("--replica-root", default=None, help="Override replica-omnisciente root.")
    ap.add_argument("--kimi-dir", default=None)
    ap.add_argument("--hermes-dir", default=None)
    ap.add_argument("--claude-dir", default=None)
    ap.add_argument("--gemini-dir", default=None)
    ap.add_argument("--qwen-dir", default=None)
    ap.add_argument("--pi-dir", default=None)
    ap.add_argument("--opencode-db", default=None)
    ap.add_argument("--summary", action="store_true", help="Print JSON summary.")
    ap.add_argument("--verbose", "-v", action="store_true")
    args = ap.parse_args()

    summary = run(args)

    if args.summary:
        print(json.dumps(summary, indent=2, ensure_ascii=False))
    else:
        t = summary["totals"]
        print(f"sources: {', '.join(summary['sources']) or '(none)'}")
        print(f"sessions: {t['sessions']}  files_written: {t['written']}  "
              f"skipped_empty: {t['skipped_empty']}  skipped_active: {t['skipped_active']}")
        if args.dry_run:
            print("(dry-run: nothing written)")
        for name, st in summary["sources"].items():
            print(f"  {name:14s} sessions={st['sessions']:4d}  wrote={st['written']:4d}  "
                  f"empty={st['skipped_empty']:3d}  active={st['skipped_active']:3d}")
    return 0


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