fix(cron): malformed next_run_at no longer freezes the scheduler

One bad next_run_at value in jobs.json aborts the due-jobs scan with
ValueError from fromisoformat, before any save_jobs, so siblings lose
progress (fast-forwards etc).

Early normalization in _get_due_jobs_locked + defensive parses in
compute_next_run / _recoverable_oneshot_run_at.

Added test_bad_next_run_at_does_not_crash_or_block_sibling_jobs.

cron/jobs.py

@@ -568,7 +568,10 @@ def _recoverable_oneshot_run_at(
     if not run_at:
         return None
 
-    run_at_dt = _ensure_aware(datetime.fromisoformat(run_at))
+    try:
+        run_at_dt = _ensure_aware(datetime.fromisoformat(run_at))
+    except Exception:
+        return None
     if run_at_dt >= now - timedelta(seconds=ONESHOT_GRACE_SECONDS):
         return run_at
     return None
@@ -630,9 +633,11 @@ def compute_next_run(schedule: Dict[str, Any], last_run_at: Optional[str] = None
         if minutes is None:
             return None
         if last_run_at:
-            # Next run is last_run + interval
-            last = _ensure_aware(datetime.fromisoformat(last_run_at))
-            next_run = last + timedelta(minutes=minutes)
+            try:
+                last = _ensure_aware(datetime.fromisoformat(last_run_at))
+                next_run = last + timedelta(minutes=minutes)
+            except Exception:
+                next_run = now + timedelta(minutes=minutes)
         else:
             # First run is now + interval
             next_run = now + timedelta(minutes=minutes)
@@ -657,7 +662,10 @@ def compute_next_run(schedule: Dict[str, Any], last_run_at: Optional[str] = None
         # rather than to an arbitrary restart time.
         base_time = now
         if last_run_at:
-            base_time = _ensure_aware(datetime.fromisoformat(last_run_at))
+            try:
+                base_time = _ensure_aware(datetime.fromisoformat(last_run_at))
+            except Exception:
+                base_time = now
         cron = croniter(expr, base_time)
         next_run = cron.get_next(datetime)
         return next_run.isoformat()
@@ -1702,6 +1710,63 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
         if not isinstance(rj.get("schedule"), dict):
             rj["schedule"] = {}
             needs_save = True
+
+    # Normalize malformed "next_run_at" records (direct jobs.json edit,
+    # corruption, migration, or buggy writer). If present but not a valid
+    # ISO string, datetime.fromisoformat(next_run) later raises and aborts
+    # the entire scan *before* save_jobs(). Healthy siblings then lose any
+    # fast-forwarded next_run_at (same class of bug as bad "id" or "schedule").
+    # Strip the bad value so the existing "no next_run_at" recovery path
+    # recomputes a sane value and persists it for this job.
+    for j in jobs:
+        nr = j.get("next_run_at")
+        if nr is not None:
+            if not isinstance(nr, str):
+                j.pop("next_run_at", None)
+                needs_save = True
+            else:
+                try:
+                    datetime.fromisoformat(nr)
+                except Exception:
+                    j.pop("next_run_at", None)
+                    needs_save = True
+    for rj in raw_jobs:
+        nr = rj.get("next_run_at")
+        if nr is not None:
+            if not isinstance(nr, str):
+                rj.pop("next_run_at", None)
+                needs_save = True
+            else:
+                try:
+                    datetime.fromisoformat(nr)
+                except Exception:
+                    rj.pop("next_run_at", None)
+                    needs_save = True
+
+    # Same treatment for last_run_at (used as base in recovery / compute_next_run).
+    for j in jobs:
+        lr = j.get("last_run_at")
+        if lr is not None and not isinstance(lr, str):
+            j.pop("last_run_at", None)
+            needs_save = True
+        elif isinstance(lr, str):
+            try:
+                datetime.fromisoformat(lr)
+            except Exception:
+                j.pop("last_run_at", None)
+                needs_save = True
+    for rj in raw_jobs:
+        lr = rj.get("last_run_at")
+        if lr is not None and not isinstance(lr, str):
+            rj.pop("last_run_at", None)
+            needs_save = True
+        elif isinstance(lr, str):
+            try:
+                datetime.fromisoformat(lr)
+            except Exception:
+                rj.pop("last_run_at", None)
+                needs_save = True
+
     # Resolve the one-shot running-claim stale-recovery TTL once per scan
     # (derived from HERMES_CRON_TIMEOUT). See _oneshot_run_claim_ttl_seconds.
     _run_claim_ttl = _oneshot_run_claim_ttl_seconds()

tests/cron/test_jobs.py

@@ -1446,6 +1446,63 @@ class TestMarkJobRunConcurrency:
             )
 
 
+class TestBadNextRunAtRecovery:
+    """Regression: malformed next_run_at must not crash the due scan or starve siblings.
+
+    Mirrors the id-less and non-dict-schedule patterns: a single bad persisted
+    record in jobs.json must not abort _get_due_jobs_locked before save.
+    """
+
+    def test_bad_next_run_at_does_not_crash_or_block_sibling_jobs(self, tmp_cron_dir):
+        """One job with unparseable next_run_at + one healthy due sibling.
+
+        get_due_jobs must succeed and return the healthy job; the bad record
+        must be repaired (next_run_at cleared so recovery can set a sane value).
+        """
+        from datetime import timezone, timedelta as td
+        now = datetime.now(timezone.utc)
+        past = (now - td(seconds=30)).isoformat()
+        future = (now + td(days=1)).isoformat()
+
+        # Bad record: next_run_at is not a valid ISO string (e.g. from hand-edit or corruption)
+        # Healthy sibling is past due with good schedule.
+        bad_job = {
+            "id": "bad-next",
+            "schedule": {"kind": "interval", "minutes": 60},
+            "next_run_at": "not-a-valid-iso-timestamp!!!",
+            "enabled": True,
+            "created_at": past,
+        }
+        good_job = {
+            "id": "good-sibling",
+            "schedule": {"kind": "interval", "minutes": 5},
+            "next_run_at": past,
+            "enabled": True,
+            "created_at": past,
+        }
+        save_jobs([bad_job, good_job])
+
+        # Must not raise
+        due = get_due_jobs()
+
+        # The healthy job must still be returned
+        ids = [j["id"] for j in due]
+        assert "good-sibling" in ids, f"healthy sibling missing from due jobs: {ids}"
+        assert "bad-next" not in ids  # bad one may be repaired and/or not yet due after repair
+
+        # Bad job should have been auto-repaired (next_run_at stripped or fixed)
+        repaired = get_job("bad-next")
+        assert repaired is not None
+        nr = repaired.get("next_run_at")
+        if nr is not None:
+            # If still present it must now be parseable
+            datetime.fromisoformat(nr)
+
+        # Calling again must remain stable (no crash on re-scan)
+        due2 = get_due_jobs()
+        assert any(j["id"] == "good-sibling" for j in due2)
+
+
 class TestSaveJobOutput:
     def test_creates_output_file(self, tmp_cron_dir):
         output_file = save_job_output("test123", "# Results\nEverything ok.")