simbiot-harness/tests/test_addressing.py

# -*- coding: utf-8 -*-
"""Адресация в группах (harness/addressing.py, аудит 2026-10-17) — hermetic.

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

Проверяет: правила (личка, @бот, reply боту, имя «Коди», reply человеку);
серая зона — через decision-модель (порог), модель недоступна → молчать;
триаж пишет MessageTriaged/MessageIgnored(not_addressed), неадресованное не
попадает в _pending_inbox и не даёт сигнала гейту; личка не трогается;
история для модели — без ответов бота на неадресованное; повторный триаж
не дублирует; receive_message ловит @упоминание и reply на бота.
"""
import os
import sys
import tempfile

sys.path.insert(0, "/root/simbiot-harness")
os.environ["SIMBIOT_DB"] = os.path.join(tempfile.mkdtemp(prefix="simbiot_addr_"), "harness.db")
for _v in ("SIMBIOT_BOT_NAMES", "SIMBIOT_ADDRESSING_MIN_P", "SIMBIOT_DECISION"):
    os.environ.pop(_v, None)

from harness import addressing, decision, es, scheduler, tg_inbox
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)


BOT_ID = 7245454391
tg_inbox._BOT.update(id=BOT_ID, username="juno42bot")
tg_inbox.mark_seen = lambda mid, **k: SEEN.append(mid) or True
SEEN = []
G = -1004402779881


def g(text="", **kw):
    return {"chat_type": "supergroup", "chat_id": G, "text": text, **kw}


rv = lambda m: addressing.rule_verdict(m, BOT_ID)
check("личка → всегда", rv({"chat_type": "private", "text": "привет"}) == (True, "private"))
check("@juno42bot → да", rv(g("глянь", mentions_bot=True))[0] is True)
check("reply на бота → да", rv(g("а это?", reply_to_user_id=BOT_ID, reply_to_is_bot=True))[0] is True)
check("имя «Коди» → да", rv(g("Коди, сделай отчёт"))[1] == "name")
check("«cody» латиницей → да", rv(g("cody pls"))[0] is True)
check("слово-подстрока не ловится («кодировка»)", rv(g("кодировка сломалась"))[0] is None)
check("reply человеку → нет", rv(g("согласен", reply_to_user_id=156156021))
      == (False, "reply_to_human"))
check("просто реплика → серая зона", rv(g("Доброе утро!")) == (None, None))

# ── триаж в сторе ──
CALLS = []


def fake_choose(purpose, options, state, instructions):
    CALLS.append(state)
    if FAKE["down"]:
        return None
    p = FAKE["p"]
    return {"choice": "bot" if p >= .5 else "group", "applied": True, "confidence": 0.5,
            "probabilities": {"bot": p, "group": 1 - p}}


FAKE = {"p": 0.9, "down": False}
decision.choose = fake_choose


def recv(uid, text, chat_type="supergroup", **kw):
    es.append_event("inbox", f"m-{uid}", "MessageReceived", {
        "update_id": uid, "chat_id": G if chat_type != "private" else 7569223131,
        "chat_type": chat_type, "first_name": kw.pop("who", "Vlad"), "text": text,
        "kind": "text", **kw})


recv(1, "Доброе утро!")                                   # серая зона, p низкий
recv(2, "Коди, что по задачам?", who="Alex")              # имя
recv(3, "ты я вижу хочешь подключить", reply_to_user_id=156156021, who="Alex")
recv(4, "привет из лички", chat_type="private", who="Alex")
FAKE["p"] = 0.1
n = addressing.triage(es.run_projections())
model = es.run_projections()
ib = model["inbox"]
check("триаж: 3 групповых вердикта, личка не трогается", n == 3 and "addressed" not in ib["m-4"], n)
check("«Доброе утро» → проигнорировано (модель)", ib["m-1"].get("ignored")
      and ib["m-1"].get("ignore_reason") == "not_addressed")
check("«Коди, …» → адресовано без модели", ib["m-2"].get("addressed") is True
      and ib["m-2"]["triage"]["rule"] == "name")
check("reply человеку → проигнорировано без модели", ib["m-3"].get("ignored"))
check("модель спрошена только по серой зоне", len(CALLS) == 1, len(CALLS))
check("👀 только на адресованное", SEEN == ["m-2"], SEEN)
pend = {m["update_id"] for m in scheduler._pending_inbox(model)}
check("в инбокс LLM: адресованное + личка", pend == {2, 4}, pend)
check("повторный триаж — без вердиктов", addressing.triage(model) == 0)

# порог и недоступная модель
recv(5, "посмотри отчёт пожалуйста")
FAKE["p"] = 0.8
addressing.triage(es.run_projections())
check("серая зона p≥0.5 → адресовано", es.run_projections()["inbox"]["m-5"].get("addressed") is True)
recv(6, "ну что там")
FAKE["down"] = True
addressing.triage(es.run_projections())
m6 = es.run_projections()["inbox"]["m-6"]
check("модель недоступна → молчим", m6.get("ignored") and m6.get("ignore_reason") == "not_addressed")
FAKE["down"] = False

# гейт: только неадресованное в группе → тишина
only = {"inbox": {"m-1": {**ib["m-1"]}}, "tasks": {}}
check("гейт: неадресованное не даёт сигнала",
      not any(s[1] == "unreplied_inbox" for s in scheduler.collect_signals(LEVELS["min"], only)))
untriaged = {"inbox": {"m-9": {"update_id": 9, "chat_type": "supergroup", "text": "эй"}}, "tasks": {}}
check("без вердикта групповое не будит LLM", not scheduler._pending_inbox(untriaged))

# история для модели: ответ бота на НЕадресованное не попадает
es.append_event("inbox", "m-1", "ReplySent", {"text": "Доброе утро 🙂 (встрял)"})
es.append_event("inbox", "m-2", "ReplySent", {"text": "По задачам: 3 в работе"})
recv(7, "а что с дедлайном")
CALLS.clear()
addressing.triage(es.run_projections())
st = CALLS[-1] if CALLS else ""
check("история: ответ на адресованное есть", "По задачам: 3 в работе" in st, st[:300])
check("история: «встревание» вырезано", "встрял" not in st, st[:300])

# приём: упоминание и reply на бота
tg_inbox.save_document = lambda msg: None
mid = tg_inbox.receive_message({"update_id": 100, "message": {
    "message_id": 1, "date": 1, "chat": {"id": G, "type": "supergroup", "title": "Правление"},
    "from": {"id": 1, "first_name": "Vlad"}, "text": "@juno42bot глянь",
    "entities": [{"type": "mention", "offset": 0, "length": 10}]}})
mid2 = tg_inbox.receive_message({"update_id": 101, "message": {
    "message_id": 2, "date": 1, "chat": {"id": G, "type": "supergroup", "title": "Правление"},
    "from": {"id": 1, "first_name": "Vlad"}, "text": "а так?",
    "reply_to_message": {"message_id": 1, "text": "ответ бота",
                         "from": {"id": BOT_ID, "is_bot": True, "first_name": "Cody"}}}})
ib = es.run_projections()["inbox"]
check("приём: @упоминание бота", ib[mid].get("mentions_bot") is True)
check("приём: reply на бота", ib[mid2].get("reply_to_is_bot") is True
      and ib[mid2].get("reply_to_user_id") == BOT_ID)
check("приём: обычный текст без упоминания", tg_inbox._mentions_bot({"text": "привет"}) is False)

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