аудит @juno42bot 2026-10-16: медиа без подписи в триаже, мин-тик по входящему, тишина сек-уровня, изоляция тестов

- фото/документ/видео без подписи выпадали из триажа (_pending_inbox и
  build_context пускали только текст/голос): висели до 23 ч, все 4
  неотвеченных за неделю — такие. Один фильтр _is_pending, медиа с тегом
  и путём файла; окно свежести SIMBIOT_MEDIA_PENDING_H=6 — старые не
  будят LLM задним числом; стикер без текста — не повод.
- MinNudge: мин-тик через 3 с тишины после MessageReceived/Transcribed, не
  ждём границы минуты (было: медиана 36 с до старта LLM, p90 93 с при
  24 с самой модели). Слот следующего такта, граница его пропускает;
  SIMBIOT_MIN_NUDGE=0 — выключить.
- сек-уровень: CycleTicked+ReflexObserved каждую секунду = 1.35 млн
  подписанных событий/нед (98% стора 1.37 ГБ). Теперь heartbeat раз в
  60 с + такты с событием (алерт, команды, каналы); «reflex ok» не
  спамит журнал.
- 6 тестов писали в боевой стор под SIMBIOT_HOME тика (TTFTWatchdogRetry
  с бюджетом 0.5 с в проде — это они): изоляция SIMBIOT_DB.
Тест: tests/test_tick_latency_noise.py 20 PASS; полный набор — без регрессий.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

harness/scheduler.py

@@ -226,13 +226,42 @@ def _gate_threshold(level_cfg) -> str:
     return _GATE_DEFAULTS.get(level_cfg.name, "off")
 
 
+# медиа с содержимым ждут ответа и без подписи: фото/документ/видео без
+# текста раньше выпадали из триажа — висели до 23 ч или без ответа вовсе
+# (аудит 2026-10-16: все 4 неотвеченных за неделю — фото/документы).
+# Стикер без текста — не повод будить LLM. video_note при приёме → "video".
+_MEDIA_KINDS = ("voice", "audio", "video", "video_note", "photo", "document", "animation")
+
+
+def _media_fresh(m: dict) -> bool:
+    """Медиа без подписи ждёт ответа SIMBIOT_MEDIA_PENDING_H часов (6): старые
+    фото/документы, выпавшие из триажа до фикса, не будят LLM задним числом."""
+    try:
+        hours = float(os.environ.get("SIMBIOT_MEDIA_PENDING_H", "6"))
+        ts = datetime.fromisoformat(m.get("ts") or "")
+    except ValueError:
+        return False
+    if ts.tzinfo is None:
+        ts = ts.replace(tzinfo=timezone.utc)
+    return _now() - ts <= timedelta(hours=hours)
+
+
+def _is_pending(m: dict) -> bool:
+    """Входящее ждёт ответа LLM: не отвечено/не проигнорировано, не команда,
+    есть текст или свежее медиа с содержимым."""
+    from . import tg_commands
+    if m.get("replied") or m.get("ignored") or tg_commands.parse(m.get("text") or ""):
+        return False
+    if m.get("text"):
+        return True
+    if m.get("kind") in ("voice", "audio"):
+        return True  # как раньше: голосовое без расшифровки — «не расслышал»
+    return m.get("kind") in _MEDIA_KINDS and _media_fresh(m)
+
+
 def _pending_inbox(model: dict):
     """Входящие TG без ответа (не команды) — материал для минутного LLM."""
-    from . import tg_commands
+    return [m for m in (model.get("inbox") or {}).values() if _is_pending(m)]
-    return [m for m in (model.get("inbox") or {}).values()
-            if not m.get("replied") and not m.get("ignored")
-            and not tg_commands.parse(m.get("text") or "")
-            and (m.get("text") or m.get("kind") in ("voice", "audio", "video_note"))]
 
 
 def collect_signals(level_cfg, model: dict, load: float = None):
