migrate_hermes: Гермес → Simbiot (t-261001124130) — память/персона/история/config/cron → пакет + PolicyChanged в стор целевого инстанса + RAG-индексация; dry-run по умолчанию

scripts/migrate_hermes.py

@@ -0,0 +1,409 @@
+# -*- coding: utf-8 -*-
+"""Simbiot Harness — миграция данных агента Гермес → Simbiot (t-261001124130).
+
+Читает каталог Гермеса (~/.hermes): память (memories/MEMORY.md, USER.md),
+персону (SOUL.md), историю диалогов (state.db: sessions/messages), настройки
+(config.yaml), кроны (cron/jobs.json) — и собирает пакет для нового агента
+Simbiot:
+
+  OUT/identity.md        — склейка персоны+памяти → policy.identity
+  OUT/memory/MEMORY.md   — факты памяти (как были, в формате Гермеса)
+  OUT/memory/USER.md     — профиль пользователя
+  OUT/history/*.md       — история диалогов, активные сессии (по файлу на сессию)
+  OUT/history-index.md   — индекс сессий
+  OUT/config-hermes.yaml — настройка (секреты замаскированы)
+  OUT/cron-map.json      — джобы крона → эквиваленты Simbiot (задачи/таймеры)
+  OUT/report.json        — отчёт миграции (counts, notes, events)
+
+Режимы:
+  --apply                — применить (по умолчанию dry-run: только отчёт, ничего
+                           не пишет в RAG/стор)
+  --harness-dir DIR      — при --apply: записать PolicyChanged/MemoryIngested в
+                           event store ЭТОГО инстанса Simbiot (агент получает
+                           память сразу в контекст тиков)
+  --rag-db PATH --rag-py P --rag-script P — при --apply проиндексировать
+                           память/историю в RAG-базу TARGET-хоста
+
+Идемпотентность: события с фиксированными aggregate_id (harness/policy-h-<hash>),
+RAG-индексация дедуплицируется по sha (вендор), файлы_OUT перезаписываются —
+повторный прогон безопасен.
+
+Запуск:  python3 scripts/migrate_hermes.py [--hermes /root/.hermes] [--apply ...]
+"""
+import argparse
+import glob
+import hashlib
+import json
+import os
+import re
+import sqlite3
+import sys
+import time
+from datetime import datetime, timezone
+
+MAX_HISTORY_SESSIONS = 40     # сколько последних сессий с диалогом переносим
+MAX_HISTORY_MSGS_PER_SESSION = 40
+MAX_TOTAL_HISTORY_MSGS = 150
+SECRET_RE = re.compile(
+    r"(sk-[A-Za-z0-9_\-]{16,}|sk-or-v1-[A-Za-z0-9_\-]{16,}|agt_[A-Za-z0-9_\-]{10,}|"
+    r"[0-9]{8,10}:[A-Za-z0-9_\-]{30,}|xox[bpars]-[A-Za-z0-9_\-]{10,}|"
+    r"gh[pousr]_[A-Za-z0-9_\-]{20,})")
+
+
+def redact(text: str) -> str:
+    return SECRET_RE.sub("<REDACTED>", text or "")
+
+
+def _jlog(msg: str):
+    print(f"[migrate] {msg}", flush=True)
+
+
+# ─── источники ───────────────────────────────────────────────────────────────
+
+def probe_sources(hermes_dir: str) -> dict:
+    src = {"hermes_dir": hermes_dir, "files": {}, "state_db": {}, "cron": {}}
+    cfg = os.path.join(hermes_dir, "config.yaml")
+    src["files"]["config.yaml"] = os.path.getsize(cfg) if os.path.exists(cfg) else None
+    for name, ext in (("MEMORY", "memories/MEMORY.md"), ("USER", "memories/USER.md"),
+                      ("SOUL", "SOUL.md")):
+        p = os.path.join(hermes_dir, ext)
+        src["files"][ext] = os.path.getsize(p) if os.path.exists(p) else None
+    state = os.path.join(hermes_dir, "state.db")
+    src["files"]["state.db"] = os.path.getsize(state) if os.path.exists(state) else None
+    if os.path.exists(state):
+        db = sqlite3.connect(f"file:{state}?mode=ro", uri=True)
+        db.row_factory = sqlite3.Row
+        for t in ("messages", "sessions", "system_prompts"):
+            try:
+                src["state_db"][t] = db.execute(f"SELECT COUNT(*) c FROM {t}").fetchone()["c"]
+            except sqlite3.Error:
+                src["state_db"][t] = None
+        try:
+            r = db.execute("SELECT COUNT(*) c FROM messages WHERE active=1 "
+                           "AND role IN ('user','assistant')").fetchone()
+            src["state_db"]["messages_active_user_assistant"] = r["c"]
+        except sqlite3.Error:
+            pass
+        db.close()
+    jobs = os.path.join(hermes_dir, "cron", "jobs.json")
+    if os.path.exists(jobs):
+        try:
+            with open(jobs) as f:
+                data = json.load(f)
+            jlist = data.get("jobs") if isinstance(data, dict) else data
+            src["cron"]["jobs_total"] = len(jlist or [])
+            src["cron"]["jobs_enabled"] = sum(1 for j in (jlist or []) if j.get("enabled"))
+        except Exception as e:
+            src["cron"]["error"] = str(e)[:200]
+    return src
+
+
+def extract_memory(hermes_dir: str) -> dict:
+    """Персона + память целиком (§-факты Гермеса и профиль юзера)."""
+    out = {}
+    parts_id = []
+    soul = os.path.join(hermes_dir, "SOUL.md")
+    if os.path.exists(soul):
+        with open(soul, encoding="utf-8", errors="replace") as f:
+            out["SOUL_raw_md"] = f.read()
+        parts_id.append("## Персона (SOUL, из Гермеса)\n" + out["SOUL_raw_md"])
+    mem = os.path.join(hermes_dir, "memories", "MEMORY.md")
+    if os.path.exists(mem):
+        with open(mem, encoding="utf-8", errors="replace") as f:
+            out["MEMORY_raw_md"] = f.read()
+        parts_id.append("## Память (MEMORY, из Гермеса, § = разделитель фактов)\n" + out["MEMORY_raw_md"])
+    user = os.path.join(hermes_dir, "memories", "USER.md")
+    if os.path.exists(user):
+        with open(user, encoding="utf-8", errors="replace") as f:
+            out["USER_raw_md"] = f.read()
+        parts_id.append("## Профиль пользователя (USER, из Гермеса)\n" + out["USER_raw_md"])
+    out["identity_md"] = "\n\n".join(parts_id)
+    return out
+
+
+def extract_history(hermes_dir: str, out_dir: str) -> dict:
+    """Активные сессии с реальным диалогом (user+assistant) → markdown по файлам.
+
+    Крон-сессии (по 1 сообщению-запуску) не тянут — это шум; живые чаты
+    (telegram/matrix/cli/subagent) переносим с капами (капы в шапке).
+    """
+    state = os.path.join(hermes_dir, "state.db")
+    res = {"sessions_total": 0, "sessions_written": 0, "messages_written": 0,
+           "skipped_single_ping": 0, "files": []}
+    if not os.path.exists(state):
+        res["error"] = "state.db не найден"
+        return res
+    db = sqlite3.connect(f"file:{state}?mode=ro", uri=True)
+    db.row_factory = sqlite3.Row
+    sessions = db.execute("""
+        SELECT s.id, s.source, s.chat_type, s.display_name, s.title,
+               s.started_at, s.last_activity_at, COUNT(m.id) n
+        FROM sessions s JOIN messages m ON m.session_id = s.id
+        WHERE m.active = 1 AND m.role IN ('user', 'assistant')
+        GROUP BY s.id
+        HAVING n >= 2
+        ORDER BY COALESCE(s.last_activity_at, s.started_at) DESC
+        LIMIT ?""", (MAX_HISTORY_SESSIONS,)).fetchall()
+    res["sessions_total"] = len(sessions)
+    hdir = os.path.join(out_dir, "history")
+    os.makedirs(hdir, exist_ok=True)
+    idx_lines = ["# История диалогов Гермеса → Simbiot",
+                 "", "сессии с реальным диалогом (active=1, ≥2 сообщения),",
+                 f"капы: {MAX_HISTORY_MSGS_PER_SESSION}/сессия, {MAX_TOTAL_HISTORY_MSGS} всего", ""]
+    total = 0
+    for s in sessions:
+        if total >= MAX_TOTAL_HISTORY_MSGS:
+            idx_lines.append(f"- (усечено: достигнут глобальный кап {MAX_TOTAL_HISTORY_MSGS} сообщений)")
+            break
+        msgs = db.execute("""SELECT role, content, timestamp FROM messages
+                             WHERE session_id=? AND active=1
+                               AND role IN ('user','assistant')
+                             ORDER BY timestamp ASC""",
+                          (s["id"],)).fetchall()
+        if len(msgs) < 2:
+            res["skipped_single_ping"] += 1
+            continue
+        msgs = msgs[-MAX_HISTORY_MSGS_PER_SESSION:]
+        title = (s["title"] or s["display_name"] or s["chat_type"] or s["id"])[:90]
+        when = datetime.fromtimestamp(s["started_at"] or 0, tz=timezone.utc).strftime("%Y-%m-%d")
+        body = [f"# {title}", "",
+                f"session: {s['id']} · source: {s['source'] or '?'} · "
+                f"type: {s['chat_type'] or '-'} · начать: {when} UTC", ""]
+        for m in msgs:
+            ts = datetime.fromtimestamp(m["timestamp"] or 0, tz=timezone.utc).strftime("%Y-%m-%d %H:%M")
+            body.append(f"**[{ts} UTC] {m['role']}:** {redact(m['content'] or '')}")
+            body.append("")
+        fname = f"{s['id']}.md"
+        path = os.path.join(hdir, fname)
+        with open(path, "w", encoding="utf-8") as f:
+            f.write("\n".join(body))
+        res["files"].append(path)
+        res["sessions_written"] += 1
+        res["messages_written"] += len(msgs)
+        total += len(msgs)
+        idx_lines.append(f"- [{title}]({os.path.basename(path)}) — "
+                         f"{s['source'] or '?'}, {len(msgs)} сообщ., "
+                         f"последнее {datetime.fromtimestamp(s['last_activity_at'] or s['started_at'] or 0, tz=timezone.utc).strftime('%Y-%m-%d')}")
+    db.close()
+    with open(os.path.join(out_dir, "history-index.md"), "w", encoding="utf-8") as f:
+        f.write("\n".join(idx_lines) + "\n")
+    return res
+
+
+def extract_config(hermes_dir: str, out_dir: str) -> dict:
+    cfg = os.path.join(hermes_dir, "config.yaml")
+    res = {"file": None, "keys": []}
+    if not os.path.exists(cfg):
+        res["error"] = "config.yaml не найден"
+        return res
+    with open(cfg, encoding="utf-8", errors="replace") as f:
+        raw = f.read()
+    safe = redact(raw)
+    dest = os.path.join(out_dir, "config-hermes.yaml")
+    with open(dest, "w", encoding="utf-8") as f:
+        f.write("# Настройка Гермеса (секреты замаскированы <REDACTED>)\n" + safe)
+    res["file"] = dest
+    res["keys"] = [ln.split(":", 1)[0] for ln in raw.splitlines()
+                   if ln and not ln[0].isspace() and not ln.startswith("#")
+                   and ln.endswith(":")]
+    # маппинг в Simbiot (.env) — подсказки в отчёт
+    mapping = []
+    try:
+        import yaml
+        cfgd = yaml.safe_load(raw) or {}
+        model = ((cfgd.get("model") or {}).get("default")
+                 if isinstance(cfgd.get("model"), dict) else None) or cfgd.get("model")
+        if isinstance(model, str) and model:
+            mapping.append({"hermes": "model", "simbiot": "SIMBIOT_MODEL", "value": model})
+        tz = cfgd.get("timezone")
+        if tz:
+            mapping.append({"hermes": "timezone", "simbiot": "(env TZ хоста)", "value": tz})
+    except Exception as e:
+        mapping.append({"warning": f"yaml не разобран: {str(e)[:150]}"})
+    res["simbiot_env_mapping"] = mapping
+    return res
+
+
+def extract_cron(hermes_dir: str, out_dir: str) -> dict:
+    jobs_path = os.path.join(hermes_dir, "cron", "jobs.json")
+    res = {"jobs_total": 0, "jobs_enabled": 0, "mapped": []}
+    if not os.path.exists(jobs_path):
+        res["error"] = "cron/jobs.json не найден"
+        return res
+    with open(jobs_path) as f:
+        data = json.load(f)
+    jlist = data.get("jobs") if isinstance(data, dict) else data
+    res["jobs_total"] = len(jlist or [])
+    rows = []
+    for j in (jlist or []):
+        if not j.get("enabled"):
+            continue
+        res["jobs_enabled"] += 1
+        sched = j.get("schedule") or {}
+        expr = sched.get("expr") or sched.get("display") or ""
+        rows.append({
+            "id": (j.get("id") or "")[:16], "name": j.get("name"),
+            "cadence": "cron " + expr if expr else sched.get("kind"),
+            "prompt_sha": hashlib.sha256((j.get("prompt") or "").encode()).hexdigest()[:12],
+            "prompt_chars": len(j.get("prompt") or ""),
+            "simbiot_equivalent": "задача на доске по такту / systemd timer — см. docs",
+        })
+    res["mapped"] = rows
+    with open(os.path.join(out_dir, "cron-map.json"), "w", encoding="utf-8") as f:
+        json.dump(res, f, ensure_ascii=False, indent=2)
+    return res
+
+
+def write_identity_files(mem: dict, out_dir: str) -> list:
+    os.makedirs(os.path.join(out_dir, "memory"), exist_ok=True)
+    written = []
+    id_path = os.path.join(out_dir, "identity.md")
+    with open(id_path, "w", encoding="utf-8") as f:
+        f.write((mem.get("identity_md") or "").strip() + "\n")
+    written.append(id_path)
+    if mem.get("MEMORY_raw_md"):
+        p = os.path.join(out_dir, "memory", "MEMORY.md")
+        with open(p, "w", encoding="utf-8") as f:
+            f.write(mem["MEMORY_raw_md"])
+        written.append(p)
+    if mem.get("USER_raw_md"):
+        p = os.path.join(out_dir, "memory", "USER.md")
+        with open(p, "w", encoding="utf-8") as f:
+            f.write(mem["USER_raw_md"])
+        written.append(p)
+    return written
+
+
+def rag_index(files: list, rag_db: str, rag_py: str, rag_script: str) -> dict:
+    """Память/история/личность → целевая RAG-база (вендор дедупит по sha)."""
+    res = {"indexed": [], "failed": []}
+    if not (os.path.exists(rag_py) and os.path.exists(rag_script)):
+        res["error"] = "RAG-вендор не найден (rag-venv/rag.py) — индексация пропущена"
+        return res
+    for path in files:
+        env = dict(os.environ, RAG_DB=rag_db)
+        import subprocess
+        try:
+            proc = subprocess.run([rag_py, rag_script, "add", path],
+                                  capture_output=True, text=True, timeout=120, env=env)
+            if proc.returncode == 0:
+                res["indexed"].append(os.path.basename(path))
+            else:
+                res["failed"].append({"file": os.path.basename(path),
+                                      "err": (proc.stderr or proc.stdout)[-200:]})
+        except Exception as e:
+            res["failed"].append({"file": os.path.basename(path), "err": str(e)[:200]})
+    return res
+
+
+def apply_to_store(harness_dir: str, identity_md: str, counts: dict) -> dict:
+    """PolicyChanged с памятью Гермеса в identity → тики нового агента видят её."""
+    res = {"events": [], "error": None}
+    try:
+        sys.path.insert(0, harness_dir)
+        from harness import es as target_es
+        pol_marker = "policy-hermes-" + hashlib.sha256(identity_md.encode()).hexdigest()[:10]
+        existing = {}
+        try:
+            model = target_es.run_projections()
+            existing = model.get("policy") or {}
+        except Exception:
+            pass
+        if existing.get("policy_hermes_marker") == pol_marker:
+            res["events"].append({"event": "PolicyChanged", "status": "skip-duplicate"})
+            return res
+        identity = (
+            "Миграция Гермеса → Simbiot: у агента есть перенесённая память и история "
+            f"(сессий: {counts.get('sessions_written', 0)}). Ниже — персона и память. "
+            "Факты выше считаются проверенными данными организации.\n\n" + identity_md)
+        target_es.append_event("harness", "policy", "PolicyChanged", {
+            **(existing or {}),
+            "identity": (identity or "")[:12000],
+            "policy_hermes_marker": pol_marker,
+            "hermes_migration": {"applied_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+                                 "history_sessions": counts.get("sessions_written", 0)},
+        })
+        res["events"].append({"event": "PolicyChanged", "status": "applied",
+                              "marker": pol_marker})
+    except Exception as e:
+        res["error"] = str(e)[:300]
+    return res
+
+
+def main():
+    ap = argparse.ArgumentParser(description="Гермес → Simbiot")
+    ap.add_argument("--hermes", default=os.path.expanduser("~/.hermes"))
+    ap.add_argument("--out", default=None, help="каталог пакета (default: OUT/hermes-migration)")
+    ap.add_argument("--apply", action="store_true", help="писать (default: dry-run)")
+    ap.add_argument("--harness-dir", default=None, help="инстанс Simbiot для события PolicyChanged")
+    ap.add_argument("--rag-db", default=None)
+    ap.add_argument("--rag-py", default="/root/rag-venv/bin/python")
+    ap.add_argument("--rag-script", default="/root/rag/rag.py")
+    args = ap.parse_args()
+
+    simbiot_repo = os.path.dirname(os.path.abspath(__file__))
+    out_dir = args.out or os.path.join(simbiot_repo, "hermes-migration")
+    if args.apply:
+        os.makedirs(out_dir, exist_ok=True)
+
+    report = {"hermes_dir": args.hermes, "mode": "apply" if args.apply else "dry-run",
+              "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
+
+    # 0) dry-run: ничего не создаём, только отчёт-превью в stdout
+    if not args.apply:
+        report["sources"] = probe_sources(args.hermes)
+        mem = extract_memory(args.hermes)
+        report["memory_preview"] = {k: len(v) for k, v in mem.items()
+                                    if isinstance(v, str)}
+        report["note"] = "dry-run: запусти с --apply (и --harness-dir <симбиот-инсталл>)"
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+        return 0
+
+    # 1) карта источников (фиксируем ДО работы)
+    report["sources"] = probe_sources(args.hermes)
+    _jlog("sources: " + json.dumps(report["sources"], ensure_ascii=False)[:400])
+
+    # 2) память/личность
+    mem = extract_memory(args.hermes)
+    files = write_identity_files(mem, out_dir)
+    _jlog(f"identity files: {len(files)}")
+
+    # 3) история
+    hist = extract_history(args.hermes, out_dir)
+    report["history"] = {k: v for k, v in hist.items() if k != "files"}
+    _jlog(f"history: sessions={hist.get('sessions_written')}, msgs={hist.get('messages_written')}")
+    files = files + hist.get("files", []) + [os.path.join(out_dir, "history-index.md")]
+
+    # 4) config.yaml
+    report["config"] = extract_config(args.hermes, out_dir)
+    if report["config"].get("file"):
+        files.append(report["config"]["file"])
+
+    # 5) cron
+    report["cron"] = extract_cron(args.hermes, out_dir)
+    files.append(os.path.join(out_dir, "cron-map.json"))
+
+    # 6) RAG (память+история)
+    if args.rag_db:
+        report["rag"] = rag_index([f for f in files if f.endswith(".md")],
+                                  args.rag_db, args.rag_py, args.rag_script)
+        _jlog("rag: indexed=" + str(len(report["rag"].get("indexed", [])))
+              + " failed=" + str(len(report["rag"].get("failed", []))))
+
+    # 7) события нового инстанса
+    counts = {"sessions_written": hist.get("sessions_written", 0)}
+    report["store"] = apply_to_store(args.harness_dir if args.harness_dir
+                                     else simbiot_repo, mem.get("identity_md", ""), counts)
+
+    # 8) отчёт
+    report["written"] = {"files": len(files), "out_dir": out_dir,
+                         "identity_chars": len(mem.get("identity_md") or "")}
+    with open(os.path.join(out_dir, "report.json"), "w", encoding="utf-8") as f:
+        json.dump(report, f, ensure_ascii=False, indent=2)
+    _jlog("report.json записан")
+    print(json.dumps(report, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())