#!/usr/bin/env python3 """Simbiot — инициализация нового изолированного инстанса (t-261004113351). Мультитенантность: память инстанса (ES-база, RAG-индекс, .env, доки) живёт в отдельном SIMBIOT_HOME, репозиторий остаётся чистым кодом. Скрипт: 1) создаёт структуру хома (db/ rag/ env/ inbox/ digests/ audit/ docs/ …) 2) инициализирует ЧИСТУЮ ES-базу (никаких событий старого инстанса) 3) пишет PolicyChanged с идентичностью нового владельца 4) копирует его проектные документы в /docs/ (+ RAG-индексация, если вендор доступен) 5) опционально: селективный импорт знаний из инстанса-источника — последние выжимки (--import-digests) и выбранные доки (--import-docs) 6) печатает, как поднять: демоном (systemd) и разово (CLI) Пример (инстанс Юры на той же машине): venv/bin/python scripts/init_instance.py --home /var/lib/simbiot-yura \ --name Юра --identity "Ты — агент Юры…" \ --docs /tmp/yura-project/*.md --env-from /root/yura.env На машине Юры (после install.sh): ~/simbiot-harness/venv/bin/python scripts/init_instance.py \ --home ~/simbiot-data --name Юра --env-from ~/yura.env --docs ~/my-project """ import argparse import glob import json import os import shutil import subprocess import sys import tempfile _REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) _RAG_VENV_PY = "/root/rag-venv/bin/python" _RAG_SCRIPT = "/root/rag/rag.py" # точечные переопределения окружения вызывающего НЕ должны протекать в новый # инстанс (например, SIMBIOT_DB боя, выставленная воркером часа) _POINT_ENV = ("SIMBIOT_DB", "SIMBIOT_RAG_DB", "SIMBIOT_AUDIT_KEY_DIR", "SIMBIOT_VOICE_DIR") def _jlog(msg: str) -> None: print(json.dumps({"ts": __import__("datetime").datetime.now().isoformat(), "msg": msg}, ensure_ascii=False)) def _rag_index(files: list, rag_db: str) -> dict: """Индексация файлов в RAG-базу инстанса (вендор дедупит по sha).""" res = {"indexed": [], "failed": [], "error": None} if not (os.path.exists(_RAG_VENV_PY) and os.path.exists(_RAG_SCRIPT)): res["error"] = "rag-вендор не найден — документы скопированы, индексация пропущена" return res os.makedirs(os.path.dirname(rag_db), exist_ok=True) env = dict(os.environ, RAG_DB=rag_db) for path in files: try: proc = subprocess.run([_RAG_VENV_PY, _RAG_SCRIPT, "add", path], capture_output=True, text=True, timeout=120, env=env) (res["indexed"] if proc.returncode == 0 else res["failed"]).append( os.path.basename(path)) except Exception as e: res["failed"].append({"file": os.path.basename(path), "err": str(e)[:200]}) return res def _expand_docs(patterns: list) -> list: out = [] for p in patterns or []: for m in (glob.glob(p) if any(c in p for c in "*?[") else [p]): if os.path.isfile(m): out.append(os.path.abspath(m)) return out def _export_digests(src_db: str, per_level: int, out_dir: str) -> list: """Выжимки инстанса-источника → md-файлы (для селективного переноса).""" import sqlite3 made = [] os.makedirs(out_dir, exist_ok=True) src = sqlite3.connect(f"file:{src_db}?mode=ro", uri=True) src.row_factory = sqlite3.Row for lvl in ("min", "hour", "day"): rows = src.execute( "SELECT timestamp, data FROM events WHERE event_type='DigestWritten'" " AND json_extract(data,'$.level')=? AND json_extract(data,'$.stub')" " IS NULL ORDER BY id DESC LIMIT ?", (lvl, per_level)).fetchall() for r in rows: d = json.loads(r["data"]) fn = os.path.join(out_dir, f"digest-{lvl}-{r['timestamp'][:16].replace(':','')}.md") with open(fn, "w", encoding="utf-8") as f: f.write(f"# Выжимка {lvl} {r['timestamp']}\n\n{d.get('text','')}\n") made.append(fn) src.close() return made def main() -> int: ap = argparse.ArgumentParser(description="init изолированного инстанса Simbiot") ap.add_argument("--home", required=True, help="SIMBIOT_HOME нового инстанса") ap.add_argument("--name", default="agent", help="имя владельца (для идентичности)") ap.add_argument("--identity", default=None, help="идентичность (PolicyChanged); по умолчанию — по имени") ap.add_argument("--docs", nargs="*", default=[], help="проектные документы нового владельца (файлы/глоб)") ap.add_argument("--env-from", default=None, help="готовый .env (токены/ключи) → /env/.env (0600)") ap.add_argument("--src-db", default=None, help="ES-база инстанса-источника (для селективного импорта)") ap.add_argument("--import-digests", type=int, default=0, help="сколько последних выжимок на уровень перенести из --src-db") ap.add_argument("--import-docs", nargs="*", default=[], help="файлы-источники знаний (доки/выжимки) → RAG нового инстанса") ap.add_argument("--force", action="store_true", help="работать даже если хом не пуст") args = ap.parse_args() home = os.path.abspath(os.path.expanduser(args.home)) # SIMBIOT_HOME должен быть в env ДО импорта harness.* (paths читает при старте) for k in _POINT_ENV: os.environ.pop(k, None) os.environ["SIMBIOT_HOME"] = home sys.path.insert(0, _REPO) if os.path.isdir(home) and os.listdir(home) and not args.force: print(f"ОТКАЗ: {home} не пуст (запуск повторно?). --force чтобы продолжить.") return 2 os.makedirs(home, exist_ok=True) from harness import paths, es, commands # после выставления SIMBIOT_HOME made = paths.ensure_instance_dirs() if made: _jlog({"home_dirs": len(made), "home": home}) # 1) чистая база + идентичность es.init_db() n_events = es._conn().execute("SELECT COUNT(*) c FROM events").fetchone()["c"] if n_events: print(f"ОТКАЗ: в {paths.db_path()} уже {n_events} событий — это не чистый инстанс.") return 2 identity = args.identity or ( f"Ты — Simbiot-агент {args.name}. Помогаешь {args.name} в его проектах; " "чужих проектов и контекстов у тебя нет — если спрашивают о том, чего " "нет в твоей базе, так и говори.") commands.set_policy(identity=identity) _jlog({"identity_set": True, "db": paths.db_path()}) # 2) .env инстанса if args.env_from: dst = paths.env_path() os.makedirs(os.path.dirname(dst), exist_ok=True) shutil.copyfile(args.env_from, dst) os.chmod(dst, 0o600) _jlog({"env": dst}) # 3) доки владельца → /docs (+RAG) report = {"docs_copied": [], "rag": None, "digests_imported": []} docs = _expand_docs(args.docs) for p in docs: dst = os.path.join(paths.docs_dir(), os.path.basename(p)) shutil.copyfile(p, dst) report["docs_copied"].append(os.path.basename(p)) # 4) селективный импорт знаний to_index = list(docs and [os.path.join(paths.docs_dir(), os.path.basename(p)) for p in docs] or []) if args.import_digests and args.src_db: exported = _export_digests(args.src_db, args.import_digests, os.path.join(home, "digests", "imported")) report["digests_imported"] = [os.path.basename(x) for x in exported] to_index.extend(exported) to_index.extend(_expand_docs(args.import_docs)) if to_index: rag = _rag_index(to_index, paths.rag_db()) if not rag["failed"]: rag.pop("failed") # пустой список — не шумим в отчёте report["rag"] = rag print(json.dumps({"ok": True, "home": home, "db": paths.db_path(), "env": paths.env_path(), "rag_db": paths.rag_db(), "docs_dir": paths.docs_dir(), **report}, ensure_ascii=False, indent=2)) print("\nКак поднять:") print(f" разово: SIMBIOT_HOME={home} {_REPO}/venv/bin/python {_REPO}/cli.py status") print(f" демон: SIMBIOT_HOME={home} {_REPO}/venv/bin/python -m harness.scheduler") print(" systemd: скопируй scripts/simbiot-harness.service, добавь строку") print(f" Environment=SIMBIOT_HOME={home}") return 0 if __name__ == "__main__": sys.exit(main())