simbiot-harness/tests/test_kodi_core.py

# -*- coding: utf-8 -*-
"""Ядро Коди (harness/kodi_core.py), голова addressing — hermetic.

Запуск: ./venv/bin/python tests/test_kodi_core.py

Эмбеддинги подменены детерминированными (тема «боту» vs «людям» + шум),
Ollama/Telegram не трогаются. Проверяет: датасет (серая зона, приоритет
меток human > feedback > teacher), обучение и промоут чемпиона с
подписанным событием, отказ от худшего претендента, подмена файла модели
→ ученика нет, откат; теневой прогноз пишется один раз; роутер;
вопрос Alex с лимитом в сутки и кнопка → метка человека (чужой — нет);
неявная обратная связь (встревание → 0, «дёрнул бота» → 1); триаж: в
shadow решает учитель, в on уверенный знакомый ученик решает сам.
"""
import hashlib
import json
import os
import sys
import tempfile

sys.path.insert(0, "/root/simbiot-harness")
_tmp = tempfile.mkdtemp(prefix="simbiot_core_")
os.environ["SIMBIOT_DB"] = os.path.join(_tmp, "harness.db")
os.environ["SIMBIOT_CORE_DIR"] = os.path.join(_tmp, "core")
for _v in ("SIMBIOT_CORE_ADDRESSING", "SIMBIOT_CORE_ASK_PER_DAY", "SIMBIOT_DECISION"):
    os.environ.pop(_v, None)

import numpy as np

from harness import addressing, decision, es, kodi_core, tg_inbox

es.init_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)


A = np.zeros(768, np.float32); A[0] = 1
B = np.zeros(768, np.float32); B[1] = 1


def fake_embed(texts):
    out = []
    for t in texts:
        h = int(hashlib.sha1(t.encode()).hexdigest(), 16)
        rng = np.random.default_rng(h % (2 ** 32))
        base = A if "#bot" in t else B
        v = base + 0.02 * rng.standard_normal(768).astype(np.float32)
        out.append(v / np.linalg.norm(v))
    return out


kodi_core.embed = fake_embed
API = []
tg_inbox._api = lambda method, params, timeout=35: API.append((method, dict(params))) or {"ok": True, "result": True}
tg_inbox._load_env = lambda: None
tg_inbox._default_chat_id = lambda: 7569223131
tg_inbox.mark_seen = lambda *a, **k: True
G = -1004402779881
UID = [1000]


def recv(text, who="Vlad", ts=None):
    UID[0] += 1
    mid = f"m-{UID[0]}"
    es.append_event("inbox", mid, "MessageReceived", {
        "update_id": UID[0], "chat_id": G, "chat_type": "supergroup", "chat_title": "Правление",
        "first_name": who, "text": text, "kind": "text", "message_id": UID[0]})
    return mid


# ── 1. опыт: 60 размеченных учителем сообщений серой зоны + правило + метки ──
hm = recv("#bot спорное, учитель ошибся")
es.append_event("core", f"label:addressing:{hm}:teacher", "CoreLabel",
                {"head": "addressing", "msg": hm, "label": 0.2, "source": "teacher_backfill"})
es.append_event("core", f"label:addressing:{hm}:human", "CoreLabel",
                {"head": "addressing", "msg": hm, "label": 1, "source": "human"})
for i in range(60):
    bot = i % 2 == 0
    mid = recv(f"{'#bot сделай отчёт' if bot else 'обсуждаем между собой'} {i}")
    es.append_event("core", f"label:addressing:{mid}:teacher", "CoreLabel", {
        "head": "addressing", "msg": mid, "label": 0.9 if bot else 0.1, "source": "teacher_backfill"})
rule_mid = recv("Коди, привет")                          # правило — не в датасете
es.append_event("inbox", rule_mid, "MessageTriaged", {"addressed": True, "rule": "name"})
rows = kodi_core.dataset()
ids = {r[0]["id"] for r in rows}
check("датасет: серая зона с метками (61)", len(rows) == 61, len(rows))
check("датасет: решённое правилом не входит", rule_mid not in ids)
hrow = next(r for r in rows if r[0]["id"] == hm)
check("метка человека сильнее учителя (y=1, вес 5)", hrow[2] == 1 and hrow[3] == 5.0 and hrow[4] == "human")

