simbiot-harness/tests/test_kodi_core_tasks.py

# -*- coding: utf-8 -*-
"""Ядро Коди, голова task_fit: «влезет ли S-задача в мин-тик» — hermetic.

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

Проверяет: метки из истории (1 — закрыта на min с одного захвата; 0 —
повторы/эскалация/другой уровень; None — исход не ясен, M-задачи);
обучение и чемпион головы; shadow — прогноз пишется, батч не меняется;
on — уверенное «не влезет» → S→M до мин-захвата, «влезет» остаётся;
повторно (после мин-попытки) прогноз не делается; статус по головам.
"""
import hashlib
import json
import os
import sys
import tempfile

sys.path.insert(0, "/root/simbiot-harness")
_tmp = tempfile.mkdtemp(prefix="simbiot_core_tasks_")
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_TASK_FIT", "SIMBIOT_DECISION"):
    os.environ.pop(_v, None)

import numpy as np

from harness import es, kodi_core, scheduler

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:
        rng = np.random.default_rng(int(hashlib.sha1(t.encode()).hexdigest(), 16) % (2 ** 32))
        v = (A if "тяжёлая" in t else B) + 0.02 * rng.standard_normal(768).astype(np.float32)
        out.append(v / np.linalg.norm(v))
    return out


kodi_core.embed = fake_embed
N = [0]


def task(title, size="S", claims=(), done=None, promote=None):
    N[0] += 1
    tid = f"t-{N[0]:04d}"
    es.append_event("task", tid, "TaskSubmitted", {"id": tid, "title": title, "size": size,
                                                   "created": f"2026-10-01T00:{N[0] % 60:02d}:00"})
    for lv in claims:
        es.append_event("task", tid, "TaskClaimed", {"level": lv})
        if promote and lv == "min":
            es.append_event("task", tid, "TaskPromoted", {"from": "S", "to": "S", "reason": "guard"})
    if promote == "esc":
        es.append_event("task", tid, "TaskPromoted", {"from": "S", "to": "M", "reason": "эскалация"})
    if done:
        es.append_event("task", tid, "TaskCompleted", {"result": "ok"})
    return tid


# ── метки ──
fit = task("лёгкая: проверь бэкап", claims=["min"], done=True)
loop = task("тяжёлая: найди плашку в портале", claims=["min", "min", "min"], done=True, promote="guard")
esc = task("тяжёлая: миграция", claims=["min", "min", "hour"], done=True, promote="esc")
open_ = task("лёгкая: в работе", claims=["min"])
m_ = task("M-задача", size="M", claims=["hour"], done=True)
H = kodi_core.task_history()
check("метка 1: закрыта на min с одного захвата", kodi_core.task_label(H[fit]) == 1.0)
check("метка 0: три захвата на min", kodi_core.task_label(H[loop]) == 0.0)
check("метка 0: эскалация к часу", kodi_core.task_label(H[esc]) == 0.0)
check("исход не ясен — без метки", kodi_core.task_label(H[open_]) is None)
check("M-задача — не для этой головы", kodi_core.task_label(H[m_]) is None)

# ── обучение ──
for i in range(30):
    task(f"лёгкая: мелочь {i}", claims=["min"], done=True)
    task(f"тяжёлая: исследование {i}", claims=["min", "min"], done=True, promote="guard")
rep = kodi_core.train(log=lambda *_: None, head="task_fit")
check("обучение task_fit → чемпион", rep["promoted"] and rep["holdout"]["acc"] >= .9, rep)
check("модели голов — в разных каталогах",
      os.path.isfile(os.path.join(os.environ["SIMBIOT_CORE_DIR"], "task_fit", "champion.json")))

# ── shadow: прогноз есть, батч не трогаем ──
heavy = task("тяжёлая: новая про портал")
light = task("лёгкая: новая проверка")
model = es.run_projections()
batch = [model["tasks"][heavy], model["tasks"][light]]
out = scheduler._core_task_fit(list(batch))
check("shadow: батч не изменён", [t["id"] for t in out] == [heavy, light])
preds = {json.loads(e["data"])["task"]: json.loads(e["data"]) for e in es.get_events(0, 10**6)
         if e["event_type"] == "CorePrediction"}