@@ -354,11 +383,7 @@ def build_context(level_cfg, model: dict, batch=None):
     # сек-рефлекс (t-260929210755), мин-LLM на них не тратится. Фильтруем
     # даже без ответа — иначе гонка на границе тика (команда пришла,
     # сек ещё не отвечал) или лежащий TG заставляли минуту отвечать LLM.
-    from . import tg_commands
+    inbox = _pending_inbox(model)
-    inbox = [m for m in (model.get("inbox") or {}).values()
-             if not m.get("replied") and not m.get("ignored")
-             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 = []
@@ -375,6 +400,10 @@ def build_context(level_cfg, model: dict, batch=None):
                     tag = (" [голосовое, расшифровка не удалась — попроси прислать текстом]"
                            if m.get("transcribed") else
                            " [голосовое, ещё не расшифровано]")
+            elif m.get("kind") in _MEDIA_KINDS and m.get("kind") != "voice":
+                fp = m.get("file_path") or m.get("file_name")
+                tag = (f" [{m.get('kind')}" + (f", файл: {fp}" if fp else ", файл не сохранён")
+                       + ("" if (m.get("text") or "").strip() else ", без подписи") + "]")
             lines.append(f"{i}. [{m_id}] от {who}{tag}"
                          # t-261002085926: Alex ответил (reply) на конкретное
                          # сообщение — покажи якорь, иначе триаж не понимает,
@@ -844,18 +873,24 @@ def _maybe_escalate_repo_drift(level_cfg, today: str, behind: int, ahead):
     })
 
 
-def process_level(level_cfg, model: dict):
+def process_level(level_cfg, model: dict, seq: int = None):
-    """Тик одного уровня: занят? → absorb; иначе работа."""
+    """Тик одного уровня: занят? → absorb; иначе работа.
+
+    seq — номер такта от демона (внеочередной мин-тик по входящему берёт
+    слот следующего такта); None — текущий такт по часам.
+    """
     st = _runners.setdefault(level_cfg.name, {"busy": False, "task": None})
-    if st["busy"]:
+    if seq is None:
         seq = tick_number(level_cfg, int(time.time()))
+    if st["busy"]:
         deterministic_digest_stub(level_cfg, seq, "busy")
         es.append_event("cycle", f"absorbed:{level_cfg.name}:{seq}", "TickAbsorbed",
                         {"level": level_cfg.name, "reason": "busy"})
         return "absorbed"
 
-    seq = tick_number(level_cfg, int(time.time()))
+    if level_cfg.name == "sec" and not _sec_heartbeat_due(seq):
-    if not emit_tick(level_cfg, seq):
+        pass  # сек-такт без CycleTicked: идемпотентность держит last_seq демона
+    elif not emit_tick(level_cfg, seq):
         return "already"  # тик этого такта уже был (идемпотентность)
 
     # северная звезда (t-260929161138): дневной уровень раз в сутки пишёт
@@ -963,6 +998,26 @@ def process_level(level_cfg, model: dict):
         st["busy"] = False
 
 
