# -*- coding: utf-8 -*- """t-260930064924: батч-обработка входящих TG-сообщений (hermetic, без LLM/TG). Запуск: ./venv/bin/python tests/test_inbox_batch.py Минутный тик отвечает на ВСЕ неотвеченные сообщения одним сводным ответом (или несколькими реплаями за тик), а не по одному за тик (m-516173222..228 — 7 тиков подряд на один разговор). Проверяет: батч-секция build_context (все сообщения + готовая команда), send_batch_reply (одна отправка → ReplySent на каждое; гонки; ignored; разные чаты), cli tg-reply -c (батч-режим; отказ без флага; совместимость одиночного режима), гейт после батча, отсутствие дублей ReplySent. """ import io import json import os import re import sys # ── изоляция: своя БД в /tmp (инбокс пишем сами, prod не нужен) ── _DST = "/tmp/simbiot-test-inbox-batch.db" for p in (_DST, _DST + "-wal", _DST + "-shm"): if os.path.exists(p): os.unlink(p) os.environ["SIMBIOT_DB"] = _DST sys.path.insert(0, "/root/simbiot-harness") from harness import es, tg_inbox from harness.scheduler import build_context, collect_signals, LEVELS fails = [] def check(name, cond, extra=""): print(("PASS" if cond else "FAIL"), name, ("| " + extra) if not cond else "") if not cond: fails.append(name) def _recv(update_id, text, chat_id=42, date=1780000000): """MessageReceived в Telegram-формате payload (chat:{id}).""" tg_inbox.receive_message({"update_id": update_id, "message": { "message_id": update_id % 100000, "date": date, "chat": {"id": chat_id}, "from": {"id": 1, "username": "alex", "first_name": "Alex"}, "text": text}}) es.init_db() # ── 1. build_context: батч-секция показывает ВСЕ неотвеченные + готовую команду ── fake = {"inbox": {}, "tasks": {}, "escalations": {}, "policy": {}} for i in range(12): fake["inbox"][f"m-700{i:02d}"] = { "update_id": 70000 + i, "chat_id": 42, "text": f"сообщение {i} из разговора", "kind": "text", "replied": False, "ignored": False, "ts": f"2026-09-30T07:{i:02d}:00+00:00", } ctx = build_context(LEVELS["min"], fake) m = re.search(r"ВХОДЯЩИЕ СООБЩЕНИЯ.*?(?:\n\n|$)", ctx, re.S) sec = m.group(0) if m else "" check("build_context: батч-секция есть", bool(sec)) check("build_context: ВСЕ 12 в батче (кап 50, не 10)", all(f"m-700{i:02d}" in sec for i in range(12)), sec[:200]) check("build_context: счётчик батча", "БАТЧ из 12" in sec, sec[:150]) ids_line = ",".join(f"m-700{i:02d}" for i in range(12)) check("build_context: ГОТОВАЯ команда со ВСЕМИ id", f"cli.py tg-reply {ids_line} -c" in sec, sec[:400]) check("build_context: инструкция «за ЭТОТ тик»", "именно в ЭТОМ тике" in sec) # ── 2. send_batch_reply: одна отправка → ReplySent на каждое входящее ── # сценарий из постановки: 7 сообщений одного разговора об экономии тиков for i, text in enumerate([ "думаешь, стоит экономить тики?", "или не париться?", "просто интересно", "сколько мы уже потратили", "на тики за сегодня", "примерно", "ну и на неделю если считать", ]): _recv(800 + i, text, date=1780000000 + i) sends, actions = [], [] tg_inbox.send_message = lambda chat_id, text, reply_to=None, timeout=35: ( sends.append((chat_id, text)) or {"ok": True}) tg_inbox._api = lambda method, params, timeout=35: ( actions.append((method, params.get("chat_id"))) or {"ok": True}) batch = [f"m-{800+i}" for i in range(7)] before = es.count_events() head = tg_inbox.send_batch_reply(batch, "Сводный ответ: экономить стоит, посчитал — 5 тиков за час.") check("send_batch_reply: возвращает головной id", head == "m-800", head) check("send_batch_reply: ОДНА отправка TG", len(sends) == 1, str(sends)) check("send_batch_reply: чат правильный", sends and sends[0][0] == 42, str(sends)) check("send_batch_reply: 7 событий ReplySent", es.count_events() - before == 7, str(es.count_events() - before)) model = es.run_projections() replied = [model["inbox"][f"m-{800+i}"]["replied"] for i in range(7)] check("send_batch_reply: проекция закрыла ВСЕ 7", all(replied), str(replied)) head_text = model["inbox"]["m-800"]["reply_text"] tail_text = model["inbox"]["m-806"]["reply_text"] check("send_batch_reply: полный текст у головного", head_text.startswith("Сводный ответ"), head_text[:80]) check("send_batch_reply: у хвостовых — ссылка (нет дубля текста)", "сводный ответ батча" in tail_text and "см. m-800" in tail_text, tail_text) # ── 3. гонка и ignored: батч закрывает только живых кандидатов ── _recv(810, "первое из новой пачки") _recv(811, "второе") _recv(812, "третье — доживёт до батча") # m-810 уже отвечено кем-то раньше (гонка): одиночный ReplySent без батч-поля es.append_event("inbox", "m-810", "ReplySent", {"to_chat": 42, "text": "гонка: успел раньше", "tg_ok": True}) # m-811 — спам/чужой: MessageIgnored (проекция ставит ignored=True) es.append_event("inbox", "m-811", "MessageIgnored", {"reason": "spam"}) sends.clear() before = es.count_events() head2 = tg_inbox.send_batch_reply(["m-810", "m-811", "m-812"], "Сводный по живым сообщениям пачки.") check("гонка: закрыт только живой кандидат (1 ReplySent)", es.count_events() - before == 1, str(es.count_events() - before)) check("гонка: одна отправка на батч", len(sends) == 1, str(sends)) model = es.run_projections() check("гонка: живое закрыто", model["inbox"]["m-812"]["replied"] is True) check("гонка: голова батча = живое сообщение", head2 == "m-812", head2) # повтор батча: все уже отвечены → ValueError (тихий пропуск тика) try: tg_inbox.send_batch_reply(["m-812"], "повтор") check("гонка: повтор батча → ValueError", False, "не бросил") except ValueError as e: check("гонка: повтор батча → ValueError", "уже отвечено" in str(e), str(e)) # неизвестный id → ValueError с именем try: tg_inbox.send_batch_reply(["m-999999"], "нет такого") check("гонка: неизвестный id → ValueError", False, "не бросил") except ValueError as e: check("гонка: неизвестный id → ValueError", "m-999999" in str(e), str(e)) # ── 4. разные чаты: батч делится, каждому — своя отправка ── _recv(900, "в личке", chat_id=42) _recv(901, "в другом чате", chat_id=77) sends.clear() before = es.count_events() head3 = tg_inbox.send_batch_reply(["m-900", "m-901"], "Сводный по двум чатам.") check("чаты: 2 отправки (по одной на чат)", len(sends) == 2, str(sends)) check("чаты: 2 события ReplySent", es.count_events() - before == 2, str(es.count_events() - before)) chats = sorted(c for c, _ in sends) check("чаты: каждому свой chat_id", chats == [42, 77], str(sends)) model = es.run_projections() check("чаты: оба входящих закрыты", model["inbox"]["m-900"]["replied"] and model["inbox"]["m-901"]["replied"]) # ── 5. гейт: после сводного ответа минуту не будит ── sigs = collect_signals(LEVELS["min"], model) check("гейт: unreplied_inbox не будит минуту после батча", all(s[1] != "unreplied_inbox" for s in sigs), str(sigs)) # ── 6. cli tg-reply: батч-режим ── import importlib.util _spec = importlib.util.spec_from_file_location("cli", "/root/simbiot-harness/cli.py") cli = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(cli) def _cli(argv): sys.argv = argv buf = io.StringIO() sys.stdout = buf try: cli.main() return 0, buf.getvalue() except SystemExit as e: return e.code, buf.getvalue() finally: sys.stdout = sys.__stdout__ for i in range(3): _recv(950 + i, f"cli-сообщение {i}", date=1780000030 + i) sends.clear() rc, out = _cli(["cli.py", "tg-reply", "m-950,m-951,m-952", "-c", "Ответ из CLI на батч"]) check("cli: батч через запятую + -c → OK", rc == 0 and out.startswith("OK: сводный батч-ответ"), f"rc={rc} | {out}") check("cli: одна отправка, закрыто 3", len(sends) == 1 and "закрыто 3)" in out, f"{sends} | {out}") # без -c и несколько id → отказ (защита от случайного суммарного закрытия) rc, out = _cli(["cli.py", "tg-reply", "m-950,m-951", "текст без флага"]) check("cli: несколько id БЕЗ -c → отказ", rc == 1, f"rc={rc} | {out}") # одиночный id с -c — валидный батч из 1 _recv(960, "одиночное", date=1780000040) sends.clear() rc, out = _cli(["cli.py", "tg-reply", "m-960", "-c", "Одиночный ответ в батч-режиме"]) check("cli: одиночный id + -c → OK", rc == 0 and out.startswith("OK: сводный батч-ответ"), f"rc={rc} | {out}") check("cli: одиночный батч закрыл 1", len(sends) == 1 and "закрыто 1)" in out, out) # старый одиночный режим без флага работает как раньше (совместимость) _recv(970, "обычное", date=1780000050) sends.clear() rc, out = _cli(["cli.py", "tg-reply", "m-970", "Старый одиночный режим"]) check("cli: одиночный режим без флага — совместимость", rc == 0 and out.startswith("OK: ответ на m-970 отправлен"), f"rc={rc} | {out}") # ── 7. команды бота не попадают в батч-секцию ── fake2 = json.loads(json.dumps(fake)) fake2["inbox"]["m-990001"] = { "update_id": 990001, "chat_id": 42, "text": "/лестница", "kind": "text", "replied": False, "ignored": False, "ts": "2026-09-30T08:00:00+00:00"} ctx2 = build_context(LEVELS["min"], fake2) m2 = re.search(r"ВХОДЯЩИЕ СООБЩЕНИЯ.*?(?:\n\n|$)", ctx2, re.S) sec2 = m2.group(0) if m2 else "" check("фильтр: команда бота НЕ в батче", "m-990001" not in sec2 and "990001" not in sec2) # ── 8. стор: ReplySent по одному на входящее, дублей нет; батчи размечены ── db = es._conn() rows = db.execute( "SELECT aggregate_id, COUNT(*) c FROM events " "WHERE aggregate='inbox' AND event_type='ReplySent' " "GROUP BY aggregate_id HAVING c > 1").fetchall() check("стор: нет дублей ReplySent на входящее", rows == [], str([dict(r) for r in rows])) batches = db.execute( "SELECT COUNT(*) c FROM events WHERE aggregate='inbox' AND event_type='ReplySent' " "AND data LIKE '%\"batch\":%'").fetchone()["c"] check("стор: батч-события размечены (batch-поле)", batches >= 12, str(batches)) db.close() print() if fails: print("ПРОВАЛЕНО:", fails) sys.exit(1) print("ВСЕ ТЕСТЫ ПРОЙДЕНЫ (hermetic, БД:", _DST + ")")