replica-omnisciente/scripts/brain-to-gbrain.py

#!/usr/bin/env python3
"""
Bridge: Réplica brain sessions → gbrain conversation pages.

Reads .aurelio/brain/session-*/ directories and pushes each session as a
gbraidge 'conversation' page via the gbrain CLI.

Output per session:
  slug:  conversations/<source>/<session-id>
  body:  YAML frontmatter + formatted transcript

Usage:
  python3 scripts/brain-to-gbrain.py [--dry-run] [--source NAME]...
                                      [--replica-root PATH]
                                      [--ssh HOST] [--gbrain-bin PATH]
                                      [--summary] [--limit N]

Modes:
  --ssh USER@HOST    Push via SSH to remote gbrain (CT 208).
                     Default: root@192.168.0.38 (Proxmox → pct 208)
  (local)            Use local gbrain CLI (must be initialized locally).
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any

_here = Path(__file__).resolve()
for _candidate in (_here.parent, *_here.parents):
    if (_candidate / ".aurelio").is_dir():
        REPLICA_ROOT = _candidate
        break
else:
    REPLICA_ROOT = _here.parents[2]

BRAIN_DIR = REPLICA_ROOT / ".aurelio" / "brain"

VALID_SOURCES = {
    "kimi", "hermes", "claude-code", "antigravity", "qwen-code",
    "mimocode", "pi", "opencode",
}

DEFAULT_SSH = "root@192.168.0.38"
REMOTE_GBRAIN_BIN = "/opt/pf-services-208/aurelio-gbrain/bin/gbrain"
REMOTE_CT = "208"


def _load_json(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None


def _load_chat_history(session_dir: Path) -> dict | None:
    ch = session_dir / ".system_generated" / "chat_history.json"
    return _load_json(ch)


def _load_session_meta(session_dir: Path) -> dict | None:
    sj = session_dir / "session.jsonl"
    if not sj.exists():
        return None
    lines = sj.read_text(encoding="utf-8").strip().split("\n")
    meta = {}
    for line in lines:
        try:
            rec = json.loads(line)
        except json.JSONDecodeError:
            continue
        if rec.get("type") == "session_start":
            meta.update(rec.get("data", {}))
            meta["created"] = rec.get("timestamp")
        elif rec.get("type") == "session_end":
            meta["totalEvents"] = rec.get("data", {}).get("totalEvents", 0)
    return meta or None


def _format_transcript(chat_history: dict) -> str:
    messages = chat_history.get("messages", [])
    lines = []
    for msg in messages:
        role = msg.get("role", "unknown")
        content = msg.get("content", "")
        if not content:
            continue
        model = msg.get("model") or msg.get("modelType", "")
        if role == "user":
            lines.append(f"## User\n\n{content}\n")
        elif role == "assistant":
            header = f"## Assistant"
            if model:
                header += f" ({model})"
            lines.append(f"{header}\n\n{content}\n")
        elif role in ("tool_call", "tool_result"):
            lines.append(f"### {role}\n\n{content[:800]}\n")
        elif role == "system":
            lines.append(f"### System\n\n{content[:500]}\n")
    return "\n".join(lines)


def _make_page_content(session_dir: Path, chat_history: dict, meta: dict | None) -> str:
    source = chat_history.get("source", "unknown")
    title = chat_history.get("title", f"{source} session")
    created = chat_history.get("created", "")
    modified = chat_history.get("modified", "")
    workspace = chat_history.get("workspacePath", "")
    session_id = chat_history.get("id", session_dir.name)
    source_locator = chat_history.get("sourceLocator", "")
    model_type = chat_history.get("modelType", source)
    messages = chat_history.get("messages", [])
    total_events = meta.get("totalEvents", len(messages)) if meta else len(messages)

    tags = [source, "agent-session"]
    if workspace:
        tags.append(Path(workspace).name if workspace else "")
    tags = [t for t in tags if t]

    frontmatter_lines = [
        "---",
        "type: conversation",
        f'title: "{title.replace(chr(34), chr(39))}"',
        f"date: {created[:10] if created else 'unknown'}",
        f"source: {source}",
        f"model: {model_type}",
        f"tags: [{', '.join(tags)}]",
        f"workspace: {workspace or 'unknown'}",
        f"brain_session_id: {session_id}",
        f"message_count: {len(messages)}",
        f"total_events: {total_events}",
    ]
    if source_locator:
        frontmatter_lines.append(f"source_locator: \"{source_locator}\"")
    if modified:
        frontmatter_lines.append(f"modified: {modified}")
    frontmatter_lines.append("---")
    frontmatter_lines.append("")

    body_lines = [
        f"# {title}",
        "",
        f"> Source: **{source}** | Model: **{model_type}** | Messages: **{len(messages)}**",
    ]
    if workspace:
        body_lines.append(f"> Workspace: `{workspace}`")
    body_lines.append("")
    body_lines.append("## Transcript")
    body_lines.append("")
    body_lines.append(_format_transcript(chat_history))

    return "\n".join(frontmatter_lines + body_lines)


# ---------------------------------------------------------------------------
# Remote SSH mode: stage files locally, then import on CT 208 via --no-embed
# ---------------------------------------------------------------------------

def _ssh_gbrain_import(ssh_host: str, staging_dir: Path, dry_run: bool) -> dict:
    """SCP staging dir to CT 208, run gbrain import --no-embed."""
    if dry_run:
        count = len(list(staging_dir.glob("*.md")))
        return {"ok": True, "imported": count, "dry_run": True}

    try:
        # Tar + scp for efficiency — extract to /tmp so files land at /tmp/<staging_dir_name>/
        tar_cmd = f"tar czf - -C {staging_dir.parent} {staging_dir.name}"
        ssh_cmd = f"ssh -o ConnectTimeout=10 {ssh_host} pct exec {REMOTE_CT} -- tar xzf - -C /tmp"
        tar_proc = subprocess.Popen(tar_cmd, shell=True, stdout=subprocess.PIPE)
        ssh_proc = subprocess.Popen(ssh_cmd, shell=True, stdin=tar_proc.stdout, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        tar_proc.stdout.close()
        tar_proc.wait()
        ssh_proc.wait()

        if tar_proc.returncode != 0 or ssh_proc.returncode != 0:
            return {"ok": False, "error": f"tar/scp failed: {ssh_proc.stderr.decode()[:200]}"}

        remote_import_path = f"/tmp/{staging_dir.name}"

        # Run gbrain import --no-embed
        import_cmd = (
            f"pct exec {REMOTE_CT} -- {REMOTE_GBRAIN_BIN} "
            f"import {remote_import_path} --no-embed"
        )
        r = subprocess.run(
            ["ssh", "-o", "ConnectTimeout=10", ssh_host, import_cmd],
            capture_output=True, text=True, timeout=600,
        )

        # Cleanup remote
        subprocess.run(
            ["ssh", "-o", "ConnectTimeout=5", ssh_host,
             f"pct exec {REMOTE_CT} -- rm -rf {remote_import_path}"],
            capture_output=True, timeout=15,
        )

        if r.returncode == 0:
            output = r.stdout + r.stderr
            imported = 0
            for line in output.split("\n"):
                if "imported" in line and "pages" in line:
                    parts = line.split()
                    for i, p in enumerate(parts):
                        if p == "imported" and i > 0:
                            try:
                                imported = int(parts[i - 1])
                            except ValueError:
                                pass
            return {"ok": True, "imported": imported, "output": output.strip()[:500]}
        # Even with non-zero exit, check if pages were imported (warnings cause exit 1)
        output = (r.stdout or "") + (r.stderr or "")
        imported = 0
        for line in output.split("\n"):
            if "imported" in line and "pages" in line:
                parts = line.split()
                for i, p in enumerate(parts):
                    if p == "imported" and i > 0:
                        try:
                            imported = int(parts[i - 1])
                        except ValueError:
                            pass
        if imported > 0:
            return {"ok": True, "imported": imported, "output": output.strip()[:500]}
        return {"ok": False, "error": output.strip()[:500]}

    except (FileNotFoundError, subprocess.TimeoutExpired) as e:
        return {"ok": False, "error": str(e)}


# ---------------------------------------------------------------------------
# Local mode: put pages via local gbrain CLI
# ---------------------------------------------------------------------------

def _local_gbrain_put(gbrain_bin: str | None, slug: str, content: str, dry_run: bool) -> dict:
    if dry_run:
        return {"ok": True, "slug": slug, "dry_run": True, "content_length": len(content)}

    cmd = [gbrain_bin or "gbrain", "put", slug]
    try:
        r = subprocess.run(cmd, input=content, capture_output=True, text=True, timeout=120)
        if r.returncode == 0:
            return {"ok": True, "slug": slug}
        return {"ok": False, "slug": slug, "error": r.stderr.strip() or r.stdout.strip()}
    except (FileNotFoundError, subprocess.TimeoutExpired) as e:
        return {"ok": False, "slug": slug, "error": str(e)}


def discover_sessions(brain_dir: Path, source_filter: set[str] | None = None) -> list[Path]:
    sessions = []
    for d in sorted(brain_dir.iterdir()):
        if not d.is_dir() or not d.name.startswith("session-"):
            continue
        parts = d.name.split("-", 2)
        if len(parts) < 2:
            continue
        source = parts[1]
        if source_filter and source not in source_filter:
            continue
        if source not in VALID_SOURCES:
            continue
        sessions.append(d)
    return sessions


def run(args: argparse.Namespace) -> dict[str, Any]:
    replica_root = Path(args.replica_root).resolve() if args.replica_root else REPLICA_ROOT
    brain_dir = replica_root / ".aurelio" / "brain"

    source_filter = set(args.source) if args.source else None
    sessions = discover_sessions(brain_dir, source_filter)

    if args.limit:
        sessions = sessions[:args.limit]

    ssh_host = args.ssh
    use_remote = bool(ssh_host)

    summary: dict[str, Any] = {
        "replica_root": str(replica_root),
        "brain_dir": str(brain_dir),
        "dry_run": args.dry_run,
        "mode": "remote-ssh" if use_remote else "local",
        "remote": ssh_host or None,
        "total_sessions": len(sessions),
        "pushed": 0,
        "failed": 0,
        "errors": [],
    }

    if use_remote:
        # Stage all files locally, then bulk import
        staging = replica_root / ".aurelio" / "gbraidge-staging"
        staging.mkdir(parents=True, exist_ok=True)

        for sess_dir in sessions:
            chat_history = _load_chat_history(sess_dir)
            if not chat_history:
                continue
            meta = _load_session_meta(sess_dir)
            source = chat_history.get("source", "unknown")
            session_id = chat_history.get("id", sess_dir.name)
            slug = f"conversations/{source}/{session_id.removeprefix('session-')}"
            content = _make_page_content(sess_dir, chat_history, meta)

            out = staging / f"{slug.replace('/', '_')}.md"
            out.parent.mkdir(parents=True, exist_ok=True)
            if not args.dry_run:
                out.write_text(content, encoding="utf-8")
            summary["pushed"] += 1

        # Bulk import
        if not args.dry_run:
            print(f"Staged {summary['pushed']} files. Importing to gbrain...", file=sys.stderr)
            result = _ssh_gbrain_import(ssh_host, staging, args.dry_run)
            summary["import_result"] = result
            if not result.get("ok"):
                summary["failed"] = summary["pushed"]
                summary["errors"].append({"error": result.get("error", "unknown")})
            else:
                summary["imported"] = result.get("imported", 0)

        # Cleanup staging
        if not args.dry_run:
            import shutil
            shutil.rmtree(staging, ignore_errors=True)
    else:
        for sess_dir in sessions:
            chat_history = _load_chat_history(sess_dir)
            if not chat_history:
                continue
            meta = _load_session_meta(sess_dir)
            source = chat_history.get("source", "unknown")
            session_id = chat_history.get("id", sess_dir.name)
            slug = f"conversations/{source}/{session_id.removeprefix('session-')}"
            content = _make_page_content(sess_dir, chat_history, meta)

            result = _local_gbrain_put(args.gbrain_bin, slug, content, args.dry_run)
            if result["ok"]:
                summary["pushed"] += 1
                if args.verbose:
                    print(f"  OK  {slug}")
            else:
                summary["failed"] += 1
                err = result.get("error", "unknown")
                summary["errors"].append({"slug": slug, "error": err})
                if args.verbose:
                    print(f"  FAIL {slug}: {err}", file=sys.stderr)

    return summary


def main() -> int:
    ap = argparse.ArgumentParser(description="Bridge brain sessions → gbrain conversation pages.")
    ap.add_argument("--dry-run", action="store_true", help="Report what would be pushed without touching gbrain.")
    ap.add_argument("--source", action="append", help="Limit to specific sources (repeatable).")
    ap.add_argument("--replica-root", default=None)
    ap.add_argument("--ssh", default=None,
                    help=f"SSH target for remote gbrain (default: {DEFAULT_SSH}). "
                         "Set to empty string for local mode.")
    ap.add_argument("--gbrain-bin", default=None, help="Path to local gbrain binary.")
    ap.add_argument("--limit", type=int, default=None, help="Max sessions to process.")
    ap.add_argument("--summary", action="store_true", help="Print JSON summary.")
    ap.add_argument("--verbose", "-v", action="store_true")
    args = ap.parse_args()

    if args.ssh is None:
        args.ssh = DEFAULT_SSH

    summary = run(args)

    if args.summary:
        print(json.dumps(summary, indent=2, ensure_ascii=False))
    else:
        print(f"mode: {summary['mode']}  remote: {summary['remote'] or 'local'}")
        print(f"sessions: {summary['total_sessions']}  pushed: {summary['pushed']}  failed: {summary['failed']}")
        if summary.get("imported"):
            print(f"imported: {summary['imported']}")
        if args.dry_run:
            print("(dry-run: nothing pushed)")
        for err in summary.get("errors", []):
            print(f"  FAIL: {err.get('slug', 'batch')}: {err.get('error', 'unknown')}", file=sys.stderr)

    return 1 if summary.get("failed", 0) > 0 else 0


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