feat(cron): acked failure signatures stop re-pinging — durable incidents + ack CLI (salvage #94692) (#95017)
* feat(cron): durable failure incidents with signature dedup and ack Introduce a durable cron incident store (cron_incidents in the shared cron/executions.db) that groups "same job + same error signature" across runs, so a known recurring failure stops re-pinging the operator every run once it has been acknowledged. - cron/incidents.py: lazily-created incident table (detected -> alerted -> reviewed -> closed lifecycle; closed is per-signature terminal), sha256 signature dedup over job_id + normalized error, redacted/truncated error storage, failure-type classification, and ack/list/get/count helpers. - cron/scheduler.py: record an incident on the failure delivery path and suppress the per-run failure ping when the exact signature is acked (both the normal failure path and the processing-raised retry path). Best-effort: an incident-store error never breaks the cron run or delivery. Streak nudge, alert-once markers, and delivery-error behavior are untouched. - hermes_cli: add `hermes cron incidents [--state ...]` and `hermes cron incidents ack <id>`. - tests/cron/test_cron_incidents.py: dedup, lifecycle, redaction, classification, lazy-schema, scheduler gating, and CLI coverage. Non-goals deferred to later slices: Discord buttons/review view, HMAC action tokens, owner-agent review launch, approval-gated fixes, incident playbooks. * refactor(cron): tighten incident lifecycle, wire alerted state and suppressed_acked outcome Follow-ups on top of the salvaged #94692: - Drop the dead 'reviewed' state and the SQLite CHECK (state validity lives in INCIDENT_STATES so future slices can add states without a table rebuild); lifecycle is detected -> alerted -> closed. - Actually mark incidents 'alerted' after a failure ping reaches delivery, on both the normal and exception delivery paths. - Record ack-suppressed runs with a distinct 'suppressed_acked' delivery outcome (registered in cron_health monitoring) instead of the ambiguous generic 'suppressed'. - Drift-skip alerts explicitly bypass the ack gate (they carry the remediation command and alert once via drift_alerted already). - Docs: failure-incidents section in the cron guide. - Tests for the alerted transition + never-resurrect-closed. --------- Co-authored-by: Laura López Real <113060513+laulopezreal@users.noreply.github.com>
This commit is contained in:
@@ -24,7 +24,9 @@ from hermes_time import now as _hermes_now
|
||||
logger = logging.getLogger(__name__)
|
||||
_KNOWN_STATUSES = {"claimed", "running", "completed", "failed", "unknown"}
|
||||
_KNOWN_SOURCES = {"builtin", "direct", "external"}
|
||||
_KNOWN_DELIVERY_OUTCOMES = {"delivered", "failed", "suppressed", "not_configured"}
|
||||
_KNOWN_DELIVERY_OUTCOMES = {
|
||||
"delivered", "failed", "suppressed", "suppressed_acked", "not_configured",
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
"""Durable cron failure incidents with signature dedup and ack.
|
||||
|
||||
The executions ledger (``cron.executions``) records every attempt; this module
|
||||
groups the *failures* into durable incidents keyed by ``(job_id, error
|
||||
signature)`` so the same job failing with the same error does not re-ping the
|
||||
operator every run once they have acknowledged it.
|
||||
|
||||
Lifecycle: ``detected`` → ``alerted`` → ``closed``. Closing
|
||||
(acking) an incident is per-signature: the same job + same normalized error
|
||||
keeps resolving to the SAME incident id, so a closed incident stays closed (no
|
||||
re-alert) until the error text changes, which mints a brand-new incident.
|
||||
``detected`` means the failure was recorded; ``alerted`` means at least one
|
||||
failure ping for the signature actually reached the operator. Richer states
|
||||
(e.g. a dv9.6 ``reviewed``) are deliberately NOT reserved here — state
|
||||
validity lives in ``INCIDENT_STATES`` (Python), not a SQLite CHECK, exactly
|
||||
so a future slice can add states without a table rebuild.
|
||||
|
||||
Incidents live in the SAME ``cron/executions.db`` as ``cron.executions`` so
|
||||
there is one durable cron store per profile. The schema is lazily created on
|
||||
connect and a missing database never raises (directories are created).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import re
|
||||
import sqlite3
|
||||
import threading
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Iterator, List, Optional
|
||||
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_time import now as _hermes_now
|
||||
|
||||
# Optional test override (mirrors ``cron.executions.EXECUTIONS_FILE``).
|
||||
EXECUTIONS_FILE: Optional[Path] = None
|
||||
|
||||
INCIDENT_STATES = ("detected", "alerted", "closed")
|
||||
_FAILURE_TYPE_ORDER = (
|
||||
("rate_limit", (r"\b429\b", "rate limit", "usage limit", "quota")),
|
||||
("timeout", ("timeout", "timed out")),
|
||||
("auth", (r"\b401\b", "unauthorized", "authentication", "auth")),
|
||||
("delivery", ("delivery", "deliver", "delivering")),
|
||||
("config", ("config", "configuration", "validation")),
|
||||
("script", ("script", "no_agent")),
|
||||
("agent", ("agent", "model", "provider", "inference")),
|
||||
)
|
||||
MAX_ERROR_CHARS = 500
|
||||
_MAX_SIGNATURE_ERROR_CHARS = 200
|
||||
|
||||
_lock = threading.RLock()
|
||||
|
||||
|
||||
def _connect() -> sqlite3.Connection:
|
||||
path = _db_path()
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
return sqlite3.connect(path, timeout=5)
|
||||
|
||||
|
||||
def _db_path() -> Path:
|
||||
"""Resolve the shared cron DB path.
|
||||
|
||||
Prefer the ``cron.executions`` override when one is installed so an
|
||||
operator/test that redirects the executions ledger also redirects the
|
||||
incident table — they must stay in the SAME database. Falls back to this
|
||||
module's own override, then the canonical profile home.
|
||||
"""
|
||||
try:
|
||||
from cron.executions import EXECUTIONS_FILE as _EXEC_OVERRIDE
|
||||
|
||||
if _EXEC_OVERRIDE is not None:
|
||||
return Path(_EXEC_OVERRIDE)
|
||||
except Exception:
|
||||
pass
|
||||
if EXECUTIONS_FILE is not None:
|
||||
return Path(EXECUTIONS_FILE)
|
||||
return get_hermes_home().resolve() / "cron" / "executions.db"
|
||||
|
||||
|
||||
def _initialize_schema(conn: sqlite3.Connection) -> None:
|
||||
from hermes_state import apply_wal_with_fallback
|
||||
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("PRAGMA busy_timeout=5000")
|
||||
apply_wal_with_fallback(conn, db_label="cron/executions.db")
|
||||
conn.execute("PRAGMA synchronous=FULL")
|
||||
conn.execute(
|
||||
"""CREATE TABLE IF NOT EXISTS cron_incidents (
|
||||
id TEXT PRIMARY KEY,
|
||||
job_id TEXT NOT NULL,
|
||||
error_sig TEXT NOT NULL,
|
||||
state TEXT NOT NULL,
|
||||
failure_type TEXT NOT NULL DEFAULT 'unknown',
|
||||
first_seen_at TEXT NOT NULL,
|
||||
last_seen_at TEXT NOT NULL,
|
||||
acked_at TEXT,
|
||||
closed_at TEXT,
|
||||
error TEXT NOT NULL,
|
||||
output_file TEXT
|
||||
)"""
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_cron_incidents_job "
|
||||
"ON cron_incidents(job_id)"
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_cron_incidents_state "
|
||||
"ON cron_incidents(state)"
|
||||
)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _transaction() -> Iterator[sqlite3.Connection]:
|
||||
"""Open a connection, commit/rollback on exit, always close.
|
||||
|
||||
Mirrors ``cron.executions._transaction``: schema init runs inside the
|
||||
``try`` so a PRAGMA/DDL failure after a successful ``connect()`` still
|
||||
closes the connection instead of leaking it.
|
||||
"""
|
||||
with _lock:
|
||||
conn = _connect()
|
||||
try:
|
||||
_initialize_schema(conn)
|
||||
with conn:
|
||||
yield conn
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _normalize_error(error: str) -> str:
|
||||
"""Strip whitespace and lowercase before signing (dedup normalization)."""
|
||||
return re.sub(r"\s+", " ", str(error or "")).strip().lower()
|
||||
|
||||
|
||||
def _redact_error(error: str) -> str:
|
||||
"""Redact secrets then bound the stored error length."""
|
||||
text = str(error or "")
|
||||
try:
|
||||
from agent.redact import redact_sensitive_text
|
||||
|
||||
text = redact_sensitive_text(text)
|
||||
except Exception:
|
||||
# Redaction is best-effort; the scheduler path never fails on it.
|
||||
pass
|
||||
return text[:MAX_ERROR_CHARS]
|
||||
|
||||
|
||||
def _error_signature(job_id: str, error: str) -> str:
|
||||
"""Dedup key: stable for same job + same normalized error prefix."""
|
||||
normalized = _normalize_error(error)[:_MAX_SIGNATURE_ERROR_CHARS]
|
||||
digest = hashlib.sha256(job_id.encode() + normalized.encode()).hexdigest()
|
||||
return digest[:12]
|
||||
|
||||
|
||||
def _incident_id(job_id: str, error_sig: str) -> str:
|
||||
return f"{job_id[:6]}_{error_sig}"
|
||||
|
||||
|
||||
def _classify_failure_type(error: str) -> str:
|
||||
"""Classify a failure from error-text keywords; ``unknown`` is the default."""
|
||||
text = _normalize_error(error)
|
||||
if not text:
|
||||
return "unknown"
|
||||
for kind, patterns in _FAILURE_TYPE_ORDER:
|
||||
for pattern in patterns:
|
||||
if pattern.startswith("\\b") and pattern.endswith("\\b"):
|
||||
if re.search(pattern, text):
|
||||
return kind
|
||||
elif pattern in text:
|
||||
return kind
|
||||
return "unknown"
|
||||
|
||||
|
||||
def upsert_incident(
|
||||
job_id: str,
|
||||
error: str,
|
||||
*,
|
||||
job_name: Optional[str] = None,
|
||||
failure_type: Optional[str] = None,
|
||||
output_file: Optional[str] = None,
|
||||
) -> tuple[str, bool]:
|
||||
"""Record (or refresh) the incident for ``job_id`` + ``error``.
|
||||
|
||||
Returns ``(incident_id, is_new)``. A row for the same signature already
|
||||
existing refreshes ``last_seen_at``/``error``/``output_file`` and keeps its
|
||||
current state — a ``closed`` (acked) incident stays closed for the same
|
||||
signature. A changed error text mints a new incident automatically.
|
||||
"""
|
||||
job_id = str(job_id or "")
|
||||
sig = _error_signature(job_id, error)
|
||||
stored_error = _redact_error(error)
|
||||
incident_id = _incident_id(job_id, sig)
|
||||
now = _hermes_now().isoformat()
|
||||
failure_type = failure_type or _classify_failure_type(error)
|
||||
output_file = str(output_file) if output_file is not None else None
|
||||
|
||||
with _transaction() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT id FROM cron_incidents WHERE id=?", (incident_id,)
|
||||
).fetchone()
|
||||
if row is not None:
|
||||
conn.execute(
|
||||
"""UPDATE cron_incidents
|
||||
SET last_seen_at=?, error=?, output_file=?
|
||||
WHERE id=?""",
|
||||
(now, stored_error, output_file, incident_id),
|
||||
)
|
||||
return incident_id, False
|
||||
conn.execute(
|
||||
"""INSERT INTO cron_incidents
|
||||
(id, job_id, error_sig, state, failure_type,
|
||||
first_seen_at, last_seen_at, error, output_file)
|
||||
VALUES (?, ?, ?, 'detected', ?, ?, ?, ?, ?)""",
|
||||
(incident_id, job_id, sig, failure_type, now, now,
|
||||
stored_error, output_file),
|
||||
)
|
||||
return incident_id, True
|
||||
|
||||
|
||||
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: no transition (including back
|
||||
to ``alerted``) leaves it — re-open happens by the error changing and
|
||||
minting a NEW incident. Unknown states are rejected (no-op, ``False``).
|
||||
"""
|
||||
if state not in INCIDENT_STATES:
|
||||
return False
|
||||
now = _hermes_now().isoformat()
|
||||
with _transaction() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT state FROM cron_incidents WHERE id=?", (incident_id,)
|
||||
).fetchone()
|
||||
if row is None or row["state"] == state:
|
||||
return False
|
||||
if row["state"] == "closed":
|
||||
return False
|
||||
if state == "closed":
|
||||
conn.execute(
|
||||
"""UPDATE cron_incidents
|
||||
SET state='closed', closed_at=?, acked_at=?
|
||||
WHERE id=? AND state != 'closed'""",
|
||||
(now, now, incident_id),
|
||||
)
|
||||
else:
|
||||
conn.execute(
|
||||
"UPDATE cron_incidents SET state=? WHERE id=?",
|
||||
(state, incident_id),
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def ack_incident(incident_id: str) -> bool:
|
||||
"""Acknowledge (close) an incident; return whether the state changed.
|
||||
|
||||
A no-op (``False``) when the incident does not exist or is already closed.
|
||||
"""
|
||||
return set_incident_state(incident_id, "closed")
|
||||
|
||||
|
||||
def list_incidents(state: Optional[str] = None) -> List[Dict[str, Any]]:
|
||||
"""Return incidents, newest-activity first, optionally filtered by state."""
|
||||
if state is not None and state not in INCIDENT_STATES:
|
||||
return []
|
||||
with _transaction() as conn:
|
||||
if state is None:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM cron_incidents "
|
||||
"ORDER BY last_seen_at DESC, id DESC"
|
||||
).fetchall()
|
||||
else:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM cron_incidents WHERE state=? "
|
||||
"ORDER BY last_seen_at DESC, id DESC",
|
||||
(state,),
|
||||
).fetchall()
|
||||
return [dict(row) for row in rows]
|
||||
|
||||
|
||||
def get_incident(incident_id: str) -> Optional[Dict[str, Any]]:
|
||||
with _transaction() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT * FROM cron_incidents WHERE id=?", (incident_id,)
|
||||
).fetchone()
|
||||
return dict(row) if row is not None else None
|
||||
|
||||
|
||||
def count_incidents(state: Optional[str] = None) -> int:
|
||||
if state is not None and state not in INCIDENT_STATES:
|
||||
return 0
|
||||
with _transaction() as conn:
|
||||
if state is None:
|
||||
row = conn.execute("SELECT COUNT(*) AS n FROM cron_incidents").fetchone()
|
||||
else:
|
||||
row = conn.execute(
|
||||
"SELECT COUNT(*) AS n FROM cron_incidents WHERE state=?",
|
||||
(state,),
|
||||
).fetchone()
|
||||
return int(row["n"]) if row is not None else 0
|
||||
+133
-32
@@ -342,6 +342,61 @@ def _summarize_cron_failure_for_delivery(job: dict, error: str | None) -> str:
|
||||
return f"⚠️ Cron '{job_name}' failed: {cleaned}"
|
||||
|
||||
|
||||
def _upsert_incident_for_failure(
|
||||
job: dict, error: str, *, output_file: Optional[Any] = None
|
||||
) -> tuple[bool, Optional[str]]:
|
||||
"""Record a durable failure incident for this run.
|
||||
|
||||
The incident store groups "same job + same error signature" across runs so
|
||||
an operator-acked failure stops re-pinging every run. Returns
|
||||
``(acked, incident_id)``: ``acked`` is True when the incident for this
|
||||
exact signature is already ``closed`` (acked) — the per-run failure ping
|
||||
should be suppressed. ``incident_id`` lets the caller mark the incident
|
||||
``alerted`` after the ping actually goes out. The streak nudge and
|
||||
``_summarize_cron_failure_for_delivery`` text stay intact for un-acked
|
||||
failures.
|
||||
|
||||
Best-effort: an incident-store error must never break the cron run or the
|
||||
delivery path — failures are logged at debug and the caller delivers as if
|
||||
no incident 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") == "closed")
|
||||
return acked, incident_id
|
||||
except Exception as exc:
|
||||
logger.debug(
|
||||
"Incident store unavailable for job %s (delivery unaffected): %s",
|
||||
job["id"], exc,
|
||||
)
|
||||
return False, None
|
||||
|
||||
|
||||
def _mark_incident_alerted(incident_id: Optional[str]) -> None:
|
||||
"""Record that a failure ping for this incident reached delivery.
|
||||
|
||||
Best-effort like the upsert: bookkeeping never breaks the cron run.
|
||||
``set_incident_state`` is a no-op for closed incidents, so this can
|
||||
never resurrect an acked signature.
|
||||
"""
|
||||
if not incident_id:
|
||||
return
|
||||
try:
|
||||
from cron.incidents import set_incident_state
|
||||
|
||||
set_incident_state(incident_id, "alerted")
|
||||
except Exception as exc:
|
||||
logger.debug("Failed marking incident %s alerted: %s", incident_id, exc)
|
||||
|
||||
|
||||
class CronPromptInjectionBlocked(Exception):
|
||||
"""Raised by _build_job_prompt when the fully-assembled prompt trips the
|
||||
injection scanner. Caught in run_job so the operator sees a clean
|
||||
@@ -6768,6 +6823,11 @@ def _run_one_job_body(
|
||||
execution_id = create_execution(job["id"], source="direct")["id"]
|
||||
delivery_attempted = False
|
||||
delivery_error = None
|
||||
# Durable failure-incident bookkeeping for this run (see cron.incidents):
|
||||
# set on the failure paths below; consumed by the delivery_outcome
|
||||
# computation and the post-delivery "alerted" transition.
|
||||
incident_acked = False
|
||||
failure_incident_id = None
|
||||
try:
|
||||
# Pre-run dispatch claim (issue #38758): atomically commit a finite
|
||||
# one-shot's dispatch BEFORE its side effect runs, so a tick that dies
|
||||
@@ -6943,15 +7003,35 @@ def _run_one_job_body(
|
||||
"the configuration is fixed."
|
||||
)
|
||||
else:
|
||||
deliver_content = final_response if success else (
|
||||
_summarize_cron_failure_for_delivery(job, error)
|
||||
+ _failure_streak_nudge(job)
|
||||
)
|
||||
if success:
|
||||
deliver_content = final_response
|
||||
else:
|
||||
# Durable failure incident: record this job+error
|
||||
# signature once and, when the operator already acked it,
|
||||
# suppress the per-run failure ping (the streak nudge and
|
||||
# the failure summarizer stay intact for un-acked
|
||||
# failures). Best-effort — an incident-store error never
|
||||
# breaks the delivery path (see _upsert_incident_for_failure).
|
||||
incident_acked, failure_incident_id = _upsert_incident_for_failure(
|
||||
job, error or "", output_file=output_file
|
||||
)
|
||||
if incident_acked and not drift_skip:
|
||||
deliver_content = ""
|
||||
else:
|
||||
deliver_content = (
|
||||
_summarize_cron_failure_for_delivery(job, error)
|
||||
+ _failure_streak_nudge(job)
|
||||
)
|
||||
if drift_skip and not success:
|
||||
# Drift-skip alert: bypass the generic summarizer's
|
||||
# 180-char truncation (it would eat the remediation
|
||||
# command) and strip the internal marker — deliver the
|
||||
# guard's own actionable message intact.
|
||||
# Deliberately NOT gated on incident ack: a drift skip
|
||||
# means the run was never attempted and the message
|
||||
# carries the remediation command — acking the failure
|
||||
# signature silences failure pings, not drift alerts
|
||||
# (which already alert once via the drift_alerted marker).
|
||||
_drift_text = re.sub(
|
||||
r"\[drift_skip[^\]]*\]\s*", "", str(error)
|
||||
).strip()
|
||||
@@ -7092,8 +7172,18 @@ def _run_one_job_body(
|
||||
delivery_outcome = "not_configured"
|
||||
elif should_deliver and normalized_deliver != "local":
|
||||
delivery_outcome = "delivered"
|
||||
elif incident_acked and not success:
|
||||
# Distinct from plain "suppressed" (silence marker / local jobs):
|
||||
# the failure ping was withheld because the operator acked this
|
||||
# exact signature via `hermes cron incidents ack`.
|
||||
delivery_outcome = "suppressed_acked"
|
||||
else:
|
||||
delivery_outcome = "suppressed"
|
||||
if delivery_outcome in ("delivered", "not_configured") and not success:
|
||||
# The failure ping left the process (or was composed for a
|
||||
# configured target) — record it on the incident so the CLI
|
||||
# distinguishes "failure seen" from "operator was pinged".
|
||||
_mark_incident_alerted(failure_incident_id)
|
||||
finish_execution(
|
||||
execution_id,
|
||||
success=success,
|
||||
@@ -7130,34 +7220,45 @@ def _run_one_job_body(
|
||||
job.get("deliver", "local")
|
||||
)
|
||||
unresolved_origin = False
|
||||
try:
|
||||
delivery_attempted = True
|
||||
delivery_error = _deliver_result(
|
||||
job,
|
||||
# Composed exactly like the normal failure delivery above.
|
||||
# mark_job_run below records THIS run in failure_streak
|
||||
# whichever layer failed, so a job that fails before the
|
||||
# run body every tick builds a streak nobody is ever told
|
||||
# about: its alerts only ever leave through here, and the
|
||||
# nudge only ever left through there (#88655).
|
||||
_summarize_cron_failure_for_delivery(job, _err_text)
|
||||
+ _failure_streak_nudge(job),
|
||||
adapters=adapters,
|
||||
loop=loop,
|
||||
)
|
||||
except Exception as delivery_exc:
|
||||
delivery_error = str(delivery_exc)
|
||||
logger.error(
|
||||
"Delivery failed for job %s: %s", job["id"], delivery_exc
|
||||
)
|
||||
if not delivery_error and normalized_deliver == "origin":
|
||||
unresolved_origin = not _resolve_delivery_targets(job)
|
||||
if delivery_error:
|
||||
delivery_outcome = "failed"
|
||||
elif unresolved_origin:
|
||||
delivery_outcome = "not_configured"
|
||||
elif normalized_deliver != "local":
|
||||
delivery_outcome = "delivered"
|
||||
# Durable failure incident: same ack gate as the normal failure
|
||||
# delivery above — an acked signature stays silent on this path
|
||||
# too, so the retry-path alert cannot re-ping after acknowledgment.
|
||||
incident_acked, failure_incident_id = _upsert_incident_for_failure(
|
||||
job, _err_text
|
||||
)
|
||||
if incident_acked:
|
||||
delivery_outcome = "suppressed_acked"
|
||||
else:
|
||||
try:
|
||||
delivery_attempted = True
|
||||
delivery_error = _deliver_result(
|
||||
job,
|
||||
# Composed exactly like the normal failure delivery above.
|
||||
# mark_job_run below records THIS run in failure_streak
|
||||
# whichever layer failed, so a job that fails before the
|
||||
# run body every tick builds a streak nobody is ever told
|
||||
# about: its alerts only ever leave through here, and the
|
||||
# nudge only ever left through there (#88655).
|
||||
_summarize_cron_failure_for_delivery(job, _err_text)
|
||||
+ _failure_streak_nudge(job),
|
||||
adapters=adapters,
|
||||
loop=loop,
|
||||
)
|
||||
except Exception as delivery_exc:
|
||||
delivery_error = str(delivery_exc)
|
||||
logger.error(
|
||||
"Delivery failed for job %s: %s", job["id"], delivery_exc
|
||||
)
|
||||
if not delivery_error and normalized_deliver == "origin":
|
||||
unresolved_origin = not _resolve_delivery_targets(job)
|
||||
if delivery_error:
|
||||
delivery_outcome = "failed"
|
||||
elif unresolved_origin:
|
||||
delivery_outcome = "not_configured"
|
||||
elif normalized_deliver != "local":
|
||||
delivery_outcome = "delivered"
|
||||
if delivery_outcome in ("delivered", "not_configured"):
|
||||
_mark_incident_alerted(failure_incident_id)
|
||||
try:
|
||||
if not _consume_interrupted_flag(job["id"], execution_token):
|
||||
mark_kwargs = {}
|
||||
|
||||
@@ -6,6 +6,7 @@ pause/resume/run/remove, status, and tick.
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Iterable, List, Optional
|
||||
@@ -258,6 +259,104 @@ def cron_runs(job_id: Optional[str] = None, limit: int = 20):
|
||||
print(f" {record['error']}")
|
||||
|
||||
|
||||
_INCIDENT_STATE_COLORS = {
|
||||
"detected": Colors.RED,
|
||||
"alerted": Colors.YELLOW,
|
||||
"closed": Colors.GREEN,
|
||||
}
|
||||
|
||||
|
||||
def cron_incidents(args) -> int:
|
||||
"""List or acknowledge durable cron failure incidents.
|
||||
|
||||
``hermes cron incidents [--state <s>]`` lists incidents (the stored error
|
||||
is redacted and truncated at write time, safe for terminal display);
|
||||
``hermes cron incidents ack <id>`` closes one so its failure ping stays
|
||||
silent until the error signature changes.
|
||||
"""
|
||||
from cron.incidents import ack_incident, list_incidents
|
||||
|
||||
action = getattr(args, "incident_action", "list")
|
||||
if action == "ack":
|
||||
incident_id = getattr(args, "incident_id", None)
|
||||
if not incident_id:
|
||||
print(
|
||||
color(
|
||||
"✗ Incident ID required: hermes cron incidents ack <incident_id>",
|
||||
Colors.RED,
|
||||
)
|
||||
)
|
||||
return 1
|
||||
if ack_incident(incident_id):
|
||||
print(
|
||||
color(
|
||||
f"✓ Incident {incident_id} acknowledged (closed).",
|
||||
Colors.GREEN,
|
||||
)
|
||||
)
|
||||
else:
|
||||
print(
|
||||
color(
|
||||
f"Incident {incident_id} not found or already closed.",
|
||||
Colors.YELLOW,
|
||||
)
|
||||
)
|
||||
return 0
|
||||
|
||||
state = getattr(args, "state", None)
|
||||
incidents = list_incidents(state=state)
|
||||
if not incidents:
|
||||
print(color("No cron failure incidents recorded.", Colors.DIM))
|
||||
if state:
|
||||
print(color(f" (filtered by state '{state}')", Colors.DIM))
|
||||
return 0
|
||||
|
||||
print()
|
||||
print(
|
||||
color(
|
||||
"┌─────────────────────────────────────────────────────────────────────────┐",
|
||||
Colors.CYAN,
|
||||
)
|
||||
)
|
||||
print(
|
||||
color(
|
||||
"│ Cron Failure Incidents │",
|
||||
Colors.CYAN,
|
||||
)
|
||||
)
|
||||
print(
|
||||
color(
|
||||
"└─────────────────────────────────────────────────────────────────────────┘",
|
||||
Colors.CYAN,
|
||||
)
|
||||
)
|
||||
print()
|
||||
for inc in incidents:
|
||||
state_display = color(
|
||||
inc["state"], _INCIDENT_STATE_COLORS.get(inc["state"], Colors.DIM)
|
||||
)
|
||||
print(f" {color(inc['id'], Colors.YELLOW)} {state_display}")
|
||||
print(f" Job: {inc['job_id']}")
|
||||
print(f" Type: {inc.get('failure_type', 'unknown')}")
|
||||
print(f" First seen: {inc.get('first_seen_at', '?')}")
|
||||
print(f" Last seen: {inc.get('last_seen_at', '?')}")
|
||||
error_text = re.sub(r"\s+", " ", inc.get("error") or "").strip()
|
||||
if len(error_text) > 160:
|
||||
error_text = error_text[:157].rstrip() + "..."
|
||||
print(f" Error: {error_text}")
|
||||
if inc.get("output_file"):
|
||||
print(f" Output: {inc['output_file']}")
|
||||
print()
|
||||
print(
|
||||
color(
|
||||
f" {len(incidents)} incident(s) | ack one with: "
|
||||
"hermes cron incidents ack <id>",
|
||||
Colors.DIM,
|
||||
)
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def cron_status():
|
||||
"""Show cron execution status."""
|
||||
from cron.jobs import list_jobs
|
||||
@@ -680,6 +779,9 @@ def cron_command(args):
|
||||
cron_runs(getattr(args, "job_id", None), getattr(args, "limit", 20))
|
||||
return 0
|
||||
|
||||
if subcmd == "incidents":
|
||||
return cron_incidents(args)
|
||||
|
||||
if subcmd == "notepad":
|
||||
return cron_notepad(args)
|
||||
|
||||
|
||||
@@ -284,6 +284,26 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
|
||||
cron_runs.add_argument("job_id", nargs="?", help="Optional job ID filter")
|
||||
cron_runs.add_argument("--limit", type=int, default=20, help="Rows to show (1-500)")
|
||||
|
||||
# cron incidents — durable failure incidents (list/ack)
|
||||
cron_incidents = cron_subparsers.add_parser(
|
||||
"incidents", help="List or acknowledge durable cron failure incidents"
|
||||
)
|
||||
cron_incidents.add_argument(
|
||||
"--state",
|
||||
choices=["detected", "alerted", "closed"],
|
||||
help="Filter incidents by lifecycle state",
|
||||
)
|
||||
cron_incidents.add_argument(
|
||||
"incident_action",
|
||||
nargs="?",
|
||||
default="list",
|
||||
choices=["list", "ack"],
|
||||
help="Action (default: list)",
|
||||
)
|
||||
cron_incidents.add_argument(
|
||||
"incident_id", nargs="?", help="Incident ID to acknowledge (ack)"
|
||||
)
|
||||
|
||||
# cron notepad — per-job durable KV scratchpad (injected into the job
|
||||
# prompt each run; the running agent writes it via this CLI).
|
||||
cron_notepad = cron_subparsers.add_parser(
|
||||
|
||||
@@ -0,0 +1,329 @@
|
||||
"""Durable cron failure incidents: signature dedup, lifecycle, ack, CLI."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
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
|
||||
|
||||
|
||||
def _point_db(monkeypatch, tmp_path):
|
||||
"""Point the incident store at a throwaway executions.db (same file shape
|
||||
the scheduler uses). ``cron.executions.EXECUTIONS_FILE`` stays None so the
|
||||
incident store falls back to its own override."""
|
||||
monkeypatch.setattr(incidents, "EXECUTIONS_FILE", tmp_path / "cron" / "executions.db")
|
||||
return incidents
|
||||
|
||||
|
||||
def _job(**overrides):
|
||||
job = {
|
||||
"id": "incident-gating-test",
|
||||
"name": "incident gating test",
|
||||
"prompt": "hello",
|
||||
"enabled": True,
|
||||
"state": "scheduled",
|
||||
"schedule": {"kind": "interval", "minutes": 5, "display": "every 5m"},
|
||||
"deliver": "local",
|
||||
"model": None,
|
||||
"provider": None,
|
||||
"provider_snapshot": "openrouter",
|
||||
"base_url": None,
|
||||
}
|
||||
job.update(overrides)
|
||||
return job
|
||||
|
||||
|
||||
def _tick_failing(job, tmp_path, deliveries, error="boom unrelated"):
|
||||
"""Run one run_one_job tick whose agent raises ``error`` (the failure
|
||||
path that composes the per-run failure ping). Mirrors the drift-alert-once
|
||||
harness so the incident gating is exercised through the real scheduler."""
|
||||
fake_db = MagicMock()
|
||||
|
||||
def fake_deliver(jb, content, adapters=None, loop=None):
|
||||
deliveries.append(content)
|
||||
return None
|
||||
|
||||
with cron_jobs.use_cron_store(tmp_path), \
|
||||
patch("cron.scheduler._hermes_home", tmp_path), \
|
||||
patch("cron.scheduler._resolve_origin", return_value=None), \
|
||||
patch("hermes_cli.env_loader.load_hermes_dotenv"), \
|
||||
patch("hermes_cli.env_loader.reset_secret_source_cache"), \
|
||||
patch("hermes_state.SessionDB", return_value=fake_db), \
|
||||
patch("tools.mcp_tool.discover_mcp_tools", return_value=[]), \
|
||||
patch("hermes_cli.runtime_provider.resolve_runtime_provider",
|
||||
return_value={
|
||||
"api_key": "test-key",
|
||||
"base_url": "https://example.invalid/v1",
|
||||
"provider": "openrouter",
|
||||
"api_mode": "chat_completions",
|
||||
}), \
|
||||
patch.object(sched, "_deliver_result", side_effect=fake_deliver), \
|
||||
patch("run_agent.AIAgent") as mock_agent_cls:
|
||||
mock_agent = MagicMock()
|
||||
mock_agent.run_conversation.side_effect = RuntimeError(error)
|
||||
mock_agent_cls.return_value = mock_agent
|
||||
sched.run_one_job(dict(job))
|
||||
return mock_agent_cls.called
|
||||
|
||||
|
||||
# ── Store + dedup ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_new_failure_creates_incident_and_is_new(monkeypatch, tmp_path):
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
|
||||
inc_id, is_new = inc.upsert_incident("job-1", "Provider timeout: read timed out")
|
||||
|
||||
assert is_new is True
|
||||
assert inc_id.startswith("job-1_")
|
||||
row = inc.get_incident(inc_id)
|
||||
assert row is not None
|
||||
assert row["job_id"] == "job-1"
|
||||
assert row["state"] == "detected"
|
||||
assert row["failure_type"] == "timeout"
|
||||
assert row["first_seen_at"] == row["last_seen_at"]
|
||||
assert inc.count_incidents() == 1
|
||||
|
||||
|
||||
def test_same_signature_dedups_same_incident(monkeypatch, tmp_path):
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
|
||||
id1, new1 = inc.upsert_incident("job-1", "Provider timeout: read timed out")
|
||||
id2, new2 = inc.upsert_incident("job-1", "PROVIDER TIMEOUT: read timed out ")
|
||||
|
||||
assert id1 == id2, "normalized (case/whitespace) signature must dedup"
|
||||
assert new1 is True
|
||||
assert new2 is False
|
||||
assert inc.count_incidents() == 1
|
||||
# Refresh updates last_seen but never resets an open state.
|
||||
assert inc.get_incident(id1)["state"] == "detected"
|
||||
|
||||
|
||||
def test_error_change_mints_new_incident(monkeypatch, tmp_path):
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
|
||||
id1, _ = inc.upsert_incident("job-1", "provider timeout")
|
||||
id2, new2 = inc.upsert_incident("job-1", "provider rate limit 429")
|
||||
|
||||
assert id1 != id2
|
||||
assert new2 is True
|
||||
assert inc.count_incidents() == 2
|
||||
|
||||
|
||||
# ── Redaction / classification ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_redaction_applied_to_incident_error(monkeypatch, tmp_path):
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
secret = "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij"
|
||||
|
||||
inc_id, _ = inc.upsert_incident("job-1", f"failed: {secret} boom")
|
||||
|
||||
row = inc.get_incident(inc_id)
|
||||
assert secret not in row["error"]
|
||||
assert "boom" in row["error"]
|
||||
|
||||
|
||||
def test_error_truncated_to_bounded_length(monkeypatch, tmp_path):
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
long_error = "x" * 2000
|
||||
|
||||
inc_id, _ = inc.upsert_incident("job-1", long_error)
|
||||
|
||||
assert len(inc.get_incident(inc_id)["error"]) <= 500
|
||||
|
||||
|
||||
def test_failure_type_classification(monkeypatch, tmp_path):
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
cases = [
|
||||
("delivery failed for telegram chat", "delivery"),
|
||||
("Provider read timed out after 60s", "timeout"),
|
||||
("authentication failed: invalid API key", "auth"),
|
||||
("HTTP 429: rate limit exceeded", "rate_limit"),
|
||||
("configuration validation blocked the run", "config"),
|
||||
("script exited with code 1", "script"),
|
||||
("agent crashed mid-conversation", "agent"),
|
||||
("something completely unexpected happened", "unknown"),
|
||||
]
|
||||
for error, expected in cases:
|
||||
assert inc._classify_failure_type(error) == expected, (error, expected)
|
||||
|
||||
|
||||
# ── Lifecycle / ack ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_lifecycle_transitions(monkeypatch, tmp_path):
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
inc_id, _ = inc.upsert_incident("job-1", "boom")
|
||||
|
||||
assert inc.get_incident(inc_id)["state"] == "detected"
|
||||
assert inc.set_incident_state(inc_id, "alerted") is True
|
||||
assert inc.set_incident_state(inc_id, "closed") is True
|
||||
row = inc.get_incident(inc_id)
|
||||
assert row["state"] == "closed"
|
||||
assert row["acked_at"] and row["closed_at"]
|
||||
|
||||
# Closed is terminal for that signature: no re-open, no re-transition.
|
||||
assert inc.set_incident_state(inc_id, "alerted") is False
|
||||
assert inc.set_incident_state(inc_id, "closed") is False
|
||||
assert inc.ack_incident(inc_id) is False
|
||||
assert inc.get_incident(inc_id)["state"] == "closed"
|
||||
|
||||
# Invalid states are rejected, not raised.
|
||||
assert inc.set_incident_state(inc_id, "bogus") is False
|
||||
assert inc.list_incidents(state="bogus") == []
|
||||
assert inc.count_incidents(state="bogus") == 0
|
||||
|
||||
|
||||
def test_acked_signature_stays_closed_on_refresh(monkeypatch, tmp_path):
|
||||
"""Ack is per-signature: upserting the same error after ack must NOT
|
||||
resurrect the incident — a changed error is what mints a new one."""
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
inc_id, _ = inc.upsert_incident("job-1", "same failure text")
|
||||
inc.ack_incident(inc_id)
|
||||
|
||||
same_id, is_new = inc.upsert_incident("job-1", "SAME FAILURE TEXT")
|
||||
|
||||
assert same_id == inc_id
|
||||
assert is_new is False
|
||||
assert inc.get_incident(inc_id)["state"] == "closed"
|
||||
|
||||
|
||||
# ── Missing DB / lazy schema ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_missing_db_no_crash(monkeypatch, tmp_path):
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
|
||||
inc_id, is_new = inc.upsert_incident("job-1", "boom")
|
||||
|
||||
assert is_new is True
|
||||
assert (tmp_path / "cron" / "executions.db").is_file()
|
||||
assert inc.list_incidents() == [inc.get_incident(inc_id)]
|
||||
assert inc.count_incidents() == 1
|
||||
assert inc.get_incident("nope") is None
|
||||
|
||||
|
||||
# ── Scheduler gating ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_unacked_failure_still_alerts(monkeypatch, tmp_path):
|
||||
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="unacked boom")
|
||||
_tick_failing(job, tmp_path, deliveries, error="unacked boom")
|
||||
|
||||
assert len(deliveries) == 2, "unacked failures must keep alerting per run"
|
||||
rows = inc.list_incidents()
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["state"] == "detected"
|
||||
|
||||
|
||||
def test_ack_suppresses_alert_until_signature_changes(monkeypatch, tmp_path):
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
deliveries = []
|
||||
job = _job()
|
||||
with cron_jobs.use_cron_store(tmp_path):
|
||||
cron_jobs.save_jobs([job])
|
||||
# First failure: alert delivered, incident minted.
|
||||
_tick_failing(job, tmp_path, deliveries, error="boom signature A")
|
||||
assert len(deliveries) == 1
|
||||
rows = inc.list_incidents()
|
||||
assert len(rows) == 1 and rows[0]["state"] == "detected"
|
||||
inc_id = rows[0]["id"]
|
||||
|
||||
# Acknowledge it.
|
||||
assert inc.ack_incident(inc_id) is True
|
||||
|
||||
# Same signature: alert suppressed, incident stays closed.
|
||||
_tick_failing(job, tmp_path, deliveries, error="boom signature A")
|
||||
assert len(deliveries) == 1, "acked signature must not re-ping"
|
||||
assert inc.get_incident(inc_id)["state"] == "closed"
|
||||
|
||||
# Changed signature: new incident, alert again.
|
||||
_tick_failing(job, tmp_path, deliveries, error="boom signature B")
|
||||
assert len(deliveries) == 2, "changed signature must re-alert"
|
||||
assert inc.count_incidents() == 2
|
||||
|
||||
|
||||
def test_mark_incident_alerted_sets_state_never_resurrects(monkeypatch, tmp_path):
|
||||
"""The post-delivery 'alerted' transition records that a ping went out,
|
||||
and is a no-op on a closed (acked) incident — it can never resurrect one."""
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
|
||||
inc_id, _ = inc.upsert_incident("job-1", "boom")
|
||||
sched._mark_incident_alerted(inc_id)
|
||||
assert inc.get_incident(inc_id)["state"] == "alerted"
|
||||
|
||||
inc.ack_incident(inc_id)
|
||||
sched._mark_incident_alerted(inc_id)
|
||||
assert inc.get_incident(inc_id)["state"] == "closed"
|
||||
|
||||
# Best-effort: bad/missing ids never raise.
|
||||
sched._mark_incident_alerted(None)
|
||||
sched._mark_incident_alerted("nonexistent")
|
||||
|
||||
|
||||
def test_best_effort_incident_store_failure_returns_false(monkeypatch, tmp_path):
|
||||
"""An incident-store error must never break the cron delivery path."""
|
||||
_point_db(monkeypatch, tmp_path)
|
||||
with patch("cron.incidents.upsert_incident",
|
||||
side_effect=RuntimeError("db locked")):
|
||||
assert sched._upsert_incident_for_failure(_job(), "boom") == (False, None)
|
||||
|
||||
|
||||
# ── CLI ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_cli_list_and_ack(monkeypatch, tmp_path, capsys):
|
||||
from hermes_cli.cron import cron_incidents
|
||||
|
||||
inc = _point_db(monkeypatch, tmp_path)
|
||||
inc_id, _ = inc.upsert_incident("job-1", "provider timeout boom")
|
||||
|
||||
# List.
|
||||
list_args = argparse.Namespace(
|
||||
incident_action="list", state=None, incident_id=None
|
||||
)
|
||||
assert cron_incidents(list_args) == 0
|
||||
out = capsys.readouterr().out
|
||||
assert inc_id in out
|
||||
assert "job-1" in out
|
||||
|
||||
# State filter.
|
||||
filter_args = argparse.Namespace(
|
||||
incident_action="list", state="closed", incident_id=None
|
||||
)
|
||||
assert cron_incidents(filter_args) == 0
|
||||
out = capsys.readouterr().out
|
||||
assert "No cron failure incidents recorded." in out
|
||||
|
||||
# Ack.
|
||||
ack_args = argparse.Namespace(
|
||||
incident_action="ack", state=None, incident_id=inc_id
|
||||
)
|
||||
assert cron_incidents(ack_args) == 0
|
||||
assert inc.get_incident(inc_id)["state"] == "closed"
|
||||
out = capsys.readouterr().out
|
||||
assert "acknowledged" in out.lower()
|
||||
|
||||
# Ack again: already closed, still a clean exit.
|
||||
assert cron_incidents(ack_args) == 0
|
||||
out = capsys.readouterr().out
|
||||
assert "already closed" in out.lower()
|
||||
|
||||
# Ack with a missing id is a usage error.
|
||||
missing_args = argparse.Namespace(
|
||||
incident_action="ack", state=None, incident_id=None
|
||||
)
|
||||
assert cron_incidents(missing_args) == 1
|
||||
@@ -349,6 +349,34 @@ cron:
|
||||
failure_nudge_threshold: 3 # default; 0 disables the nudge
|
||||
```
|
||||
|
||||
### Failure incidents: acknowledge a known failure
|
||||
|
||||
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.
|
||||
|
||||
```bash
|
||||
hermes cron incidents # list incidents (newest activity first)
|
||||
hermes cron incidents --state alerted # filter: detected | alerted | closed
|
||||
hermes cron incidents ack <id> # acknowledge — stop re-pinging
|
||||
```
|
||||
|
||||
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
|
||||
failing with a *different* error a new incident is minted and alerts fire
|
||||
again. A successful run doesn't touch incidents — they are per-signature, not
|
||||
per-job.
|
||||
|
||||
Incident lifecycle: `detected` (failure recorded) → `alerted` (at least one
|
||||
failure ping reached delivery) → `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`.
|
||||
|
||||
## Delivery options
|
||||
|
||||
When scheduling jobs, you specify where the output goes:
|
||||
|
||||
Reference in New Issue
Block a user