TG: входящее взято в работу — emoji-реакция 👀 + typing на время тика (e-261002202857, t-261002205233)

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

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 <m-id>[,<m-id2>…] [-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) остались незакоммиченными — следующие часы по конвенции коммитят свои.

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=транскрипт) и отвечает по смыслу.

harness/scheduler.py

@@ -9,6 +9,7 @@
 """
 import json
 import os
+import threading
 import time
 import traceback
 from datetime import datetime, timezone, timedelta
@@ -454,6 +455,29 @@ 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:
         try:
             text, usage, trace_id = isolated_runner.run_llm_tick(
                 level_cfg, prompt,
@@ -524,6 +548,10 @@ def run_level_tick(level_cfg, seq: int, model: dict):
                 # 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):

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)
                 # голосовые/аудио/видео-заметки: расшифровка сразу при приёме,
                 # чтобы минутный уровень уже видел текст (не качает файлы сам —
                 # таймаут тика мал). Сбой транскрипции не валит поллер.

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)

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)