feat(cron): persist incident reopen_count with a truthful reopen flag

cron/incidents.py

@@ -7,7 +7,9 @@
 until the error text changes and mints a new one. ``alerted`` means a failure ping actually reached
 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).
+``cron.executions`` (one ledger file). ``reopen_count`` tracks recurrence: how many times a
+resolved incident re-opened under the same signature. Rows predating the counter carry NULL
+(*not measured*) — never a silent 0, which would falsely attest "never reopened".
 """
 
 from __future__ import annotations
@@ -88,11 +90,18 @@ def _initialize_schema(conn: sqlite3.Connection) -> None:
              alerted_at    TEXT,
              closed_at     TEXT,
              error         TEXT NOT NULL,
-             output_file   TEXT
+             output_file   TEXT,
+             reopen_count  INTEGER
            )"""
     )
     # 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")
+    # Ledgers predating the reopen counter lack ``reopen_count``. The ddl is deliberately NULLABLE
+    # with NO DEFAULT: ``add_column_if_missing`` is a raw ALTER TABLE pass-through, and a DEFAULT
+    # would backfill every existing row with the default value — silently attesting "never
+    # reopened" for rows that were never counted. Existing rows land NULL (not measured); the
+    # write path sets 0 on INSERT and increments on reopen.
+    add_column_if_missing(conn, "cron_incidents", "reopen_count", "reopen_count INTEGER")
     conn.execute(
         "CREATE INDEX IF NOT EXISTS idx_cron_incidents_job "
         "ON cron_incidents(job_id)"
@@ -158,10 +167,13 @@ def upsert_incident(
     output_file: Optional[str] = None,
 ) -> tuple[str, bool]:
     """Record (or refresh) the incident for ``job_id`` + ``error``; returns ``(incident_id,
-    is_new)``. An existing row for the signature refreshes
+    reopened)``, where ``reopened`` is True *only* when an existing ``resolved`` incident re-opened.
+    A first-ever INSERT and a plain refresh of an open incident both return False. An existing row
+    for the signature refreshes
     ``last_seen_at``/``error``/``output_file`` and keeps its state — a ``closed`` incident stays
     closed, while a ``resolved`` one (job recovered, then broke the same way again) re-opens as
-    ``detected`` so the operator is alerted once more. A changed error text mints a new incident."""
+    ``detected``, clears its alert/close stamps, and increments ``reopen_count`` so the operator is
+    alerted once more with the recurrence visible. A changed error text mints a new incident."""
     job_id = str(job_id or "")
     sig = _error_signature(job_id, error)
     stored_error = _redact_error(error)
@@ -176,25 +188,33 @@ def upsert_incident(
         ).fetchone()
         if row is not None:
             reopen = row["state"] == "resolved"
-            conn.execute(
-                """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),
-            )
+            if reopen:
+                # COALESCE: a pre-counter row (NULL) starts its history at 1, not at NULL+1.
+                conn.execute(
+                    """UPDATE cron_incidents
+                       SET last_seen_at=?, error=?, output_file=?, state='detected',
+                           alerted_at=NULL, closed_at=NULL,
+                           reopen_count=COALESCE(reopen_count, 0) + 1
+                       WHERE id=?""",
+                    (now, stored_error, output_file, incident_id),
+                )
+            else:
+                conn.execute(
+                    """UPDATE cron_incidents
+                       SET last_seen_at=?, error=?, output_file=?
+                       WHERE id=?""",
+                    (now, stored_error, output_file, incident_id),
+                )
             return incident_id, reopen
         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', ?, ?, ?, ?, ?)""",
