diff --git a/harness/scheduler.py b/harness/scheduler.py index d66c900..6d132d1 100644 --- a/harness/scheduler.py +++ b/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 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"))] + return [m for m in (model.get("inbox") or {}).values() if _is_pending(m)] 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 = [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"))] + inbox = _pending_inbox(model) 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): - """Тик одного уровня: занят? → absorb; иначе работа.""" +def process_level(level_cfg, model: dict, seq: int = None): + """Тик одного уровня: занят? → 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 not emit_tick(level_cfg, seq): + if level_cfg.name == "sec" and not _sec_heartbeat_due(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,7 +1080,9 @@ def _reflex_tick(level_cfg, seq: int, model: dict): raise except Exception as e: print(f"[agent-channel] рефлекс не смог: {e}", flush=True) - es.append_event("cycle", f"reflex:{seq}", "ReflexObserved", payload) + 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) - if state not in ("already",): + state = process_level(cfg, model, seq=seq) + 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() \ No newline at end of file diff --git a/tests/test_dash_chat.py b/tests/test_dash_chat.py index e889834..fb73f8d 100644 --- a/tests/test_dash_chat.py +++ b/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): diff --git a/tests/test_gate.py b/tests/test_gate.py index 7f8b610..0f59a1f 100644 --- a/tests/test_gate.py +++ b/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 diff --git a/tests/test_router_env_fallback.py b/tests/test_router_env_fallback.py index 3bb0251..1555113 100644 --- a/tests/test_router_env_fallback.py +++ b/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: окружение тест-процесса не должно подсказывать значения diff --git a/tests/test_tick_latency_noise.py b/tests/test_tick_latency_noise.py new file mode 100644 index 0000000..c206ca5 --- /dev/null +++ b/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) diff --git a/tests/test_timeout_kill.py b/tests/test_timeout_kill.py index 1a3c928..4cb71bb 100644 --- a/tests/test_timeout_kill.py +++ b/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 diff --git a/tests/test_tool_budget.py b/tests/test_tool_budget.py index 8133f8f..153de9e 100644 --- a/tests/test_tool_budget.py +++ b/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 = [] diff --git a/tests/test_ttft_retry.py b/tests/test_ttft_retry.py index 88e460d..35f5506 100644 --- a/tests/test_ttft_retry.py +++ b/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 = []