From f6cb17acafa397aacac80b9f0377309343421d53 Mon Sep 17 00:00:00 2001 From: h Date: Tue, 22 Sep 2026 14:21:51 +0000 Subject: [PATCH] feat(hands,jobs): watchdog ring with a neighbour agent, daily heartbeat and silence alarm --- .env.example | 3 + README.md | 61 +++++ beaver_agent/hands/__init__.py | 25 ++ beaver_agent/hands/heartbeat.py | 404 ++++++++++++++++++++++++++++++++ beaver_agent/hands/vibegram.py | 9 + beaver_agent/jobs/__init__.py | 8 + beaver_agent/jobs/heartbeat.py | 113 +++++++++ scripts/vibegram_join.py | 84 +++++++ tests/test_heartbeat.py | 322 +++++++++++++++++++++++++ 9 files changed, 1029 insertions(+) create mode 100644 beaver_agent/hands/heartbeat.py create mode 100644 beaver_agent/jobs/heartbeat.py create mode 100755 scripts/vibegram_join.py create mode 100644 tests/test_heartbeat.py diff --git a/.env.example b/.env.example index 2c63a85..d30dcac 100644 --- a/.env.example +++ b/.env.example @@ -114,6 +114,9 @@ VIBEGRAM_REPO= # Токен GitHub с доступом к репе (приватная); пусто - публичная VIBEGRAM_REPO_TOKEN= +VIBEGRAM_HB_TOKEN= +VIBEGRAM_HB_NICK= + ### Досуг (ежечасный тик мастера, с S13) ###################################### # Час активного окна (12:00-03:59) и четыре часа тихого (04:00-11:59) будят diff --git a/README.md b/README.md index feea80e..51e42a6 100644 --- a/README.md +++ b/README.md @@ -204,6 +204,67 @@ publishes no reply, so the only voice in it is the `say` tool. Switches: `досуг: выкл` as the first line of the file for every day, and `LEISURE=off` in `.env` to drop the job entirely. +## The watchdog ring (`beaver_agent/jobs/heartbeat.py`) + +An agreement with a foreign agent on someone else's home server: both print +one line a day into a vibegram room of their own, and both watch the other's +silence. If the neighbour goes quiet past the threshold, *their* human is the +one who needs to hear about it, so the alarm goes to the master thread as a +spawned triage job and the dispatcher decides whether to wake the owner. + +The room is a **second, independent** `Vibegram` client (`VIBEGRAM_HB`), not +the one the dispatcher's `vibegram` tool and the triage job share. `/api/pending` +is a cursor on the hub's side: one client for both rooms would mean the triage +job eats the heartbeats, and the heartbeats drown in conversation. The model +gets no tool for this room - the watchdog is a mechanism. + +The point of a watchdog is that it does not lie: an instrument that quietly +fails to watch is worse than no instrument, because it looks healthy. So there +are no defaults anywhere in it. + +- `отметка` (17:12 local) prints `hb at=… boot=… every=… src=…`. The + period is not a constant: it is derived from that job's own registered cron + (`run.job.cron`). A schedule that does not reduce to "every N hours" gets + `every=unknown src=none` - lying with a number is worse than refusing. +- `сторож` (17:42 local) asks for an explicit window (`pending(limit=25)`), + finds the newest `hb claude-kai …`, and compares its age against + `every * 1.5` where `every` comes **from the neighbour's own line**. The + clock it trusts is `seen` - when *we* saw the line - not the `at=` inside it, + which is the neighbour's statement about himself. +- Every run prints `ack claude-kai last=… seen=… res=… win=… armed=… thr=…` + into the room. Anything the run could not measure says so out loud there: + `every=unknown` in his line falls back to the last known period and marks the + threshold `thr=36h-fallback`; a period never read once is `thr=unknown` and + no alarm is taken. +- Never having seen a beat is not an alarm, it is `armed=no` - there is nothing + to watch yet. Seven days of `armed=no` in a row tells the human once, so an + unfinished ring does not quietly become permanent. +- If the window's highest event id went **backwards** since the last run, the + window is looking at the past: `win=stale`, and an alarm in that run is + forbidden. Same for a hub that did not answer at all. +- The alarm names all three possible causes - the neighbour's machine, his + cron, or the channel (the room itself) - and if our own write into the room + went through in that same run while neither `hb` nor `ack` came back from + him, it says to suspect the channel before the neighbour's death. It repeats + at most once a day, not once a run. + +State between runs is `мета/бобер/.состояние/сторож.json`; a lost file means +the ring disarms itself and starts the seven-day count again. + +Setup is two variables in `.env` and one manual step: + +```bash +# join the watchdog room with the invite code the neighbour sends +python3 scripts/vibegram_join.py --nick +# put the token into .env as VIBEGRAM_HB_TOKEN, then +docker compose restart gateway +``` + +`VIBEGRAM_HB_NICK` is optional - without it the nick is `VIBEGRAM_NICK`. With +no `VIBEGRAM_HB_TOKEN` both jobs do nothing and the log says once at startup +that the watchdog is not configured. The neighbour's nick is the `PEER` +constant in `beaver_agent/hands/heartbeat.py`. + ## Policy (`beaver_agent/policy.py`) `bypassPermissions` everywhere; the boundary is `PreToolUse` rules declared diff --git a/beaver_agent/hands/__init__.py b/beaver_agent/hands/__init__.py index 95d54c8..6f061c0 100644 --- a/beaver_agent/hands/__init__.py +++ b/beaver_agent/hands/__init__.py @@ -6,6 +6,7 @@ from __future__ import annotations +import logging import os from pathlib import Path @@ -19,6 +20,7 @@ from beaver_agent.hands.vibegram import Vibegram from beaver_agent.vault import TZ env = os.environ.get +_log = logging.getLogger(__name__) obsidian = McpServer.stdio( name="obsidian-fs", @@ -98,6 +100,29 @@ vibegram = ( ) vibegram_exposed = (ExposedMcp(name="vibegram"),) if VIBEGRAM else () +VIBEGRAM_HB_NICK = env("VIBEGRAM_HB_NICK") or env("VIBEGRAM_NICK") +VIBEGRAM_HB = ( + Vibegram( + hub=os.environ["VIBEGRAM_HUB"], + token=os.environ["VIBEGRAM_HB_TOKEN"], + nick=VIBEGRAM_HB_NICK or "", + tz=TZ, + ) + if env("VIBEGRAM_HB_TOKEN") and env("VIBEGRAM_HUB") and VIBEGRAM_HB_NICK + else None +) +"""Сторожевая комната: ВТОРОЙ, независимый экземпляр клиента. + +Комната на хабе определяется токеном, а `/api/pending` - курсор на стороне +хаба: общий с основной комнатой экземпляр означал бы, что отметки соседа +съедает триаж-джоб, а сами они тонут в разговоре. Модели эта рука не +выдаётся: сторож - механизм, у него нет тулзы. +""" +if VIBEGRAM_HB is None: + _log.info( + "сторож: VIBEGRAM_HB_TOKEN не задан - отметка и слежка за соседом выключены" + ) + t3code = ( McpServer.http(name="t3code", url=env("T3CODE_MCP", "http://t3code-mcp:8000/mcp")) if "t3" in env("COMPOSE_PROFILES", "").split(",") diff --git a/beaver_agent/hands/heartbeat.py b/beaver_agent/hands/heartbeat.py new file mode 100644 index 0000000..a3a40a7 --- /dev/null +++ b/beaver_agent/hands/heartbeat.py @@ -0,0 +1,404 @@ +"""Взаимный сторож с чужим агентом: отметка раз в сутки и тревога на молчание. + +Договорённость с соседом (22.09.2026): в отдельной комнате вайбграма оба +раз в сутки печатают строку `hb <ник> at=… boot=… every=… src=…`, и каждый +следит за молчанием другого. Замолчал дольше порога - человеку соседа +уходит сообщение, что цепь молчит. + +Смысл сторожа в том, чтобы он не врал: прибор, который молча не сторожит, +хуже отсутствующего - он выглядит исправным. Поэтому дефолтов здесь нет +нигде. Период своей отметки считается из фактического крона джоба +(`every_from_cron`), и расписание, которое не сводится к «раз в N часов», +даёт честное `every=unknown`, а не подставленное число. Порог тревоги - +`every * 1.5`, где `every` взят ИЗ СТРОКИ СОСЕДА, а не из своей константы. +Каждый прогон, в котором показание снять нельзя (окно смотрит в прошлое, +период неизвестен, отметок не видели ни разу), говорит это вслух в `ack`. + +Опора тревоги - `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", + "every_from_cron", + "look", + "parse_beat", +] + +PEER = "claude-kai" +"""Чей `hb` сторожим. Ник соседа стоит и в его строке, и в нашем `ack`.""" +WINDOW = 25 +"""Сколько событий просим у хаба явно: у `pending` свой дефолт (50), и молча +опираться на чужой дефолт в стороже нельзя - размер окна едет в `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 +) -> timedelta | None: + """Шаг крона, если он сводится к «раз в N часов»; иначе None. + + Считаем по стенным часам зоны, в которой планировщик читает кроны: + суточный крон - это 24 часа и в сутки перевода стрелок тоже, а вот + `0 2,8,12,16,20 * * *` не сводится ни к какому N и даёт None. + """ + 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.total_seconds() % 3600 == 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, +) -> str: + """Расписка о прогоне в комнату: что видели, чем мерили и вооружены ли.""" + mark = span(threshold) + ("-fallback" if fallback and threshold else "") + 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={'stale' if stale else window} armed={'yes' if armed else 'no'} " + f"thr={mark}" + ) + + +@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 + 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, + window: int = WINDOW, + peer: str = PEER, +) -> Report: + """Разобрать окно, обновить состояние и решить, будить ли человека. + + Состояние меняется здесь, а `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 + note = f"отметок соседа в окне: {len(beats)}" + + 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 + if stale: + pass + 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: + 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, + ), + stale=stale, + 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}, а не наше: разбудить владельца имеет смысл " + "только чтобы он написал соседу по своему каналу. Повторно об этом " + "молчании мы скажем не раньше чем через сутки." + ) diff --git a/beaver_agent/hands/vibegram.py b/beaver_agent/hands/vibegram.py index 8ef1e33..354ab2b 100644 --- a/beaver_agent/hands/vibegram.py +++ b/beaver_agent/hands/vibegram.py @@ -151,6 +151,15 @@ class Vibegram: ) return fresh + async def say(self, text: str) -> None: + """Написать в комнату; сбой хаба - исключение, а не строка. + + Тулза мастера мягче (`_send` превращает rate limit в текст), но + механизмам нужен отказ, который видно: сторож по нему отличает + мёртвый канал от мёртвого соседа. + """ + await self._post("/api/messages", {"body": text}) + def mentioned(self, events: list[Event]) -> bool: needle = f"@{self.nick}".lower() return any(needle in e.text.lower() for e in events) diff --git a/beaver_agent/jobs/__init__.py b/beaver_agent/jobs/__init__.py index 0b14706..f7ca1f5 100644 --- a/beaver_agent/jobs/__init__.py +++ b/beaver_agent/jobs/__init__.py @@ -8,6 +8,7 @@ from beaver_gateway.jobs.scheduler import Job from beaver_agent.jobs.closing import close_idle from beaver_agent.jobs.deploy import deploy +from beaver_agent.jobs.heartbeat import beat, watch from beaver_agent.jobs.komodo import komodo_alert from beaver_agent.jobs.leisure import tick from beaver_agent.jobs.logs import logs_watch @@ -27,6 +28,13 @@ jobs = [ *([] if LEISURE_OFF else [Job("досуг", tick, cron="7 * * * *")]), Job("вайбграм", vibegram, cron="*/10 * * * *", critical=False), Job("закрытие", close_idle, cron="20 4 * * *", critical=False), + # Сторожевое кольцо с чужим агентом. Кроны читаются в локальной зоне, а + # сосед печатает свою отметку в 05:00 UTC - вечер по Варшаве разводит нас + # на полсуток, и его строка лежит в комнате задолго до нашего прогона. + # Сторож идёт через полчаса после своей отметки: так «моя запись в комнату + # проходит» - свежий факт этих же суток, а не вчерашний. + Job("отметка", beat, cron="12 17 * * *", critical=False), + Job("сторож", watch, cron="42 17 * * *", critical=False), Job("память", memory, cron="30 4 * * 0", critical=False), Job("голосовой", rotate_voice, cron="5 4 * * *", critical=False), Job("куратор", curate, cron="0 2,8,12,16,20 * * *", critical=False), diff --git a/beaver_agent/jobs/heartbeat.py b/beaver_agent/jobs/heartbeat.py new file mode 100644 index 0000000..09b4fd7 --- /dev/null +++ b/beaver_agent/jobs/heartbeat.py @@ -0,0 +1,113 @@ +"""Сторожевое кольцо с чужим агентом: своя отметка и слежка за его молчанием. + +Два джоба, оба раз в сутки и оба в отдельной комнате вайбграма (второй +экземпляр клиента, `VIBEGRAM_HB`): + +- `beat` печатает `hb <ник> at=… boot=… every=… src=…`. Период не константа: + он выведен из крона ЭТОГО ЖЕ джоба (`run.job.cron`) - что зарегистрировано + у планировщика, то и написано в строке. Не сводится к «раз в N часов» - + пишем `every=unknown`, потому что соврать числом хуже, чем отказаться; +- `watch` читает окно комнаты, сверяет возраст последней отметки соседа с + порогом из ЕГО ЖЕ строки, печатает в комнату `ack` и - если цепь молчит - + отдаёт тревогу мастеру через `run.spawn_job`, как это делает триаж + вайбграма: будить ли владельца, решает диспетчер. + +Токена сторожевой комнаты нет - оба джоба молча ничего не делают, а о том, +что сторож не сконфигурирован, один раз сказано в логе при старте +(`beaver_agent/hands/__init__.py`). +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +from beaver_agent.hands import VIBEGRAM_HB +from beaver_agent.hands.heartbeat import ( + BOOT, + PEER, + WINDOW, + Watch, + alarm_text, + beat_line, + every_from_cron, + look, +) +from beaver_agent.vault import STATE_DIR + +if TYPE_CHECKING: + from beaver_gateway.jobs.scheduler import JobRun + +_log = logging.getLogger(__name__) + +STATE_FILE = STATE_DIR / "сторож.json" +"""Что видели в прошлый прогон: без файла сторож после передеплоя считает, +что соседа не видел ни разу, и разоружается сам по себе.""" + +TITLE = "сторож" + +BRIEF = ( + "Сторожевое кольцо с чужим агентом {peer} (взаимные отметки раз в сутки " + "в отдельной комнате вайбграма) подняло тревогу.\n" + "{body}\n" + 'Будить владельца - inject(conversation="master", urgency="wake", ' + "text=до 3 строк: что цепь молчит, сколько молчит и какие три причины). " + "Это не авария у нас: наша половина работает, молчит чужая. " + "Если inject вернул ошибку - это сбой доставки, а не «не срочно»: " + "повтори один раз с тем же текстом." +) + + +async def beat(run: JobRun) -> None: + """Отметка о том, что мы живы. Не ушла - джоб падает, это видно в прогонах.""" + if VIBEGRAM_HB is None: + return + line = beat_line( + nick=VIBEGRAM_HB.nick, + at=datetime.now(UTC), + boot=BOOT, + every=every_from_cron(run.job.cron, run.scheduler.tz), + ) + await VIBEGRAM_HB.say(line) + _log.info("сторож: отметка ушла - %s", line) + + +async def watch(run: JobRun) -> None: + """Прогон сторожа: окно → решение → `ack` в комнату → тревога, если молчит.""" + if VIBEGRAM_HB is None: + return + try: + events = await VIBEGRAM_HB.pending(limit=WINDOW) + except Exception: # noqa: BLE001 - любой отказ хаба это отсутствие показания + # Окно не прочитано - показания нет вовсе, и тревога тут запрещена + # ровно так же, как на съехавшем окне: молчание хаба это не молчание + # соседа. Говорим об этом в лог, состояние не трогаем. + _log.exception("сторож: комната не прочитана, прогон без показания") + return + state = Watch.load(STATE_FILE) + report = look( + state, + events, + now=datetime.now(UTC), + res=every_from_cron(run.job.cron, run.scheduler.tz), + ) + sent = True + try: + await VIBEGRAM_HB.say(report.ack) + except Exception as exc: # noqa: BLE001 - своя запись не прошла, это различитель + sent = False + _log.warning("сторож: свой ack в комнату не ушёл: %s", exc) + _log.info("сторож: %s | %s", report.ack, report.note) + if report.alarm is not None: + await run.spawn_job( + agent="beaver-triage", + title=TITLE, + text=BRIEF.format(peer=PEER, body=alarm_text(report, mine_sent=sent)), + ) + now = datetime.now(UTC) + if report.alarm == "silence": + state.alerted = now + else: + state.unarmed_told = True + state.save() diff --git a/scripts/vibegram_join.py b/scripts/vibegram_join.py new file mode 100755 index 0000000..b931b2e --- /dev/null +++ b/scripts/vibegram_join.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 + +"""Войти в комнату вайбграма по коду приглашения и получить токен агента. + +Обычный путь - CLI `vibegram join --nick <ник> --no-hooks` из клона +репы комнаты. Сторожевой комнате клон не нужен (в ней только строки `hb` и +`ack`), поэтому здесь тот же вызов хаба напрямую: `POST /api/rooms/join` +с `joinCode` и `nick`. Ответ печатается целиком - токен из него едет в +`.env` как `VIBEGRAM_HB_TOKEN`, после чего gateway надо перезапустить. + + python3 scripts/vibegram_join.py --nick <ник> + python3 scripts/vibegram_join.py --hub https://vibegram.example + +Хаб по умолчанию берётся из `VIBEGRAM_HUB` в окружении или в соседнем `.env`. +Токен печатается в терминал - не вставляй вывод в чаты и в задачи. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import urllib.error +import urllib.request +from pathlib import Path + +ENV = Path(__file__).resolve().parent.parent / ".env" + + +def hub_from_env() -> str | None: + if os.environ.get("VIBEGRAM_HUB"): + return os.environ["VIBEGRAM_HUB"] + if not ENV.exists(): + return None + for line in ENV.read_text(encoding="utf-8").splitlines(): + key, _, value = line.partition("=") + if key.strip() == "VIBEGRAM_HUB" and value.strip(): + return value.strip() + return None + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("code", help="код приглашения из комнаты") + parser.add_argument("--nick", help="ник агента в этой комнате") + parser.add_argument("--hub", default=hub_from_env(), help="адрес хаба вайбграма") + args = parser.parse_args() + if not args.hub: + print("не знаю адрес хаба: задай --hub или VIBEGRAM_HUB", file=sys.stderr) + return 2 + + body: dict[str, str] = {"joinCode": args.code} + if args.nick: + body["nick"] = args.nick + request = urllib.request.Request( # noqa: S310 - адрес хаба задаёт оператор + f"{args.hub.rstrip('/')}/api/rooms/join", + data=json.dumps(body).encode(), + headers={"content-type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=20) as response: # noqa: S310 + data = json.loads(response.read()) + except urllib.error.HTTPError as exc: + print(f"хаб отказал: {exc.code} {exc.read().decode()[:400]}", file=sys.stderr) + return 1 + except OSError as exc: + print(f"хаб не отвечает: {exc}", file=sys.stderr) + return 1 + + print(json.dumps(data, ensure_ascii=False, indent=2)) + token = next( + (data[k] for k in ("token", "agentToken", "apiToken") if data.get(k)), None + ) + if token: + print(f"\nVIBEGRAM_HB_TOKEN={token}\n\nдальше: строку в .env, затем рестарт") + else: + print("\nтокена в ответе не видно - смотри поля выше", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_heartbeat.py b/tests/test_heartbeat.py new file mode 100644 index 0000000..2caa3ed --- /dev/null +++ b/tests/test_heartbeat.py @@ -0,0 +1,322 @@ +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from typing import Any, cast +from zoneinfo import ZoneInfo + +from beaver_agent.hands.heartbeat import ( + PEER, + Watch, + alarm_text, + beat_line, + every_from_cron, + look, + parse_beat, + parse_span, + span, +) +from beaver_agent.hands.vibegram import Event + +WARSAW = ZoneInfo("Europe/Warsaw") +NOW = datetime(2026, 9, 22, 15, 42, tzinfo=UTC) + + +def hb(at, *, every="24h", nick=PEER, eid=100): + return Event( + id=eid, + nick=nick, + kind="message", + text=( + f"{nick}: hb {nick} at={at.strftime('%Y-%m-%dT%H:%M:%SZ')} " + f"boot=2026-09-01T03:00:00Z every={every} src=cron+uptime-s" + ), + at="09-22 07:00", + ) + + +def state(tmp_path, **kw): + return Watch(path=tmp_path / "сторож.json", **kw) + + +def test_parse_beat_reads_fields_and_unknown_every(): + beat = parse_beat( + "claude-kai: hb claude-kai at=2026-09-22T05:00:00Z " + "boot=2026-09-01T03:00:00Z every=24h src=cron+uptime-s" + ) + assert beat is not None + assert beat.nick == "claude-kai" + assert beat.at == datetime(2026, 9, 22, 5, 0, tzinfo=UTC) + assert beat.boot == datetime(2026, 9, 1, 3, 0, tzinfo=UTC) + assert beat.every == timedelta(hours=24) + + unknown = parse_beat( + "claude-kai: hb claude-kai at=2026-09-22T05:00:00Z " + "boot=? every=unknown src=none" + ) + assert unknown is not None + assert unknown.every is None and unknown.every_raw == "unknown" + assert unknown.boot is None + + assert parse_beat("claude-kai: просто сообщение") is None + assert parse_beat("claude-kai: ack beaver last=x", "ack") is not None + + +def test_span_round_trip(): + assert span(timedelta(hours=36)) == "36h" + assert span(timedelta(minutes=90)) == "90m" + assert span(None) == "unknown" + assert parse_span("36h") == timedelta(hours=36) + assert parse_span("unknown") is None + assert parse_span("") is None + + +def test_every_from_cron_only_when_schedule_is_every_n_hours(): + assert every_from_cron("12 17 * * *", WARSAW, now=NOW) == timedelta(hours=24) + assert every_from_cron("0 */6 * * *", WARSAW, now=NOW) == timedelta(hours=6) + # Перевод стрелок не делает суточный крон 23-часовым: считаем по стенным часам + march = datetime(2026, 3, 28, 12, tzinfo=UTC) + assert every_from_cron("12 17 * * *", WARSAW, now=march) == timedelta(hours=24) + # Не сводится к «раз в N часов» - и не сводим + assert every_from_cron("0 2,8,12,16,20 * * *", WARSAW, now=NOW) is None + assert every_from_cron("*/10 * * * *", WARSAW, now=NOW) is None + # Недельный крон тоже сводится к N часам - это честные 168h, а не unknown + assert every_from_cron("30 4 * * 0", WARSAW, now=NOW) == timedelta(hours=168) + assert every_from_cron(None, WARSAW, now=NOW) is None + assert every_from_cron("не крон", WARSAW, now=NOW) is None + + +def test_beat_line_says_unknown_instead_of_guessing(): + line = beat_line( + nick="beaver", + at=datetime(2026, 9, 22, 15, 12, tzinfo=UTC), + boot=datetime(2026, 9, 20, 1, 2, 3, tzinfo=UTC), + every=timedelta(hours=24), + ) + assert line == ( + "hb beaver at=2026-09-22T15:12:00Z boot=2026-09-20T01:02:03Z every=24h src=cron" + ) + blind = beat_line(nick="beaver", at=NOW, boot=NOW, every=None) + assert "every=unknown src=none" in blind + + +def test_fresh_beat_arms_and_stays_quiet(tmp_path): + w = state(tmp_path) + report = look(w, [hb(NOW - timedelta(hours=10))], now=NOW, res=timedelta(hours=24)) + assert report.armed and report.alarm is None and not report.stale + assert report.threshold == timedelta(hours=36) + assert "armed=yes" in report.ack and "win=25" in report.ack + assert "res=24h" in report.ack and "thr=36h" in report.ack + assert w.seen == NOW and w.top == 100 + + +def test_threshold_is_every_times_one_and_a_half(tmp_path): + # 30 часов при every=24h - ещё не тревога: одна потерянная строка не в счёт + quiet = state(tmp_path, seen=NOW - timedelta(hours=30), every=timedelta(hours=24)) + assert look(quiet, [], now=NOW, res=timedelta(hours=24)).alarm is None + # 37 часов - уже за порогом 36 + loud = state(tmp_path, seen=NOW - timedelta(hours=37), every=timedelta(hours=24)) + report = look(loud, [], now=NOW, res=timedelta(hours=24)) + assert report.alarm == "silence" + # Порог берётся из строки соседа, а не из нашей константы + slow = state(tmp_path, seen=NOW - timedelta(hours=37), every=timedelta(hours=48)) + assert look(slow, [], now=NOW, res=timedelta(hours=24)).alarm is None + + +def test_last_beat_wins_over_gaps(tmp_path): + w = state(tmp_path, seen=NOW - timedelta(hours=40), every=timedelta(hours=24)) + report = look(w, [hb(NOW - timedelta(hours=1), eid=120)], now=NOW, res=None) + assert report.alarm is None + assert report.age is not None and report.age < timedelta(hours=2) + assert "res=unknown" in report.ack + + +def test_unknown_every_falls_back_to_last_known_and_says_so(tmp_path): + w = state(tmp_path, every=timedelta(hours=24)) + report = look(w, [hb(NOW, every="unknown", eid=130)], now=NOW, res=None) + assert report.threshold == timedelta(hours=36) + assert "thr=36h-fallback" in report.ack + assert w.every == timedelta(hours=24) + + +def test_no_period_ever_means_no_measurement_and_no_alarm(tmp_path): + w = state(tmp_path, seen=NOW - timedelta(days=9)) + report = look(w, [], now=NOW, res=timedelta(hours=24)) + assert report.alarm is None and report.threshold is None + assert "thr=unknown" in report.ack + assert "мерить нечем" in report.note + + +def test_never_seen_is_armed_no_not_alarm(tmp_path): + w = state(tmp_path) + report = look(w, [], now=NOW, res=timedelta(hours=24)) + assert not report.armed and report.alarm is None + assert "armed=no" in report.ack and "last=never seen=never" in report.ack + assert w.unarmed_since == NOW + + +def test_seven_days_unarmed_tells_the_human_once(tmp_path): + w = state(tmp_path, unarmed_since=NOW - timedelta(days=7)) + assert look(w, [], now=NOW, res=timedelta(hours=24)).alarm == "unarmed" + w.unarmed_told = True + assert look(w, [], now=NOW, res=timedelta(hours=24)).alarm is None + # Пришла первая отметка - кольцо вооружилось, счётчик забыт + look(w, [hb(NOW, eid=140)], now=NOW, res=timedelta(hours=24)) + assert w.unarmed_since is None + + +def test_stale_window_forbids_alarm(tmp_path): + w = state( + tmp_path, seen=NOW - timedelta(hours=40), every=timedelta(hours=24), top=500 + ) + report = look(w, [hb(NOW - timedelta(hours=40), eid=7)], now=NOW, res=None) + assert report.stale and report.alarm is None + assert "win=stale" in report.ack + assert w.top == 500 # съехавшее окно не двигает отметку максимума + # Окно снова вперёд - тревога берётся + fresh = look(w, [hb(NOW - timedelta(hours=40), eid=501)], now=NOW, res=None) + assert not fresh.stale + + +def test_alarm_not_repeated_within_a_day(tmp_path): + w = state( + tmp_path, + seen=NOW - timedelta(hours=40), + every=timedelta(hours=24), + alerted=NOW - timedelta(hours=5), + ) + assert look(w, [], now=NOW, res=None).alarm is None + w.alerted = NOW - timedelta(hours=25) + assert look(w, [], now=NOW, res=None).alarm == "silence" + + +def test_alarm_text_names_three_causes_and_suspects_the_channel(tmp_path): + w = state(tmp_path, seen=NOW - timedelta(hours=40), every=timedelta(hours=24)) + report = look(w, [], now=NOW, res=timedelta(hours=24)) + silent = alarm_text(report, mine_sent=True) + assert "машина соседа" in silent and "крон" in silent and "канал" in silent + assert "канал подозревать надо раньше" in silent + # Сосед хоть чем-то отметился - канал скорее жив + heard = look( + state(tmp_path, seen=NOW - timedelta(hours=40), every=timedelta(hours=24)), + [ + Event( + id=9, + nick=PEER, + kind="message", + text=f"{PEER}: ack beaver last=x", + at="09-22 07:00", + ) + ], + now=NOW, + res=None, + ) + assert heard.heard + assert "канал скорее жив" in alarm_text(heard, mine_sent=True) + + +def test_state_survives_a_round_trip(tmp_path): + w = state( + tmp_path, + at=NOW, + seen=NOW, + every=timedelta(hours=24), + every_unknown=True, + top=42, + alerted=NOW, + unarmed_since=None, + unarmed_told=True, + ) + w.save() + back = Watch.load(w.path) + assert back.at == NOW and back.seen == NOW and back.top == 42 + assert back.every == timedelta(hours=24) and back.every_unknown + assert back.alerted == NOW and back.unarmed_told + assert Watch.load(tmp_path / "нет.json").seen is None + + +class FakeRoom: + def __init__(self, events, *, nick="beaver", send_fails=False): + self.events = events + self.nick = nick + self.send_fails = send_fails + self.said = [] + self.limit = None + + async def pending(self, limit=50): + self.limit = limit + return self.events + + async def say(self, text): + if self.send_fails: + msg = "хаб молчит" + raise RuntimeError(msg) + self.said.append(text) + + +class FakeRun: + def __init__(self, cron="42 17 * * *"): + self.job = SimpleNamespace(cron=cron, name="сторож") + self.scheduler = SimpleNamespace(tz=WARSAW) + self.spawned = [] + + async def spawn_job(self, *, agent, text, title=None): + self.spawned.append((agent, title, text)) + return None + + +async def test_watch_job_prints_ack_and_spawns_alarm(tmp_path, monkeypatch): + from beaver_agent.jobs import heartbeat as job + + monkeypatch.setattr(job, "STATE_FILE", tmp_path / "сторож.json") + Watch( + path=tmp_path / "сторож.json", + seen=datetime.now(UTC) - timedelta(hours=40), + every=timedelta(hours=24), + ).save() + room = FakeRoom([]) + monkeypatch.setattr(job, "VIBEGRAM_HB", room) + run = FakeRun() + + await job.watch(cast("Any", run)) + + assert room.limit == 25 # окно запрошено явным размером + assert room.said and room.said[0].startswith(f"ack {PEER} ") + assert "res=24h" in room.said[0] + assert len(run.spawned) == 1 + agent, title, text = run.spawned[0] + assert agent == "beaver-triage" and title == "сторож" + assert "канал подозревать надо раньше" in text + # Тревога записана в состояние - следующий прогон в те же сутки молчит + assert Watch.load(tmp_path / "сторож.json").alerted is not None + await job.watch(cast("Any", run)) + assert len(run.spawned) == 1 + + +async def test_beat_job_sends_period_from_its_own_cron(monkeypatch): + from beaver_agent.jobs import heartbeat as job + + room = FakeRoom([], nick="beaver") + monkeypatch.setattr(job, "VIBEGRAM_HB", room) + await job.beat(cast("Any", FakeRun(cron="12 17 * * *"))) + assert room.said[0].startswith("hb beaver at=") + assert "every=24h src=cron" in room.said[0] + + odd = FakeRoom([], nick="beaver") + monkeypatch.setattr(job, "VIBEGRAM_HB", odd) + await job.beat(cast("Any", FakeRun(cron="0 2,8,12,16,20 * * *"))) + assert "every=unknown src=none" in odd.said[0] + + +async def test_watch_job_keeps_quiet_when_the_room_is_unreadable(tmp_path, monkeypatch): + from beaver_agent.jobs import heartbeat as job + + class Dead(FakeRoom): + async def pending(self, limit=50): + msg = "хаб не отвечает" + raise RuntimeError(msg) + + monkeypatch.setattr(job, "STATE_FILE", tmp_path / "сторож.json") + monkeypatch.setattr(job, "VIBEGRAM_HB", Dead([])) + run = FakeRun() + await job.watch(cast("Any", run)) + assert not run.spawned + assert not (tmp_path / "сторож.json").exists()