feat(hands,jobs): job-run watchdog looks a week back and alarms on jobs that stopped running

beaver_agent/hands/heartbeat.py

@@ -151,13 +151,19 @@ def parse_beat(text: str, kind: str = "hb") -> Beat | None:
 
 
 def every_from_cron(
-    expression: str | None, tz: tzinfo, *, now: datetime | None = None
+    expression: str | None,
+    tz: tzinfo,
+    *,
+    now: datetime | None = None,
+    grain: timedelta = timedelta(hours=1),
 ) -> timedelta | None:
-    """Шаг крона, если он сводится к «раз в N часов»; иначе None.
+    """Шаг крона, если он сводится к «раз в N `grain`»; иначе None.
 
     Считаем по стенным часам зоны, в которой планировщик читает кроны:
     суточный крон - это 24 часа и в сутки перевода стрелок тоже, а вот
-    `0 2,8,12,16,20 * * *` не сводится ни к какому N и даёт None.
+    `0 2,8,12,16,20 * * *` не сводится ни к какому N и даёт None. `grain`
+    по умолчанию час - такой период печатается в отметке соседу; сторожу
+    прогонов нужен и `*/10`, он просит минуту.
     """
     if not expression:
         return None
@@ -171,7 +177,7 @@ def every_from_cron(
     if len(steps) != 1:
         return None
     step = steps.pop()
-    return step if step.total_seconds() % 3600 == 0 else None
+    return step if step % grain == timedelta(0) else None
 
 
 def beat_line(

beaver_agent/hands/runs.py

@@ -11,6 +11,17 @@
 - тревога берётся только с ХВОСТА: три и более `failed` подряд, считая от
   последнего прогона джоба. Одиночный упавший прогон - сетевой чих, а не
   болезнь, и будить на него человека значит выучить его не смотреть;
+- окно счёта - неделя, а не сутки: иначе суточный джоб не наберёт трёх
+  падений подряд никогда, и `отметка`, `закрытие`, `голосовой`, `память`
+  не охранялись бы вовсе;
+- отсутствие записей - тоже показание. Джоб с регулярным кроном, у которого
+  последний прогон старше `FACTOR` его периодов, «не запускался» - это
+  отдельная тревога, не «лежит». Период выводится из крона тем же
+  `every_from_cron`, что у сторожа соседа, и расписание, которое не
+  сводится к «раз в N часов», в давности не участвует и называется в логе
+  неохраняемым: подставить ему число значило бы соврать. Джобы без крона
+  (вебхуки, события) давностью не меряются вовсе, а джоб, которого в
+  журнале нет ни разу, молча ждёт своей первой записи;
 - пустая выдача - это показание о запросе, а не о здоровье джобов. Ноль
   строк за сутки (кроме своих) означает, что планировщик вообще не
   работал, и это отдельная тревога с отдельным текстом, а не тишина;
@@ -38,13 +49,19 @@ from datetime import UTC, datetime, timedelta
 from pathlib import Path  # noqa: TC003 - в аннотации dataclass
 from typing import TYPE_CHECKING
 
+from beaver_gateway.jobs.scheduler import next_run
 from beaver_gateway.security.redact import MASK, is_secret_name, redact
 
+from beaver_agent.hands.heartbeat import every_from_cron
+
 if TYPE_CHECKING:
-    from collections.abc import Sequence
+    from collections.abc import Mapping, Sequence
+    from datetime import tzinfo
 
 __all__ = [
+    "DAY",
     "ERROR_CHARS",
+    "FACTOR",
     "FAILED",
     "FORGET",
     "MAX_LINES",
@@ -52,22 +69,39 @@ __all__ = [
     "STREAK",
     "WINDOW",
     "Clear",
+    "Quiet",
     "Report",
+    "Resumed",
     "Run",
     "Runs",
     "Streak",
     "alarm_text",
+    "awake",
     "clear_text",
     "empty_text",
     "first_line",
     "look",
     "mask",
+    "quiet_text",
     "streak",
 ]
 
-WINDOW = timedelta(hours=24)
-"""Окно журнала: сутки. Джоб реже трёх раз в сутки за него не наберёт серии
-из трёх падений - это слепая зона, названная в README."""
+WINDOW = timedelta(days=7)
+"""Окно журнала: неделя. Суточный джоб набирает в нём три падения подряд за
+трое суток, недельный (`память`) - нет: у него три падения это три недели,
+и его охраняет давность, а не серия."""
+DAY = timedelta(hours=24)
+"""Окно тревоги о пустом журнале: сутки без единого чужого прогона."""
+FACTOR = 2
+"""Давность = столько периодов джоба без записи. Два - это «пропущено два
+срабатывания подряд»: один пропуск случается честно (передеплой пришёлся на
+минуту крона, а догонять пропущенное планировщик не умеет), два подряд -
+уже не случайность."""
+GRAIN = timedelta(minutes=1)
+"""Шаг, в котором период крона должен сводиться к целому: крон минутный."""
+LAG = timedelta(minutes=10)
+"""Запись в журнал ложится по КОНЦУ прогона: срабатывание моложе этого
+могло ещё не дописаться, и пропуском его не считаем."""
 STREAK = 3
 """Сколько `failed` подряд считаем «джоб лежит»."""
 REPEAT = timedelta(hours=24)
@@ -80,6 +114,7 @@ ERROR_CHARS = 160
 MAX_LINES = 10
 """Сколько джобов перечисляем в одной тревоге: сразу всё не читают."""
 FAILED = "failed"
+_DAYS_FROM = timedelta(hours=48)
 STATE_VERSION = 1
 
 _PAIR = re.compile(r"(?=([A-Za-z][\w.-]{2,})\s*([:=])\s*([^\s,;)\]}\"']{6,}))")
@@ -117,14 +152,29 @@ def _plural(count: int, forms: tuple[str, str, str]) -> str:
 
 
 def _age(delta: timedelta) -> str:
-    """Возраст человеку: «4 часа», «45 минут». Секунды тут не нужны никому."""
+    """Возраст человеку: «4 часа», «45 минут», «15 суток». Секунды не нужны."""
     minutes = max(int(delta.total_seconds()) // 60, 0)
     if minutes < 60:
         return f"{minutes} {_plural(minutes, ('минуту', 'минуты', 'минут'))}"
     hours = minutes // 60
+    if delta >= _DAYS_FROM:
+        days = hours // 24
+        return f"{days} {_plural(days, ('сутки', 'суток', 'суток'))}"
     return f"{hours} {_plural(hours, ('час', 'часа', 'часов'))}"
 
 
+def _period(delta: timedelta) -> str:
+    """Период крона: «10 минут», «24 часа», «7 суток» - без округления."""
+    minutes = int(delta.total_seconds()) // 60
+    if minutes % (24 * 60) == 0 and delta >= _DAYS_FROM:
+        days = minutes // (24 * 60)
+        return f"{days} {_plural(days, ('сутки', 'суток', 'суток'))}"
+    if minutes % 60 == 0:
+        hours = minutes // 60
+        return f"{hours} {_plural(hours, ('час', 'часа', 'часов'))}"
+    return f"{minutes} {_plural(minutes, ('минута', 'минуты', 'минут'))}"
+
+
 def _mask_pairs(text: str) -> str:
     out: list[str] = []
     pos = 0
@@ -182,7 +232,25 @@ class Streak:
     error: str
     capped: bool
     """Серия упёрлась в край окна: падений было НЕ МЕНЬШЕ `count`, а сколько
