hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(cron): remind once per cooldown after the alert-once gate; migrate alerted_at

Browse Source 
Builds on the previous commit (alerted treated like closed): a permanently
silent `alerted` incident would hide a job that stays broken for days, so the
gate now withholds only inside `cron.failure_repeat_alert_hours` (default 6,
0 = re-alert on every failing run) and lets exactly one reminder through, which
re-stamps the window.

- cron/incidents.py: `alerted_at` column (added in place to existing ledgers via
  add_column_if_missing); set_incident_state(..., "alerted") stamps it every time
  so a reminder restarts the cooldown; resolved->detected re-open clears it so
  the same error after a green run alerts immediately.
- cron/scheduler.py: `_repeat_alert_withheld` reads the stamp; a missing or
  unparseable stamp (pre-migration row) delivers rather than swallowing the
  alert; `closed` still wins; the unreadable-ledger fail-open is unchanged. The
  crash path (`_deliver_crash_failure`) shares the gate through
  `_upsert_incident_for_failure`.
- hermes_cli/config_defaults.py + website/docs cron page: document the key.
- tests: fold the contributor's two tests and the old "unacked failures keep
  alerting per run" change-detector into two invariants (unit gate incl. 0 /
  legacy row / closed; end-to-end alert once -> reminder once -> recovery re-arms).

Fixes #113665
Co-authored-by: Yagna Vudathu <yagnavudathu@gmail.com>
This commit is contained in:
teknium1
2026-09-18 00:46:13 -07:00
committed by Teknium
parent 7aa466665b
commit 454f9c1421
5 changed files with 149 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
cron/incidents.py
View File

