From 26edbb76c8aa18c73ea6140cbb5fae05e487f127 Mon Sep 17 00:00:00 2001 From: Simbiot Harness Date: Fri, 2 Oct 2026 21:15:48 +0000 Subject: [PATCH] =?UTF-8?q?TG:=20=D0=B2=D1=85=D0=BE=D0=B4=D1=8F=D1=89?= =?UTF-8?q?=D0=B5=D0=B5=20=D0=B2=D0=B7=D1=8F=D1=82=D0=BE=20=D0=B2=20=D1=80?= =?UTF-8?q?=D0=B0=D0=B1=D0=BE=D1=82=D1=83=20=E2=80=94=20emoji-=D1=80=D0=B5?= =?UTF-8?q?=D0=B0=D0=BA=D1=86=D0=B8=D1=8F=20=F0=9F=91=80=20+=20typing=20?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=B2=D1=80=D0=B5=D0=BC=D1=8F=20=D1=82=D0=B8?= =?UTF-8?q?=D0=BA=D0=B0=20(e-261002202857,=20t-261002205233)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cli.py | 13 ++ docs/digests/hour-2026-10-02-2110.md | 21 +++ harness/es.py | 10 +- harness/scheduler.py | 162 ++++++++++++---------- harness/tg_inbox.py | 76 +++++++++++ tests/test_seen_indication.py | 192 +++++++++++++++++++++++++++ tests/test_task_batch.py | 4 + 7 files changed, 410 insertions(+), 68 deletions(-) create mode 100644 docs/digests/hour-2026-10-02-2110.md create mode 100644 tests/test_seen_indication.py diff --git a/cli.py b/cli.py index ec0c17d..7236d98 100644 --- a/cli.py +++ b/cli.py @@ -58,6 +58,10 @@ def main(): help="сколько последних показать") p_tgl.add_argument("--all", action="store_true", help="без ограничения по количеству") + p_tgs = sub.add_parser("tg-seen", help="emoji-реакция 👀: маркер «сообщение взято в работу»") + p_tgs.add_argument("msg_id", help="id входящего (m-NNN), несколько — через запятую") + p_tgs.add_argument("-e", "--emoji", default="👀", help="эмодзи реакции (по умолчанию 👀)") + p_lad = sub.add_parser("ladder", help="срез лестницы: уровни, занятость, задачи, эскалации") p_not = sub.add_parser("tg-notify", help="отправить сообщение в чат харнеса (уведомление)") p_not.add_argument("text") @@ -198,6 +202,15 @@ def main(): else: mid = tg_inbox.send_reply(args.msg_id, args.text) print(f"OK: ответ на {mid} отправлен") + elif args.cmd == "tg-seen": + es.init_db() + from harness import tg_inbox + ids = [x.strip() for x in args.msg_id.split(",") if x.strip()] + done = [] + for mid in ids: + ok = tg_inbox.mark_seen(mid, emoji=args.emoji) + done.append(f"{mid}:{'ok' if ok else 'skip(нет/уже seen/Ошибка TG)'}") + print("OK: " + ", ".join(done) if done else "OK: пусто") elif args.cmd == "tg-transcribe": es.init_db() from harness import voice diff --git a/docs/digests/hour-2026-10-02-2110.md b/docs/digests/hour-2026-10-02-2110.md new file mode 100644 index 0000000..ce37539 --- /dev/null +++ b/docs/digests/hour-2026-10-02-2110.md @@ -0,0 +1,21 @@ +# Дайджест часа 2026-10-02 21:10 (UTC) + +## События часа +- **Реализована индикация взятия сообщений в работу** (t-261002205233-7b65b1, эскалация e-261002202857-73bbaf): + - поллер при приёме каждого входящего TG ставит emoji-реакцию 👀 (`setMessageReaction`) и пишет событие `MessageSeen` — Alex сразу видит, что сообщение принято, без «тишины на весь тик»; + - пока LLM-тик гоняется по неотвеченному инбоксу, фоновый поток держит `typing` в TG-чате («печатает…» виден весь тик); поток гасится в `finally` на любом исходе тика (успех/сбой/исключение), переживать тик не может; + - клипбот-команда `cli.py tg-seen [,…] [-e emoji]` — явный маркер для уровня при батч-обработке; + - идемпотентность/ошибки: флаг `seen` в проекции → повторный маркер SKIP; TG-сбой → `False` без события, следующий маркер ретраит. +- **Фикс hermeticity теста**: `test_task_batch` (секция 11) падал от протухшей `/tmp/simbiot_batch_e2e.db` (старые подписи под чужим ключом валили аудит) — добавлена чистка перед e2e. На чистой БД — два прогона ALL PASS. + +## Статусы +- Задача t-261002205233-7b65b1 [M] — завершена (см. выше). +- Эскалация e-261002202857-73bbaf (info, min→hour) — закрыта реализацией фичи. +- Тесты: 14/14 сидов зелёные, включая новый `tests/test_seen_indication.py` (20 проверок). +- Деплой: коммит + push в origin, рестарт `simbiot-harness-tg` (безопасен — отдельная единица; рестарт основного `simbiot-harness` не требовался: поллер — единственный, кто ставит реакцию при приёме). + +## Из минутных выжимок (контекст, не подняли) +- m-516173494 (Alex: «поставил заново, команда не работает») — минимальный уровень запросил три конкретики (команда / полный вывод / сервер); ответа пока нет — ждём, вслепую не гадаем. + +## Мелочи +- S: у `HourDigest`-файла дайджесты предыдущего часа (1720) остались незакоммиченными — следующие часы по конвенции коммитят свои. \ No newline at end of file diff --git a/harness/es.py b/harness/es.py index 7d007ba..46b85e4 100644 --- a/harness/es.py +++ b/harness/es.py @@ -501,7 +501,7 @@ def _apply_event(model: dict, ev: dict): elif etype == "MessageReceived": model.setdefault("inbox", {})[aid] = { **data, "replied": False, "ignored": False, - "ts": ev["timestamp"], + "seen": False, "ts": ev["timestamp"], } inbox = model["inbox"] if len(inbox) > 100: # кап инбокса: последние 100 входящих @@ -512,6 +512,14 @@ def _apply_event(model: dict, ev: dict): if m: m["replied"] = True m["reply_text"] = data.get("text", "") + elif etype == "MessageSeen": + # индикация взятия в работу (e-261002202857): флаг seen → reaction 👀 + # ставится ровно один раз, повторные маркеры сообщений не найдут + m = model.setdefault("inbox", {}).get(aid) + if m: + m["seen"] = True + if isinstance(data, dict) and data.get("emoji"): + m["seen_emoji"] = data["emoji"] elif etype == "MessageTranscribed": # голосовое → текст: минутный уровень видит его как обычное # входящее (kind=voice, text=транскрипт) и отвечает по смыслу. diff --git a/harness/scheduler.py b/harness/scheduler.py index d76b86a..4ca0bd6 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -9,6 +9,7 @@ """ import json import os +import threading import time import traceback from datetime import datetime, timezone, timedelta @@ -454,76 +455,103 @@ def run_level_tick(level_cfg, seq: int, model: dict): if not claimed: claimed = None prompt = build_context(level_cfg, model, batch) + # Индикация «взял и работаю» (e-261002202857/73bbaf): пока LLM-тик гоняется, + # фоновый поток держит sendChatAction typing в TG-чатах инбокса (обновление + # каждые ~2с) — «печатает…» виден весь тик. Реакция 👀 на конкретные + # входящие — поллером при приёме (MessageSeen) и по требованию cli.py tg-seen. + # Останов потока — в finally в конце функции. + from . import tg_inbox + keepalive = None + keep_stop = threading.Event() + pending = [m for m in _pending_inbox(model) if not m.get("seen")] + if pending: + # typing — только TG-чаты (портал-каналу typing API недоступен) + tg_chat_ids = list({m.get("chat_id") for m in pending + if m.get("chat_id") + and not str(m.get("chat_id")).startswith("portal")}) + if tg_chat_ids: + keepalive = threading.Thread( + target=tg_inbox.typing_keepalive, + args=(tg_chat_ids, keep_stop), + name=f"keepalive-{level_cfg.name}-{seq}", daemon=True) + keepalive.start() + # весь хвост тика — под finally: keepalive гасим на любом исходе + # (успех / worker-сбой / исключение), поток не переживает свой тик. try: - text, usage, trace_id = isolated_runner.run_llm_tick( - level_cfg, prompt, - session_id=f"s-{level_cfg.name}-{seq}", - trace_prefix=f"{level_cfg.name}-{seq}", - ) - except Exception as e: - # e-260930162600-cbb7e2/cbb7e2-семейство: TimeoutExpired от isolated_runner - # пролетал МИМО блока text-is-None ниже — claimed-задачи висели в running - # до сирот-чистки (час+), очередь казалась «не стартующей» (85246b). - # Возвращаем claim в очередь сразу (TaskPromoted — тот же механизм, что - # orphan-requeue: ready + сброс уровня; следующий тик возьмёт заново), - # исключение летит дальше — CycleError/эскалация как раньше. - for tid in (claimed or []): - size = next((t.get("size", "M") for t in batch if t.get("id") == tid), "M") - es.append_event("task", tid, "TaskPromoted", - {"to": size, "from": size, - "reason": f"тик-воркер умер ({str(e)[:120]}) — " - f"claim возвращён в очередь без потери"}) - raise - if text is None: - # воркер упал (таймаут/краш) — НИ одна задача не выполнена с его - # точки зрения, но executor мог частично поработать инструментами. - # По одной TaskFailed на каждую claimed — иначе батч висит в running - # до orphan_requeue (час+). Сироты всё равно подстрахуют. - for tid in (claimed or []): - es.append_event("task", tid, "TaskFailed", { - "error": str(usage.get("error", "llm worker failed"))[:500], - }) - raise RuntimeError(str(usage.get("error", "llm worker failed"))) - # дайджест тика - period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc) - es.append_event("cycle", f"digest:{level_cfg.name}:{seq}", "DigestWritten", { - "level": level_cfg.name, "period": period.isoformat(), - "summary_ref": trace_id, "text": (text or "")[:2000], - }) - # BudgetSpent воркер уже записал сам — инкрементально, после каждого - # модельного вызова (устраняет потерю spend при таймауте воркера). - # Если usage в результате пуст (воркер не смог разобрать историю — - # старый воркер / нетекстовая структура), дублируем итог, чтобы тик - # не терялся в spend вовсе: дедуп по trace_id в read model. - if usage and not usage.get("model_calls"): - es.append_event("harness", "budget", "BudgetSpent", { - "level": level_cfg.name, "trace_id": trace_id, "source": "runner", - **{k: v for k, v in usage.items() if k != "error"}, - }) - # БАТЧ-ЗАВЕРШЕНИЕ (t-260930101325): TaskCompleted по КАЖДОЙ задаче - # пачки (общий результат-дайджест — answer каждой; per-task отчёт - # исполнитель пишет в дайджесте). Доставка — той же командой - # commands.deliver_result (идемпотентна: без флага/уже sent — Skip). - # Ответ урезаем на задачу, чтобы 6 задач не раздули 6×2000. - # GUARD (e-261001074527-f0c1b4): финальный текст тика «бюджет исчерпан» — - # стоп-сообщение воркера, а не результат; ночь 20727 им фейково закрылись - # 5 L-задач. Такой тик работы не завершал — задачи возвращаются в ready. - from .commands import looks_like_budget_stub - if (text or "") and looks_like_budget_stub(text): - return _guard_budget_stub_requeue(claimed, text) - for tid in (claimed or []): - es.append_event("task", tid, "TaskCompleted", { - "result_ref": trace_id, "answer": (text or "")[:2000], - }) try: - from . import commands as _cmd - if _cmd.deliver_result(tid, text or "", owner=f"tick:{level_cfg.name}"): - _log(f"delivery: результат {tid} доставлен в TG") + text, usage, trace_id = isolated_runner.run_llm_tick( + level_cfg, prompt, + session_id=f"s-{level_cfg.name}-{seq}", + trace_prefix=f"{level_cfg.name}-{seq}", + ) except Exception as e: - # доставка — best effort: не валить тик; DeliverySent допишет - # delivery-демон (см. deliver_pending) — задача не потеряется - _log(f"delivery: отправка {tid} не удалась ({e}) — доставит delivery-демон") - return text, claimed or [] + # e-260930162600-cbb7e2/cbb7e2-семейство: TimeoutExpired от isolated_runner + # пролетал МИМО блока text-is-None ниже — claimed-задачи висели в running + # до сирот-чистки (час+), очередь казалась «не стартующей» (85246b). + # Возвращаем claim в очередь сразу (TaskPromoted — тот же механизм, что + # orphan-requeue: ready + сброс уровня; следующий тик возьмёт заново), + # исключение летит дальше — CycleError/эскалация как раньше. + for tid in (claimed or []): + size = next((t.get("size", "M") for t in batch if t.get("id") == tid), "M") + es.append_event("task", tid, "TaskPromoted", + {"to": size, "from": size, + "reason": f"тик-воркер умер ({str(e)[:120]}) — " + f"claim возвращён в очередь без потери"}) + raise + if text is None: + # воркер упал (таймаут/краш) — НИ одна задача не выполнена с его + # точки зрения, но executor мог частично поработать инструментами. + # По одной TaskFailed на каждую claimed — иначе батч висит в running + # до orphan_requeue (час+). Сироты всё равно подстрахуют. + for tid in (claimed or []): + es.append_event("task", tid, "TaskFailed", { + "error": str(usage.get("error", "llm worker failed"))[:500], + }) + raise RuntimeError(str(usage.get("error", "llm worker failed"))) + # дайджест тика + period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc) + es.append_event("cycle", f"digest:{level_cfg.name}:{seq}", "DigestWritten", { + "level": level_cfg.name, "period": period.isoformat(), + "summary_ref": trace_id, "text": (text or "")[:2000], + }) + # BudgetSpent воркер уже записал сам — инкрементально, после каждого + # модельного вызова (устраняет потерю spend при таймауте воркера). + # Если usage в результате пуст (воркер не смог разобрать историю — + # старый воркер / нетекстовая структура), дублируем итог, чтобы тик + # не терялся в spend вовсе: дедуп по trace_id в read model. + if usage and not usage.get("model_calls"): + es.append_event("harness", "budget", "BudgetSpent", { + "level": level_cfg.name, "trace_id": trace_id, "source": "runner", + **{k: v for k, v in usage.items() if k != "error"}, + }) + # БАТЧ-ЗАВЕРШЕНИЕ (t-260930101325): TaskCompleted по КАЖДОЙ задаче + # пачки (общий результат-дайджест — answer каждой; per-task отчёт + # исполнитель пишет в дайджесте). Доставка — той же командой + # commands.deliver_result (идемпотентна: без флага/уже sent — Skip). + # Ответ урезаем на задачу, чтобы 6 задач не раздули 6×2000. + # GUARD (e-261001074527-f0c1b4): финальный текст тика «бюджет исчерпан» — + # стоп-сообщение воркера, а не результат; ночь 20727 им фейково закрылись + # 5 L-задач. Такой тик работы не завершал — задачи возвращаются в ready. + from .commands import looks_like_budget_stub + if (text or "") and looks_like_budget_stub(text): + return _guard_budget_stub_requeue(claimed, text) + for tid in (claimed or []): + es.append_event("task", tid, "TaskCompleted", { + "result_ref": trace_id, "answer": (text or "")[:2000], + }) + try: + from . import commands as _cmd + if _cmd.deliver_result(tid, text or "", owner=f"tick:{level_cfg.name}"): + _log(f"delivery: результат {tid} доставлен в TG") + except Exception as e: + # доставка — best effort: не валить тик; DeliverySent допишет + # delivery-демон (см. deliver_pending) — задача не потеряется + _log(f"delivery: отправка {tid} не удалась ({e}) — доставит delivery-демон") + return text, claimed or [] + finally: + keep_stop.set() # индикатор «работаю» гасим на любом исходе тика + if keepalive is not None and keepalive.is_alive(): + keepalive.join(timeout=3) def process_level(level_cfg, model: dict): diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py index 6c116a7..8cba361 100644 --- a/harness/tg_inbox.py +++ b/harness/tg_inbox.py @@ -66,6 +66,72 @@ def send_message(chat_id, text: str, reply_to=None, timeout: int = 35): raise +def set_reaction(chat_id, message_id: int, emoji: str = "👀", is_big: bool = False): + """setMessageReaction: видимый маркер «агент взял/смотрит» (e-261002202857). + + Идентичность Telegram: повторная постановка ТОГО ЖЕ эмодзи — no-op API; + смена эмодзи на другой видна как изменение реакции. Идемпотентный повтор + безопасен. Реакция допустима на сообщениях не старше 48 часов. + """ + params = {"chat_id": chat_id, "message_id": message_id, + "reaction_type": json.dumps({"type": "emoji", "emoji": emoji}), + "is_big": json.dumps(is_big)} + return _api("setMessageReaction", params, timeout=10) + + +def mark_seen(msg_id: str, emoji: str = "👀", is_big: bool = False): + """Реакция на входящее по id + событие MessageSeen. + + Один флаг seen в проекции → реакция ставится ровно один раз на + сообщение (двумя контурами: поллер при приёме и мин-LLM после + взятия батча; флаг общий, гонки нет — тот, кто первый). + Успех события НЕ зависит от доступности TG на момент постановки: + ставим MessageSeen, только если реакция ушла в TG успешно; иначе + флаг не подаём и следующий маркер повторит попытку. + """ + _load_env() + model = es.run_projections() + m = (model.get("inbox") or {}).get(msg_id) + if m is None: + return False + if m.get("seen"): + return False # уже помечено — идемпотентный SKIP (реакция уже стоит) + chat_id = m.get("chat_id") + tg_mid = m.get("message_id") or (m.get("update_id") or 0) % 100000 + if str(chat_id or "").startswith("portal"): + try: + from . import portal_channel + portal_channel.mark_seen(msg_id) + except Exception: + pass + es.append_event("inbox", msg_id, "MessageSeen", {"emoji": "portal"}) + return True + try: + set_reaction(chat_id, tg_mid, emoji=emoji, is_big=is_big) + except Exception: + return False # TG недоступен: флаг не подаём — повторит следующий маркер + es.append_event("inbox", msg_id, "MessageSeen", {"emoji": emoji}) + return True + + +def typing_keepalive(chat_ids: list, stop, interval: float = 2.0): + """Фоновый индикатор «работаю» на длинных тиках (e-261002202857). + + sendChatAction typing живёт ~5с и сам не обновляется — обновляем каждые + interval секунд, пока caller не остановит (threading.Event в finally + run_level_tick). Живой пульс виден весь тик: «печатает…» == «взял, + готовлю ответ». Индикатор — best effort: сбои TG не валят тик. + """ + _load_env() # токен для _api: в процессе scheduler может быть не загружен + while not stop.wait(max(0.5, min(interval, 5.0))): + for chat_id in set(chat_ids): + try: + _api("sendChatAction", {"chat_id": chat_id, "action": "typing"}, + timeout=10) + except Exception: + pass + + def receive_message(update: dict): """MessageReceived в стор. Идемпотентно по update_id.""" msg = update.get("message") or update.get("edited_message") or {} @@ -99,6 +165,9 @@ def receive_message(update: dict): data = { "update_id": update["update_id"], "chat_id": chat.get("id"), + # tg message_id нужен для setMessageReaction (реакция «seen», e-261002202857): + # update_id ≠ message_id (глобальный стрим vs чат), фолбэк — старый hash + "message_id": msg.get("message_id"), "user_id": user.get("id"), "username": user.get("username"), "first_name": user.get("first_name"), @@ -277,6 +346,13 @@ def poll_loop(): mid = receive_message(u) print(f"[tg] {mid} от {(msg.get('from') or {}).get('username') or user}: " f"{(msg.get('text') or '')[:60]}", flush=True) + # Реакция 👀 — Alex сразу ВИДИТ, что сообщение принято и взято в + # работу, без «тишины на весь тик» (e-261002202857). + # best effort: недоступность TG не мешает приёму/обработке. + try: + mark_seen(mid) + except Exception as e: + print(f"[tg] {mid} реакция 👀 не удалась: {e}", flush=True) # голосовые/аудио/видео-заметки: расшифровка сразу при приёме, # чтобы минутный уровень уже видел текст (не качает файлы сам — # таймаут тика мал). Сбой транскрипции не валит поллер. diff --git a/tests/test_seen_indication.py b/tests/test_seen_indication.py new file mode 100644 index 0000000..ffce9e7 --- /dev/null +++ b/tests/test_seen_indication.py @@ -0,0 +1,192 @@ +# -*- coding: utf-8 -*- +"""t-261002205233-7b65b1: индикация взятия входящих TG в работу, hermetic. + +Запуск: ./venv/bin/python tests/test_seen_indication.py + +Проверяет: +- set_reaction шлёт setMessageReaction с корректным reaction_type (JSON); +- mark_seen: реакция + MessageSeen (проекция seen=True), повтор — skip + (нет ни реакции, ни второго события), TG-сбой — False без события; +- receive_message сохраняет message_id (нативный, не хэш-фолбэк); +- typing_keepalive: держит sendChatAction typing и останавливается по Event; +- run_level_tick: поднимает keepalive по TG-чатам неотвеченного инбокса, + исключает портал-чаты, гасит поток в finally (на любом исходе возврата). +""" +import json +import os +import sys +import threading +import time + +_DST = "/tmp/simbiot-test-seen-indication.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 run_level_tick +from harness.levels import 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) + + +es.init_db() + +# ── 1. set_reaction: setMessageReaction с JSON reaction_type ── +calls = [] +tg_inbox._api = lambda method, params, timeout=35: calls.append((method, params)) or {"ok": True} +tg_inbox.set_reaction(42, 777, emoji="👀") +check("set_reaction: метод setMessageReaction", calls and calls[0][0] == "setMessageReaction") +rt = json.loads(calls[0][1]["reaction_type"]) +check("set_reaction: reaction_type emoji 👀", + rt == {"type": "emoji", "emoji": "👀"} and calls[0][1]["message_id"] == 777, str(calls)) + +# ── 2. receive_message хранит message_id ── +tg_inbox.receive_message({"update_id": 516173900, "message": { + "message_id": 777, "date": 1780000000, "chat": {"id": 42}, + "from": {"id": 1, "username": "alex", "first_name": "Alex"}, "text": "привет"}}) +m = es.run_projections()["inbox"]["m-516173900"] +check("receive_message: message_id в проекции", m.get("message_id") == 777, str(m.get("message_id"))) +check("receive_message: seen по умолчанию False", m.get("seen") is False) + +# ── 3. mark_seen: реакция + MessageSeen, повтор — skip ── +calls.clear() +before = es.count_events() +ok = tg_inbox.mark_seen("m-516173900") +check("mark_seen: True при постановке", ok is True) +check("mark_seen: setMessageReaction отправлен", + any(c[0] == "setMessageReaction" for c in calls), str([c[0] for c in calls])) +check("mark_seen: MessageSeen записан", + es.count_events() - before == 1, str(es.count_events() - before)) +check("mark_seen: проекция seen=True", + es.run_projections()["inbox"]["m-516173900"].get("seen") is True) +check("mark_seen: seen_emoji 👀", + es.run_projections()["inbox"]["m-516173900"].get("seen_emoji") == "👀") +# повтор: флаг seen → ни реакции, ни события (идемпотентность) +calls.clear() +before = es.count_events() +ok2 = tg_inbox.mark_seen("m-516173900") +check("mark_seen: повтор skip (реакции нет)", ok2 is False and calls == [], str(calls)) +check("mark_seen: повтор без нового события", es.count_events() == before) + +# ── 4. TG-сбой: False, БЕЗ MessageSeen (флаг не подаём) ── +tg_inbox.receive_message({"update_id": 516173901, "message": { + "message_id": 778, "date": 1780000000, "chat": {"id": 42}, + "from": {"id": 1, "username": "alex"}, "text": "второе"}}) + +def _boom(method, params, timeout=35): + raise ConnectionError("tg down") +tg_inbox._api = _boom +before = es.count_events() +ok = tg_inbox.mark_seen("m-516173901") +check("mark_seen: TG-сбой → False", ok is False) +check("mark_seen: TG-сбой без MessageSeen (retry позже)", + es.count_events() == before and + es.run_projections()["inbox"]["m-516173901"].get("seen") is not True) + +# ── 5. typing_keepalive: держит typing, останавливается по Event ── +typing_calls, reaction_calls = [], [] +def _api_counter(method, params, timeout=35): + (typing_calls if method == "sendChatAction" else reaction_calls).append(params) + return {"ok": True} +tg_inbox._api = _api_counter +stop = threading.Event() +th = threading.Thread(target=tg_inbox.typing_keepalive, args=([42], stop), + kwargs={"interval": 0.5}, daemon=True) +th.start() +time.sleep(1.6) +stop.set() +th.join(timeout=3) +check("typing_keepalive: sendChatAction держится (>2 за 1.6с)", + len(typing_calls) >= 2, str(len(typing_calls))) +check("typing_keepalive: только typing, реакций нет", + reaction_calls == [] and all(p.get("action") == "typing" for p in typing_calls)) +check("typing_keepalive: останов по Event (join < 1с)", + not th.is_alive(), "поток жив после stop.set()") + +# ── 6. run_level_tick: keepalive по TG-чатам, портал исключён, stop в finally ── +import harness.isolated_runner as _ir_module + +from harness import scheduler +_orig_es = scheduler.es +_orig_tg_typing_backup = tg_inbox.typing_keepalive +ka_seen = {} +def _fake_keepalive(chat_ids, stop, interval=2.0): + ka_seen["ids"] = list(chat_ids) + ka_seen["stop"] = stop +tg_inbox.typing_keepalive = _fake_keepalive + +written = [] +class _FakeES: + def __getattr__(self, name): + def _stub(*a, **k): + return None + return _stub + def append_event(self, aggregate, aid, evtype, data={}): + written.append((evtype, aid)) + return "x" +class _FakeRunner: + @staticmethod + def run_llm_tick(level_cfg, prompt, session_id=None, trace_prefix=None, timeout=None): + return "дайджест", {"model_calls": 1}, "trace-x" + +model = { + "tasks": {"t-x": {"id": "t-x", "title": "задача", "size": "S", "status": "ready", + "created": "2026-10-02T20:00:00", "details": ""}}, + "digests": {}, + "inbox": { + "m-1": {"update_id": 1, "chat_id": 42, "text": "вопрос", "kind": "text", + "replied": False, "ignored": False, "ts": "2026-10-02T20:01:00+00:00"}, + "m-2": {"update_id": 2, "chat_id": "portal:web", "text": "портал", "kind": "text", + "replied": False, "ignored": False, "ts": "2026-10-02T20:02:00+00:00"}, + }, + "escalations": {}, "policy": {}, +} +try: + scheduler.es = _FakeES() + _ir_module.run_llm_tick = staticmethod(_FakeRunner.run_llm_tick) + run_level_tick(LEVELS["hour"], 4242, model) + check("tick: keepalive поднят ровно по TG-чату (портал исключён)", + ka_seen.get("ids") == [42], str(ka_seen)) + check("tick: finally погасил keepalive (stop set)", + ka_seen.get("stop") is not None and ka_seen.get("stop").is_set()) + + # worker умер → finally всё равно гасит поток + ka_seen.clear() + class _DeadRunner: + @staticmethod + def run_llm_tick(*a, **k): + raise RuntimeError("worker dead") + _ir_module.run_llm_tick = staticmethod(_DeadRunner.run_llm_tick) + try: + scheduler.run_level_tick(LEVELS["hour"], 4243, model) + except RuntimeError: + pass + check("tick: worker упал → finally всё равно гасит keepalive", + ka_seen.get("stop") is not None and ka_seen.get("stop").is_set()) + + # все сообщения уже seen → keepalive не поднимается вовсе + ka_seen.clear() + model2 = dict(model) + model2["inbox"] = { + "m-1": {**model["inbox"]["m-1"], "seen": True}, + "m-2": {**model["inbox"]["m-2"], "seen": True}, + } + _ir_module.run_llm_tick = staticmethod(_FakeRunner.run_llm_tick) + run_level_tick(LEVELS["hour"], 4244, model2) + check("tick: всё seen → keepalive не поднимается", ka_seen == {}, str(ka_seen)) +finally: + scheduler.es = _orig_es + tg_inbox.typing_keepalive = _orig_tg_typing_backup + +print() +print("FAILURES:", fails if fails else "нет — все проверки пройдены") +sys.exit(1 if fails else 0) \ No newline at end of file diff --git a/tests/test_task_batch.py b/tests/test_task_batch.py index 3a1bba3..fcfad4e 100644 --- a/tests/test_task_batch.py +++ b/tests/test_task_batch.py @@ -229,6 +229,10 @@ check("11: cli complete существует", r.returncode == 0 and "-a" in r.s # e2e по реальному стору БЕЗ TG (нет чата → deliver_result False, но TaskCompleted есть) os.environ["SIMBIOT_DB"] = "/tmp/simbiot_batch_e2e.db" +# hermeticity: протухшая /tmp-БД (подписи под старым ключом) валила аудит — чистим +for _p in ("/tmp/simbiot_batch_e2e.db", "/tmp/simbiot_batch_e2e.db-wal", "/tmp/simbiot_batch_e2e.db-shm"): + if os.path.exists(_p): + os.unlink(_p) import harness.es as es import importlib importlib.reload(es)