-    на самом деле - за сутками не видно, и врать точным числом нельзя."""
+    на самом деле - за неделей не видно, и врать точным числом нельзя."""
+
+
+@dataclass(frozen=True, slots=True)
+class Quiet:
+    """Джоб с кроном, который перестал запускаться."""
+
+    job: str
+    age: timedelta
+    """Сколько прошло с начала последнего прогона - любого, хоть упавшего."""
+    period: timedelta
+
+
+@dataclass(frozen=True, slots=True)
+class Resumed:
+    """Джоб, о молчании которого говорили, снова запустился."""
+
+    job: str
+    status: str
 
 
 @dataclass(frozen=True, slots=True)
@@ -203,6 +271,8 @@ class Runs:
     path: Path
     alerted: dict[str, datetime] = field(default_factory=dict)
     """Джоб → когда о нём ушла тревога. Он же признак «сейчас в тревоге»."""
+    quiet: dict[str, datetime] = field(default_factory=dict)
+    """То же для «не запускался»: отдельно, у тревог разные отбои."""
     empty_alerted: datetime | None = None
 
     @classmethod
@@ -213,14 +283,10 @@ class Runs:
             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,
+            alerted=_stamps(raw.get("alerted")),
+            quiet=_stamps(raw.get("quiet")),
             empty_alerted=_parse_iso(raw.get("empty_alerted")),
         )
 