@@ -5,7 +5,9 @@ by ``(job_id, error signature)`` so the same job failing with the same error doe
operator every run once acknowledged. Lifecycle: ``detected`` → ``alerted`` → ``closed``. The same
job + same normalized error resolves to the SAME incident id, so a closed incident stays closed
until the error text changes and mints a new one. ``alerted`` means a failure ping actually reached
the operator. Incidents share ``cron/executions.db`` with ``cron.executions`` (one ledger file).
the operator (``alerted_at`` = when the latest one did; the scheduler withholds repeats until
``cron.failure_repeat_alert_hours`` have passed). Incidents share ``cron/executions.db`` with
``cron.executions`` (one ledger file).
"""
from __future__ import annotations
@@ -66,6 +68,8 @@ def _connect() -> sqlite3.Connection:
def _initialize_schema(conn: sqlite3.Connection) -> None:
from hermes_cli.sqlite_util import add_column_if_missing
conn.execute(
"""CREATE TABLE IF NOT EXISTS cron_incidents (
id TEXT PRIMARY KEY,
@@ -76,11 +80,14 @@ def _initialize_schema(conn: sqlite3.Connection) -> None:
first_seen_at TEXT NOT NULL,
last_seen_at TEXT NOT NULL,
acked_at TEXT,
alerted_at TEXT,
closed_at TEXT,
error TEXT NOT NULL,
output_file TEXT
)"""
)
# Ledgers created before the alert-once gate lack ``alerted_at``; add it in place.
add_column_if_missing(conn, "cron_incidents", "alerted_at", "alerted_at TEXT")
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_cron_incidents_job "
"ON cron_incidents(job_id)"
@@ -168,6 +175,7 @@ def upsert_incident(
"""UPDATE cron_incidents
SET last_seen_at=?, error=?, output_file=?,
state=CASE WHEN state='resolved' THEN 'detected' ELSE state END,
alerted_at=CASE WHEN state='resolved' THEN NULL ELSE alerted_at END,
closed_at=CASE WHEN state='resolved' THEN NULL ELSE closed_at END
WHERE id=?""",
(now, stored_error, output_file, incident_id),
@@ -187,7 +195,8 @@ def upsert_incident(
def set_incident_state(incident_id: str, state: str) -> bool:
"""Transition an incident's lifecycle state; return whether it changed. ``closed`` is terminal
for that signature (re-open happens by a changed error minting a NEW incident). Unknown states
are rejected (no-op, ``False``)."""
are rejected (no-op, ``False``). ``alerted`` also stamps ``alerted_at`` — every time, so the
cooldown reminder (see ``cron.scheduler._upsert_incident_for_failure``) restarts its window."""
if state not in INCIDENT_STATES:
return False
now = _hermes_now().isoformat()
@@ -195,7 +204,15 @@ def set_incident_state(incident_id: str, state: str) -> bool:
row = conn.execute(
"SELECT state FROM cron_incidents WHERE id=?", (incident_id,)
).fetchone()
if row is None or row["state"] in (state, "closed"):
if row is None or row["state"] == "closed":
return False
if state == "alerted":
conn.execute(
"UPDATE cron_incidents SET state='alerted', alerted_at=? WHERE id=?",
(now, incident_id),
)
return True
if row["state"] == state:
return False
if state == "closed":
conn.execute(

52
cron/scheduler.py
View File

@@ -17,7 +17,7 @@ import threading
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
# fcntl is Unix-only; Windows uses msvcrt
try:
@@ -285,21 +285,53 @@ def _summarize_cron_failure_for_delivery(job: dict, error: str | None) -> str:
return message
DEFAULT_FAILURE_REPEAT_ALERT_HOURS = 6.0
def _failure_repeat_alert_hours() -> float:
"""``cron.failure_repeat_alert_hours``: how long an ``alerted`` incident stays silent before one
reminder ping. ``0`` (or negative) re-alerts on every failing run (the pre-gate behaviour)."""
from cron.jobs import _cron_config_number
return _cron_config_number("failure_repeat_alert_hours", DEFAULT_FAILURE_REPEAT_ALERT_HOURS, float)
def _repeat_alert_withheld(incident: dict) -> bool:
"""An ``alerted`` incident withholds the per-run ping until the reminder cooldown has elapsed
since its last ping. A missing/unparseable ``alerted_at`` (ledger written before the column
existed) delivers rather than swallowing an alert."""
hours = _failure_repeat_alert_hours()
if hours <= 0:
return False
alerted_at = incident.get("alerted_at")
if not alerted_at:
return False
try:
from cron.jobs import _ensure_aware
last = _ensure_aware(datetime.fromisoformat(str(alerted_at)))
except (TypeError, ValueError):
return False
return _hermes_now() - last < timedelta(hours=hours)
def _upsert_incident_for_failure(
job: dict, error: str, *, output_file: Optional[Any] = None
) -> tuple[bool, Optional[str]]:
"""Record a durable failure incident (grouped by job + error signature). Returns
``(acked, incident_id)``; acked=True when the signature's incident is already ``closed``
(operator ack) or ``alerted`` (a ping already went out) -> suppress the per-run ping.
Store errors log at debug; the caller delivers as if none existed."""
``(withheld, incident_id)``; withheld=True when the signature's incident is already ``closed``
(operator ack) or ``alerted`` inside the ``cron.failure_repeat_alert_hours`` cooldown (a ping
already went out) -> suppress the per-run ping. Store errors log at debug; the caller delivers
as if none existed."""
try:
from cron.incidents import get_incident, upsert_incident
incident_id, _is_new = upsert_incident(
job["id"], str(error or ""), job_name=job.get("name"), output_file=output_file)
incident = get_incident(incident_id)
acked = bool(incident and incident.get("state") in ("closed", "alerted"))
return acked, incident_id
state = incident.get("state") if incident else None
withheld = state == "closed" or (state == "alerted" and _repeat_alert_withheld(incident))
return withheld, incident_id
except Exception as exc:
logger.debug(
"Incident store unavailable for job %s (delivery unaffected): %s",
@@ -2562,7 +2594,8 @@ def _classify_delivery_outcome(
if should_deliver and normalized_deliver != "local":
return "delivered"
if incident_acked and not success:
# Failure ping withheld: operator acked this exact signature (vs. plain "suppressed").
# Failure ping withheld for a known signature: operator acked it, or it was already
# alerted inside the reminder cooldown (vs. plain "suppressed").
return "suppressed_acked"
return "suppressed"
@@ -2590,8 +2623,9 @@ def _compose_run_delivery(
deliver_content = final_response
_resolve_incidents_for_recovered_job(job)
else:
# Record the job+error signature once; if already acked by the operator, suppress the
# per-run ping. Best-effort: a ledger failure never breaks delivery.
# Record the job+error signature once; withhold the per-run ping while the operator
# already acked it (closed) or was already told (alerted, inside the reminder cooldown).
# Best-effort: a ledger failure never breaks delivery.
incident_acked, failure_incident_id = _upsert_incident_for_failure(
job, error or "", output_file=output_file
)

6
hermes_cli/config_defaults.py
View File

@@ -1752,6 +1752,12 @@ DEFAULT_CONFIG = {
# cron jobs as a direct external subprocess (warns once; no cgroup isolation), true
# fails closed with the enable-linger remedy. Kanban always requires a scope.
"require_restart_safe_scope": False,
# A job failing with the SAME error alerts once, then stays silent for this many hours
# before one reminder ping (the run is still recorded; `hermes cron incidents` shows it).
# A green run or a different error alerts again immediately; `hermes cron incidents ack`
# silences a signature for good. 0 = re-alert on every failing run. Keep in sync with
# cron.scheduler.DEFAULT_FAILURE_REPEAT_ALERT_HOURS.
"failure_repeat_alert_hours": 6,
},
# Kanban multi-agent coordination. The dispatcher ticks every N seconds, reclaims stale claims,
# promotes dependency-satisfied todos to ready, and fires `hermes -p <assignee> chat -q ...` per

86
tests/cron/test_cron_incidents.py
View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import argparse
import sys
from datetime import timedelta
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -12,6 +13,7 @@ sys.path.insert(0, str(Path(__file__).parent.parent.parent))
import cron.incidents as incidents
import cron.jobs as cron_jobs
import cron.scheduler as sched
from hermes_time import now as _hermes_now
def _point_db(monkeypatch, tmp_path):
@@ -222,19 +224,39 @@ def test_missing_db_no_crash(monkeypatch, tmp_path):
# ── Scheduler gating ───────────────────────────────────────────────────────
def test_unacked_failure_still_alerts(monkeypatch, tmp_path):
def test_repeat_failure_alerts_once_then_reminds_after_cooldown(monkeypatch, tmp_path):
"""Same job + same error: the first failing run delivers, the repeat is withheld (but still
recorded as a run), one reminder goes out once ``cron.failure_repeat_alert_hours`` has
elapsed, and a green run re-arms the signature so the same error alerts again."""
inc = _point_db(monkeypatch, tmp_path)
deliveries = []
job = _job()
# A real (non-local) lane: the ping leaves the process, so the incident is marked alerted.
job = _job(deliver="telegram:123")
(tmp_path / "config.yaml").write_text("cron:\n preflight: false\n", encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
with cron_jobs.use_cron_store(tmp_path):
cron_jobs.save_jobs([job])
_tick_failing(job, tmp_path, deliveries, error="unacked boom")
_tick_failing(job, tmp_path, deliveries, error="unacked boom")
_tick_failing(job, tmp_path, deliveries, error="repeat boom")
_tick_failing(job, tmp_path, deliveries, error="repeat boom")
assert len(deliveries) == 1, "an alerted signature must not re-ping on every run"
rows = inc.list_incidents()
assert len(rows) == 1 and rows[0]["state"] == "alerted" and rows[0]["alerted_at"]
stored = [j for j in cron_jobs.load_jobs() if j["id"] == job["id"]][0]
assert stored["last_status"] == "error", "the withheld run is still recorded"
assert len(deliveries) == 2, "unacked failures must keep alerting per run"
rows = inc.list_incidents()
assert len(rows) == 1
assert rows[0]["state"] == "detected"
# Cooldown elapsed: exactly one reminder, then silent again.
stale = (_hermes_now() - timedelta(hours=7)).isoformat()
with inc._transaction() as conn:
conn.execute("UPDATE cron_incidents SET alerted_at=?", (stale,))
_tick_failing(job, tmp_path, deliveries, error="repeat boom")
_tick_failing(job, tmp_path, deliveries, error="repeat boom")
assert len(deliveries) == 2, "one reminder after the cooldown, not one per run"
# Recovery re-arms: the same error after a green run alerts immediately.
sched._resolve_incidents_for_recovered_job(job)
_tick_failing(job, tmp_path, deliveries, error="repeat boom")
assert len(deliveries) == 3
assert inc.count_incidents() == 1
def test_ack_suppresses_alert_until_signature_changes(monkeypatch, tmp_path):
@@ -335,34 +357,32 @@ def test_cli_list_and_ack(monkeypatch, tmp_path, capsys):
incident_action="ack", state=None, incident_id=None
)
assert cron_incidents(missing_args) == 1
def test_alerted_signature_suppresses_repeat_ping(monkeypatch, tmp_path):
"""Once a failure ping has gone out (``alerted``), the same signature stays
silent on later runs until the job recovers or the error changes (#113665).
Unit level: the upsert gate treats ``alerted`` like ``closed``."""
def test_alerted_gate_honours_cooldown_opt_out_and_legacy_rows(monkeypatch, tmp_path):
"""Unit level: ``alerted`` withholds only inside the cooldown; ``0`` restores per-run
alerts; a pre-migration row (state alerted, no ``alerted_at``) delivers rather than
swallowing the alert; ``closed`` still wins regardless of the cooldown."""
inc = _point_db(monkeypatch, tmp_path)
monkeypatch.setattr(sched, "_failure_repeat_alert_hours", lambda: 6.0)
job = _job()
acked, inc_id = sched._upsert_incident_for_failure(job, "repeat boom")
assert acked is False and inc_id is not None
withheld, inc_id = sched._upsert_incident_for_failure(job, "repeat boom")
assert withheld is False and inc_id is not None
sched._mark_incident_alerted(inc_id)
assert inc.get_incident(inc_id)["state"] == "alerted"
acked_again, same_id = sched._upsert_incident_for_failure(job, "repeat boom")
assert acked_again is True, "alerted signature must not re-ping"
assert same_id == inc_id
assert sched._upsert_incident_for_failure(job, "repeat boom") == (True, inc_id)
monkeypatch.setattr(sched, "_failure_repeat_alert_hours", lambda: 0)
assert sched._upsert_incident_for_failure(job, "repeat boom") == (False, inc_id)
def test_second_run_silent_once_ping_went_out(monkeypatch, tmp_path):
"""End to end through the scheduler tick: first failure delivers, and once
the post-delivery mark fires the same failure goes silent (#113665)."""
inc = _point_db(monkeypatch, tmp_path)
deliveries = []
job = _job()
with cron_jobs.use_cron_store(tmp_path):
cron_jobs.save_jobs([job])
_tick_failing(job, tmp_path, deliveries, error="repeat boom")
assert len(deliveries) == 1
rows = inc.list_incidents()
assert len(rows) == 1
# Production marks the incident alerted once the ping leaves the process.
sched._mark_incident_alerted(rows[0]["id"])
_tick_failing(job, tmp_path, deliveries, error="repeat boom")
assert len(deliveries) == 1, "alerted signature must not re-ping per run"
monkeypatch.setattr(sched, "_failure_repeat_alert_hours", lambda: 6.0)
with inc._transaction() as conn:
conn.execute("UPDATE cron_incidents SET alerted_at=NULL WHERE id=?", (inc_id,))
assert sched._upsert_incident_for_failure(job, "repeat boom") == (False, inc_id)
# Re-alerting restarts the window (the reminder stamps alerted_at again).
sched._mark_incident_alerted(inc_id)
assert inc.get_incident(inc_id)["alerted_at"]
assert sched._upsert_incident_for_failure(job, "repeat boom") == (True, inc_id)
assert inc.ack_incident(inc_id) is True
monkeypatch.setattr(sched, "_failure_repeat_alert_hours", lambda: 0)
assert sched._upsert_incident_for_failure(job, "repeat boom") == (True, inc_id)

43
website/docs/user-guide/features/cron.md
View File

@@ -444,22 +444,36 @@ cron:
retry_unreachable: false # default true; disables the automatic re-runs
```
### Failure incidents: acknowledge a known failure
### Failure incidents: alert once, remind on a cooldown, acknowledge
A recurring job that keeps failing with the *same* error pings you on every
run. Each failure is also recorded as a durable **incident**, keyed by the
job plus a normalized signature of the error text, in the same per-profile
ledger database as the execution history.
A recurring job that keeps failing with the *same* error alerts you **once**,
not on every run. Each failure is recorded as a durable **incident**, keyed by
the job plus a normalized signature of the error text, in the same per-profile
ledger database as the execution history; the first failure of a signature is
always delivered, and repeats are then withheld while the incident is `alerted`
(the run is still recorded — `hermes cron runs` and the failure streak see it,
only the ping is held back).
```yaml
cron:
failure_repeat_alert_hours: 6 # still broken after this long → one reminder ping,
# then silent again; 0 = alert on every failing run
```
Anything that changes the picture alerts immediately: a *different* error mints
its own incident and pings at once, and a successful run re-arms the signature
so the same error after a green run alerts again. If the incident ledger cannot
be read, the ping is delivered rather than swallowed.
```bash
hermes cron incidents # list incidents (newest activity first)
hermes cron incidents --state alerted # filter: detected | alerted | resolved | closed
hermes cron incidents ack <id> # acknowledge — stop re-pinging
hermes cron incidents ack <id> # acknowledge — silence this signature for good
```
Acknowledging an incident silences the per-run failure ping for that exact
signature only. Nothing else changes: the run history still records every
failure, the failure streak keeps counting, and the moment the job starts
Acknowledging an incident silences the failure ping for that exact signature
only, reminders included. Nothing else changes: the run history still records
every failure, the failure streak keeps counting, and the moment the job starts
failing with a *different* error a new incident is minted and alerts fire
again.
@@ -470,13 +484,10 @@ the job later fails with the *same* error, the resolved incident re-opens as
the exception: a success leaves them alone, and a repeat stays silent.
Incident lifecycle: `detected` (failure recorded) → `alerted` (at least one
failure ping reached delivery) → `resolved` (the job ran OK afterwards;
re-opens on a repeat) or `closed` (acknowledged; terminal for that
signature). Stored error text is secret-redacted and truncated before it is
written.
Recording is always on and costs nothing to ignore — no ping is ever
suppressed until you explicitly `ack`.
failure ping reached delivery; `alerted_at` is the latest one and starts the
reminder cooldown) → `resolved` (the job ran OK afterwards; re-opens on a
repeat) or `closed` (acknowledged; terminal for that signature). Stored error
text is secret-redacted and truncated before it is written.
### Fleet health check: `hermes cron doctor`