check("shadow: прогнозы записаны (тяжёлая < 0.5 < лёгкая)",
      preds[heavy]["p_fit"] < .5 < preds[light]["p_fit"] and not preds[heavy]["act"], preds)

# ── on: уверенное «не влезет» → S→M до клейма ──
os.environ["SIMBIOT_CORE_TASK_FIT"] = "on"
heavy2 = task("тяжёлая: ещё одна")
light2 = task("лёгкая: ещё одна")
model = es.run_projections()
fam = kodi_core.load_champion("task_fit")["nov_q"]
kodi_core.load_champion("task_fit")["nov_q"] = np.asarray([1.0, 1.0])  # всё «знакомо» — проверяем связку
out = scheduler._core_task_fit([model["tasks"][heavy2], model["tasks"][light2]])
t2 = es.run_projections()["tasks"]
check("on: тяжёлая убрана из мин-батча и стала M", [t["id"] for t in out] == [light2]
      and t2[heavy2]["size"] == "M", ([t["id"] for t in out], t2[heavy2]["size"]))
check("on: лёгкая осталась S", t2[light2]["size"] == "S")
es.append_event("task", light2, "TaskClaimed", {"level": "min"})
n_before = sum(1 for e in es.get_events(0, 10**6) if e["event_type"] == "CorePrediction")
scheduler._core_task_fit([es.run_projections()["tasks"][light2]])
check("после мин-попытки прогноз не повторяется",
      sum(1 for e in es.get_events(0, 10**6) if e["event_type"] == "CorePrediction") == n_before)
os.environ.pop("SIMBIOT_CORE_TASK_FIT")

# ── честное сравнение: без свежих данных чемпион остаётся; на свежих, где он
#    систематически ошибается (мир поменялся), — снимается ──
kodi_core._forget("task_fit")
rep_same = kodi_core.train(log=lambda *_: None, head="task_fit")
check("без свежих — сравнение нечестное, чемпион остаётся",
      not rep_same["promoted"] and not rep_same["holdout"]["fair_vs_champion"]
      and kodi_core._champion_meta("task_fit").get("file"), rep_same)
check("чемпион помнит, до какого момента видел данные",
      kodi_core._champion_meta("task_fit").get("trained_upto"))
for i in range(14):  # инверсия: «тяжёлые» теперь влезают, «лёгкие» — нет
    task(f"тяжёлая: теперь простая {i}", claims=["min"], done=True)
    task(f"лёгкая: теперь сложная {i}", claims=["min", "min"], done=True, promote="guard")
kodi_core._forget("task_fit")
rep_flip = kodi_core.train(log=lambda *_: None, head="task_fit")
check("свежие → честное сравнение", rep_flip["holdout"]["fair_vs_champion"], rep_flip["holdout"])
check("чемпион на свежих хуже базы — снят",
      rep_flip["retired"] and kodi_core._champion_meta("task_fit").get("file") is None
      and kodi_core.load_champion("task_fit") is None, rep_flip)
check("снятие — подписанное событие",
      any(e["event_type"] == "CoreModelRetired" for e in es.get_events(0, 10**6)))
check("снятого можно вернуть откатом", kodi_core.rollback("task_fit")["ok"]
      and kodi_core._champion_meta("task_fit").get("file"))

# ── база: ученик на шуме не становится чемпионом, плохой чемпион снимается ──
import random
random.seed(7)
os.environ["SIMBIOT_CORE_DIR"] = os.path.join(_tmp, "core-noise")
kodi_core._forget("task_fit")
for i in range(60):  # метка случайна относительно текста — учить нечего
    if random.random() < .5:
        task(f"лёгкая: шум {i}", claims=["min"], done=True)
    else:
        task(f"лёгкая: шум {i}", claims=["min", "min"], done=True, promote="guard")
rep_noise = kodi_core.train(log=lambda *_: None, head="task_fit")
check("шум: претендент не лучше базы — не промоутится",
      rep_noise["holdout"]["logloss"] >= rep_noise["holdout"]["baseline_logloss"] - 1e-4
      and not rep_noise["promoted"] or rep_noise["beats_baseline"], rep_noise)
os.environ["SIMBIOT_CORE_DIR"] = os.path.join(_tmp, "core")
kodi_core._forget("task_fit")

st = kodi_core.status()
check("статус: обе головы", set(st) == {"addressing", "task_fit"} and st["task_fit"]["champion"], st.keys())

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