88bb2c0a95
harness/decision.py — клиент /v1/systemone по схеме router42 бенча: механика — быстрый путь, decision-модель — арбитр, недоступна → фолбэк. - порядок батча задач: из 9 старейших первой — самая ценная для AUM (on); - тема инициативной реплики: выбор вместо round-robin (on); - серая зона гейта (только warn/info): P(нужен LLM) < 0.35 → тихий такт (shadow); - cli.py decide: статус бэкендов, вопрос 2–9 вариантов / P(да), --stats. Бэкенды: облако clef:27b (~0.7 с, SIMBIOT_DECISION_API_KEY) → nimble на CPU (10–30 с); предохранитель на бэкенд, кеш детерминированных ответов, DecisionMade (backend/model/ms/applied) — policy trace для GA-сшивки. Дефолт SIMBIOT_DECISION=off — старые тесты герметичны. Тест: tests/test_decision.py 34 PASS (фейковый эндпоинт); полный набор — без регрессий. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
225 lines
10 KiB
Python
225 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""decision-модель (harness/decision.py) — hermetic, фейковый /v1/systemone.
|
||
|
||
Запуск: ./venv/bin/python tests/test_decision.py
|
||
|
||
Проверяет: off — эндпоинт не зовётся и очередь FIFO; on — выбор модели
|
||
встаёт первым в батче, остальные FIFO; shadow — решение пишется, но не
|
||
применяется; шортлист ≤ 9; кеш (повтор не идёт в сеть, стор не спамится);
|
||
предохранитель (мёртвый эндпоинт → None быстро, второй вызов без сети);
|
||
гейт: серая зона p<порога → False, p≥порога → True, гейт по умолчанию shadow;
|
||
тема инициативы — выбор модели, при off — round-robin.
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||
|
||
sys.path.insert(0, "/root/simbiot-harness")
|
||
|
||
_tmp = tempfile.mkdtemp(prefix="simbiot_decision_")
|
||
os.environ["SIMBIOT_DB"] = os.path.join(_tmp, "harness.db")
|
||
for _v in [k for k in os.environ if k.startswith("SIMBIOT_DECISION")] + [
|
||
"SIMBIOT_BATCH_MAX_MIN", "SIMBIOT_BATCH_MAX_HOUR", "SIMBIOT_BATCH_MAX_DAY"]:
|
||
os.environ.pop(_v, None)
|
||
|
||
from harness import decision, es, scheduler, proactive
|
||
from harness.levels import LEVELS
|
||
|
||
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)
|
||
|
||
|
||
# ── фейковый эндпоинт: choice → ключ из FAKE["pick"] (или последний), noul → FAKE["p"]
|
||
FAKE = {"pick": None, "p": 0.2, "calls": [], "delay": 0.0}
|
||
|
||
|
||
class H(BaseHTTPRequestHandler):
|
||
def log_message(self, *a):
|
||
pass
|
||
|
||
def do_POST(self):
|
||
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
||
body["_auth"] = self.headers.get("Authorization")
|
||
body["_path"] = self.path
|
||
FAKE["calls"].append(body)
|
||
time.sleep(FAKE["delay"])
|
||
ans = {}
|
||
for qid, q in body["questions"].items():
|
||
if q["type"] == "choice":
|
||
keys = list(q["criteria"])
|
||
pick = FAKE["pick"] if FAKE["pick"] in keys else keys[-1]
|
||
ans[qid] = {"type": "choice", "choice": pick, "confidence": 0.5,
|
||
"probabilities": {k: (0.9 if k == pick else 0.1 / (len(keys) - 1))
|
||
for k in keys}}
|
||
else:
|
||
ans[qid] = {"type": "noul", "noul": FAKE["p"]}
|
||
raw = json.dumps({"model": body["model"], "answers": ans}).encode()
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.send_header("Content-Length", str(len(raw)))
|
||
self.end_headers()
|
||
self.wfile.write(raw)
|
||
|
||
|
||
srv = HTTPServer(("127.0.0.1", 0), H)
|
||
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
||
os.environ["SIMBIOT_DECISION_URL"] = f"http://127.0.0.1:{srv.server_port}/v1/systemone"
|
||
|
||
|
||
def reset():
|
||
FAKE.update(pick=None, p=0.2, calls=[], delay=0.0)
|
||
decision._CACHE.clear()
|
||
decision._BREAKERS.clear()
|
||
|
||
|
||
def _task(tid, created, title="задача", size="M"):
|
||
return {"id": tid, "title": title, "size": size, "status": "ready",
|
||
"created": created, "details": ""}
|
||
|
||
|
||
def _model(n):
|
||
tasks = {f"t{i}": _task(f"t{i}", f"2026-10-0{i}T00:00:00" if i < 10
|
||
else f"2026-10-{i}T00:00:00", title=f"задача {i}")
|
||
for i in range(1, n + 1)}
|
||
return {"tasks": tasks, "policy": {"aum_target": "AUM ≥ +1%/год"}}
|
||
|
||
|
||
def _decisions():
|
||
return [json.loads(e["data"]) for e in es.get_events(0, 100000)
|
||
if e["event_type"] == "DecisionMade"]
|
||
|
||
|
||
hour = LEVELS["hour"]
|
||
|
||
# 1) off по умолчанию: сеть не трогаем, FIFO
|
||
reset()
|
||
check("mode off по умолчанию", decision.mode() == "off" and decision.mode("tasks") == "off")
|
||
ids = scheduler.select_task_batch(hour, _model(4), prioritize=True)
|
||
check("off: FIFO", ids[:1] == ["t1"], ids)
|
||
check("off: эндпоинт не вызывался", not FAKE["calls"])
|
||
|
||
# 2) on: выбор модели — первым, остальные FIFO
|
||
os.environ["SIMBIOT_DECISION"] = "on"
|
||
reset()
|
||
FAKE["pick"] = "t3"
|
||
ids = scheduler.select_task_batch(hour, _model(4), prioritize=True)
|
||
check("on: выбор модели первым", ids[0] == "t3", ids)
|
||
check("on: остальные FIFO", ids[1:] == ["t1", "t2", "t4"], ids)
|
||
check("on: без prioritize — чистый FIFO",
|
||
scheduler.select_task_batch(hour, _model(4))[0] == "t1")
|
||
d = _decisions()
|
||
check("DecisionMade записан", d and d[-1]["purpose"] == "tasks" and d[-1]["choice"] == "t3"
|
||
and d[-1]["applied"] is True, d[-1:] if d else None)
|
||
|
||
# 3) кеш: та же очередь — без сети и без нового события
|
||
n_calls, n_ev = len(FAKE["calls"]), len(_decisions())
|
||
scheduler.select_task_batch(hour, _model(4), prioritize=True)
|
||
check("кеш: повтор без сети", len(FAKE["calls"]) == n_calls)
|
||
check("кеш: стор не спамится", len(_decisions()) == n_ev)
|
||
|
||
# 4) шортлист ≤ 9 старейших
|
||
reset()
|
||
scheduler.select_task_batch(hour, _model(14), prioritize=True)
|
||
crit = FAKE["calls"][-1]["questions"]["c"]["criteria"]
|
||
check("шортлист ≤ 9", len(crit) == 9, len(crit))
|
||
check("шортлист — старейшие", list(crit) == [f"t{i}" for i in range(1, 10)], list(crit))
|
||
check("keep_alive передан", FAKE["calls"][-1].get("keep_alive"))
|
||
|
||
# 5) shadow: решение записано, но FIFO
|
||
os.environ["SIMBIOT_DECISION"] = "shadow"
|
||
reset()
|
||
FAKE["pick"] = "t2"
|
||
ids = scheduler.select_task_batch(hour, _model(3), prioritize=True)
|
||
check("shadow: FIFO остаётся", ids[0] == "t1", ids)
|
||
check("shadow: решение записано с applied=False",
|
||
_decisions()[-1]["choice"] == "t2" and _decisions()[-1]["applied"] is False)
|
||
|
||
# 6) предохранитель: мёртвый эндпоинт → None, второй вызов без сети
|
||
os.environ["SIMBIOT_DECISION"] = "on"
|
||
reset()
|
||
os.environ["SIMBIOT_DECISION_URL"] = "http://127.0.0.1:9/v1/systemone"
|
||
ids = scheduler.select_task_batch(hour, _model(3), prioritize=True)
|
||
check("мёртвый эндпоинт: FIFO", ids[0] == "t1", ids)
|
||
check("предохранитель взведён", decision._BREAKERS["primary"]["until"] > time.time())
|
||
t0 = time.time()
|
||
r = decision.choose("tasks", {"a": "x", "b": "y"}, "s", "q")
|
||
check("предохранитель: мгновенный None", r is None and time.time() - t0 < 0.5)
|
||
os.environ["SIMBIOT_DECISION_URL"] = f"http://127.0.0.1:{srv.server_port}/v1/systemone"
|
||
|
||
# 6b) облако + локальный запасной: ключ в Authorization, без keep_alive;
|
||
# облако упало → ответ запасного, бэкенд записан в DecisionMade
|
||
reset()
|
||
port = srv.server_port
|
||
os.environ["SIMBIOT_DECISION_API_KEY"] = "k-test"
|
||
os.environ["SIMBIOT_DECISION_URL"] = f"http://127.0.0.1:{port}/cloud/v1/systemone"
|
||
os.environ["SIMBIOT_DECISION_FALLBACK_URL"] = f"http://127.0.0.1:{port}/local/v1/systemone"
|
||
bes = decision.backends()
|
||
check("с ключом: облако clef:27b + запасная nimble",
|
||
[b["model"] for b in bes] == ["clef:27b", "nimble"], bes)
|
||
FAKE["pick"] = "b"
|
||
r = decision.choose("agent", {"a": "x", "b": "y"}, "s-cloud", "q")
|
||
c = FAKE["calls"][-1]
|
||
check("облако: выбор", r and r["choice"] == "b", r)
|
||
check("облако: Bearer-ключ", c["_auth"] == "Bearer k-test", c["_auth"])
|
||
check("облако: без keep_alive", "keep_alive" not in c)
|
||
check("облако: путь основного", c["_path"].startswith("/cloud/"), c["_path"])
|
||
reset()
|
||
os.environ["SIMBIOT_DECISION_URL"] = "http://127.0.0.1:9/v1/systemone"
|
||
FAKE["pick"] = "a"
|
||
r = decision.choose("agent", {"a": "x", "b": "y"}, "s-fallback", "q")
|
||
c = FAKE["calls"][-1] if FAKE["calls"] else {}
|
||
check("облако упало → запасной ответил", r and r["choice"] == "a", r)
|
||
check("запасной: путь локального, без ключа",
|
||
c.get("_path", "").startswith("/local/") and c.get("_auth") is None, c.get("_path"))
|
||
check("запасной: предохранитель облака взведён",
|
||
decision._BREAKERS.get("primary", {}).get("until", 0) > time.time())
|
||
dm = _decisions()[-1]
|
||
check("DecisionMade: backend=fallback, model=nimble, ошибка облака сохранена",
|
||
dm.get("backend") == "fallback" and dm.get("model") == "nimble" and dm.get("error"), dm)
|
||
for _v in ("SIMBIOT_DECISION_API_KEY", "SIMBIOT_DECISION_FALLBACK_URL"):
|
||
os.environ.pop(_v)
|
||
os.environ["SIMBIOT_DECISION_URL"] = f"http://127.0.0.1:{port}/v1/systemone"
|
||
check("без ключа: только локальная (как раньше)",
|
||
[b["name"] for b in decision.backends()] == ["primary"])
|
||
|
||
# 7) гейт: по умолчанию shadow (не влияет), с GATE=on — по порогу
|
||
reset()
|
||
sigs = [("warn", "escalation:e-1", "что-то не так")]
|
||
check("гейт по умолчанию shadow", decision.mode("gate") == "shadow")
|
||
v, p = decision.gate_wants_llm("hour", sigs)
|
||
check("гейт shadow: вердикта нет, p записан", v is None and abs(p - 0.2) < 1e-9, (v, p))
|
||
os.environ["SIMBIOT_DECISION_GATE"] = "on"
|
||
reset()
|
||
FAKE["p"] = 0.1
|
||
check("гейт on: p<порога → False", decision.gate_wants_llm("hour", sigs)[0] is False)
|
||
reset()
|
||
FAKE["p"] = 0.8
|
||
check("гейт on: p≥порога → True", decision.gate_wants_llm("hour", sigs)[0] is True)
|
||
os.environ.pop("SIMBIOT_DECISION_GATE")
|
||
|
||
# 8) тема инициативы: выбор модели; при off — round-robin
|
||
reset()
|
||
FAKE["pick"] = "cyberbank"
|
||
check("тема: выбор модели", proactive.pick_topic({"tasks": {}})["id"] == "cyberbank")
|
||
os.environ["SIMBIOT_DECISION"] = "off"
|
||
check("тема off: round-robin", proactive.pick_topic({"tasks": {}})["id"] == proactive.TOPICS[0]["id"])
|
||
|
||
# 9) глобальный shadow — потолок для точечного on
|
||
os.environ["SIMBIOT_DECISION"] = "shadow"
|
||
os.environ["SIMBIOT_DECISION_TASKS"] = "on"
|
||
check("глобальный shadow сильнее точечного on", decision.mode("tasks") == "shadow")
|
||
os.environ.pop("SIMBIOT_DECISION_TASKS")
|
||
|
||
srv.shutdown()
|
||
print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)")
|
||
sys.exit(1 if fails else 0)
|