fix(cron): surface missed-fire catch-up lateness in hermes cron list/status

cron/jobs.py

@@ -1213,6 +1213,29 @@ def _compute_grace_seconds(schedule: dict) -> int:
     return max(MIN_GRACE, min(grace, MAX_GRACE))
 
 
+# Missed-run visibility (#99879): a recurring dispatch within this many
+# seconds of its scheduled instant renders as "on time". The built-in ticker
+# runs once a minute and a busy tick can push dispatch a couple of minutes
+# past the scheduled instant — that is normal cadence, not gateway downtime.
+_LATE_DISPATCH_TOLERANCE_SECONDS = 300
+
+
+def _classify_dispatch_lateness(lateness_seconds: float, grace_seconds: int) -> str:
+    """Classify a recurring dispatch by how late it fired.
+
+    ``on_time``  — within normal ticker cadence slack;
+    ``late``     — missed the scheduled instant but within the catch-up
+                   grace window (e.g. gateway briefly down);
+    ``catch_up`` — beyond the grace window; the due-scan skipped the
+                   accumulated misses and executed once now.
+    """
+    if lateness_seconds > grace_seconds:
+        return "catch_up"
+    if lateness_seconds > _LATE_DISPATCH_TOLERANCE_SECONDS:
+        return "late"
+    return "on_time"
+
+
 # Durable (persisted-state) recovery counter for a recurring job wedged in a
 # stale ``last_status == "error"`` state with ``next_run_at`` parked in the
 # future.  This is the restart-surviving half of the recurring-cron wedge
@@ -4185,6 +4208,28 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
                             needs_save = True
                             break
 
+                # Missed-run visibility (#99879): persist scheduled-vs-actual
+                # dispatch timing on the job record so `hermes cron list` /
+                # `hermes cron status` (separate CLI processes) can show a
+                # late catch-up run as such instead of an ordinary on-time
+                # run. Recurring schedules only — one-shots beyond grace are
+                # retired above, and manual triggers have no scheduled
+                # instant to be late against.
+                if not manual_run and kind in {"cron", "interval"}:
+                    lateness = max(0.0, (now - next_run_dt).total_seconds())
+                    dispatch_stamp = {
+                        "scheduled_at": next_run,
+                        "dispatched_at": now.isoformat(),
+                        "lateness_seconds": round(lateness, 1),
+                        "kind": _classify_dispatch_lateness(lateness, grace),
+                    }
+                    job["last_dispatch"] = dispatch_stamp
+                    for rj in raw_jobs:
+                        if rj["id"] == job["id"]:
+                            rj["last_dispatch"] = dispatch_stamp
+                            needs_save = True
+                            break
+
                 due.append(job)
         except Exception:
             logger.exception(

hermes_cli/cron.py

@@ -133,6 +133,53 @@ def _warn_if_gateway_not_running() -> None:
     print(color("     Check status:  hermes cron status", Colors.DIM))
 
 
+def _format_lateness(seconds: float) -> str:
+    """Render a lateness duration compactly: '31m', '2h 30m', '45s'."""
+    try:
+        seconds = max(0, int(seconds))
+    except (TypeError, ValueError):
+        return "?"
+    if seconds < 60:
+        return f"{seconds}s"
+    minutes, _ = divmod(seconds, 60)
+    hours, minutes = divmod(minutes, 60)
+    days, hours = divmod(hours, 24)
+    parts = []
+    if days:
+        parts.append(f"{days}d")
+    if hours:
+        parts.append(f"{hours}h")
+    if minutes and not days:
+        parts.append(f"{minutes}m")
+    return " ".join(parts) or "0m"
+
+
+def _dispatch_display(dispatch: dict) -> Optional[str]:
+    """One-line scheduled-vs-actual dispatch summary for a job (#99879).
+
+    Returns None when the stamp is malformed. On-time dispatches render a
+    dim confirmation; late/catch-up dispatches render loudly so a run that
+    fired 30–150 min after gateway downtime no longer looks like an
+    ordinary on-time success.
+    """
+    if not isinstance(dispatch, dict):
+        return None
+    scheduled = dispatch.get("scheduled_at")
+    actual = dispatch.get("dispatched_at")
+    kind = dispatch.get("kind")
+    if not scheduled or not actual or not kind:
+        return None
+    lateness = _format_lateness(dispatch.get("lateness_seconds", 0))
+    if kind == "on_time":
+        return color(f"on time (scheduled {scheduled})", Colors.DIM)
+    label = "catch-up after missed fire" if kind == "catch_up" else "late"
+    return (
+        color(f"⚠ {label}: ", Colors.YELLOW)
+        + f"scheduled {scheduled}, ran {actual} "
+        + color(f"({lateness} late)", Colors.YELLOW)
+    )
+
+
 def cron_list(show_all: bool = False):
     """List all scheduled jobs."""
     from cron.jobs import list_jobs
@@ -224,6 +271,10 @@ def cron_list(show_all: bool = False):
                     status_display += color(f"  ({streak} failures in a row)", Colors.RED)
             print(f"    Last run:  {last_run}  {status_display}")
 
+        dispatch_line = _dispatch_display(job.get("last_dispatch"))
+        if dispatch_line:
+            print(f"    Dispatch:  {dispatch_line}")
+
         latest_execution = job.get("latest_execution")
         if latest_execution:
             print(
@@ -548,6 +599,31 @@ def _print_active_jobs_summary(jobs) -> None:
         print(f"  {len(jobs)} active job(s)")
         if next_runs:
             print(f"  Next run: {min(next_runs)}")
+        # Missed-run visibility (#99879): call out jobs whose LAST dispatch
+        # was late or a catch-up so post-downtime late fires are visible at
+        # status level, not just buried per-job in `hermes cron list`.
+        late = [
+            j for j in jobs
+            if isinstance(j.get("last_dispatch"), dict)
+            and j["last_dispatch"].get("kind") in ("late", "catch_up")
+        ]
+        if late:
+            print()
+            print(color(
+                f"  ⚠ {len(late)} job(s) last fired late (missed-fire catch-up):",
+                Colors.YELLOW,
+            ))
+            for j in late:
+                d = j["last_dispatch"]
+                print(
+                    f"    {j.get('id', '?')}  {j.get('name', '(unnamed)')}: "
+                    f"scheduled {d.get('scheduled_at', '?')}, "
+                    f"ran {d.get('dispatched_at', '?')} "
+                    + color(
+                        f"({_format_lateness(d.get('lateness_seconds', 0))} late)",
+                        Colors.YELLOW,
+                    )
+                )
     else:
         print("  No active jobs")
 

tests/cron/test_dispatch_lateness_stamp.py

@@ -0,0 +1,153 @@
+"""Missed-run dispatch visibility (#99879).
+
+The catch-up machinery already ran late fires after gateway downtime, but
+the run then looked like an ordinary on-time success: nothing recorded the
+scheduled instant vs the actual dispatch time. The due-scan now persists a
+``last_dispatch`` stamp (scheduled_at / dispatched_at / lateness_seconds /
+kind) on every recurring dispatch so `hermes cron list` and
+`hermes cron status` can surface late catch-ups.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from cron.jobs import (
+    _classify_dispatch_lateness,
+    get_due_jobs,
+    load_jobs,
+    save_jobs,
+)
+
+FIXED_NOW = datetime(2026, 9, 1, 9, 31, 0, tzinfo=timezone.utc)
+
+
+@pytest.fixture()
+def cron_store(tmp_path, monkeypatch):
+    """Redirect cron storage to a temp dir and pin the clock."""
+    monkeypatch.setattr("cron.jobs.CRON_DIR", tmp_path / "cron")
+    monkeypatch.setattr("cron.jobs.JOBS_FILE", tmp_path / "cron" / "jobs.json")
+    monkeypatch.setattr("cron.jobs.OUTPUT_DIR", tmp_path / "cron" / "output")
+    monkeypatch.setattr("cron.jobs._hermes_now", lambda: FIXED_NOW)
+    return tmp_path
+
+
+def _daily_job(jid, next_run_dt, **extra):
+    job = {
+        "id": jid,
+        "name": jid,
+        "prompt": "x",
+        "schedule": {"kind": "cron", "expr": "0 9 * * *"},
+        "next_run_at": next_run_dt.isoformat(),
+        "last_run_at": None,
+        "enabled": True,
+        "state": "scheduled",
+        "repeat": {"times": None, "completed": 0},
+        "deliver": "local",
+    }
+    job.update(extra)
+    return job
+
+
+def _interval_job(jid, next_run_dt, **extra):
+    job = {
+        "id": jid,
+        "name": jid,
+        "prompt": "x",
+        "schedule": {"kind": "interval", "minutes": 60},
+        "next_run_at": next_run_dt.isoformat(),
+        "last_run_at": None,
+        "enabled": True,
+        "state": "scheduled",
+        "repeat": {"times": None, "completed": 0},
+        "deliver": "local",
+    }
+    job.update(extra)
+    return job
+
+
+class TestDueScanDispatchStamp:
+    def test_catch_up_beyond_grace_stamped_and_persisted(self, cron_store):
+        # Scheduled yesterday 09:00, now is today 09:31 — far beyond the
+        # 2h max grace for a daily job: the catch-up path fires once now.
+        scheduled = FIXED_NOW - timedelta(hours=24, minutes=31)
+        save_jobs([_daily_job("daily", scheduled)])
+
+        due = get_due_jobs()
+
+        assert [d["id"] for d in due] == ["daily"]
+        stamp = due[0]["last_dispatch"]
+        assert stamp["kind"] == "catch_up"
+        assert stamp["scheduled_at"] == scheduled.isoformat()
+        assert stamp["dispatched_at"] == FIXED_NOW.isoformat()
+        expected_late = (FIXED_NOW - scheduled).total_seconds()
+        assert stamp["lateness_seconds"] == pytest.approx(expected_late, abs=1)
+        # Persisted, so a separate `hermes cron list` process can read it.
+        persisted = load_jobs()[0]
+        assert persisted["last_dispatch"] == stamp
+
+    def test_late_within_grace_stamped_late(self, cron_store):
+        # 31 minutes late: within the daily 2h grace window but beyond the
+        # on-time ticker tolerance — the shape from issue #99879's report.
+        scheduled = FIXED_NOW - timedelta(minutes=31)
+        save_jobs([_daily_job("daily", scheduled)])
+
+        due = get_due_jobs()
+
+        stamp = due[0]["last_dispatch"]
+        assert stamp["kind"] == "late"
+        assert stamp["lateness_seconds"] == pytest.approx(31 * 60, abs=1)
+
+    def test_on_time_dispatch_stamped_on_time(self, cron_store):
+        # Interval schedule: no cron-expr matching guard, dispatch 30s late
+        # is normal ticker cadence.
+        scheduled = FIXED_NOW - timedelta(seconds=30)
+        save_jobs([_interval_job("hourly", scheduled)])
+
+        due = get_due_jobs()
+
+        stamp = due[0]["last_dispatch"]
+        assert stamp["kind"] == "on_time"
+        assert stamp["lateness_seconds"] == pytest.approx(30, abs=1)
+
+    def test_manual_trigger_not_stamped(self, cron_store):
+        # A manual trigger stamps manual_run_at == next_run_at; there is no
+        # scheduled instant to be late against, so no dispatch stamp.
+        run_at = FIXED_NOW - timedelta(hours=5)
+        save_jobs([
+            _daily_job("manual", run_at, manual_run_at=run_at.isoformat())
+        ])
+
+        due = get_due_jobs()
+
+        assert [d["id"] for d in due] == ["manual"]
+        assert "last_dispatch" not in due[0]
+
+    def test_new_dispatch_overwrites_stale_stamp(self, cron_store):
+        scheduled = FIXED_NOW - timedelta(seconds=10)
+        stale = {
+            "scheduled_at": "2026-08-30T09:00:00+00:00",
+            "dispatched_at": "2026-08-30T11:00:00+00:00",
+            "lateness_seconds": 7200.0,
+            "kind": "catch_up",
+        }
+        save_jobs([_interval_job("hourly", scheduled, last_dispatch=stale)])
+
+        due = get_due_jobs()
+
+        stamp = due[0]["last_dispatch"]
+        assert stamp["kind"] == "on_time"
+        assert stamp["scheduled_at"] == scheduled.isoformat()
+
+
+class TestClassifyDispatchLateness:
+    def test_within_tolerance_is_on_time(self):
+        assert _classify_dispatch_lateness(0, 7200) == "on_time"
+        assert _classify_dispatch_lateness(299, 7200) == "on_time"
+
+    def test_beyond_tolerance_within_grace_is_late(self):
+        assert _classify_dispatch_lateness(301, 7200) == "late"
+        assert _classify_dispatch_lateness(7200, 7200) == "late"
+
+    def test_beyond_grace_is_catch_up(self):
+        assert _classify_dispatch_lateness(7201, 7200) == "catch_up"