+# Сек-уровень писал CycleTicked + ReflexObserved КАЖДУЮ секунду: 1.35 млн
+# подписанных событий/нед — 98% стора (аудит 2026-10-16, стор 1.37 ГБ).
+# Теперь в стор — heartbeat раз в SIMBIOT_SEC_HEARTBEAT с (60) и такты, где
+# что-то случилось (алерт нагрузки, команды бота, каналы портала/агентов).
+# router42 бенча и так считает эти события шумом; дашборд видит heartbeat.
+SEC_HEARTBEAT = int(os.environ.get("SIMBIOT_SEC_HEARTBEAT", "60"))
+_SEC_HB = {"last": None, "seq": None}
+
+
+def _sec_heartbeat_due(seq: int) -> bool:
+    """Heartbeat-такт сек-уровня: первый после старта и раз в SEC_HEARTBEAT.
+    Решение кешируется на seq — process_level и _reflex_tick видят одно."""
+    if _SEC_HB["seq"] == seq:
+        return True
+    if SEC_HEARTBEAT <= 1 or _SEC_HB["last"] is None or seq - _SEC_HB["last"] >= SEC_HEARTBEAT:
+        _SEC_HB["last"] = _SEC_HB["seq"] = seq
+        return True
+    return False
+
+
 def _reflex_tick(level_cfg, seq: int, model: dict):
     """Секундный рефлекс: телеметрия + алерты + без-LLM команды, без LLM.
 
@@ -1025,6 +1080,8 @@ def _reflex_tick(level_cfg, seq: int, model: dict):
         raise
     except Exception as e:
         print(f"[agent-channel] рефлекс не смог: {e}", flush=True)
+    if (_sec_heartbeat_due(seq) or payload.get("alert") or payload.get("commands")
+            or payload.get("portal_channel") or payload.get("agent_channel")):
         es.append_event("cycle", f"reflex:{seq}", "ReflexObserved", payload)
     return payload, None
 
@@ -1210,11 +1267,23 @@ def main_loop():
     dt.start()
     last_seq = {}
     threads = {}
+    nudge = MinNudge()
+
+    def _idle(name):
+        t = threads.get(name)
+        return not _runners.get(name, {}).get("busy") and not (t is not None and t.is_alive())
+
     while True:
         for name in ACTIVE:
             cfg = LEVELS[name]
             seq = tick_number(cfg, int(time.time()))
-            if last_seq.get(name) == seq:
+            if name == "min" and nudge.due() and _idle(name):
+                # внеочередной мин-тик на входящее: слот следующего такта
+                # (свой seq уже отработан на границе) — граница его пропустит
+                seq = max(seq, last_seq.get(name, seq - 1) + 1)
+                nudge.fired()
+                _log(f"min #{seq}: тик по входящему (не ждём границы минуты)")
+            elif last_seq.get(name, -1) >= seq:
                 continue  # этого такта ещё не было — ждём границу
             last_seq[name] = seq
             if cfg.uses_llm and _runners.get(name, {}).get("busy"):
@@ -1226,14 +1295,64 @@ def main_loop():
                 continue
             def _run(cfg=cfg, seq=seq):
                 model = es.run_projections()
-                state = process_level(cfg, model)
+                state = process_level(cfg, model, seq=seq)
-                if state not in ("already",):
+                if state not in ("already",) and not str(state).startswith("reflex ok"):
                     _log(f"{cfg.name} #{seq}: {state}")
             t = threading.Thread(target=_run, daemon=True, name=f"tick-{name}")
             threads[name] = t
             t.start()
+        nudge.poll()
         time.sleep(POLL)
 
 
+class MinNudge:
+    """Тик мин-уровня по входящему, а не по границе минуты (аудит 2026-10-16).
+
+    Ответ в TG ждал границы такта: медиана 36 с от приёма до старта LLM
+    (p90 93 с) при 24 с самой модели. Демон раз в POLL дёшево смотрит хвост
+    стора (диапазон по PK от последнего виденного id) на новые
+    MessageReceived; через DEBOUNCE секунд тишины после последнего —
+    внеочередной тик (серия сообщений Alex уходит одним батчем). Занят
+    уровень — нудж ждёт его конца. Гейт важности работает как обычно:
+    команда/чужой/уже отвеченное → тихий такт без LLM.
+    SIMBIOT_MIN_NUDGE=0 — выключить; SIMBIOT_MIN_NUDGE_DEBOUNCE (3 с).
+    """
+
+    def __init__(self):
+        self.enabled = os.environ.get("SIMBIOT_MIN_NUDGE", "1") != "0"
+        self.debounce = float(os.environ.get("SIMBIOT_MIN_NUDGE_DEBOUNCE", "3"))
+        self.last_id = None
+        self.pending_since = None  # время последнего нового входящего
+        self._db = None
+
+    def poll(self):
+        if not self.enabled:
+            return
+        try:
+            if self._db is None:
+                self._db = es._conn()
+            db = self._db
+            if self.last_id is None:
+                self.last_id = db.execute("SELECT COALESCE(MAX(id), 0) FROM events").fetchone()[0]
+                return
+            row = db.execute(
+                "SELECT MAX(id), SUM(event_type IN ('MessageReceived', 'MessageTranscribed')) "
+                "FROM events WHERE id > ?", (self.last_id,)).fetchone()
+        except Exception:
+            self._db = None  # переоткроем на следующем проходе
+            return  # нудж — ускоритель, не критичный путь
+        if row and row[0]:
+            self.last_id = row[0]
+            if row[1]:
+                self.pending_since = time.time()
+
+    def due(self) -> bool:
+        return (self.pending_since is not None
+                and time.time() - self.pending_since >= self.debounce)
+
+    def fired(self):
+        self.pending_since = None
+
+
 if __name__ == "__main__":
     main_loop()

tests/test_dash_chat.py

@@ -47,6 +47,12 @@ def _row(msg_id, sender, text, ts, consumed=0):
     return (msg_id, sender, text, ts, consumed)
 
 
+# изоляция стора (аудит 2026-10-16): тест не пишет в боевой стор, даже
+# когда его гоняет тик харнеса под SIMBIOT_HOME (TTFTWatchdogRetry с
+# бюджетом 0.5с из этого набора оседали в проде)
+import tempfile as _iso_tf  # noqa: E402
+os.environ["SIMBIOT_DB"] = os.path.join(_iso_tf.mkdtemp(prefix="simbiot_test_"), "harness.db")
+
 class DashChatTest(unittest.TestCase):
     @classmethod
     def setUpClass(cls):

tests/test_gate.py

@@ -16,6 +16,12 @@ sys.path.insert(0, "/root/simbiot-harness")
 for _v in ("SIMBIOT_GATE_MIN", "SIMBIOT_GATE_HOUR", "SIMBIOT_GATE_DAY"):
     os.environ.pop(_v, None)
 
+# изоляция стора (аудит 2026-10-16): тест не пишет в боевой стор, даже
+# когда его гоняет тик харнеса под SIMBIOT_HOME (TTFTWatchdogRetry с
+# бюджетом 0.5с из этого набора оседали в проде)
+import tempfile as _iso_tf  # noqa: E402
+os.environ["SIMBIOT_DB"] = os.path.join(_iso_tf.mkdtemp(prefix="simbiot_test_"), "harness.db")
+
 from harness import scheduler
 from harness.levels import LEVELS
 

tests/test_router_env_fallback.py

@@ -26,6 +26,12 @@ def check(name, cond, extra=""):
         fails.append(name)
 
 
+# изоляция стора (аудит 2026-10-16): тест не пишет в боевой стор, даже
+# когда его гоняет тик харнеса под SIMBIOT_HOME (TTFTWatchdogRetry с
+# бюджетом 0.5с из этого набора оседали в проде)
+import tempfile as _iso_tf  # noqa: E402
+os.environ["SIMBIOT_DB"] = os.path.join(_iso_tf.mkdtemp(prefix="simbiot_test_"), "harness.db")
+
 from harness import runtime
 
 # hermetic: окружение тест-процесса не должно подсказывать значения

tests/test_tick_latency_noise.py

@@ -0,0 +1,124 @@
+# -*- coding: utf-8 -*-
+"""Аудит 2026-10-16: медиа без подписи в триаже, тик по входящему, тишина сек-уровня.
+
+Запуск: ./venv/bin/python tests/test_tick_latency_noise.py  (hermetic, свой стор)
+
+Проверяет: фото/документ/видео без подписи — ожидают ответа (свежие ≤ 6 ч),
+старые — нет; стикер без текста — нет; команда — нет; контекст тика
+показывает тег медиа с путём файла; MinNudge срабатывает после debounce на
+MessageReceived/MessageTranscribed, игнорирует прочие события, выключается
+env; сек-уровень пишет CycleTicked/ReflexObserved только heartbeat'ом и в
+тактах с событием (команды/алерт), а тихие такты — без записи в стор.
+"""
+import json
+import os
+import sys
+import tempfile
+import time
+from datetime import datetime, timedelta, timezone
+
+sys.path.insert(0, "/root/simbiot-harness")
+os.environ["SIMBIOT_DB"] = os.path.join(tempfile.mkdtemp(prefix="simbiot_latnoise_"), "harness.db")
+for _v in ("SIMBIOT_MIN_NUDGE", "SIMBIOT_MIN_NUDGE_DEBOUNCE", "SIMBIOT_MEDIA_PENDING_H",
+           "SIMBIOT_SEC_HEARTBEAT", "SIMBIOT_DECISION"):
+    os.environ.pop(_v, None)
+
+from harness import es, scheduler
+from harness.levels import LEVELS
+
+es.init_db()
+fails = []
+
+
+def check(name, cond, extra=""):
+    print(("PASS" if cond else "FAIL"), name, ("| " + str(extra)) if not cond else "")
+    if not cond:
+        fails.append(name)
+
+
+now = datetime.now(timezone.utc)
+
+
+def msg(uid, kind="text", text="", age_h=0.0, **kw):
+    return {"update_id": uid, "kind": kind, "text": text, "replied": False, "ignored": False,
+            "ts": (now - timedelta(hours=age_h)).isoformat(), "chat_id": 1,
+            "first_name": "Alex", **kw}
+
+
+# ── 1. медиа без подписи в триаже ──
+inbox = {f"m-{m['update_id']}": m for m in [
+    msg(1, "photo", age_h=0.1, file_path="inbox/1.jpg"),
+    msg(2, "document", age_h=1, file_name="dogovor.pdf", file_path="inbox/dogovor.pdf"),
+    msg(3, "video", age_h=0.2),
+    msg(4, "photo", age_h=30),            # старое, выпало до фикса
+    msg(5, "sticker", age_h=0.1),
+    msg(6, "text", "/статус", age_h=0.1),
+    msg(7, "text", "привет", age_h=50),   # текст — как раньше, без окна свежести
+    msg(8, "voice", age_h=40),            # голосовое — как раньше
+    msg(9, "photo", age_h=0.1, replied=True),
+]}
+ids = {m["update_id"] for m in scheduler._pending_inbox({"inbox": inbox})}
+check("фото/документ/видео без подписи — ждут ответа", {1, 2, 3} <= ids, ids)
+check("старое медиа (>6ч) — не будит LLM", 4 not in ids, ids)
+check("стикер без текста — не ждёт", 5 not in ids, ids)
+check("команда — не для LLM", 6 not in ids, ids)
+check("текст и голосовое — как раньше", {7, 8} <= ids, ids)
+check("отвеченное — нет", 9 not in ids, ids)
+os.environ["SIMBIOT_MEDIA_PENDING_H"] = "48"
+check("окно свежести настраивается env", 4 in {m["update_id"] for m in scheduler._pending_inbox({"inbox": inbox})})
+os.environ.pop("SIMBIOT_MEDIA_PENDING_H")
+sig = scheduler.collect_signals(LEVELS["min"], {"inbox": {"m-1": inbox["m-1"]}, "tasks": {}})
+check("фото без подписи → critical-сигнал мин-уровня", any(s[1] == "unreplied_inbox" for s in sig), sig)
+ctx = scheduler.build_context(LEVELS["min"], {"inbox": {"m-2": inbox["m-2"]}, "tasks": {},
+                                              "escalations": {}, "policy": {}}, batch=[])
+check("контекст: тег документа с путём и «без подписи»",
+      "[document, файл: inbox/dogovor.pdf, без подписи]" in ctx, ctx[:400])
+
+# ── 2. MinNudge ──
+os.environ["SIMBIOT_MIN_NUDGE_DEBOUNCE"] = "0.3"
+n = scheduler.MinNudge()
+n.poll()  # первый проход — только запоминает хвост
+check("nudge: на старте не срабатывает", not n.due())
+es.append_event("cycle", "noise:1", "DigestWritten", {"level": "min"})
+n.poll()
+check("nudge: прочие события не будят", n.pending_since is None)
+es.append_event("tg", "m-777", "MessageReceived", {"update_id": 777, "text": "эй"})
+n.poll()
+check("nudge: входящее замечено, но debounce", n.pending_since is not None and not n.due())
+time.sleep(0.35)
+check("nudge: после debounce — пора", n.due())
+n.fired()
+check("nudge: fired сбрасывает", not n.due())
+es.append_event("tg", "m-777", "MessageTranscribed", {"text": "расшифровка"})
+n.poll(); time.sleep(0.35)
+check("nudge: расшифровка голосового тоже будит", n.due())
+os.environ["SIMBIOT_MIN_NUDGE"] = "0"
+n2 = scheduler.MinNudge(); n2.poll()
+es.append_event("tg", "m-778", "MessageReceived", {"update_id": 778, "text": "эй"})
+n2.poll(); time.sleep(0.35)
+check("nudge: SIMBIOT_MIN_NUDGE=0 выключает", not n2.due())
+os.environ.pop("SIMBIOT_MIN_NUDGE")
+
+# ── 3. сек-уровень: heartbeat + такты с событием ──
+def count(et):
+    return sum(1 for e in es.get_events(0, 100000) if e["event_type"] == et)
+
+sec = LEVELS["sec"]
+scheduler._SEC_HB.update(last=None, seq=None)
+base_t, base_r = count("CycleTicked"), count("ReflexObserved")
+start = 1_000_000
+for i in range(130):
+    scheduler.process_level(sec, {"inbox": {}, "tasks": {}, "policy": {"cpu_alert_threshold": 1e9}},
+                            seq=start + i)
+dt, dr = count("CycleTicked") - base_t, count("ReflexObserved") - base_r
+check("сек: 130 тихих тактов → 3 heartbeat CycleTicked (0, 60, 120)", dt == 3, dt)
+check("сек: ReflexObserved только heartbeat", dr == 3, dr)
+scheduler.process_level(sec, {"inbox": {}, "tasks": {}, "policy": {"cpu_alert_threshold": 0}},
+                        seq=start + 131)
+check("сек: алерт нагрузки пишется вне heartbeat", count("ReflexObserved") - base_r == 4)
+last = [json.loads(e["data"]) for e in es.get_events(0, 100000)
+        if e["event_type"] == "ReflexObserved"][-1]
+check("сек: в записи алерта есть alert", "alert" in last, last)
+
+print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)")
+sys.exit(1 if fails else 0)

tests/test_timeout_kill.py

@@ -14,6 +14,12 @@ import time
 import unittest
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+# изоляция стора (аудит 2026-10-16): тест не пишет в боевой стор, даже
+# когда его гоняет тик харнеса под SIMBIOT_HOME (TTFTWatchdogRetry с
+# бюджетом 0.5с из этого набора оседали в проде)
+import tempfile as _iso_tf  # noqa: E402
+os.environ["SIMBIOT_DB"] = os.path.join(_iso_tf.mkdtemp(prefix="simbiot_test_"), "harness.db")
+
 from harness import isolated_runner
 
 

tests/test_tool_budget.py

@@ -12,6 +12,12 @@ import sys
 
 sys.path.insert(0, "/root/simbiot-harness")
 
+# изоляция стора (аудит 2026-10-16): тест не пишет в боевой стор, даже
+# когда его гоняет тик харнеса под SIMBIOT_HOME (TTFTWatchdogRetry с
+# бюджетом 0.5с из этого набора оседали в проде)
+import tempfile as _iso_tf  # noqa: E402
+os.environ["SIMBIOT_DB"] = os.path.join(_iso_tf.mkdtemp(prefix="simbiot_test_"), "harness.db")
+
 from harness import _llm_worker
 
 fails = []

tests/test_ttft_retry.py

@@ -16,6 +16,12 @@ import types
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
+# изоляция стора (аудит 2026-10-16): тест не пишет в боевой стор, даже
+# когда его гоняет тик харнеса под SIMBIOT_HOME (TTFTWatchdogRetry с
+# бюджетом 0.5с из этого набора оседали в проде)
+import tempfile as _iso_tf  # noqa: E402
+os.environ["SIMBIOT_DB"] = os.path.join(_iso_tf.mkdtemp(prefix="simbiot_test_"), "harness.db")
+
 from harness import _llm_worker as W  # noqa: E402
 
 FAILS = []