diff --git a/agent/monitoring/cron_health.py b/agent/monitoring/cron_health.py
index 2ed0f1cb9b..2c4fe8d35f 100644
--- a/agent/monitoring/cron_health.py
+++ b/agent/monitoring/cron_health.py
@@ -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)
diff --git a/cron/incidents.py b/cron/incidents.py
new file mode 100644
index 0000000000..96f8a48c2a
--- /dev/null
+++ b/cron/incidents.py
@@ -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
diff --git a/cron/scheduler.py b/cron/scheduler.py
index b721b57217..11b3e91314 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -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 = {}
diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py
index 3bf2823d6b..486eaba951 100644
--- a/hermes_cli/cron.py
+++ b/hermes_cli/cron.py
@@ -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)
 
diff --git a/hermes_cli/subcommands/cron.py b/hermes_cli/subcommands/cron.py
index 2588ad071d..93441703ef 100644
--- a/hermes_cli/subcommands/cron.py
+++ b/hermes_cli/subcommands/cron.py
@@ -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(
diff --git a/tests/cron/test_cron_incidents.py b/tests/cron/test_cron_incidents.py
new file mode 100644
index 0000000000..4e1b93dc4c
--- /dev/null
+++ b/tests/cron/test_cron_incidents.py
@@ -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
diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md
index cb91709320..7e8e477be5 100644
--- a/website/docs/user-guide/features/cron.md
+++ b/website/docs/user-guide/features/cron.md
@@ -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: