fix(cron): compare due instants across DST folds

Use UTC-normalized comparisons and elapsed durations across due scans, claim ages, recovery, and the external misfire backstop. Add deterministic fall-back regressions for both folds and ordinary intervals.

Co-authored-by: Jaimin <95100522+Jaiminp007@users.noreply.github.com>

cron/jobs.py

@@ -855,6 +855,26 @@ def _ensure_aware(dt: datetime) -> datetime:
     return dt.astimezone(target_tz)
 
 
+def _elapsed_seconds(later: datetime, earlier: datetime) -> float:
+    """Return elapsed seconds between aware instants, independent of wall time."""
+    return (later.astimezone(timezone.utc) - earlier.astimezone(timezone.utc)).total_seconds()
+
+
+def _instant_after(left: datetime, right: datetime) -> bool:
+    """Whether *left* is a later absolute instant than *right*."""
+    return left.astimezone(timezone.utc) > right.astimezone(timezone.utc)
+
+
+def _instant_at_or_before(left: datetime, right: datetime) -> bool:
+    """Whether *left* is at or before *right* as an absolute instant."""
+    return left.astimezone(timezone.utc) <= right.astimezone(timezone.utc)
+
+
+def _instant_before(left: datetime, right: datetime) -> bool:
+    """Whether *left* is an earlier absolute instant than *right*."""
+    return left.astimezone(timezone.utc) < right.astimezone(timezone.utc)
+
+
 def _parse_aware(value: Any) -> Optional[datetime]:
     """``_ensure_aware(datetime.fromisoformat(value))``, or None when *value* is not a parseable ISO
     string."""
