diff --git a/README.md b/README.md index 16df94f..d4f3f82 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,8 @@ Strands Harness SDK как исполнительное ядро, модели ./venv/bin/python cli.py escalations --open # открытые эскалации ./venv/bin/python cli.py escalation-ack -n "что сделал" # закрыть ./venv/bin/python cli.py tg-reply "текст" # ответ на входящее TG +# батч-режим (t-260930064924): один сводный ответ закрывает несколько входящих +./venv/bin/python cli.py tg-reply m-1,m-2,m-3 -c "текст" # ReplySent на каждое, одна отправка ``` ## Автодоставка артефактов (t-260929205927) @@ -88,6 +90,14 @@ journalctl -u simbiot-harness -f отвечает `cli.py tg-reply` (Telegram-отправка + `ReplySent` закрывает входящее). Белый список чатов в `.env` (`SIMBIOT_TG_ALLOWED`), чужие → `MessageIgnored`. +Батч-триаж (t-260930064924): минутный тик видит ВСЕ неотвеченные сообщения +как один батч (секция инбокса в контексте, кап 50) и обязан закрыть их за +ОДИН тик — сводным ответом `cli.py tg-reply ,,… -c "текст"` +(одна TG-отправка, `ReplySent` на каждое входящее; полный текст хранится у +головного, у хвостовых — ссылка) или отдельными реплаями за тот же тик. +Гейт важности после сводного ответа больше не будит минуту на «хвосты». +Раньше m-516173222..228 обрабатывались 7 тиками подряд — по одному за тик. + ## Без-LLM команды бота (t-260929210755) `/лестница` `/статус` `/задачи` `/помощь` — мгновенные ответы из read model: их закрывает **секундный рефлекс** (`harness/tg_commands.py` в `_reflex_tick`), diff --git a/cli.py b/cli.py index 916bb4e..c62b6e0 100644 --- a/cli.py +++ b/cli.py @@ -34,8 +34,11 @@ def main(): p_pr.add_argument("-r", "--reason", default="manual promote") p_tgr = sub.add_parser("tg-reply", help="ответить на входящее TG-сообщение") - p_tgr.add_argument("msg_id", help="id входящего (m-NNN)") + p_tgr.add_argument("msg_id", help="id входящего (m-NNN), для батча — несколько через запятую") p_tgr.add_argument("text") + p_tgr.add_argument("-c", "--close", action="store_true", + help="батч-режим: один сводный ответ закрывает все указанные входящие " + "(t-260930064924)") p_taw = sub.add_parser("task-await", help="пометить задачу: результат доставить в TG при готовности") p_taw.add_argument("task_id") @@ -127,8 +130,16 @@ def main(): if low in ("/лестница", "/ladder", "/статус", "/status", "лестница", "status"): # срез лестницы по запросу из TG — видимостью управляет сам харнес args.text = "📊 " + tg_inbox.ladder_snapshot() - mid = tg_inbox.send_reply(args.msg_id, args.text) - print(f"OK: ответ на {mid} отправлен") + ids = [x.strip() for x in args.msg_id.split(",") if x.strip()] + if len(ids) > 1 or (args.close and len(ids) == 1): + if not args.close: + print("ERROR: несколько id без -c/--close — укажите флаг для сводного батч-ответа") + sys.exit(1) + mid = tg_inbox.send_batch_reply(ids, args.text) + print(f"OK: сводный батч-ответ отправлен (голова {mid}, закрыто {len(ids)})") + else: + mid = tg_inbox.send_reply(args.msg_id, args.text) + print(f"OK: ответ на {mid} отправлен") elif args.cmd == "tg-transcribe": es.init_db() from harness import voice diff --git a/harness/levels.py b/harness/levels.py index d454255..dcfcfc1 100644 --- a/harness/levels.py +++ b/harness/levels.py @@ -45,10 +45,15 @@ LEVELS = { name="min", label="Минута · пульс", every=MIN, uses_llm=True, model="auto", effort="low", ctx_tokens=4_000, task_classes=("S",), mandate=("Ты — минутный пульс харнеса Simbiot. Триаж входящих TG-сообщений: " - "если во ВХОДЯЩИХ есть сообщение без ответа — ответь на него командой " - "cli.py tg-reply \"текст\" (тон: кратко, по-русски, тёпло; можно " - "выполнить просьбу инструментами и ответить результатом). Быстрые " - "мелкие фиксы (S-задачи до 5 минут) тоже твои. Не растекайся. " + "все неотвеченные сообщения приходят тебе ОДНИМ батчем (секция " + "входящих в контексте) — обработай его ЦЕЛИКОМ за этот тик: " + "сгруппируй в один сводный ответ по смыслу всего разговора " + "(готовая команда — в заголовке секции, tg-reply ,,… -c " + "\"текст\" закрывает все сразу) или сделай отдельные реплаи, но НИ " + "ОДНО сообщение не должно остаться без ответа на следующий тик. " + "Тон: кратко, по-русски, тёпло; " + "можно выполнить просьбу инструментами и ответить результатом. " + "Быстрые мелкие фиксы (S-задачи до 5 минут) тоже твои. Не растекайся. " "У тебя есть инструмент rag_lookup — семантический поиск по базе " "знаний кооператива (контекст, проекты, порталы, встречи, документы). " "Если вопрос про кооператив/продукты/документы — сначала rag_lookup " diff --git a/harness/scheduler.py b/harness/scheduler.py index 4e13214..161e87b 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -212,9 +212,12 @@ def build_context(level_cfg, model: dict, task=None): and not tg_commands.parse(m.get("text") or "") and (m.get("text") or m.get("kind") in ("voice", "audio", "video_note"))] if inbox: + inbox = sorted(inbox, key=lambda x: x.get("ts") or "")[-50:] # кап: не хвост, а ВЕСЬ батч lines = [] - for m in sorted(inbox, key=lambda x: x.get("ts") or "")[-10:]: + ids = [] + for i, m in enumerate(inbox, 1): m_id = "m-" + str(m.get("update_id")) + ids.append(m_id) who = m.get("username") or m.get("first_name") or m.get("user_id") tag = "" if m.get("kind") == "voice": @@ -224,8 +227,20 @@ def build_context(level_cfg, model: dict, task=None): tag = (" [голосовое, расшифровка не удалась — попроси прислать текстом]" if m.get("transcribed") else " [голосовое, ещё не расшифровано]") - lines.append(f"- [{m_id}] от {who}{tag}: {m.get('text','')[:300]}") - parts.append("ВХОДЯЩИЕ СООБЩЕНИЯ (ответь через: ./venv/bin/python cli.py tg-reply \"текст\"):\n" + "\n".join(lines)) + lines.append(f"{i}. [{m_id}] от {who}{tag}: {m.get('text','')[:300]}") + batch_ids = ",".join(ids) + # t-260930064924: батч-триаж — ВСЕ неотвеченные обрабатываются ОДНИМ тиком. + # Раньше мин-уровень отвечал по одному за тик (m-516173222..228 — 7 тиков + # подряд на один разговор). Теперь в контексте лежит ГОТОВАЯ команда + # сводного ответа: -c/--close пишет ReplySent на каждое входящее батча — + # без хвостов, на которые гейт важности будил бы минуту снова. + parts.append("ВХОДЯЩИЕ СООБЩЕНИЯ — БАТЧ из " + str(len(inbox)) + ": это ОДИН разговор, " + "НЕ отвечай по одному за тик и не оставляй на следующие тики. " + "Сгруппируй всё в один осмысленный ответ по смыслу всего разговора " + "(или сделай отдельные реплаи за этот же тик — но ВСЕ сообщения должны " + "получить ответ именно в ЭТОМ тике). " + "Сводный ответ: ./venv/bin/python cli.py tg-reply " + batch_ids + + " -c \"текст\"\n" + "\n".join(lines)) # доска задач tasks = [t for t in model["tasks"].values() if t["status"] in ("ready", "running")] diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py index 5305550..157e672 100644 --- a/harness/tg_inbox.py +++ b/harness/tg_inbox.py @@ -141,6 +141,60 @@ def send_reply(msg_id: str, text: str, timeout: int = 35): return msg_id +def send_batch_reply(msg_ids: list, text: str, timeout: int = 35): + """Сводный ответ на БАТЧ входящих (t-260930064924): одна Telegram-отправка + закрывает несколько сообщений одного разговора. + + Минутный тик собирает ВСЕ неотвеченные сообщения разом (секция инбокса) + и отвечает одним сводным текстом — вместо по одному за тик + (m-516173222..228 обрабатывались 7 тиками подряд). Событие ReplySent + пишем на КАЖДОЕ закрываемое входящее (проекция ставит replied=True) — + сводный ответ не должен оставлять «хвостов», на которые гейт важности + потом будит минуту снова. Полный текст храним у головного (первого по + ts), у остальных — ссылку на него, чтобы не дублировать текст в сторе. + + Сообщения разных чатов не смешиваются: батч делится по chat_id, каждому + чату — своя отправка (сводный текст один, белый список — не только личка). + + Идемпотентность: перед отправкой перечитываем read model; уже отвеченные + (гонка с сек-рефлексом/ручным tg-reply) выпадают из батча. Если выпали + все — ValueError (тихий пропуск, не ошибка тика). + """ + model = es.run_projections() + inbox = model.get("inbox") or {} + pending = [] + for mid in msg_ids: + m = inbox.get(mid) + if m is None: + raise ValueError(f"входящее {mid} не найдено") + if m.get("replied") or m.get("ignored"): + continue # гонка: уже закрыто/чужое — не трогаем + pending.append((mid, m)) + if not pending: + raise ValueError("на все сообщения батча уже отвечено") + _load_env() + head_mid = pending[0][0] + batch_id = f"b-{pending[0][1].get('update_id')}-{len(pending)}" + by_chat = {} + for mid, m in pending: + by_chat.setdefault(m["chat_id"], []).append((mid, m)) + for chat_id, items in by_chat.items(): + try: # индикатор — best effort, как в send_reply + _api("sendChatAction", {"chat_id": chat_id, "action": "typing"}, timeout=10) + except Exception: + pass + r = send_message(chat_id, text, reply_to=None, timeout=timeout) + for mid, m in items: + es.append_event("inbox", mid, "ReplySent", { + "to_chat": chat_id, + "text": (text if mid == head_mid + else f"(сводный ответ батча {batch_id} — см. {head_mid})")[:2000], + "tg_ok": bool(r.get("ok")), + "batch": {"id": batch_id, "size": len(pending), "head": head_mid}, + }) + return head_mid + + def poll_loop(): """Long polling getUpdates → MessageReceived. Отдельный процесс.""" _load_env() diff --git a/tests/test_inbox_batch.py b/tests/test_inbox_batch.py new file mode 100644 index 0000000..b6925e5 --- /dev/null +++ b/tests/test_inbox_batch.py @@ -0,0 +1,230 @@ +# -*- 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 + ")") \ No newline at end of file