feat(hands,jobs): job-run watchdog reads job_runs and alarms on three failures in a row

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, и не заведена ли база заново - на пустой базе эта же "
+        "тревога выглядит так же."
+    )

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),

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)