diff --git a/harness/addressing.py b/harness/addressing.py new file mode 100644 index 0000000..51cc042 --- /dev/null +++ b/harness/addressing.py @@ -0,0 +1,153 @@ +"""Simbiot Harness — адресация в группах: будить ли LLM на сообщение (аудит 2026-10-17). + +Кейс: в «Правлении Коллективного Разума» люди переписываются между собой, +а Коди (@juno42bot) отвечал на каждое сообщение — «Доброе утро!», реплики +Alex для Vlad, ответы Vlad на вопросы Alex. Фильтра не было: любое +сообщение разрешённого человека в группе поднимало минутный LLM-тик. + +Триаж — ДО гейта, без LLM. Личка/портал/агенты/TUI — всегда адресовано. +В группе: + 1) детерминированно ДА: @juno42bot, reply на сообщение бота, имя бота + в тексте (Коди/Cody/… — SIMBIOT_BOT_NAMES); + 2) детерминированно НЕТ: reply на сообщение другого человека; + 3) серая зона — decision-модель (harness/decision.py): choice «кому + адресовано последнее сообщение» — бот / конкретный человек / группа + в целом; P(бот) ≥ SIMBIOT_ADDRESSING_MIN_P (0.5) → ДА. В историю + идут реплики людей и ответы бота ТОЛЬКО на адресованное ему: ответы + «встревания» сбивали модель (замер на истории правления: choice без + них — адресованное 0.93, остальные 14 ≤ 0.22; noul с ними — 4 ошибки). + Модель недоступна/выключена → НЕТ: в группе молчать безопаснее. + +Итог пишется событием: MessageTriaged{addressed: True} (+ 👀) или +MessageIgnored{reason: not_addressed} — тогда ни один уровень его не видит. +""" +import os +import re +import threading + +from . import es + +GROUP_TYPES = ("group", "supergroup") +_LOCK = threading.Lock() + + +def _names_re(): + names = os.environ.get("SIMBIOT_BOT_NAMES", "коди,cody,juno,джуно") + alts = "|".join(re.escape(n.strip()) for n in names.split(",") if n.strip()) + return re.compile(rf"(? float: + try: + return float(os.environ.get("SIMBIOT_ADDRESSING_MIN_P", "0.5")) + except ValueError: + return 0.5 + + +def needs_triage(m: dict) -> bool: + """Групповое входящее без вердикта (и не отвеченное/не проигнорированное).""" + return (m.get("chat_type") in GROUP_TYPES and "addressed" not in m + and not m.get("replied") and not m.get("ignored")) + + +def rule_verdict(m: dict, bot_id=None): + """Детерминированная часть: (True|False, правило) или (None, None) — серая зона.""" + if m.get("chat_type") not in GROUP_TYPES: + return True, "private" + if m.get("mentions_bot"): + return True, "mention" + if m.get("reply_to_is_bot") or (bot_id and m.get("reply_to_user_id") == bot_id): + return True, "reply_to_bot" + if _names_re().search(m.get("text") or ""): + return True, "name" + if m.get("reply_to_user_id") and not m.get("reply_to_is_bot"): + return False, "reply_to_human" + return None, None + + +def _history(m: dict, inbox: dict, limit: int = 8) -> list: + """Реплики того же чата до сообщения: люди + ответы бота на адресованное.""" + same = sorted((x for x in inbox.values() + if x.get("chat_id") == m.get("chat_id") and x.get("ts") + and x.get("ts") < (m.get("ts") or "")), + key=lambda x: x["ts"])[-limit:] + out = [] + for x in same: + out.append(f"{x.get('first_name') or x.get('username') or '?'}: " + f"{(x.get('text') or '[' + (x.get('kind') or '?') + ']')[:300]}") + r = x.get("reply_text") or "" + if x.get("addressed") and r and not r.startswith("(сводный"): + out.append(f"Cody (bot): {r[:200]}") + return out + + +def model_verdict(m: dict, inbox: dict): + """Серая зона: P(адресовано боту) от decision-модели или None.""" + from . import decision + hist = _history(m, inbox) + who = m.get("first_name") or m.get("username") or "?" + people = sorted({x.split(":", 1)[0] for x in hist if not x.startswith("Cody (bot)")} - {who}) + crit = {"bot": "Cody, the AI assistant bot: the message asks Cody to answer, " + "explain or do something"} + for nm in people[:6]: + crit["p_" + nm] = f"{nm} (a human in the chat)" + crit["group"] = ("the group in general / nobody specific (greetings, remarks, " + "sharing links, thinking aloud)") + rt = (f" [in reply to: {(m.get('reply_to_text') or m.get('reply_to_desc') or '')[:150]}]" + if m.get("reply_to_message_id") else "") + state = (f"Group chat '{m.get('chat_title') or ''}': humans talk to each other; Cody " + "(@juno42bot) is an AI assistant bot that must speak ONLY when addressed.\n" + "Recent messages:\n" + "\n".join(hist) + + f"\nLAST message — {who}{rt}: " + f"{(m.get('text') or '[' + (m.get('kind') or '?') + ']')[:600]}") + r = decision.choose("addressing", crit, state, "Who is the LAST message addressed to?") + if not r: + return None + # в shadow решение пишется, но вердикт всё равно нужен — порог по P(бот) + return float(r["probabilities"].get("bot", 0.0)) + + +def triage(model: dict, log=None) -> int: + """Вердикт по всем групповым входящим без него. Возвращает число вердиктов.""" + inbox = (model or {}).get("inbox") or {} + todo = [m for m in inbox.values() if needs_triage(m)] + if not todo: + return 0 + with _LOCK: + # перечитать: параллельный уровень мог уже вынести вердикт + inbox = (es.run_projections().get("inbox") or {}) + todo = [m for m in inbox.values() if needs_triage(m)] + bot_id = None + try: + from . import tg_inbox + bot_id = tg_inbox.bot_identity().get("id") + except Exception: + pass + n = 0 + for m in sorted(todo, key=lambda x: x.get("ts") or ""): + mid = f"m-{m.get('update_id')}" + ok, rule = rule_verdict(m, bot_id) + p = None + if ok is None: + p = model_verdict(m, inbox) + ok, rule = ((p >= _min_p(), "decision") if p is not None + else (False, "no_decision")) + if ok: + es.append_event("inbox", mid, "MessageTriaged", + {"addressed": True, "rule": rule, "p": p}) + m["addressed"] = True # для истории следующих в этом же проходе + try: + from . import tg_inbox + tg_inbox.mark_seen(mid) + except Exception: + pass + else: + es.append_event("inbox", mid, "MessageIgnored", + {"reason": "not_addressed", "rule": rule, "p": p, + "chat_id": m.get("chat_id"), "user_id": m.get("user_id")}) + m["addressed"] = False + if log: + log(f"addressing: {mid} {'→ боту' if ok else '— не боту'} ({rule}" + + (f", p={p:.2f}" if p is not None else "") + ")") + n += 1 + return n diff --git a/harness/es.py b/harness/es.py index 8fdbbca..86f1355 100644 --- a/harness/es.py +++ b/harness/es.py @@ -691,12 +691,20 @@ def _apply_event(model: dict, ev: dict): "engine": data.get("engine"), "replied": False, "ignored": False, "ts": ev["timestamp"], } + elif etype == "MessageTriaged": + # адресация в группе (harness/addressing.py): addressed=True — будить LLM + m = model.setdefault("inbox", {}).get(aid) + if m: + m["addressed"] = bool(data.get("addressed")) + m["triage"] = {"rule": data.get("rule"), "p": data.get("p")} elif etype == "MessageIgnored": # канал портала (t-261001123121): доставка реплая помечена ReplySent'ом # (portal ReplySent — признак push'а) — отдельный маркер не нужен m = model.setdefault("inbox", {}).get(aid) if m: m["ignored"] = True + if data.get("reason"): + m["ignore_reason"] = data.get("reason") else: model["inbox"][aid] = {"ignored": True, "reason": data.get("reason"), "ts": ev["timestamp"]} elif etype == "CycleError": diff --git a/harness/scheduler.py b/harness/scheduler.py index 6d132d1..0c76cfd 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -252,6 +252,10 @@ def _is_pending(m: dict) -> bool: from . import tg_commands if m.get("replied") or m.get("ignored") or tg_commands.parse(m.get("text") or ""): return False + # группа: только адресованное боту (harness/addressing.py, аудит 2026-10-17); + # без вердикта триажа — ещё не входящее для LLM + if m.get("chat_type") in ("group", "supergroup") and not m.get("addressed"): + return False if m.get("text"): return True if m.get("kind") in ("voice", "audio"): @@ -917,6 +921,13 @@ def process_level(level_cfg, model: dict, seq: int = None): # только если есть событие ≥ порога уровня; тихий такт отрабатывает # детерминированной механикой (как sec) — без расхода кредитов роутера. if level_cfg.uses_llm: + # адресация в группах — ДО гейта: чужой разговор не будит LLM + try: + from . import addressing + if addressing.triage(model, log=_log): + model = es.run_projections() + except Exception as e: + _log(f"addressing: триаж не удался ({e}) — групповые ждут следующего такта") try: load = os.getloadavg()[0] except Exception: diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py index 915a1ab..0ddb3be 100644 --- a/harness/tg_inbox.py +++ b/harness/tg_inbox.py @@ -258,6 +258,36 @@ def _resolve_reply_target(chat_id, reply_to_message_id: int): return None +_BOT = {} + + +def bot_identity() -> dict: + """getMe один раз на процесс: {"id", "username"} (пусто, если TG недоступен).""" + if not _BOT: + try: + _load_env() + r = (_api("getMe", {}, timeout=15) or {}).get("result") or {} + _BOT.update(id=r.get("id"), username=(r.get("username") or "").lower()) + except Exception: + return {} + return _BOT + + +def _mentions_bot(msg: dict) -> bool: + """@username бота в тексте/подписи или text_mention на бота.""" + me = bot_identity() + text = msg.get("text") or msg.get("caption") or "" + for ent in (msg.get("entities") or []) + (msg.get("caption_entities") or []): + if ent.get("type") == "mention" and me.get("username"): + frag = text[ent.get("offset", 0): ent.get("offset", 0) + ent.get("length", 0)] + if frag.lstrip("@").lower() == me["username"]: + return True + if ent.get("type") == "text_mention" and me.get("id") \ + and (ent.get("user") or {}).get("id") == me["id"]: + return True + return bool(me.get("username")) and ("@" + me["username"]) in text.lower() + + def receive_message(update: dict): """MessageReceived в стор. Идемпотентно по update_id.""" msg = update.get("message") or update.get("edited_message") or {} @@ -358,6 +388,15 @@ def receive_message(update: dict): "date": data.get("date"), } data["received_at"] = data.get("date") + # адресация в группах (аудит 2026-10-17, «Коди встревает в разговор»): + # кому reply (боту или человеку) и упомянут ли бот — сырьё для + # harness/addressing.py, решающего, будить ли LLM вообще + rfrom = (msg.get("reply_to_message") or {}).get("from") or {} + if rfrom: + data["reply_to_user_id"] = rfrom.get("id") + data["reply_to_is_bot"] = bool(rfrom.get("is_bot")) + data["reply_to_first_name"] = rfrom.get("first_name") + data["mentions_bot"] = _mentions_bot(msg) es.append_event("inbox", f"m-{update['update_id']}", "MessageReceived", data) return f"m-{update['update_id']}" @@ -629,10 +668,13 @@ def poll_loop(): # Реакция 👀 — Alex сразу ВИДИТ, что сообщение принято и взято в # работу, без «тишины на весь тик» (e-261002202857). # best effort: недоступность TG не мешает приёму/обработке. - try: - mark_seen(mid) - except Exception as e: - print(f"[tg] {mid} реакция 👀 не удалась: {e}", flush=True) + # В группах 👀 ставит триаж адресации (harness/addressing.py) — + # и только на адресованное боту: не метим чужие разговоры. + if (msg.get("chat") or {}).get("type") not in ("group", "supergroup"): + try: + mark_seen(mid) + except Exception as e: + print(f"[tg] {mid} реакция 👀 не удалась: {e}", flush=True) # голосовые/аудио/видео-заметки: расшифровка сразу при приёме, # чтобы минутный уровень уже видел текст (не качает файлы сам — # таймаут тика мал). Сбой транскрипции не валит поллер. diff --git a/tests/test_addressing.py b/tests/test_addressing.py new file mode 100644 index 0000000..d00ddd6 --- /dev/null +++ b/tests/test_addressing.py @@ -0,0 +1,148 @@ +# -*- 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)