diff --git a/README.md b/README.md index 34aec89..b241da8 100644 --- a/README.md +++ b/README.md @@ -268,6 +268,58 @@ 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`. +## The job-run watchdog (`beaver_agent/jobs/runs.py`) + +The scheduler has always written one row per run into `job_runs` - name, +status, the head of the traceback in `error`. Nobody ever read it. The channel +to the neighbour agent was dead for 38 hours straight, the scheduler recorded +that 231 times, and nothing reached a human: the outage surfaced ten days later, +by hand, out of curiosity. The record existed, the reader did not. This job is +the reader. + +- `прогоны` (`:27` every hour) reads `job_runs` for the last 24 hours through + the same connection the scheduler itself writes with + (`run.conversations.db.session()` and the `JobRunRecord` model) - no second + truth about where the journal lives. +- An alarm is taken only off the **tail**: three or more `failed` in a row + counting back from a job's latest run, i.e. the job is *lying down* rather + than having blinked once. A single failure is a network hiccup, and waking + a human for it teaches them not to look. +- The alarm says the job's name, how many runs in a row died, when the series + started, and the **first line** of the error - never the traceback. Replayed + over the real outage it reads: «джоб `вайбграм` лежит 23 часа и дольше, + 144 прогона подряд с 12.09 22:00 UTC: HubError: vibegram 530 bad_json». It repeats at most once a day per job, and + when the job recovers one short `отбой` line goes out and the counter resets. +- **An empty answer is a statement about the query, not about health.** Zero + rows in 24 hours (other than this job's own) means no job ran at all, and + that is its own alarm with its own text - the scheduler is what to look at, + not any one job. A failed query is no reading whatsoever: the run logs it and + stays silent, because a database that did not answer is not evidence of + health or of illness. +- Everything that goes out is masked (`mask`): the gateway's shared `redact` + plus two local rules for what it cannot catch by construction - a + `NAME=value` pair in the middle of a line (a traceback puts them there) and + the query string of any URL, which is dropped whole because the parameter + carrying the key can be named anything. + +State between runs is `мета/бобер/.состояние/прогоны.json` (who was told and +when). A lost file means a lying job is reported again and an `отбой` for +something reported before the loss never comes - the evidence for it is gone. + +Two blind spots, both deliberate: + +- **The watchdog cannot report its own death.** Its own runs are excluded from + the count (`run.job.name`), because a job that is down will not file a report + about itself - an observer inside the object does not report the object's + death. If `прогоны` itself starts failing, nothing here will say so; that + needs an outside eye, like the watchdog ring above. +- **The window is 24 hours, so a job that runs less than three times a day can + never accumulate three consecutive failures inside it.** `отметка`, + `закрытие`, `голосовой` (daily) and `память` (weekly) are therefore + unwatched by this job in practice; `вайбграм`, `логи`, `куратор`, `досуг`, + `ротация` and `сторож` are inside it. Widening the counting window past the + query window is the fix, when it is worth it. + ## Policy (`beaver_agent/policy.py`) `bypassPermissions` everywhere; the boundary is `PreToolUse` rules declared diff --git a/beaver_agent/hands/runs.py b/beaver_agent/hands/runs.py new file mode 100644 index 0000000..37eafc1 --- /dev/null +++ b/beaver_agent/hands/runs.py @@ -0,0 +1,395 @@ +"""Сторож собственных прогонов: журнал `job_runs` наконец кто-то читает. + +Планировщик gateway пишет строку на каждый прогон джоба (`job_runs`: имя, +статус, хвост трейсбека в `error`). Запись была всегда, читателя не было +ни одного: канал к соседнему агенту лежал 38 часов подряд, планировщик +честно записал это 231 раз, и до человека не доехало ничего. Этот модуль - +вторая половина прибора: показание из таблицы превращается в тревогу. + +Что здесь считается показанием: + +- тревога берётся только с ХВОСТА: три и более `failed` подряд, считая от + последнего прогона джоба. Одиночный упавший прогон - сетевой чих, а не + болезнь, и будить на него человека значит выучить его не смотреть; +- пустая выдача - это показание о запросе, а не о здоровье джобов. Ноль + строк за сутки (кроме своих) означает, что планировщик вообще не + работал, и это отдельная тревога с отдельным текстом, а не тишина; +- упавший запрос к БД - не показание вовсе: решает джоб, здесь такого + входа просто нет. + +Свои собственные прогоны из счёта выкинуты (`me`): джоб, который лежит, +о себе не доложит - наблюдатель внутри объекта не сообщает о смерти +объекта. Это дыра прибора, и она названа в README, а не заклеена. + +Текст тревоги проходит `mask`: из `error` берётся первая строка, режется +по длине и чистится - общим `redact` gateway плюс два своих правила на то, +что `redact` по конструкции не ловит внутри строки (`ИМЯ=значение` не с +начала строки и хвост запроса у URL). Тревога уезжает в чат к диспетчеру, +логи гейтвея видны в чатах - секрету тут негде быть. +""" + +from __future__ import annotations + +import itertools +import json +import re +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta +from pathlib import Path # noqa: TC003 - в аннотации dataclass +from typing import TYPE_CHECKING + +from beaver_gateway.security.redact import MASK, is_secret_name, redact + +if TYPE_CHECKING: + from collections.abc import Sequence + +__all__ = [ + "ERROR_CHARS", + "FAILED", + "FORGET", + "MAX_LINES", + "REPEAT", + "STREAK", + "WINDOW", + "Clear", + "Report", + "Run", + "Runs", + "Streak", + "alarm_text", + "clear_text", + "empty_text", + "first_line", + "look", + "mask", + "streak", +] + +WINDOW = timedelta(hours=24) +"""Окно журнала: сутки. Джоб реже трёх раз в сутки за него не наберёт серии +из трёх падений - это слепая зона, названная в README.""" +STREAK = 3 +"""Сколько `failed` подряд считаем «джоб лежит».""" +REPEAT = timedelta(hours=24) +"""Чаще этого об одном и том же джобе не говорим.""" +FORGET = timedelta(days=7) +"""Джоб, о котором тревожились, исчез из окна насовсем - забываем его молча: +починки мы не видели, врать «отбой» не о чем.""" +ERROR_CHARS = 160 +"""Сколько букв первой строки ошибки уезжает человеку.""" +MAX_LINES = 10 +"""Сколько джобов перечисляем в одной тревоге: сразу всё не читают.""" +FAILED = "failed" +STATE_VERSION = 1 + +_PAIR = re.compile(r"(?=([A-Za-z][\w.-]{2,})\s*([:=])\s*([^\s,;)\]}\"']{6,}))") +"""`ИМЯ=значение` где угодно в строке: общий `redact` ловит такие пары только +с начала строки, а трейсбек кладёт их в середину. Просмотр нулевой длины - +чтобы `RuntimeError: MSOS_MCP=…` не съедало секрет первой же парой: обычный +`sub` откусил бы `RuntimeError:` вместе со значением и до вложенной пары +не дошёл.""" +_QUERY = re.compile(r"(https?://[^\s?#]+)\?[^\s#]*") +"""Хвост запроса у URL сносим целиком: имя параметра с ключом может быть любым.""" + + +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 _plural(count: int, forms: tuple[str, str, str]) -> str: + if count % 10 == 1 and count % 100 != 11: + return forms[0] + if count % 10 in (2, 3, 4) and count % 100 not in (12, 13, 14): + return forms[1] + return forms[2] + + +def _age(delta: timedelta) -> str: + """Возраст человеку: «4 часа», «45 минут». Секунды тут не нужны никому.""" + minutes = max(int(delta.total_seconds()) // 60, 0) + if minutes < 60: + return f"{minutes} {_plural(minutes, ('минуту', 'минуты', 'минут'))}" + hours = minutes // 60 + return f"{hours} {_plural(hours, ('час', 'часа', 'часов'))}" + + +def _mask_pairs(text: str) -> str: + out: list[str] = [] + pos = 0 + for match in _PAIR.finditer(text): + if match.start(3) < pos or not is_secret_name(match[1]): + continue + out.append(text[pos : match.start(3)]) + out.append(MASK) + pos = match.end(3) + out.append(text[pos:]) + return "".join(out) + + +def mask(text: str) -> str: + """Убрать из строки всё, что похоже на секрет, прежде чем её показывать.""" + return _QUERY.sub(r"\1?" + MASK, _mask_pairs(redact(text))) + + +def first_line(error: str | None) -> str: + """Первая строка ошибки - чистая и короткая. Трейсбек человеку не нужен.""" + for line in mask(error or "").splitlines(): + head = line.strip() + if not head: + continue + if len(head) > ERROR_CHARS: + head = head[: ERROR_CHARS - 1].rstrip() + "…" + return head + return "без текста ошибки" + + +@dataclass(frozen=True, slots=True) +class Run: + """Одна строка журнала: ровно то, на что мы смотрим, без ORM.""" + + id: int + job: str + status: str + started_at: datetime + error: str | None = None + + @property + def failed(self) -> bool: + return self.status == FAILED + + +@dataclass(frozen=True, slots=True) +class Streak: + """Хвост подряд упавших прогонов одного джоба.""" + + job: str + count: int + since: datetime + """Начало серии - первое падение из тех, что видно в окне.""" + age: timedelta + error: str + capped: bool + """Серия упёрлась в край окна: падений было НЕ МЕНЬШЕ `count`, а сколько + на самом деле - за сутками не видно, и врать точным числом нельзя.""" + + +@dataclass(frozen=True, slots=True) +class Clear: + """Джоб, о котором мы тревожились, больше не лежит.""" + + job: str + status: str + """Статус последнего прогона.""" + count: int + """Сколько подряд упавших осталось в хвосте: 0 - чисто.""" + + +@dataclass(slots=True) +class Runs: + """Состояние между прогонами: кому и когда уже сказали.""" + + path: Path + alerted: dict[str, datetime] = field(default_factory=dict) + """Джоб → когда о нём ушла тревога. Он же признак «сейчас в тревоге».""" + empty_alerted: datetime | None = None + + @classmethod + def load(cls, path: Path) -> Runs: + 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) + alerted = {} + for job, when in dict(raw.get("alerted") or {}).items(): + parsed = _parse_iso(str(when)) + if parsed is not None: + alerted[str(job)] = parsed + return cls( + path=path, + alerted=alerted, + empty_alerted=_parse_iso(raw.get("empty_alerted")), + ) + + def save(self) -> None: + body = { + "version": STATE_VERSION, + "alerted": {job: _iso(when) for job, when in sorted(self.alerted.items())}, + "empty_alerted": _iso(self.empty_alerted), + } + 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: + """Что решил один прогон сторожа журнала.""" + + alarms: tuple[Streak, ...] + clears: tuple[Clear, ...] + empty: bool + """Прогонов за сутки не было вовсе - тревога о планировщике, не о джобе.""" + seen: int + """Сколько чужих прогонов было в окне.""" + mine: int + """Сколько своих - их в счёт не берём, но в тексте называем.""" + note: str + + +def streak(runs: Sequence[Run], *, now: datetime) -> Streak | None: + """Хвост падений одного джоба, считая от последнего прогона. + + `runs` - прогоны одного джоба, новейший первым. Серия, упёршаяся в край + окна, помечается `capped`: её настоящую длину мы не знаем. + """ + failures = list(itertools.takewhile(lambda run: run.failed, runs)) + if not failures: + return None + oldest = failures[-1] + return Streak( + job=failures[0].job, + count=len(failures), + since=oldest.started_at, + age=now - oldest.started_at, + error=first_line(failures[0].error), + capped=len(failures) == len(runs), + ) + + +def look(state: Runs, runs: Sequence[Run], *, now: datetime, me: str) -> Report: + """Разобрать окно журнала и решить, о чём говорить человеку. + + Состояние здесь только чистится от забытых джобов; отметки «сказали» и + «сказали про пустоту» ставит джоб - и только когда тревога действительно + ушла: несостоявшаяся доставка не должна заглушать следующий прогон. + """ + mine = sum(1 for run in runs if run.job == me) + others = sorted( + (run for run in runs if run.job != me), key=lambda run: run.id, reverse=True + ) + if not others: + quiet = state.empty_alerted is not None and now - state.empty_alerted < REPEAT + return Report( + alarms=(), + clears=(), + empty=not quiet, + seen=0, + mine=mine, + note=( + f"чужих прогонов за сутки нет вовсе (своих {mine})" + + ("; о пустоте уже говорили меньше суток назад" if quiet else "") + ), + ) + + by_job: dict[str, list[Run]] = {} + for run in others: + by_job.setdefault(run.job, []).append(run) + + alarms: list[Streak] = [] + clears: list[Clear] = [] + held: list[str] = [] + for job, rows in by_job.items(): + tail = streak(rows, now=now) + if tail is not None and tail.count >= STREAK: + told = state.alerted.get(job) + if told is None or now - told >= REPEAT: + alarms.append(tail) + else: + held.append(job) + elif job in state.alerted: + clears.append( + Clear( + job=job, + status=rows[0].status, + count=0 if tail is None else tail.count, + ) + ) + + forgotten = [ + job + for job, told in state.alerted.items() + if job not in by_job and now - told >= FORGET + ] + for job in forgotten: + del state.alerted[job] + + note = f"прогонов за сутки {len(others)} по {len(by_job)} джобам (своих {mine})" + if alarms: + note += "; лежат: " + ", ".join(sorted(tail.job for tail in alarms)) + if held: + note += "; уже говорили меньше суток назад: " + ", ".join(sorted(held)) + if clears: + note += "; починились: " + ", ".join(sorted(clear.job for clear in clears)) + if forgotten: + note += "; забыты (пропали из окна): " + ", ".join(sorted(forgotten)) + + return Report( + alarms=tuple(sorted(alarms, key=lambda tail: tail.count, reverse=True)), + clears=tuple(sorted(clears, key=lambda clear: clear.job)), + empty=False, + seen=len(others), + mine=mine, + note=note, + ) + + +def streak_line(tail: Streak) -> str: + """«джоб `вайбграм` лежит 4 часа, 24 прогона подряд с 12.09 18:00 UTC: …».""" + runs = f"{tail.count} {_plural(tail.count, ('прогон', 'прогона', 'прогонов'))}" + return ( + f"джоб `{tail.job}` лежит {_age(tail.age)}" + f"{' и дольше' if tail.capped else ''}, {runs} подряд с " + f"{tail.since.astimezone(UTC).strftime('%d.%m %H:%M UTC')}" + + (" (это всё окно суток, серия началась раньше)" if tail.capped else "") + + f": {tail.error}" + ) + + +def alarm_text(report: Report) -> str: + """Тревога человеку: что лежит, сколько и с какой ошибкой.""" + lines = [f"- {streak_line(tail)}" for tail in report.alarms[:MAX_LINES]] + rest = len(report.alarms) - len(lines) + if rest > 0: + lines.append(f"- и ещё {rest} {_plural(rest, ('джоб', 'джоба', 'джобов'))}") + return "\n".join(lines) + + +def clear_line(clear: Clear) -> str: + if clear.count == 0: + return f"отбой: джоб `{clear.job}` снова проходит ({clear.status})" + return ( + f"отбой: джоб `{clear.job}` больше не лежит подряд " + f"(упавших в хвосте {clear.count}, порог {STREAK})" + ) + + +def clear_text(report: Report) -> str: + return "\n".join(f"- {clear_line(clear)}" for clear in report.clears) + + +def empty_text(report: Report) -> str: + """Отдельная тревога: за сутки не отработал вообще никто.""" + return ( + f"За последние сутки в журнале прогонов (`job_runs`) нет ни одной записи, " + f"кроме моих собственных (их {report.mine}): не отработал ни один джоб. " + "Это показание о планировщике, а не о каком-то одном джобе. Смотреть " + "стоит три вещи: поднялся ли планировщик вообще (в логе gateway " + "`scheduler: no postgres driver - cron and webhooks are off`), живы ли " + "кроны в pgqueuer, и не заведена ли база заново - на пустой базе эта же " + "тревога выглядит так же." + ) diff --git a/beaver_agent/jobs/__init__.py b/beaver_agent/jobs/__init__.py index 72faf7d..fcb7ab9 100644 --- a/beaver_agent/jobs/__init__.py +++ b/beaver_agent/jobs/__init__.py @@ -14,6 +14,7 @@ from beaver_agent.jobs.leisure import tick from beaver_agent.jobs.logs import logs_watch from beaver_agent.jobs.memory import curate, memory from beaver_agent.jobs.rotation import rotate +from beaver_agent.jobs.runs import watch_runs from beaver_agent.jobs.t3code import t3code_event from beaver_agent.jobs.vibegram import vibegram from beaver_agent.jobs.voice import rotate_voice @@ -37,6 +38,13 @@ jobs = [ # ради чего заведён; чтение 25 событий стоит копейки. Job("отметка", beat, cron="12 17 * * *", critical=False), Job("сторож", watch, cron="42 */6 * * *", critical=False), + # Сторож собственного журнала прогонов. Час - это компромисс: окно счёта + # сутки, и джоб, упавший три раза подряд, лежит уже заметно дольше часа, + # так что чаще смысла нет, а реже - тревога стареет на свой же период. + # ":27" выбрана пустой: на ":00" ротация и куратор, на ":40" логи, на + # ":42" сторож соседа, а собственный прогон этого джоба не должен ждать + # очереди за чужим спавном. + Job("прогоны", watch_runs, cron="27 * * * *", 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/runs.py b/beaver_agent/jobs/runs.py new file mode 100644 index 0000000..5ed231f --- /dev/null +++ b/beaver_agent/jobs/runs.py @@ -0,0 +1,146 @@ +"""Сторож собственных прогонов: раз в час читает `job_runs` и будит человека. + +Подключение к базе - то же самое, которым пишет сам планировщик +(`run.conversations.db.session()` и модель `JobRunRecord`): своей строки +подключения у джоба нет, и второй правды о том, где журнал, тоже. + +Устроен как сторож соседа (`beaver_agent/jobs/heartbeat.py`): состояние +файлом в `мета/бобер/.состояние/`, тревога - спавном джоба к триажу, и +запись «сказали» ставится только после того, как тревога ушла. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +from beaver_gateway.storage.models import JobRunRecord +from sqlmodel import col, select + +from beaver_agent.hands.runs import ( + STREAK, + WINDOW, + Report, + Run, + Runs, + alarm_text, + clear_text, + empty_text, + 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 = "прогоны" + +LIMIT = 5000 +"""Потолок строк за окно. Сутки нормальной работы - это сотни строк; упереться +в потолок можно только при шторме, и тогда серии считаются по новейшим, а +недостающий хвост честно виден как `capped` в тексте тревоги.""" + +BRIEF = ( + "Сторож собственных прогонов (таблица `job_runs`, окно сутки) нашёл " + "лежащие джобы: у каждого {streak} и больше падений подряд, считая от " + "последнего прогона.\n" + "{body}\n" + 'Будить владельца - inject(conversation="master", urgency="wake", ' + "text=до 3 строк: какой джоб лежит, сколько прогонов подряд и что в " + "ошибке). Это авария у НАС, а не у соседа: джоб пишется в журнал каждый " + "прогон, и раз он падает подряд, сам он не починится. Подробности - " + "в логах gateway. Если inject вернул ошибку - это сбой доставки, а не " + "«не срочно»: повтори один раз с тем же текстом." +) + +CLEAR_BRIEF = ( + "Сторож собственных прогонов: то, о чём он тревожился, починилось.\n" + "{body}\n" + 'Владельца НЕ будить: inject(conversation="master", urgency="normal", ' + "text=одна строка «отбой»). Человек знает про тревогу, ему нужно знать, " + "что она кончилась." +) + +EMPTY_BRIEF = ( + "Сторож собственных прогонов не нашёл в журнале НИЧЕГО.\n" + "{body}\n" + 'Будить владельца - inject(conversation="master", urgency="wake", ' + "text=до 3 строк: что за сутки не отработал ни один джоб и куда смотреть). " + "Молчащий планировщик - это молчание всех джобов разом, включая тех, что " + "носят человеку почту." +) + + +async def recent(run: JobRun, *, since: datetime) -> list[Run]: + """Прогоны всех джобов за окно, новейшие первыми. + + `started_at` в базе лежит наивным UTC (так его кладёт планировщик), и + сравнение в `where` тоже наивное - как это делает gateway в своих + запросах по времени. + """ + stmt = ( + select(JobRunRecord) + .where( + col(JobRunRecord.started_at) >= since.astimezone(UTC).replace(tzinfo=None) + ) + .order_by(col(JobRunRecord.id).desc()) + .limit(LIMIT) + ) + async with run.conversations.db.session() as session: + rows = (await session.exec(stmt)).all() + return [ + Run( + id=row.id or 0, + job=row.job, + status=row.status, + started_at=( + row.started_at + if row.started_at.tzinfo + else row.started_at.replace(tzinfo=UTC) + ), + error=row.error, + ) + for row in rows + ] + + +async def watch_runs(run: JobRun) -> None: + """Прогон сторожа: журнал → решение → тревога, «отбой» или молчание.""" + now = datetime.now(UTC) + try: + rows = await recent(run, since=now - WINDOW) + except Exception: # noqa: BLE001 - отказ запроса это отсутствие показания + # Журнал не прочитан - сказать о здоровье джобов нечего, и молчание + # базы выдавать за здоровье нельзя ровно так же, как за болезнь. + _log.exception("прогоны: журнал не прочитан, прогон без показания") + return + state = Runs.load(STATE_FILE) + report = look(state, rows, now=now, me=run.job.name) + _log.info("прогоны: %s", report.note) + await _tell(run, report, state, now=now) + state.save() + + +async def _tell(run: JobRun, report: Report, state: Runs, *, now: datetime) -> None: + if report.empty: + await _spawn(run, EMPTY_BRIEF.format(body=empty_text(report))) + state.empty_alerted = now + if report.alarms: + await _spawn(run, BRIEF.format(body=alarm_text(report), streak=STREAK)) + for tail in report.alarms: + state.alerted[tail.job] = now + if report.clears: + await _spawn(run, CLEAR_BRIEF.format(body=clear_text(report))) + for clear in report.clears: + state.alerted.pop(clear.job, None) + + +async def _spawn(run: JobRun, text: str) -> None: + await run.spawn_job(agent="beaver-triage", title=TITLE, text=text) diff --git a/tests/test_runs.py b/tests/test_runs.py new file mode 100644 index 0000000..0085c56 --- /dev/null +++ b/tests/test_runs.py @@ -0,0 +1,366 @@ +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from typing import Any, cast + +import pytest + +from beaver_agent.hands.runs import ( + STREAK, + Run, + Runs, + alarm_text, + clear_text, + empty_text, + first_line, + look, + mask, +) + +NOW = datetime(2026, 9, 22, 15, 42, tzinfo=UTC) +BOOM = ( + "HubError: vibegram 530 bad_json: \n" + "Traceback (most recent call last):\n" + ' File "/app/src/beaver_gateway/jobs/scheduler.py", line 170, in _dispatch\n' + "HubError: vibegram 530 bad_json" +) + + +def run(rid, job, status, *, ago, error=None): + return Run( + id=rid, + job=job, + status=status, + started_at=NOW - timedelta(minutes=ago), + error=error if error is not None or status != "failed" else BOOM, + ) + + +def series(job, statuses, *, step=10, start=1000): + """Прогоны джоба от старых к новым: statuses[-1] - последний прогон.""" + ago = step * len(statuses) + out = [] + for i, status in enumerate(statuses): + out.append(run(start + i, job, status, ago=ago - step * i)) + return out + + +def state(tmp_path, **kw): + return Runs(path=tmp_path / "прогоны.json", **kw) + + +def test_three_in_a_row_is_an_alarm(tmp_path): + rows = series("вайбграм", ["done", "failed", "failed", "failed"]) + report = look(state(tmp_path), rows, now=NOW, me="прогоны") + assert not report.empty and report.seen == 4 + (tail,) = report.alarms + assert tail.job == "вайбграм" and tail.count == 3 + assert tail.since == NOW - timedelta(minutes=30) + assert not tail.capped + text = alarm_text(report) + assert "джоб `вайбграм` лежит 30 минут, 3 прогона подряд" in text + assert "с 22.09 15:12 UTC" in text + # Первая строка ошибки, а не трейсбек + assert "HubError: vibegram 530 bad_json" in text + assert "Traceback" not in text and "scheduler.py" not in text + + +def test_two_in_a_row_stays_quiet(tmp_path): + rows = series("вайбграм", ["done", "failed", "failed"]) + report = look(state(tmp_path), rows, now=NOW, me="прогоны") + assert not report.alarms and not report.clears and not report.empty + # Одиночное падение - тем более молчание + once = look(state(tmp_path), series("логи", ["failed"]), now=NOW, me="прогоны") + assert not once.alarms + + +def test_only_the_tail_counts(tmp_path): + # Падало много, но последний прогон прошёл - джоб не лежит + rows = series("логи", ["failed", "failed", "failed", "failed", "done"]) + assert not look(state(tmp_path), rows, now=NOW, me="прогоны").alarms + + +def test_alarm_not_repeated_within_a_day(tmp_path): + rows = series("вайбграм", ["done", "failed", "failed", "failed", "failed"]) + told = state(tmp_path, alerted={"вайбграм": NOW - timedelta(hours=5)}) + quiet = look(told, rows, now=NOW, me="прогоны") + assert not quiet.alarms + assert "уже говорили меньше суток назад" in quiet.note + older = state(tmp_path, alerted={"вайбграм": NOW - timedelta(hours=25)}) + assert look(older, rows, now=NOW, me="прогоны").alarms + + +def test_repair_clears_and_resets(tmp_path): + rows = series("вайбграм", ["failed", "failed", "failed", "done"]) + told = state(tmp_path, alerted={"вайбграм": NOW - timedelta(hours=2)}) + report = look(told, rows, now=NOW, me="прогоны") + assert not report.alarms + (clear,) = report.clears + assert clear.job == "вайбграм" and clear.count == 0 and clear.status == "done" + assert "отбой: джоб `вайбграм` снова проходит (done)" in clear_text(report) + # Сброс делает джоб; после него тот же вход снова может поднять тревогу + del told.alerted["вайбграм"] + again = look(told, series("вайбграм", ["failed"] * 3), now=NOW, me="прогоны") + assert again.alarms + + +def test_repair_is_told_once_not_every_run(tmp_path): + rows = series("вайбграм", ["failed", "failed", "failed", "done"]) + clean = state(tmp_path) + assert not look(clean, rows, now=NOW, me="прогоны").clears + + +def test_flapping_job_is_a_clear_not_an_alarm(tmp_path): + # После тревоги джоб моргает: хвост короче порога - он больше не лежит + rows = series("вайбграм", ["done", "done", "failed", "failed"]) + told = state(tmp_path, alerted={"вайбграм": NOW - timedelta(hours=2)}) + report = look(told, rows, now=NOW, me="прогоны") + (clear,) = report.clears + assert clear.count == 2 + assert f"порог {STREAK}" in clear_text(report) + + +def test_streak_running_past_the_window_says_at_least(tmp_path): + rows = series("вайбграм", ["failed"] * 144, step=10) + report = look(state(tmp_path), rows, now=NOW, me="прогоны") + (tail,) = report.alarms + assert tail.capped and tail.count == 144 + text = alarm_text(report) + assert "лежит 24 часа и дольше, 144 прогона подряд" in text + assert "это всё окно суток, серия началась раньше" in text + + +def test_empty_window_is_its_own_alarm(tmp_path): + report = look(state(tmp_path), [], now=NOW, me="прогоны") + assert report.empty and not report.alarms and report.seen == 0 + assert "не отработал ни один джоб" in empty_text(report) + assert "планировщике" in empty_text(report) + # И повторяется не чаще раза в сутки + told = state(tmp_path, empty_alerted=NOW - timedelta(hours=5)) + assert not look(told, [], now=NOW, me="прогоны").empty + stale = state(tmp_path, empty_alerted=NOW - timedelta(hours=25)) + assert look(stale, [], now=NOW, me="прогоны").empty + + +def test_own_runs_do_not_count_as_life_or_as_illness(tmp_path): + mine = series("прогоны", ["failed", "failed", "failed", "done"]) + report = look(state(tmp_path), mine, now=NOW, me="прогоны") + # Свои прогоны в окне есть, чужих нет: это всё равно пустое окно + assert report.empty and not report.alarms and report.mine == 4 + assert "их 4" in empty_text(report) + + +def test_forgotten_job_leaves_the_state_silently(tmp_path): + rows = series("логи", ["done", "done"]) + told = state( + tmp_path, + alerted={ + "исчезнувший": NOW - timedelta(days=8), + "недавний": NOW - timedelta(hours=3), + }, + ) + report = look(told, rows, now=NOW, me="прогоны") + assert not report.clears # починки мы не видели, «отбой» врать не о чем + assert "исчезнувший" not in told.alerted + assert "недавний" in told.alerted + + +def test_several_broken_jobs_come_in_one_alarm(tmp_path): + rows = [ + *series("вайбграм", ["failed"] * 4, start=100), + *series("логи", ["failed"] * 3, start=200), + *series("досуг", ["done"], start=300), + ] + report = look(state(tmp_path), rows, now=NOW, me="прогоны") + assert len(report.alarms) == 2 + assert report.alarms[0].job == "вайбграм" # длиннее серия - выше в тексте + assert alarm_text(report).count("\n") == 1 + + +def test_first_line_masks_secrets_and_cuts_the_tail(): + assert first_line(None) == "без текста ошибки" + assert first_line("") == "без текста ошибки" + # Токен в строке ошибки не уезжает человеку + assert "s3cr3tvalue" not in first_line("RuntimeError: MSOS_MCP=s3cr3tvalue died") + assert "MSOS_MCP=<…>" in first_line("RuntimeError: MSOS_MCP=s3cr3tvalue died") + assert "hunter2" not in first_line("ошибка: password=hunter2") + assert "abc123" not in first_line("GET https://hub/api?t=abc123 -> 500") + assert "https://hub/api?<…>" in first_line("GET https://hub/api?t=abc123 -> 500") + assert "Bearer abcdefghij" not in first_line("Authorization: Bearer abcdefghij") + # Длинная строка режется + long = first_line("E: " + "x" * 500) + assert len(long) == 160 and long.endswith("…") + # Имя без запаха секрета остаётся читаемым + assert mask("status=530 job=вайбграм") == "status=530 job=вайбграм" + + +def test_state_survives_a_round_trip(tmp_path): + saved = state( + tmp_path, + alerted={"вайбграм": NOW, "логи": NOW - timedelta(hours=3)}, + empty_alerted=NOW, + ) + saved.save() + back = Runs.load(saved.path) + assert back.alerted["вайбграм"] == NOW + assert back.alerted["логи"] == NOW - timedelta(hours=3) + assert back.empty_alerted == NOW + assert Runs.load(tmp_path / "нет.json").alerted == {} + (tmp_path / "мусор.json").write_text("не json", encoding="utf-8") + assert Runs.load(tmp_path / "мусор.json").empty_alerted is None + + +class FakeRun: + def __init__(self, database=None, *, name="прогоны"): + self.job = SimpleNamespace(name=name, cron="27 * * * *") + self.conversations = SimpleNamespace(db=database) + self.spawned = [] + + async def spawn_job(self, *, agent, text, title=None): + self.spawned.append((agent, title, text)) + return None + + +async def make_db(tmp_path, rows): + from beaver_gateway.storage.db import Database + from beaver_gateway.storage.models import JobRunRecord + + database = Database(f"sqlite+aiosqlite:///{tmp_path / 'journal.db'}") + await database.create_all() + async with database.session() as session: + for row in rows: + session.add(row) + await session.commit() + return database + + +def record(job, status, *, ago, error=None): + from beaver_gateway.storage.models import JobRunRecord + + started = (datetime.now(UTC) - timedelta(minutes=ago)).replace(tzinfo=None) + return JobRunRecord( + job=job, + trigger="cron", + started_at=started, + finished_at=started, + status=status, + error=error if error is not None or status != "failed" else BOOM, + ) + + +async def test_job_reads_the_real_journal_and_spawns(tmp_path, monkeypatch): + from beaver_agent.jobs import runs as job + + monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json") + database = await make_db( + tmp_path, + [ + record("вайбграм", "failed", ago=30), + record("вайбграм", "failed", ago=20), + record("вайбграм", "failed", ago=10), + record("досуг", "done", ago=5), + record("прогоны", "done", ago=3), + # Старше суток - в окно не попадает + record("логи", "failed", ago=60 * 30), + ], + ) + run_ = FakeRun(database) + + await job.watch_runs(cast("Any", run_)) + + assert len(run_.spawned) == 1 + agent, title, text = run_.spawned[0] + assert agent == "beaver-triage" and title == "прогоны" + assert "джоб `вайбграм` лежит" in text and "3 прогона подряд" in text + assert "логи" not in text # за окном + assert "прогоны" not in text.split("\n")[1] # сам себя не считает + # Тревога записана - следующий прогон в те же сутки молчит + assert Runs.load(tmp_path / "прогоны.json").alerted["вайбграм"] is not None + await job.watch_runs(cast("Any", run_)) + assert len(run_.spawned) == 1 + await database.dispose() + + +async def test_job_says_otboy_when_the_job_recovers(tmp_path, monkeypatch): + from beaver_agent.jobs import runs as job + + monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json") + Runs( + path=tmp_path / "прогоны.json", + alerted={"вайбграм": datetime.now(UTC) - timedelta(hours=2)}, + ).save() + database = await make_db( + tmp_path, + [ + record("вайбграм", "failed", ago=30), + record("вайбграм", "failed", ago=20), + record("вайбграм", "done", ago=10), + ], + ) + run_ = FakeRun(database) + + await job.watch_runs(cast("Any", run_)) + + assert len(run_.spawned) == 1 + _, _, text = run_.spawned[0] + assert "отбой: джоб `вайбграм` снова проходит" in text + assert 'urgency="normal"' in text # отбой не будит + assert Runs.load(tmp_path / "прогоны.json").alerted == {} + # Сброшено - второй «отбой» не уходит + await job.watch_runs(cast("Any", run_)) + assert len(run_.spawned) == 1 + await database.dispose() + + +async def test_job_alarms_on_an_empty_journal(tmp_path, monkeypatch): + from beaver_agent.jobs import runs as job + + monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json") + database = await make_db(tmp_path, [record("прогоны", "done", ago=5)]) + run_ = FakeRun(database) + + await job.watch_runs(cast("Any", run_)) + + assert len(run_.spawned) == 1 + _, _, text = run_.spawned[0] + assert "нет ни одной записи" in text and "не отработал ни один джоб" in text + assert Runs.load(tmp_path / "прогоны.json").empty_alerted is not None + await database.dispose() + + +async def test_unreadable_journal_is_not_a_reading(tmp_path, monkeypatch): + from beaver_agent.jobs import runs as job + + class Dead: + def session(self): + msg = "база не отвечает" + raise RuntimeError(msg) + + monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json") + run_ = FakeRun(Dead()) + + await job.watch_runs(cast("Any", run_)) + + assert not run_.spawned + assert not (tmp_path / "прогоны.json").exists() + + +@pytest.mark.parametrize("name", ["прогоны", "иначе-названный"]) +async def test_the_watcher_never_reports_itself(tmp_path, monkeypatch, name): + from beaver_agent.jobs import runs as job + + monkeypatch.setattr(job, "STATE_FILE", tmp_path / "прогоны.json") + database = await make_db( + tmp_path, + [ + record(name, "failed", ago=30), + record(name, "failed", ago=20), + record(name, "failed", ago=10), + record("досуг", "done", ago=5), + ], + ) + run_ = FakeRun(database, name=name) + + await job.watch_runs(cast("Any", run_)) + + assert not run_.spawned + await database.dispose()