"""Взаимный сторож с чужим агентом: отметка раз в сутки и тревога на молчание. Договорённость с соседом (22.09.2026): в отдельной комнате вайбграма оба раз в сутки печатают строку `hb <ник> at=… boot=… every=… src=…`, и каждый следит за молчанием другого. Замолчал дольше порога - человеку соседа уходит сообщение, что цепь молчит. Смысл сторожа в том, чтобы он не врал: прибор, который молча не сторожит, хуже отсутствующего - он выглядит исправным. Поэтому дефолтов здесь нет нигде. Период своей отметки считается из фактического крона джоба (`every_from_cron`), и расписание, которое не сводится к «раз в N часов», даёт честное `every=unknown`, а не подставленное число. Порог тревоги - `every * 1.5`, где `every` взят ИЗ СТРОКИ СОСЕДА, а не из своей константы. Каждый прогон, в котором показание снять нельзя (окно смотрит в прошлое, окно не дочитано, период неизвестен, отметок не видели ни разу), говорит это вслух в `ack` и в лог: «в окне никого» и «окно не прочитано до конца» - это два РАЗНЫХ показания, и выглядеть одинаково они не должны. Окно - последние `WINDOW` СООБЩЕНИЙ этой комнаты (`Vibegram.window`), а не «что нового с курсора»: курсор отдаёт строку соседа один раз, и следующий прогон на живой комнате увидел бы пустое окно. Опора тревоги - `seen`, то есть когда Я увидел строку: `at=` внутри строки это показание соседа о себе, и остановившиеся у него часы не должны выглядеть как живая цепь. """ from __future__ import annotations import itertools import json import re from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path # noqa: TC003 - в аннотации dataclass from typing import TYPE_CHECKING, Literal from croniter import CroniterError, croniter if TYPE_CHECKING: from collections.abc import Sequence from datetime import tzinfo from beaver_agent.hands.vibegram import Event __all__ = [ "BOOT", "FACTOR", "PEER", "WINDOW", "Beat", "Report", "Watch", "ack_line", "alarm_text", "beat_line", "drill", "every_from_cron", "look", "parse_beat", ] PEER = "claude-kai" """Чей `hb` сторожим. Ник соседа стоит и в его строке, и в нашем `ack`.""" WINDOW = 25 """Сколько последних сообщений комнаты просим явно: молча опираться на чужой дефолт в стороже нельзя - размер окна едет в `ack`.""" FACTOR = 1.5 """Порог молчания = заявленный соседом период × это. Одна потерянная строка не должна поднимать тревогу, две подряд - должны.""" UNARMED_GRACE = timedelta(days=7) """Сколько терпим `armed=no`, прежде чем сказать человеку, что кольцо так и не заработало: недостроенное иначе тихо становится постоянным.""" REPEAT = timedelta(hours=24) """Тревога повторяется не чаще этого, а не каждый прогон.""" PROBES = 5 """Сколько срабатываний крона сверяем между собой, выводя период.""" STATE_VERSION = 1 BOOT = datetime.now(UTC) """Время старта процесса: модуль импортируется при сборке конфига gateway.""" _FIELD = re.compile(r"(\w+)=(\S+)") _UNITS = {"h": 3600, "m": 60, "s": 1} _SPAN = re.compile(r"^(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?$") def _iso(when: datetime | None) -> str | None: if when is None: return None return when.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") def _parse_iso(raw: str | None) -> datetime | None: if not raw: return None try: when = datetime.fromisoformat(raw) except ValueError: return None return when if when.tzinfo else when.replace(tzinfo=UTC) def parse_span(raw: str) -> timedelta | None: """`24h`, `90m`, `36h30m` - в timedelta; `unknown` и мусор - в None.""" match = _SPAN.fullmatch(raw.strip()) if match is None or not any(match.groups()): return None seconds = sum( int(value) * _UNITS[unit] for value, unit in zip(match.groups(), "hms", strict=True) if value ) return timedelta(seconds=seconds) or None def span(delta: timedelta | None) -> str: """Период в строку: неизвестный период так и называется, а не молчит.""" if delta is None: return "unknown" seconds = int(delta.total_seconds()) if seconds % 3600 == 0: return f"{seconds // 3600}h" if seconds % 60 == 0: return f"{seconds // 60}m" return f"{seconds}s" @dataclass(frozen=True, slots=True) class Beat: """Разобранная строка отметки. `at` не читается - строка всё равно жива.""" nick: str at: datetime | None boot: datetime | None every: timedelta | None every_raw: str def parse_beat(text: str, kind: str = "hb") -> Beat | None: """Строку `hb <ник> ключ=значение …` из текста события; иначе None. Текст события хаба отрендерен как `<кто>: <тело>`, поэтому ищем метку внутри строки, а не с её начала. """ head = re.search(rf"(?:^|\s){re.escape(kind)}\s+(\S+)", text) if head is None: return None fields = dict(_FIELD.findall(text[head.end() :])) every_raw = fields.get("every", "") return Beat( nick=head.group(1), at=_parse_iso(fields.get("at")), boot=_parse_iso(fields.get("boot")), every=parse_span(every_raw) if every_raw else None, every_raw=every_raw or "unknown", ) def every_from_cron( expression: str | None, tz: tzinfo, *, now: datetime | None = None, grain: timedelta = timedelta(hours=1), ) -> timedelta | None: """Шаг крона, если он сводится к «раз в N `grain`»; иначе None. Считаем по стенным часам зоны, в которой планировщик читает кроны: суточный крон - это 24 часа и в сутки перевода стрелок тоже, а вот `0 2,8,12,16,20 * * *` не сводится ни к какому N и даёт None. `grain` по умолчанию час - такой период печатается в отметке соседу; сторожу прогонов нужен и `*/10`, он просит минуту. """ if not expression: return None start = (now or datetime.now(UTC)).astimezone(tz).replace(tzinfo=None) try: clock = croniter(expression, start_time=start) fires = [clock.get_next(datetime) for _ in range(PROBES + 1)] except (CroniterError, ValueError): return None steps = {b - a for a, b in itertools.pairwise(fires)} if len(steps) != 1: return None step = steps.pop() return step if step % grain == timedelta(0) else None def beat_line( *, nick: str, at: datetime, boot: datetime, every: timedelta | None ) -> str: """Своя отметка. Период не выведен - так и пишем, числа не выдумываем.""" return ( f"hb {nick} at={_iso(at)} boot={_iso(boot)} " f"every={span(every)} src={'cron' if every else 'none'}" ) def ack_line( *, peer: str, last: datetime | None, seen: datetime | None, res: timedelta | None, window: int, stale: bool, armed: bool, threshold: timedelta | None, fallback: bool, full: bool, alert: Literal["fired", "held", "none"], ) -> str: """Расписка о прогоне в комнату: что видели, чем мерили и что решили. `win=` - про само окно: `25` это «последние 25 сообщений комнаты прочитаны целиком», `25/partial` - «дочитать историю не удалось», `stale` - «окно смотрит в прошлое». Пустое окно при `partial` ничего не доказывает. `alert=` - РЕШЕНИЕ ПРИБОРА в этом прогоне, а не факт того, что человек сосед что-то получил: `fired` - тревога (`silence` или `unarmed`) в этом прогоне взята и отдана дальше, `held` - порог пройден, но тревога подавлена как повтор (предыдущая уходила меньше `REPEAT` назад), `none` - тревоги нет. Доставка решается за пределами сторожа - мастером и диспетчером, - и её отказ виден в логе, а не тут: без арифметики соседу видно было только, что мы что-то посчитали, но не что мы из этого поняли. """ mark = span(threshold) + ("-fallback" if fallback and threshold else "") read = "stale" if stale else f"{window}" if full else f"{window}/partial" return ( f"ack {peer} last={_iso(last) or ('never' if not armed else 'unknown')} " f"seen={_iso(seen) or 'never'} res={span(res)} " f"win={read} armed={'yes' if armed else 'no'} " f"thr={mark} alert={alert}" ) def drill(path: Path, *, now: datetime, tz: tzinfo) -> bool: """Стоит ли сегодняшняя дата в файле флага учений. Файл - список дат `YYYY-MM-DD`, по одной на строку; дата - локальная дата прогона в зоне планировщика, той же, в которой читаются кроны. Файла нет, он пуст или нечитаем - учений нет, и отметка уходит как обычно: флаг выключает прибор, поэтому его отсутствие должно значить «работаем». """ try: raw = path.read_text(encoding="utf-8") except OSError: return False today = now.astimezone(tz).strftime("%Y-%m-%d") return today in {line.strip() for line in raw.splitlines()} @dataclass(slots=True) class Watch: """Состояние между прогонами: файл на диске, переживает передеплой.""" path: Path at: datetime | None = None """`at=` из последней виденной отметки соседа - его показание о себе.""" seen: datetime | None = None """Когда МЫ увидели ту отметку. На это опирается тревога.""" every: timedelta | None = None """Последний период, который сосед назвал читаемым числом.""" every_unknown: bool = False """В последней отметке стояло `every=unknown`: порог взят с запасом назад.""" top: int = 0 """Максимальный id, который мы видели. Поехал назад - окно смотрит в прошлое.""" alerted: datetime | None = None unarmed_since: datetime | None = None unarmed_told: bool = False @classmethod def load(cls, path: Path) -> Watch: try: raw = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): return cls(path=path) if raw.get("version") != STATE_VERSION: return cls(path=path) every = raw.get("every") return cls( path=path, at=_parse_iso(raw.get("at")), seen=_parse_iso(raw.get("seen")), every=timedelta(seconds=every) if every else None, every_unknown=bool(raw.get("every_unknown")), top=int(raw.get("top", 0)), alerted=_parse_iso(raw.get("alerted")), unarmed_since=_parse_iso(raw.get("unarmed_since")), unarmed_told=bool(raw.get("unarmed_told")), ) def save(self) -> None: body = { "version": STATE_VERSION, "at": _iso(self.at), "seen": _iso(self.seen), "every": int(self.every.total_seconds()) if self.every else None, "every_unknown": self.every_unknown, "top": self.top, "alerted": _iso(self.alerted), "unarmed_since": _iso(self.unarmed_since), "unarmed_told": self.unarmed_told, } self.path.parent.mkdir(parents=True, exist_ok=True) tmp = self.path.with_suffix(".tmp") tmp.write_text(json.dumps(body, ensure_ascii=False, indent=1), encoding="utf-8") tmp.replace(self.path) @dataclass(frozen=True, slots=True) class Report: """Что решил один прогон сторожа.""" ack: str stale: bool full: bool """Окно дочитано до конца истории комнаты: пустое окно - это показание.""" armed: bool heard: bool """Видели ли в окне хоть что-то от соседа - `hb` или его `ack`.""" age: timedelta | None threshold: timedelta | None alarm: Literal["silence", "unarmed"] | None note: str """Строка для лога: почему прогон решил именно так.""" def look( state: Watch, events: Sequence[Event], *, now: datetime, res: timedelta | None, full: bool, window: int = WINDOW, peer: str = PEER, ) -> Report: """Разобрать окно, обновить состояние и решить, будить ли человека. `events` - последние сообщения комнаты, `full` - дочитаны ли они до конца истории. Параметр без дефолта нарочно: «окно полное» - это показание, и подставлять его молча за того, кто окно читал, здесь нельзя. Состояние меняется здесь, а `alerted` / `unarmed_told` ставит джоб - и только когда тревога действительно ушла: несостоявшаяся доставка не должна заглушать следующий прогон. """ top = max((e.id for e in events), default=None) stale = top is not None and top < state.top beats = [ beat for beat in (parse_beat(e.text) for e in events) if beat is not None and beat.nick == peer ] heard = bool(beats) or any(e.nick == peer for e in events) if stale: note = f"окно поехало назад: максимальный id {top} против {state.top}" else: if top is not None: state.top = max(state.top, top) if beats: last = beats[-1] state.at, state.seen = last.at, now state.every_unknown = last.every is None if last.every is not None: state.every = last.every read = "история дочитана" if full else "история НЕ дочитана" note = ( f"отметок соседа в окне: {len(beats)} " f"(сообщений комнаты: {len(events)} из {window}, {read})" ) armed = state.seen is not None threshold = state.every * FACTOR if state.every is not None else None age = now - state.seen if state.seen is not None else None if armed: state.unarmed_since = None elif state.unarmed_since is None: state.unarmed_since = now alarm: Literal["silence", "unarmed"] | None = None held = False if stale: pass elif not full: # Хвост комнаты недочитан, а недостающее - как раз самое свежее: # пустое окно тут не доказывает молчания соседа, оно доказывает, # что мы не дочитали. Показания нет - тревоги нет. note += "; окно не дочитано - это не показание, тревога не берётся" elif not armed: if ( not state.unarmed_told and state.unarmed_since is not None and now - state.unarmed_since >= UNARMED_GRACE ): alarm = "unarmed" elif threshold is None: note += "; период соседа ни разу не прочитан - мерить нечем, тревога не берётся" elif age is not None and age > threshold: if state.alerted is None or now - state.alerted >= REPEAT: alarm = "silence" else: held = True note += "; тревога уже уходила меньше суток назад" return Report( ack=ack_line( peer=peer, last=state.at, seen=state.seen, res=res, window=window, stale=stale, armed=armed, threshold=threshold, fallback=state.every_unknown, full=full, alert="fired" if alarm is not None else "held" if held else "none", ), stale=stale, full=full, armed=armed, heard=heard, age=age, threshold=threshold, alarm=alarm, note=note, ) def alarm_text(report: Report, *, peer: str = PEER, mine_sent: bool) -> str: """Текст человеку соседа. Различитель канала - последним абзацем.""" if report.alarm == "unarmed": return ( f"Сторожевое кольцо с {peer} так и не заработало: за семь суток " f"в сторожевой комнате не было ни одной его отметки `hb`, а значит " f"сторожить нечего - прибор не вооружён (`armed=no`). Свою отметку " f"мы печатаем и свой `ack` тоже" + ("" if mine_sent else ", но сейчас она в комнату не ушла") + ". Проверить стоит три вещи: та ли это комната (токен " "`VIBEGRAM_HB_TOKEN` мог остаться от другой), запустил ли сосед " "свою половину, и виден ли нам вообще чужой текст в этой комнате." ) age = span(report.age) if report.age else "?" threshold = span(report.threshold) if report.threshold else "?" channel = ( "Наша собственная запись в комнату в этом же прогоне прошла, а от " f"{peer} нет НИ `hb`, НИ `ack` - значит канал подозревать надо раньше, " "чем смерть соседа: пишем мы в комнату, но чужого оттуда не читаем." if mine_sent and not report.heard else ( "Наша собственная запись в комнату в этом прогоне НЕ прошла - " "первым делом канал, а не сосед." if not mine_sent else f"Что-то от {peer} в окне было (его `ack` или другая строка), " "так что канал скорее жив, а молчит именно отметка." ) ) return ( f"Сторожевая цепь молчит: последнюю отметку `hb {peer}` мы видели " f"{age} назад при пороге {threshold} (порог - период из его же строки " "× 1.5). Причин может быть три: машина соседа, его крон, или канал - " f"сама комната. {channel}\n" f"Это дело человека {peer}, а не наше: разбудить владельца имеет смысл " "только чтобы он написал соседу по своему каналу. Повторно об этом " "молчании мы скажем не раньше чем через сутки." )