fix(cron): id-less job no longer freezes the whole scheduler

A cron record authored by a direct jobs.json edit that bypassed
add_job() can lack an "id" key (older writers used "job_id"). Every
site in _get_due_jobs_locked indexes job["id"] eagerly — both the
logging helpers (job.get("name", job["id"]) evaluates the default
argument unconditionally) and the 'for rj in raw_jobs: if rj["id"] ==
job["id"]' persistence loops. A single malformed record therefore
raised KeyError mid-tick, aborting the entire scan before save_jobs()
ran. Result: healthy jobs' fast-forwarded next_run_at was computed in
memory then discarded on the exception unwind, freezing the whole
profile's scheduler in a per-minute loop (observed dormant for weeks).

Fix: normalize id-less records at the top of _get_due_jobs_locked before
anything keys off job["id"] — recover the id from a drifted "job_id"
key when present, else synthesize one via uuid4, and persist. This
repairs the whole bug class at the source rather than guarding each of
the ~12 downstream index sites.

Adds a regression test that fails with KeyError on the current code and
passes with the fix, asserting a healthy sibling job is still returned
when an id-less record shares the store.

cron/jobs.py

@@ -1452,7 +1452,7 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None,
                             "Job '%s' (%s) could not compute next_run_at; "
                             "leaving enabled and marking state=error so the "
                             "job is not silently disabled.",
-                            job.get("name", job["id"]),
+                            job.get("name", job.get("id", "?")),
                             kind,
                         )
                     else:
@@ -1506,7 +1506,7 @@ def claim_dispatch(job_id: str) -> bool:
                 save_jobs(jobs)
                 logger.info(
                     "Job '%s': dispatch limit reached (%d/%d) — removing",
-                    job.get("name", job["id"]),
+                    job.get("name", job.get("id", "?")),
                     completed,
                     times,
                 )
@@ -1516,7 +1516,7 @@ def claim_dispatch(job_id: str) -> bool:
             save_jobs(jobs)
             logger.debug(
                 "Job '%s': claimed dispatch %d/%d",
-                job.get("name", job["id"]),
+                job.get("name", job.get("id", "?")),
                 repeat["completed"],
                 times,
             )
@@ -1653,9 +1653,25 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
     """Inner implementation of get_due_jobs(); must be called with _jobs_lock held."""
     now = _hermes_now()
     raw_jobs = load_jobs()
+    needs_save = False
+
+    # Repair id-less records BEFORE anything keys off ``job["id"]``. A direct
+    # jobs.json edit that bypassed add_job() can leave a record without an "id"
+    # (older writers used "job_id"). Every downstream site — the logging
+    # helpers and the ``for rj in raw_jobs: if rj["id"] == job["id"]``
+    # persistence loops — indexes job["id"] eagerly, so a single malformed
+    # record raised KeyError mid-tick, aborting the whole scan before
+    # save_jobs() ran. That froze the entire profile's scheduler in a
+    # per-minute fast-forward loop (healthy jobs recomputed in memory, then
+    # discarded when the exception unwound). Recover the id from the drifted
+    # "job_id" key when present, else synthesize one, and persist.
+    for rj in raw_jobs:
+        if not rj.get("id"):
+            rj["id"] = rj.pop("job_id", None) or uuid.uuid4().hex[:12]
+            needs_save = True
+
     jobs = [_apply_skill_fields(j) for j in copy.deepcopy(raw_jobs)]
     due = []
-    needs_save = False
     # 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()
@@ -1716,7 +1732,7 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
             next_run = recovered_next
             logger.info(
                 "Job '%s' had no next_run_at; recovering %s run at %s",
-                job.get("name", job["id"]),
+                job.get("name", job.get("id", "?")),
                 recovery_kind,
                 recovered_next,
             )
@@ -1757,7 +1773,7 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
                 logger.info(
                     "Job '%s' next_run_at offset changed (%s -> %s). "
                     "Recomputing cron run to preserve local wall-clock intent: %s",
-                    job.get("name", job["id"]),
+                    job.get("name", job.get("id", "?")),
                     raw_next_run_dt.utcoffset(),
                     now.utcoffset(),
                     new_next,
@@ -1785,7 +1801,7 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
                         "Job '%s' missed its scheduled time (%s, grace=%ds). "
                         "Running now; next run provisionally set to: %s "
                         "(re-anchored on completion)",
-                        job.get("name", job["id"]),
+                        job.get("name", job.get("id", "?")),
                         next_run,
                         grace,
                         new_next,
@@ -1819,7 +1835,7 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
                         logger.info(
                             "Job '%s': one-shot dispatch limit reached (%d/%d) "
                             "— removing stale due entry",
-                            job.get("name", job["id"]),
+                            job.get("name", job.get("id", "?")),
                             completed,
                             times,
                         )

tests/cron/test_jobs.py

@@ -835,6 +835,40 @@ class TestGetDueJobs:
         next_dt = _ensure_aware(datetime.fromisoformat(updated["next_run_at"]))
         assert next_dt > _hermes_now()
 
+    def test_idless_job_does_not_crash_or_block_sibling_jobs(self, tmp_cron_dir):
+        """A job missing its 'id' key must not crash the tick or freeze siblings.
+
+        Regression: jobs authored by a direct jobs.json edit (bypassing
+        create_job) sometimes used the key 'job_id' instead of 'id'. The logging
+        helpers evaluated ``job.get("name", job["id"])`` -- Python evaluates the
+        default argument ``job["id"]`` eagerly, so an id-less job raised
+        ``KeyError: 'id'`` mid-tick. That exception aborted
+        ``_get_due_jobs_locked()`` BEFORE ``save_jobs()`` ran, so every healthy
+        job's fast-forwarded next_run_at was computed in memory then discarded --
+        the whole profile's scheduler froze in a per-minute loop.
+        """
+        healthy = create_job(prompt="Healthy", schedule="every 1h")
+
+        jobs = load_jobs()
+        # Push the healthy job beyond its grace window so the fast-forward path
+        # (one of the id-less-crash sites) runs.
+        jobs[0]["next_run_at"] = (datetime.now() - timedelta(minutes=35)).isoformat()
+        # A malformed record: no 'id' key, mirroring the real corruption.
+        jobs.append({
+            "name": "idless-job",
+            "schedule": {"kind": "cron", "expr": "0 4 * * *"},
+            "enabled": True,
+            "no_agent": True,
+            "next_run_at": None,
+        })
+        save_jobs(jobs)
+
+        # Must not raise KeyError.
+        due = get_due_jobs()
+
+        # The healthy sibling is still discovered despite the malformed neighbor.
+        assert any(d.get("id") == healthy["id"] for d in due)
+
 
     def test_long_execution_does_not_perpetually_defer(self, tmp_cron_dir, monkeypatch):
         """#33315: a recurring job whose runtime exceeds interval+grace must still