@@ -228,6 +294,7 @@ class Runs:
         body = {
             "version": STATE_VERSION,
             "alerted": {job: _iso(when) for job, when in sorted(self.alerted.items())},
+            "quiet": {job: _iso(when) for job, when in sorted(self.quiet.items())},
             "empty_alerted": _iso(self.empty_alerted),
         }
         self.path.parent.mkdir(parents=True, exist_ok=True)
@@ -236,6 +303,15 @@ class Runs:
         tmp.replace(self.path)
 
 
+def _stamps(raw: dict[str, str] | None) -> dict[str, datetime]:
+    out = {}
+    for job, when in (raw or {}).items():
+        parsed = _parse_iso(str(when))
+        if parsed is not None:
+            out[str(job)] = parsed
+    return out
+
+
 @dataclass(frozen=True, slots=True)
 class Report:
     """Что решил один прогон сторожа журнала."""
@@ -249,6 +325,10 @@ class Report:
     mine: int
     """Сколько своих - их в счёт не берём, но в тексте называем."""
     note: str
+    quiet: tuple[Quiet, ...] = ()
+    resumed: tuple[Resumed, ...] = ()
+    unguarded: tuple[str, ...] = ()
+    """Джобы с кроном, у которого нет регулярного периода: давностью не меряем."""
 
 
 def streak(runs: Sequence[Run], *, now: datetime) -> Streak | None:
@@ -271,27 +351,41 @@ def streak(runs: Sequence[Run], *, now: datetime) -> Streak | None:
     )
 
 
-def look(state: Runs, runs: Sequence[Run], *, now: datetime, me: str) -> Report:
+def look(
+    state: Runs,
+    runs: Sequence[Run],
+    *,
+    now: datetime,
+    me: str,
+    crons: Mapping[str, str] | None = None,
+    latest: Mapping[str, Run] | None = None,
+    tz: tzinfo = UTC,
+    boot: datetime | None = None,
+) -> Report:
     """Разобрать окно журнала и решить, о чём говорить человеку.
 
+    `crons` - джоб → крон, только джобы с кроном; `latest` - последний прогон
+    каждого джоба за ВСЮ историю журнала, а не за окно; `boot` - старт
+    процесса. Без `crons` давность не проверяется вовсе.
+
     Состояние здесь только чистится от забытых джобов; отметки «сказали» и
     «сказали про пустоту» ставит джоб - и только когда тревога действительно
     ушла: несостоявшаяся доставка не должна заглушать следующий прогон.
     """
