simbiot-harness/tests/test_instance_isolation.py

"""Изоляция памяти инстанса Simbiot (t-261004113351-2858c1).

Тест постановки: «поднять копию с пустой памятью, спросить про проекты —
не должна вспоминать наши». Проверяем три контура утечки:

  1) ES-память: контекст тика нового инстанса (выжимки/эскалации/доска/
     AUM-цель из policy) — пустые, идентичность — нового владельца;
  2) RAG-знания: индекс нового инстанса — только его документы (вендор
     возвращает top-k на любой запрос, поэтому смотрим содержимое индекса);
     его тема — находится (положительный контроль: поиск живой);
  3) прод не задет: в боевой БД нет событий тестового инстанса.

Запуск: venv/bin/python tests/test_instance_isolation.py
"""
import json
import os
import shutil
import subprocess
import sqlite3
import sys
import tempfile

_REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_PY = os.path.join(_REPO, "venv", "bin", "python")
_PROD_DB = os.path.join(_REPO, "harness", "harness.db")

fails = []


def check(name, cond, extra=""):
    print(("PASS" if cond else "FAIL"), name, ("| " + str(extra)) if not cond else "")
    if not cond:
        fails.append(name)

# ── стенд: новый инстанс Юры с одним его документом ────────────────────────
home = tempfile.mkdtemp(prefix="simbiot-iso-yura-")
docs = tempfile.mkdtemp(prefix="simbiot-iso-docs-")
doc = os.path.join(docs, "greenhouse.md")
with open(doc, "w", encoding="utf-8") as f:
    f.write("# Проект Юры: умная теплица\n\nДатчики влажности, автополив на ESP32, "
            "урожай томатов к июлю.\n")

r = subprocess.run([_PY, os.path.join(_REPO, "scripts", "init_instance.py"),
                    "--home", home, "--name", "Юра", "--docs", doc],
                   capture_output=True, text=True, timeout=180)
out = {}
try:
    lines = r.stdout.splitlines()          # _jlog-строки (однострочный JSON)
    i = next(i for i, ln in enumerate(lines) if ln.strip() == "{")
    j = next(j for j in range(len(lines) - 1, -1, -1) if lines[j].strip() == "}")
    out = json.loads("\n".join(lines[i:j + 1]))   # итоговый pretty-JSON
except Exception:
    pass
check("init: ok", r.returncode == 0 and out.get("ok") is True,
      r.stdout[-300:] + r.stderr[-300:])

# ── 1) ES-память: контекст тика чистого инстанса ───────────────────────────
env = {k: v for k, v in os.environ.items()
       if k not in ("SIMBIOT_DB", "SIMBIOT_RAG_DB", "SIMBIOT_HOME")}
env["SIMBIOT_HOME"] = home
code = """
import sys
sys.path.insert(0, %r)
from harness import es
from harness.levels import LEVELS
from harness.scheduler import build_context
model = es.run_projections()
ctx = build_context(LEVELS["hour"], model, batch=[])
print(ctx)
""" % _REPO
r2 = subprocess.run([_PY, "-c", code], capture_output=True, text=True,
                    env=env, timeout=120)
ctx = r2.stdout
check("context: построен", bool(ctx.strip()), r2.stderr[-200:])
for marker, why in [
    ("ВЫЖИМКИ НИЖНИХ УРОВНЕЙ", "наши выжимки протекли"),
    ("ЭСКАЛАЦИИ (", "наши эскалации протекли"),
    ("ДОСКА ЗАДАЧ", "наши задачи протекли"),
    ("ЦЕЛЬ (AUM)", "наша AUM-цель протекла"),
    ("ИНБOX", "наш инбокс протёк"),
    ("кооператив 42", "наш контекст протёк"),
]:
    check(f"context: без «{marker}»", marker not in ctx, why)
check("context: идентичность Юры", "Юра" in ctx and "ИДЕНТИЧНОСТЬ" in ctx)

# события нового инстанса: только генезис (PolicyChanged), ничего нашего
db = sqlite3.connect(os.path.join(home, "db", "harness.db"))
evs = [t[0] for t in db.execute("SELECT event_type FROM events")]
db.close()
check("db: генезис только", set(evs) <= {"PolicyChanged"}, evs)

# ── 2) RAG-знания нового инстанса ──────────────────────────────────────────
# Вендор возвращает top-k ближайших на ЛЮБОЙ запрос (пустой базы не бывает),
# поэтому «наших знаний нет» проверяем по содержимому индекса: все чанки —
# только из документов владельца. Положительный контроль: его тема — находится.
# RAG-вендор (/root/rag) есть не на каждом хосте (миграция инстанса): без него
# init_instance честно пишет «индексация пропущена» — тест проверяет только
# то, что доступно (ES-изоляция, генезис, прод не задет), RAG — skip.
_RAG_AVAILABLE = os.path.exists("/root/rag-venv/bin/python") and os.path.exists("/root/rag/rag.py")
env2 = dict(env)
if _RAG_AVAILABLE:
    r3 = subprocess.run([_PY, "-c", """
import sys, json
sys.path.insert(0, %r)
from harness import rag_lookup
his = rag_lookup.rag_query("теплица автополив датчики", k=3)
print(json.dumps({"his": [c["source"] for c in his]}))
""" % _REPO], capture_output=True, text=True, env=env2, timeout=120)
    try:
        rag = json.loads(r3.stdout.strip().splitlines()[-1])
    except Exception:
        rag = {"his": []}
    rdb = sqlite3.connect(f"file:{os.path.join(home, 'rag', 'rag.db')}?mode=ro", uri=True)
    paths = [p for (p,) in rdb.execute("SELECT DISTINCT path FROM chunks")]
    rdb.close()
    check("rag: в индексе только документы владельца",
          paths and all(os.path.realpath(p).startswith(home + os.sep) for p in paths),
          paths)
    check("rag: его документ находится",
          any("greenhouse" in s for s in rag["his"]), rag["his"])
    check("rag: база инстанса отдельная",
          os.path.exists(os.path.join(home, "rag", "rag.db"))
          and os.path.realpath(os.path.join(home, "rag", "rag.db"))
          != os.path.realpath("/root/rag/rag.db"))
else:
    print("SKIP: RAG-вендор /root/rag отсутствует на этом хосте — контур RAG не проверялся")

# ── 3) прод не задет: кросс-записей нет ────────────────────────────────────
# Прод живёт и пишет события каждую секунду; кроме того, трейсы ЭТОЙ сессии
# legitimately упоминают «greenhouse» (я писал тест). Настоящая утечка — если
# генезис тестового инстанса (PolicyChanged с его идентичностью) записался
# в боевую базу. Проверяем именно её.
pdb = sqlite3.connect(f"file:{_PROD_DB}?mode=ro", uri=True)
dirty = pdb.execute(
    "SELECT COUNT(*) FROM events WHERE aggregate_id='policy' AND data LIKE ?",
    ("%агент Юра%",)).fetchone()[0]
pdb.close()
check("prod: генезис чужого инстанса не попал", dirty == 0, dirty)

shutil.rmtree(home, ignore_errors=True)
shutil.rmtree(docs, ignore_errors=True)

print(("=" * 8) + (" FAIL: " + str(len(fails)) if fails else " ALL PASS"))
sys.exit(1 if fails else 0)