# ── 2. обучение и промоут ──
rep = kodi_core.train(log=lambda *_: None)
check("первое обучение → промоут", rep["promoted"] and rep["holdout"]["acc"] >= 0.9, rep)
st = kodi_core.status()["addressing"]
check("чемпион записан", st["champion"] and st["champion_n"] == 61, st)
prom = [json.loads(e["data"]) for e in es.get_events(0, 10**6) if e["event_type"] == "CoreModelPromoted"]
check("CoreModelPromoted с sha256", prom and len(prom[-1]["sha256"]) == 64)
model = kodi_core.load_champion()
_ctx = [{"first_name": "Vlad", "text": "обсуждаем между собой 1"},
        {"first_name": "Vlad", "text": "#bot сделай отчёт 2"}]
X, em = kodi_core.featurize([({"first_name": "Vlad", "text": "#bot глянь отчёт"}, _ctx),
                             ({"first_name": "Vlad", "text": "обсуждаем между собой"}, _ctx)])
p = kodi_core.predict(model, X)
check("ученик различает «боту» и «людям»", p[0] > 0.7 and p[1] < 0.4, p)
nov = kodi_core.novelty(model, em)
check("знакомое — ниже порога «незнакомо» модели", nov.max() < model["nov_q"][1], (nov, model["nov_q"]))
check("пороги новизны откалиброваны и упорядочены", 0 <= model["nov_q"][0] <= model["nov_q"][1] < 1, model["nov_q"])
rep2 = kodi_core.train(log=lambda *_: None)
check("тот же опыт — чемпион остаётся (не лучше)", rep2["promoted"] is False, rep2)

# подмена файла модели
meta = kodi_core._champion_meta()
path = kodi_core._model_path(meta["file"])
blob = open(path, "rb").read()
open(path, "wb").write(blob[:-1] + bytes([blob[-1] ^ 1]))
kodi_core._forget("addressing")
check("файл модели подменён → ученика нет", kodi_core.load_champion() is None)
open(path, "wb").write(blob)
kodi_core._forget("addressing")

# ── 3. роутер ──
check("роутер: знакомо+уверенно → ученик", kodi_core.route(0.95, 0.1, 0.9) == "student")
check("роутер: незнакомо+учитель сомневается → человек", kodi_core.route(0.6, 0.8, 0.5) == "human")
check("роутер: иначе → учитель", kodi_core.route(0.55, 0.1, 0.9) == "teacher")

# ── 4. триаж в shadow: решает учитель, ученик пишет прогноз ──
CALLS = []
decision.choose = lambda purpose, opts, state, instr: CALLS.append(1) or {
    "choice": "group", "applied": True, "confidence": .5, "probabilities": {"bot": 0.2, "group": 0.8}}
mid = recv("#bot сделай отчёт за неделю")
addressing.triage(es.run_projections())
ib = es.run_projections()["inbox"][mid]
pred = [json.loads(e["data"]) for e in es.get_events(0, 10**6) if e["event_type"] == "CorePrediction"]
check("shadow: решение учителя (не боту, p=0.2)", ib.get("ignored") and CALLS,
      (ib.get("ignored"), ib.get("ignore_reason"), ib.get("addressed"), CALLS))
check("shadow: прогноз ученика записан, расходится с учителем",
      pred and pred[-1]["msg"] == mid and pred[-1]["p_student"] > .7 and pred[-1]["agree"] is False, pred[-1:])
kodi_core.shadow({"update_id": int(mid[2:]), "first_name": "Vlad", "text": "x"}, [], 0.2, True)
check("прогноз на сообщение пишется один раз",
      sum(1 for e in es.get_events(0, 10**6) if e["event_type"] == "CorePrediction"
          and json.loads(e["data"])["msg"] == mid) == 1)

# ── 5. on: уверенный знакомый ученик решает сам, учитель не зовётся ──
os.environ["SIMBIOT_CORE_ADDRESSING"] = "on"
CALLS.clear()
_route = kodi_core.route  # пороги роутера проверены в §3; здесь — связка с триажем
kodi_core.route = lambda p, nov, t, q=None: "student" if p is not None and abs(p - .5) >= .3 else "teacher"
mid2 = recv("#bot собери сводку")
addressing.triage(es.run_projections())
ib2 = es.run_projections()["inbox"][mid2]
check("on: ученик решил (rule=core), учитель не вызван",
      ib2.get("addressed") is True and ib2["triage"]["rule"] == "core" and not CALLS, (ib2.get("triage"), CALLS))