+                first_seen_at, last_seen_at, error, output_file, reopen_count)
+               VALUES (?, ?, ?, 'detected', ?, ?, ?, ?, ?, 0)""",
             (incident_id, job_id, sig, failure_type, now, now,
              stored_error, output_file),
         )
-        return incident_id, True
+        return incident_id, False
 
 
 def set_incident_state(incident_id: str, state: str) -> bool:

cron/scheduler.py

@@ -397,7 +397,7 @@ def _upsert_incident_for_failure(
     try:
         from cron.incidents import get_incident, upsert_incident
 
-        incident_id, _is_new = upsert_incident(
+        incident_id, _reopened = upsert_incident(
             job["id"], str(error or ""), job_name=job.get("name"), output_file=output_file)
         incident = get_incident(incident_id)
         state = incident.get("state") if incident else None

hermes_cli/cron.py

@@ -383,9 +383,13 @@ def cron_incidents(args) -> int:
         error_text = re.sub(r"\s+", " ", inc.get("error") or "").strip()
         if len(error_text) > 160:
             error_text = error_text[:157].rstrip() + "..."
+        # A pre-counter row (NULL) says so; 0 is a measured truth, not an absence.
+        reopen_display = ("pre-counter" if inc.get("reopen_count") is None
+                          else f"{inc['reopen_count']}x")
         rows = [("Job", inc["job_id"]), ("Type", inc.get("failure_type", "unknown")),
                 ("First seen", inc.get("first_seen_at", "?")),
-                ("Last seen", inc.get("last_seen_at", "?")), ("Error", error_text),
+                ("Last seen", inc.get("last_seen_at", "?")),
+                ("Reopened", reopen_display), ("Error", error_text),
                 ("Output", inc.get("output_file"))]
         print(f"  {color(inc['id'], Colors.YELLOW)}  {state_display}")
         for label, value in rows:

hermes_cli/sqlite_util.py

@@ -85,6 +85,13 @@ def add_column_if_missing(conn: sqlite3.Connection, table: str, column: str, ddl
 
     ``column`` is the human-readable name for the call site; ``ddl`` carries the actual definition. See
     #21708.
+
+    CAUTION — this is a raw ``ALTER TABLE`` pass-through: the only guard is swallowing the
+    concurrent-migrator duplicate-column error. A ``DEFAULT`` in ``ddl`` is applied to EVERY
+    EXISTING row, so a default quietly attests its value about all pre-migration history (e.g.
+    ``... NOT NULL DEFAULT 0`` makes unmeasured rows read "zero"). When old rows must read as
+    unmeasured, pass a nullable ``ddl`` with NO DEFAULT and set values explicitly on the write
+    path; the review gate is the ddl string at the call site, not this function's name.
     """
     try:
         conn.execute(f"ALTER TABLE {table} ADD COLUMN {ddl}")

tests/cron/test_cron_incidents.py

@@ -81,9 +81,11 @@ def fake_deliver(jb, content, adapters=None, loop=None, **kwargs):
 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")
+    # ``reopened`` is True only for a resolved incident re-opening; a first-ever row is a
+    # creation (proven by the row below) and returns False.
+    inc_id, reopened = inc.upsert_incident("job-1", "Provider timeout: read timed out")
 
-    assert is_new is True
+    assert reopened is False
     assert inc_id.startswith("job-1_")
     row = inc.get_incident(inc_id)
     assert row is not None
@@ -116,12 +118,11 @@ def test_incidents_list_newest_instant_first_across_dst_fall_back(monkeypatch, t
 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   ")
+    id1, reopened1 = inc.upsert_incident("job-1", "Provider timeout: read timed out")
+    id2, reopened2 = 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 reopened1 is False and reopened2 is False, "neither write re-opened anything"
     assert inc.count_incidents() == 1
     # Refresh updates last_seen but never resets an open state.
     assert inc.get_incident(id1)["state"] == "detected"
@@ -131,10 +132,10 @@ 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")
+    id2, reopened2 = inc.upsert_incident("job-1", "provider rate limit 429")
 
     assert id1 != id2
-    assert new2 is True
+    assert reopened2 is False, "a freshly minted signature is a creation, not a re-open"
     assert inc.count_incidents() == 2
 
 
@@ -231,9 +232,9 @@ def test_acked_signature_stays_closed_on_refresh(monkeypatch, tmp_path):
 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")
+    inc_id, reopened = inc.upsert_incident("job-1", "boom")
 
-    assert is_new is True
+    assert reopened is False, "creation is not a re-open; the row asserts creation"
     assert (tmp_path / "cron" / "executions.db").is_file()
     assert inc.list_incidents() == [inc.get_incident(inc_id)]
     assert inc.count_incidents() == 1
@@ -425,3 +426,77 @@ def test_alerted_gate_honours_cooldown_opt_out_and_legacy_rows(monkeypatch, tmp_
     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)
+
+
+
+# ── reopen counter (recurrence visibility) ─────────────────────────────────
+
+
+def test_reopen_count_tracks_recurrence(monkeypatch, tmp_path):
+    inc = _point_db(monkeypatch, tmp_path)
+
+    iid, reopened = inc.upsert_incident("job-rec", "chronic boom")
+    assert reopened is False and inc.get_incident(iid)["reopen_count"] == 0
+
+    # Refresh while open: not a re-open, counter stays.
+    iid2, reopened = inc.upsert_incident("job-rec", "chronic boom")
+    assert iid2 == iid and reopened is False
+    assert inc.get_incident(iid)["reopen_count"] == 0
+
+    # Recover, then break the same way again: the resolved row re-opens and counts.
+    assert inc.close_incidents_for_recovered_job("job-rec") == 1
+    _, reopened = inc.upsert_incident("job-rec", "chronic boom")
+    assert reopened is True
+    row = inc.get_incident(iid)
+    assert row["state"] == "detected" and row["reopen_count"] == 1
+    assert row["alerted_at"] is None and row["closed_at"] is None
+
+    # Chronic incidents grow a history instead of flattening into one shape.
+    inc.close_incidents_for_recovered_job("job-rec")
+    _, reopened = inc.upsert_incident("job-rec", "chronic boom")
+    assert reopened is True and inc.get_incident(iid)["reopen_count"] == 2
+
+
+def test_pre_counter_rows_migrate_to_null_not_zero(monkeypatch, tmp_path):
+    """Old-shape ledgers (no ``reopen_count``) must read *not measured*, never a silent 0: a
+    DEFAULT on the ALTER would attest "never reopened" for rows that were never counted."""
+    import sqlite3
+
+    inc = _point_db(monkeypatch, tmp_path)
+    path = inc.EXECUTIONS_FILE
+    path.parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(path)
+    conn.execute(
+        """CREATE TABLE 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,
+             alerted_at TEXT, closed_at TEXT, error TEXT NOT NULL, output_file TEXT)"""
+    )
+    conn.execute(
+        "INSERT INTO cron_incidents (id, job_id, error_sig, state, first_seen_at, "
+        "last_seen_at, error) VALUES ('old1', 'job-old', 'deadbeefcafe', 'closed', "
+        "'2026-01-01T00:00:00-05:00', '2026-01-02T00:00:00-05:00', 'ancient failure')")
+    conn.commit()
+    conn.close()
+
+    row = inc.get_incident("old1")  # opening the store runs _initialize_schema in place
+    assert row is not None and row["state"] == "closed"
+    assert row["reopen_count"] is None, "pre-counter history must not inherit a measured 0"
+
+
+def test_reopen_from_null_pre_counter_counts_one(monkeypatch, tmp_path):
+    """The COALESCE path: a re-open where the counter is NULL starts at 1, not NULL+1."""
+    import sqlite3
+
+    inc = _point_db(monkeypatch, tmp_path)
+    iid, _ = inc.upsert_incident("job-null", "boom again")
+    conn = sqlite3.connect(inc.EXECUTIONS_FILE)
+    conn.execute("UPDATE cron_incidents SET reopen_count=NULL WHERE id=?", (iid,))
+    conn.commit()
+    conn.close()
+
+    assert inc.close_incidents_for_recovered_job("job-null") == 1
+    _, reopened = inc.upsert_incident("job-null", "boom again")
+    assert reopened is True
+    assert inc.get_incident(iid)["reopen_count"] == 1