-    mine = sum(1 for run in runs if run.job == me)
+    mine = [run 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:
+    if not any(now - run.started_at < DAY for run in 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,
+            seen=len(others),
+            mine=len(mine),
             note=(
-                f"чужих прогонов за сутки нет вовсе (своих {mine})"
+                f"чужих прогонов за сутки нет вовсе (своих {len(mine)})"
                 + ("; о пустоте уже говорили меньше суток назад" if quiet else "")
             ),
         )
@@ -328,7 +422,9 @@ def look(state: Runs, runs: Sequence[Run], *, now: datetime, me: str) -> Report:
     for job in forgotten:
         del state.alerted[job]
 
-    note = f"прогонов за сутки {len(others)} по {len(by_job)} джобам (своих {mine})"
+    note = (
+        f"прогонов за неделю {len(others)} по {len(by_job)} джобам (своих {len(mine)})"
+    )
     if alarms:
         note += "; лежат: " + ", ".join(sorted(tail.job for tail in alarms))
     if held:
@@ -338,12 +434,140 @@ def look(state: Runs, runs: Sequence[Run], *, now: datetime, me: str) -> Report:
     if forgotten:
         note += "; забыты (пропали из окна): " + ", ".join(sorted(forgotten))
 
+    silence = _silence(
+        state,
+        mine,
+        now=now,
+        me=me,
+        crons=crons or {},
+        latest=latest or {},
+        tz=tz,
+        boot=boot,
+    )
+
     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,
+        mine=len(mine),
+        note=note + silence.note,
+        quiet=silence.quiet,
+        resumed=silence.resumed,
+        unguarded=silence.unguarded,
+    )
+
+
+@dataclass(frozen=True, slots=True)
+class _Silence:
+    quiet: tuple[Quiet, ...]
+    resumed: tuple[Resumed, ...]
+    unguarded: tuple[str, ...]
+    note: str
+
+
+def awake(
+    mine: Sequence[Run], *, now: datetime, step: timedelta, boot: datetime | None
+) -> datetime:
+    """С какого момента планировщик заведомо крутил некритичные джобы.
+
+    Показание - собственные прогоны сторожа: он сам некритичный, и когда
+    гейтвей лежит или бюджет зажат (`throttled` откладывает некритичный крон
+    без записи в журнал), его прогонов нет так же, как и чужих. Цепочка своих
+    прогонов без разрыва длиннее `step`, считая от сейчас, плюс старт
+    процесса - это время, за которое чужой пропуск действительно пропуск,
+    а не общий простой.
+    """
+    anchor = now
+    for run in sorted(mine, key=lambda run: run.started_at, reverse=True):
+        if anchor - run.started_at > step:
+            break
+        anchor = min(anchor, run.started_at)
+    return anchor if boot is None else max(anchor, boot)
+
+
+def _silence(
+    state: Runs,
+    mine: Sequence[Run],
+    *,
+    now: datetime,
+    me: str,
+    crons: Mapping[str, str],
+    latest: Mapping[str, Run],
+    tz: tzinfo,
+    boot: datetime | None,
+) -> _Silence:
+    """Давность: какие джобы с кроном перестали запускаться."""
+    own = every_from_cron(crons.get(me), tz, now=now, grain=GRAIN)
+    guarded: dict[str, timedelta] = {}
+    unguarded: list[str] = []
+    for job, cron in crons.items():
+        if job == me:
+            continue
+        period = every_from_cron(cron, tz, now=now, grain=GRAIN)
+        if period is None:
+            unguarded.append(job)
+        else:
+            guarded[job] = period
+
+    note = ""
+    if unguarded:
+        note += "; давностью не охраняются (крон без регулярного периода): " + (
+            ", ".join(sorted(unguarded))
+        )
+    dropped = [job for job in state.quiet if job not in guarded]
+    for job in dropped:
+        del state.quiet[job]
+    if dropped:
+        note += "; сняты с давности (нет в расписании): " + ", ".join(sorted(dropped))
+    if not guarded:
+        return _Silence((), (), tuple(sorted(unguarded)), note)
+    if own is None:
+        # Свой период не выведен - нет и опоры, когда планировщик заведомо
+        # работал. Любое число тут было бы выдумано.
+        note += "; свой крон без регулярного периода - давность не проверяется"
+        return _Silence((), (), tuple(sorted(unguarded)), note)
+
+    since = awake(mine, now=now, step=own * FACTOR, boot=boot)
+    quiet: list[Quiet] = []
+    resumed: list[Resumed] = []
+    held: list[str] = []
+    fresh: list[str] = []
+    for job, period in guarded.items():
+        last = latest.get(job)
+        if last is None:
+            fresh.append(job)
+            continue
+        age = now - last.started_at
+        # Порог не тоньше собственного: чаще раза в час сторож не смотрит, и
+        # пропуск короче двух своих периодов неотличим от его же слепоты.
+        overdue = (
+            age > max(period, own) * FACTOR
+            and next_run(crons[job], tz, max(last.started_at, since)) <= now - LAG
+        )
+        told = state.quiet.get(job)
+        if overdue:
+            if told is None or now - told >= REPEAT:
+                quiet.append(Quiet(job=job, age=age, period=period))
+            else:
+                held.append(job)
+        elif told is not None and last.started_at > told:
+            resumed.append(Resumed(job=job, status=last.status))
+
+    if fresh:
+        note += "; ещё ни разу не запускались: " + ", ".join(sorted(fresh))
+    if quiet:
+        note += "; не запускаются: " + ", ".join(sorted(q.job for q in quiet))
+    if held:
+        note += "; о молчании уже говорили меньше суток назад: " + ", ".join(
+            sorted(held)
+        )
+    if resumed:
+        note += "; снова запускаются: " + ", ".join(sorted(r.job for r in resumed))
+    return _Silence(
+        quiet=tuple(sorted(quiet, key=lambda q: q.age, reverse=True)),
+        resumed=tuple(sorted(resumed, key=lambda r: r.job)),
+        unguarded=tuple(sorted(unguarded)),
         note=note,
     )
 
@@ -355,7 +579,7 @@ def streak_line(tail: Streak) -> str:
         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 "")
+        + (" (это всё окно недели, серия началась раньше)" if tail.capped else "")
         + f": {tail.error}"
     )
 
@@ -369,6 +593,27 @@ def alarm_text(report: Report) -> str:
     return "\n".join(lines)
 
 
+def quiet_line(quiet: Quiet) -> str:
+    """«джоб `закрытие` не запускался 50 часов при периоде 24 часа»."""
+    return (
+        f"джоб `{quiet.job}` не запускался {_age(quiet.age)} "
+        f"при периоде {_period(quiet.period)}"
+    )
+
+
+def quiet_text(report: Report) -> str:
+    """Тревога о молчании: какие джобы с кроном перестали запускаться."""
+    lines = [f"- {quiet_line(quiet)}" for quiet in report.quiet[:MAX_LINES]]
+    rest = len(report.quiet) - len(lines)
+    if rest > 0:
+        lines.append(f"- и ещё {rest} {_plural(rest, ('джоб', 'джоба', 'джобов'))}")
+    return "\n".join(lines)
+
+
+def resumed_line(resumed: Resumed) -> str:
+    return f"отбой: джоб `{resumed.job}` снова запускается ({resumed.status})"
+
+
 def clear_line(clear: Clear) -> str:
     if clear.count == 0:
         return f"отбой: джоб `{clear.job}` снова проходит ({clear.status})"
@@ -379,7 +624,9 @@ def clear_line(clear: Clear) -> str:
 
 
 def clear_text(report: Report) -> str:
-    return "\n".join(f"- {clear_line(clear)}" for clear in report.clears)
+    lines = [clear_line(clear) for clear in report.clears]
+    lines += [resumed_line(resumed) for resumed in report.resumed]
+    return "\n".join(f"- {line}" for line in lines)
 
 
 def empty_text(report: Report) -> str: