feat(inbox): батч-триаж входящих TG — сводный ответ закрывает все сообщения за один тик (t-260930064924)

Минутный тик отвечал по одному сообщению за тик (m-516173222..228 —
7 тиков подряд на один разговор): контекст показывал все, но мандат
говорил «ответь на него», и главное — нечем было закрыть несколько
входящих одним ответом (tg-reply закрывает одно, «хвосты» снова будили
гейт важности).

- tg_inbox.send_batch_reply: одна TG-отправка → ReplySent на КАЖДОЕ
  входящее батча (полный текст у головного, у хвостовых — ссылка);
  гонки (replied/ignored выпадают, пустой батч → ValueError), разные
  чаты делят батч — каждому своя отправка
- cli tg-reply <id1>,<id2>,… -c: батч-режим; несколько id без -c — отказ
  (защита от случайного закрытия); одиночный режим обратно совместим
- build_context: секция инбокса = «БАТЧ из N» с готовой командой
  (все id через запятую) и инструкцией закрыть ВСЁ за этот тик; кап
  10 → 50 (показываем весь батч, а не хвост)
- мандат min: батч-семантика (сводный ответ по смыслу разговора)
- тесты tests/test_inbox_batch.py: 32 проверки (hermetic), все зелёные

README.md

@@ -52,6 +52,8 @@ Strands Harness SDK как исполнительное ядро, модели
 ./venv/bin/python cli.py escalations --open        # открытые эскалации
 ./venv/bin/python cli.py escalation-ack <esc-id> -n "что сделал"   # закрыть
 ./venv/bin/python cli.py tg-reply <msg-id> "текст"   # ответ на входящее 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 <id1>,<id2>,… -c "текст"`
+(одна TG-отправка, `ReplySent` на каждое входящее; полный текст хранится у
+головного, у хвостовых — ссылка) или отдельными реплаями за тот же тик.
+Гейт важности после сводного ответа больше не будит минуту на «хвосты».
+Раньше m-516173222..228 обрабатывались 7 тиками подряд — по одному за тик.
+
 ## Без-LLM команды бота (t-260929210755)
 `/лестница` `/статус` `/задачи` `/помощь` — мгновенные ответы из read model:
 их закрывает **секундный рефлекс** (`harness/tg_commands.py` в `_reflex_tick`),

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,6 +130,14 @@ def main():
         if low in ("/лестница", "/ladder", "/статус", "/status", "лестница", "status"):
             # срез лестницы по запросу из TG — видимостью управляет сам харнес
             args.text = "📊 " + tg_inbox.ladder_snapshot()
+        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":

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 <id> \"текст\" (тон: кратко, по-русски, тёпло; можно "
+                 "входящих в контексте) — обработай его ЦЕЛИКОМ за этот тик: "
-                 "выполнить просьбу инструментами и ответить результатом). Быстрые "
+                 "сгруппируй в один сводный ответ по смыслу всего разговора "
-                 "мелкие фиксы (S-задачи до 5 минут) тоже твои. Не растекайся. "
+                 "(готовая команда — в заголовке секции, tg-reply <id1>,<id2>,… -c "
+                 "\"текст\" закрывает все сразу) или сделай отдельные реплаи, но НИ "
+                 "ОДНО сообщение не должно остаться без ответа на следующий тик. "
+                 "Тон: кратко, по-русски, тёпло; "
+                 "можно выполнить просьбу инструментами и ответить результатом. "
+                 "Быстрые мелкие фиксы (S-задачи до 5 минут) тоже твои. Не растекайся. "
                  "У тебя есть инструмент rag_lookup — семантический поиск по базе "
                  "знаний кооператива (контекст, проекты, порталы, встречи, документы). "
                  "Если вопрос про кооператив/продукты/документы — сначала rag_lookup "

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]}")
+            lines.append(f"{i}. [{m_id}] от {who}{tag}: {m.get('text','')[:300]}")
-        parts.append("ВХОДЯЩИЕ СООБЩЕНИЯ (ответь через: ./venv/bin/python cli.py tg-reply <id> \"текст\"):\n" + "\n".join(lines))
+        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")]

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()

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 + ")")