@@ -888,7 +908,7 @@ def _recoverable_oneshot_run_at(
         return None
     run_at = schedule.get("run_at")
     run_at_dt = _parse_aware(run_at) if run_at else None
-    if run_at_dt is not None and run_at_dt >= now - timedelta(seconds=ONESHOT_GRACE_SECONDS):
+    if run_at_dt is not None and _elapsed_seconds(now, run_at_dt) <= ONESHOT_GRACE_SECONDS:
         return run_at
     return None
 
@@ -958,7 +978,7 @@ def _job_is_stale_error_recurring(
     last_run_dt = _parse_aware(last_run) if last_run else None
     if last_run_dt is None:
         return False
-    age_seconds = (now - last_run_dt).total_seconds()
+    age_seconds = _elapsed_seconds(now, last_run_dt)
     if age_seconds < 0:
         return False
     grace = _compute_grace_seconds(schedule)
@@ -2142,7 +2162,7 @@ def _claim_is_live(claim: Any, now: datetime, ttl_seconds: float) -> bool:
     if not isinstance(claim, dict) or not claim.get("at"):
         return False
     claimed_at = _parse_aware(claim["at"])
-    if claimed_at is None or not (0 <= (now - claimed_at).total_seconds() < ttl_seconds):
+    if claimed_at is None or not (0 <= _elapsed_seconds(now, claimed_at) < ttl_seconds):
         return False
     return not _claim_owner_is_dead(claim)
 
@@ -2900,7 +2920,7 @@ def _repair_timezone_shifted_cron(d: _DueJob) -> bool:
     offset change meeting the same conditions SKIPS the pending occurrence; accepted as rare."""
     now = d.scan.now
     if not (
-        d.next_run_dt <= now
+        _instant_at_or_before(d.next_run_dt, now)
         and _timezone_offset_mismatch(d.raw_next_run_dt, now)
         and _stored_wall_clock_is_future(d.raw_next_run_dt, now)
     ):
@@ -2928,7 +2948,7 @@ def _rearm_stale_error_recurring(d: _DueJob) -> datetime:
     now = d.scan.now
     if not (
         d.kind in ("cron", "interval")
-        and d.next_run_dt > now
+        and _instant_after(d.next_run_dt, now)
         and _job_is_stale_error_recurring(d.job, d.schedule, now)
     ):
         return d.next_run_dt
@@ -2938,7 +2958,10 @@ def _rearm_stale_error_recurring(d: _DueJob) -> datetime:
     else:
         recovered_next = d.recompute_next()
         recovered_next_dt = _parse_aware(recovered_next) if recovered_next else None
-    if not (recovered_next and recovered_next_dt is not None and recovered_next_dt < d.next_run_dt):
+    if not (
+        recovered_next and recovered_next_dt is not None
+        and _instant_before(recovered_next_dt, d.next_run_dt)
+    ):
         return d.next_run_dt
     jid = d.job.get("id")
     logger.warning(
@@ -2991,13 +3014,13 @@ def _fast_forward_missed_recurring(d: _DueJob, grace: int) -> bool:
     protects the crash window before mark_job_run and covers the external fire_due path, which never
     calls advance_next_run. mark_job_run re-anchors on completion, so the value is provisional.
     """
-    if (d.scan.now - d.next_run_dt).total_seconds() <= grace:
+    if _elapsed_seconds(d.scan.now, d.next_run_dt) <= grace:
         return False
     new_next = d.recompute_next()
     if not new_next:
         return False
     d.scan.persist(d.job["id"], next_run_at=new_next)
-    if (_ensure_aware(datetime.fromisoformat(new_next)) > d.scan.now
+    if (_instant_after(_ensure_aware(datetime.fromisoformat(new_next)), d.scan.now)
             and not _cron_config_number("catch_up_missed", True, lambda value: value is not False)):
         logger.info(
             "Job '%s' missed its scheduled time (%s, grace=%ds). "
@@ -3019,7 +3042,7 @@ def _retire_expired_oneshot(d: _DueJob) -> bool:
     and recovery never revives them; only the due scan used to dispatch them hours late). With no
     claim stamped, retire it with a diagnostic (never silently delete). A claim may mean a run is
     still in flight elsewhere — skip but keep the record so its mark_job_run can land."""
-    if (d.scan.now - d.next_run_dt).total_seconds() <= ONESHOT_GRACE_SECONDS:
+    if _elapsed_seconds(d.scan.now, d.next_run_dt) <= ONESHOT_GRACE_SECONDS:
         return False
     if not (d.job.get("run_claim") or d.job.get("fire_claim")):
         _write_missed_oneshot_diagnostic(d.job, d.next_run)
@@ -3128,7 +3151,7 @@ def _evaluate_due_job(job: Dict[str, Any], scan: _DueScan, run_claim_ttl: float)
     if kind == "cron" and not manual_run and _repair_timezone_shifted_cron(d):
         return False
     d.next_run_dt = _rearm_stale_error_recurring(d)
-    if d.next_run_dt > now:
+    if _instant_after(d.next_run_dt, now):
         return False
 
     # Only the dispatch snapshot carries this field; never infer it from a later stamp.
@@ -3154,7 +3177,7 @@ def _evaluate_due_job(job: Dict[str, Any], scan: _DueScan, run_claim_ttl: float)
     # late catch-up. Recurring only — expired one-shots were retired above; manual triggers aren't
     # late.
     if not manual_run and recurring:
-        lateness = max(0.0, (now - d.next_run_dt).total_seconds())
+        lateness = max(0.0, _elapsed_seconds(now, d.next_run_dt))
         # See #99879.
         dispatch_stamp = {
             "scheduled_at": next_run,

cron/scheduler_provider.py

@@ -295,7 +295,8 @@ def fire_overdue_jobs(
         return 0
 
     from cron.jobs import (
-        ONESHOT_GRACE_SECONDS, _ensure_aware, _hermes_now, is_job_runnable, load_jobs,
+        ONESHOT_GRACE_SECONDS, _elapsed_seconds, _ensure_aware, _hermes_now,
+        is_job_runnable, load_jobs,
     )
 
     if now is None:
@@ -312,7 +313,7 @@ def fire_overdue_jobs(
             due_dt = _ensure_aware(datetime.fromisoformat(next_run_at))
         except (ValueError, TypeError):
             continue
-        overdue_seconds = (now - due_dt).total_seconds()
+        overdue_seconds = _elapsed_seconds(now, due_dt)
         if overdue_seconds < grace_minutes * 60:
             continue
         job_id = str(job.get("id") or "")

tests/cron/test_due_scan_dst_fold.py

@@ -0,0 +1,110 @@
+"""Due-scan comparisons must use absolute instants across a repeated DST hour."""
+
+from datetime import datetime, timezone
+from zoneinfo import ZoneInfo
+
+import hermes_time
+import pytest
+
+from cron import jobs
+
+
+NEW_YORK = ZoneInfo("America/New_York")
+
+
+@pytest.fixture
+def dst_cron_store(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setenv("HERMES_TIMEZONE", "America/New_York")
+    hermes_time.reset_cache()
+    monkeypatch.setattr(jobs, "CRON_DIR", tmp_path / "cron")
+    monkeypatch.setattr(jobs, "JOBS_FILE", tmp_path / "cron" / "jobs.json")
+    monkeypatch.setattr(jobs, "OUTPUT_DIR", tmp_path / "cron" / "output")
+    yield tmp_path
+    hermes_time.reset_cache()
+
+
+def _instant(hour, minute):
+    return datetime(2026, 11, 1, hour, minute, tzinfo=timezone.utc).astimezone(NEW_YORK)
+
+
+def _job(next_run_at, *, kind="interval", expr=None):
+    schedule = {"kind": kind, "minutes": 60}
+    if expr is not None:
+        schedule["expr"] = expr
+    return {
+        "id": "dst-job",
+        "name": "dst-job",
+        "prompt": "fixture",
+        "schedule": schedule,
+        "next_run_at": next_run_at.isoformat(),
+        "last_run_at": None,
+        "enabled": True,
+        "state": "scheduled",
+        "repeat": {"times": None, "completed": 0},
+        "deliver": "local",
+    }
+
+
+def _due_at(monkeypatch, now, scheduled):
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now)
+    jobs.save_jobs([_job(scheduled)])
+    return [row["id"] for row in jobs.get_due_jobs()]
+
+
+def test_fold_one_interval_is_not_due_during_fold_zero(dst_cron_store, monkeypatch):
+    now = _instant(5, 1)  # 01:01 EDT, fold=0
+    scheduled = _instant(6, 0)  # 01:00 EST, fold=1
+    assert now.fold == 0
+    assert scheduled.fold == 1
+    assert scheduled.timestamp() - now.timestamp() == 59 * 60
+
+    assert _due_at(monkeypatch, now, scheduled) == []
+
+
+def test_interval_runs_when_due_in_either_fold(dst_cron_store, monkeypatch):
+    first_fold_now = _instant(5, 1)
+    first_fold_slot = _instant(5, 0)
+    second_fold_now = _instant(6, 1)
+    second_fold_slot = _instant(6, 0)
+
+    assert first_fold_now.fold == first_fold_slot.fold == 0
+    assert second_fold_now.fold == second_fold_slot.fold == 1
+    assert _due_at(monkeypatch, first_fold_now, first_fold_slot) == ["dst-job"]
+    assert _due_at(monkeypatch, second_fold_now, second_fold_slot) == ["dst-job"]
+
+
+def test_cron_future_occurrence_is_not_due_in_prior_fold(dst_cron_store, monkeypatch):
+    now = _instant(5, 31)  # 01:31 EDT, fold=0
+    scheduled = _instant(6, 30)  # 01:30 EST, fold=1
+    assert scheduled.timestamp() > now.timestamp()
+
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now)
+    jobs.save_jobs([_job(scheduled, kind="cron", expr="30 1 * * *")])
+    assert jobs.get_due_jobs() == []
+
+
+def test_hourly_interval_does_not_recur_each_minute_during_fallback(dst_cron_store, monkeypatch):
+    """A one-hour cadence remains hourly across both occurrences of local 01:00."""
+    from datetime import timedelta
+
+    start = datetime(2026, 11, 1, 4, 0, tzinfo=timezone.utc)  # 00:00 EDT
+    initial_now = (start - timedelta(hours=1)).astimezone(NEW_YORK)
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: initial_now)
+    job = jobs.create_job(
+        prompt="fixture", schedule="every 60m", model="fixture", deliver="local")
+    fired_at = []
+
+    for elapsed_minutes in range(181):
+        now = (start + timedelta(minutes=elapsed_minutes)).astimezone(NEW_YORK)
+        monkeypatch.setattr(jobs, "_hermes_now", lambda now=now: now)
+        due = jobs.get_due_jobs()
+        if due:
+            fired_at.append(now.timestamp())
+            assert [row["id"] for row in due] == [job["id"]]
+            assert jobs.claim_job_for_fire(job["id"]) is True
+            assert jobs.mark_job_run(job["id"], success=True) is True
+
+    assert fired_at == [
+        (start + timedelta(hours=offset)).timestamp() for offset in range(4)
+    ]

tests/cron/test_misfire_catchup.py

@@ -8,7 +8,8 @@ gateway housekeeping, claims and fires those jobs after a grace window.
 """
 
 import threading
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
+from zoneinfo import ZoneInfo
 
 import pytest
 
@@ -66,6 +67,27 @@ def _park_in_past(job_id, minutes):
 
 
 class TestFireOverdueJobs:
+    def test_future_fold_one_slot_is_not_overdue_in_fold_zero(self, tmp_cron_dir, monkeypatch):
+        """A repeated-hour slot stays future in the external-provider backstop too."""
+        import hermes_time
+
+        monkeypatch.setenv("HERMES_TIMEZONE", "America/New_York")
+        hermes_time.reset_cache()
+        try:
+            zone = ZoneInfo("America/New_York")
+            now = datetime(2026, 11, 1, 5, 40, tzinfo=timezone.utc).astimezone(zone)
+            next_run = datetime(2026, 11, 1, 6, 0, tzinfo=timezone.utc).astimezone(zone)
+            monkeypatch.setattr("cron.jobs._hermes_now", lambda: now)
+            create_job(prompt="p", schedule="every 1h")
+            rows = load_jobs()
+            rows[0]["next_run_at"] = next_run.isoformat()
+            save_jobs(rows)
+
+            assert now.fold == 0 and next_run.fold == 1
+            assert fire_overdue_jobs(RecordingProvider(), now=now) == 0
+        finally:
+            hermes_time.reset_cache()
+
     def test_noop_for_builtin_provider(self, tmp_cron_dir):
         """The in-process ticker self-heals past-due jobs — the sweep must
         never double-dispatch under it."""