os.environ.pop("SIMBIOT_CORE_ADDRESSING")
kodi_core.route = _route

# ── 6. вопрос Alex и кнопки ──
os.environ["SIMBIOT_CORE_ASK_PER_DAY"] = "2"
for t in ("странное 1", "странное 2", "странное 3"):
    mm = {"update_id": int(recv(t)[2:]), "first_name": "Vlad", "text": t, "chat_title": "Правление"}
    kodi_core.ask_human(mm, [])
asked = [p for m_, p in API if m_ == "sendMessage"]
check("вопрос Alex: лимит в сутки соблюдён (2)", len(asked) == 2, len(asked))
kb = json.loads(asked[0]["reply_markup"])["inline_keyboard"][0]
check("вопрос: кнопки «Да, Коди / Нет»", kb[0]["callback_data"].endswith(":1") and kb[1]["callback_data"].endswith(":0"))
target = kb[1]["callback_data"].split(":")[2]
check("кнопка от чужого — метки нет",
      kodi_core.on_callback({"id": "q1", "data": kb[1]["callback_data"], "from": {"id": 666}},
                            {"7569223131"}) and not [e for e in es.get_events(0, 10**6)
                                                     if e["event_type"] == "CoreLabel" and
                                                     json.loads(e["data"]).get("source") == "human"
                                                     and json.loads(e["data"])["msg"] == target])
kodi_core.on_callback({"id": "q2", "data": kb[1]["callback_data"], "from": {"id": 7569223131},
                       "message": {"message_id": 5, "chat": {"id": 7569223131}}}, {"7569223131"})
hl = [json.loads(e["data"]) for e in es.get_events(0, 10**6) if e["event_type"] == "CoreLabel"
      and json.loads(e["data"]).get("source") == "human" and json.loads(e["data"])["msg"] == target]
check("кнопка Alex → метка человека 0", hl and hl[-1]["label"] == 0, hl)
check("кнопки убраны после ответа", any(m_ == "editMessageReplyMarkup" for m_, _ in API))
check("чужие callback не наши", kodi_core.on_callback({"data": "other:x"}, set()) is False)

# ── 7. неявная обратная связь ──
fp = recv("ну что, как дела у всех")
es.append_event("inbox", fp, "MessageTriaged", {"addressed": True, "rule": "decision", "p": 0.6})
es.append_event("inbox", fp, "ReplySent", {"text": "Всё отлично 🙂"})
recv("Коди, не тебе", who="Alex")
fn = recv("подготовь сводку по встрече", who="Alex")
es.append_event("inbox", fn, "MessageIgnored", {"reason": "not_addressed", "rule": "decision", "p": 0.3})
poke = recv("@juno42bot", who="Alex")
es.append_event("inbox", poke, "MessageTriaged", {"addressed": True, "rule": "mention"})
n = kodi_core.harvest_feedback()
fb = {json.loads(e["data"])["msg"]: json.loads(e["data"]) for e in es.get_events(0, 10**6)
      if e["event_type"] == "CoreFeedback"}
check("встревание + «не тебе» → метка 0", fb.get(fp, {}).get("label") == 0, fb)
check("промолчал + «@juno42bot» тем же автором → метка 1", fb.get(fn, {}).get("label") == 1, fb)
check("обратная связь пишется один раз", kodi_core.harvest_feedback() == 0)

# ── 8. откат ──
for i in range(30):  # новый опыт → новый чемпион, чтобы было куда откатываться
    mid = recv(f"{'#bot' if i % 2 else ''} новое {i}")
    es.append_event("core", f"label:addressing:{mid}:human", "CoreLabel",
                    {"head": "addressing", "msg": mid, "label": 1 if i % 2 else 0, "source": "human"})
first = kodi_core._champion_meta()["file"]
r3 = kodi_core.train(log=lambda *_: None)
if r3["promoted"]:
    rb = kodi_core.rollback()
    check("откат к предыдущему чемпиону", rb["ok"] and kodi_core._champion_meta()["file"] == first, rb)
else:
    check("откат: история пуста — честный отказ", kodi_core.rollback()["ok"] is False)

print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)")
sys.exit(1 if fails else 0)