Files
beaver-agent/beaver_agent/hands/heartbeat.py
T

502 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Взаимный сторож с чужим агентом: отметка раз в сутки и тревога на молчание.
Договорённость с соседом (22.09.2026): в отдельной комнате вайбграма оба
раз в сутки печатают строку `hb <ник> at=… boot=… every=… src=…`, и каждый
следит за молчанием другого. Замолчал дольше порога - человеку соседа
уходит сообщение, что цепь молчит.
Смысл сторожа в том, чтобы он не врал: прибор, который молча не сторожит,
хуже отсутствующего - он выглядит исправным. Поэтому дефолтов здесь нет
нигде. Период своей отметки считается из фактического крона джоба
(`every_from_cron`), и расписание, которое не сводится к «раз в N часов»,
даёт честное `every=unknown`, а не подставленное число. Порог тревоги -
`every * 1.5`, где `every` взят ИЗ СТРОКИ СОСЕДА, а не из своей константы.
Каждый прогон, в котором показание снять нельзя (окно смотрит в прошлое,
окно не дочитано, период неизвестен, отметок не видели ни разу), говорит
это вслух в `ack` и в лог: «в окне никого» и «окно не прочитано до конца» -
это два РАЗНЫХ показания, и выглядеть одинаково они не должны.
Окно - последние `WINDOW` СООБЩЕНИЙ этой комнаты (`Vibegram.window`), а не
«что нового с курсора»: курсор отдаёт строку соседа один раз, и следующий
прогон на живой комнате увидел бы пустое окно.
Опора тревоги - `seen`, то есть когда Я ВПЕРВЫЕ увидел строку: `at=` внутри
строки это показание соседа о себе, и остановившиеся у него часы не должны
выглядеть как живая цепь. Окно - это последние сообщения комнаты, поэтому
одна и та же отметка видна в нём много прогонов подряд, и новой её делает
новый id сообщения, а не текст: двигать `seen` на старой отметке значит
никогда не поднять тревогу в тихой комнате.
"""
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`: порог взят с запасом назад."""
beat_id: int = 0
"""id сообщения с последней зачтённой отметкой: по нему видно, что отметка
в окне новая, а не та же. Нуль - поля в файле не было (состояние старое)
или отметок не видели ни разу."""
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")),
beat_id=int(raw.get("beat_id", 0)),
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,
"beat_id": self.beat_id,
"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 _is_new_beat(state: Watch, eid: int, beat: Beat) -> bool:
"""Впервые ли мы видим ЭТУ отметку соседа.
Различитель - id сообщения: `at=` внутри строки это показание соседа о
себе, и повторить его он может как угодно. Нуля в `beat_id` два повода:
отметок не видели ни разу (тогда любая - новая) и состояние записано до
появления поля - там единственное, чем можно сверить отметку со зачтённой,
это её `at`, и одного прогона хватает, чтобы запомнить id.
"""
if state.beat_id:
return eid > state.beat_id
if state.seen is None:
return True
return beat.at != state.at
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 = [
(event.id, beat)
for event, beat in ((event, parse_beat(event.text)) for event 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)
again = False
if beats:
eid, last = beats[-1]
again = not _is_new_beat(state, eid, last)
if not again:
state.seen = now
state.at, state.beat_id = last.at, eid
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})"
)
if again:
note += "; отметка та же, что в прошлый прогон - seen не двигаем"
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}, а не наше: разбудить владельца имеет смысл "
"только чтобы он написал соседу по своему каналу. Повторно об этом "
"молчании мы скажем не раньше чем через сутки."
)