fix(cron): use last_run_at as croniter base for cron jobs

compute_next_run() ignored the last_run_at parameter for cron-type
schedules, always computing from _hermes_now() instead. This was
inconsistent with interval jobs which DO use last_run_at as the anchor.

After a crash or restart, cron jobs would compute next_run_at from
the arbitrary restart time rather than the actual last execution time.
While the stale detection in get_due_jobs() catches most cases, using
last_run_at as the croniter base eliminates edge cases and makes the
behavior consistent across schedule types.

Salvaged from #9014 (authored by @beenherebefore) onto current main.
The original PR branch was 2+ weeks stale and would have reverted
substantial unrelated work (jobs_file_lock, workdir/context_from/
enabled_toolsets, issue #16265 state=error recovery). Kept just the
7-line substantive fix and the regression test.

cron/jobs.py

@@ -319,7 +319,14 @@ def compute_next_run(schedule: Dict[str, Any], last_run_at: Optional[str] = None
                 schedule.get("expr"),
             )
             return None
-        cron = croniter(schedule["expr"], now)
+        # Use last_run_at as the croniter base when available, consistent
+        # with interval jobs.  This ensures that after a crash/restart,
+        # the next run is anchored to the actual last execution time
+        # rather than to an arbitrary restart time.
+        base_time = now
+        if last_run_at:
+            base_time = _ensure_aware(datetime.fromisoformat(last_run_at))
+        cron = croniter(schedule["expr"], base_time)
         next_run = cron.get_next(datetime)
         return next_run.isoformat()
 

tests/cron/test_compute_next_run_last_run_at.py

@@ -0,0 +1,87 @@
+"""Test that compute_next_run uses last_run_at for cron jobs.
+
+Regression test for: cron jobs computing next_run_at from _hermes_now()
+instead of from last_run_at, making them inconsistent with interval jobs.
+"""
+import pytest
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+pytest.importorskip("croniter")
+
+from cron.jobs import compute_next_run
+
+
+class TestCronComputeNextRunUsesLastRunAt:
+    """compute_next_run MUST use last_run_at as the croniter base for cron jobs,
+    consistent with how interval jobs work."""
+
+    def test_cron_uses_last_run_at_for_every_6h_schedule(self, monkeypatch):
+        """For a schedule like 'every 6 hours', the base time matters.
+        If last_run_at is Apr 6 14:10, next should be Apr 6 18:00.
+        If now is Apr 10 22:00, next should be Apr 11 00:00.
+        compute_next_run must use last_run_at, not now."""
+        morocco = ZoneInfo("Africa/Casablanca")
+
+        # Job last ran April 6 at 14:10
+        last_run = datetime(2026, 4, 6, 14, 10, 0, tzinfo=morocco)
+
+        # But now it's April 10 at 22:00 (e.g., gateway restarted)
+        now = datetime(2026, 4, 10, 22, 0, 0, tzinfo=morocco)
+        monkeypatch.setattr("cron.jobs._hermes_now", lambda: now)
+
+        schedule = {"kind": "cron", "expr": "0 */6 * * *"}  # every 6 hours
+
+        result = compute_next_run(schedule, last_run_at=last_run.isoformat())
+        assert result is not None
+        next_dt = datetime.fromisoformat(result)
+
+        # With last_run_at as base (Apr 6 14:10), next is Apr 6 18:00.
+        # With now as base (Apr 10 22:00), next is Apr 11 00:00.
+        # The fix should use last_run_at, returning Apr 6 18:00
+        # (stale detection in get_due_jobs() fast-forwards from there).
+        assert next_dt.date().isoformat() == "2026-04-06", (
+            f"Expected next run on Apr 6 (from last_run_at), got {next_dt}"
+        )
+        assert next_dt.hour == 18
+
+    def test_cron_without_last_run_at_uses_now(self, monkeypatch):
+        """When last_run_at is NOT provided, compute_next_run falls back to
+        _hermes_now() as the croniter base (existing behavior)."""
+        morocco = ZoneInfo("Africa/Casablanca")
+
+        now = datetime(2026, 4, 10, 22, 0, 0, tzinfo=morocco)
+        monkeypatch.setattr("cron.jobs._hermes_now", lambda: now)
+
+        schedule = {"kind": "cron", "expr": "0 */6 * * *"}
+
+        result = compute_next_run(schedule)
+        assert result is not None
+        next_dt = datetime.fromisoformat(result)
+
+        # Without last_run_at, should compute from now -> Apr 11 00:00
+        assert next_dt.date().isoformat() == "2026-04-11", (
+            f"Expected next run on Apr 11 (from now), got {next_dt}"
+        )
+        assert next_dt.hour == 0
+
+    def test_cron_weekly_consistent_with_interval(self, monkeypatch):
+        """Both cron and interval jobs should anchor to last_run_at when
+        provided, producing consistent behavior after a crash/restart."""
+        morocco = ZoneInfo("Africa/Casablanca")
+
+        last_run = datetime(2026, 4, 6, 14, 10, 0, tzinfo=morocco)
+        now = datetime(2026, 4, 10, 22, 0, 0, tzinfo=morocco)
+        monkeypatch.setattr("cron.jobs._hermes_now", lambda: now)
+
+        cron_schedule = {"kind": "cron", "expr": "0 14 * * 1"}
+        interval_schedule = {"kind": "interval", "minutes": 7 * 24 * 60}
+
+        cron_result = compute_next_run(cron_schedule, last_run_at=last_run.isoformat())
+        interval_result = compute_next_run(interval_schedule, last_run_at=last_run.isoformat())
+
+        # Both should be after last_run_at
+        cron_dt = datetime.fromisoformat(cron_result)
+        interval_dt = datetime.fromisoformat(interval_result)
+        assert cron_dt > last_run, f"Cron next {cron_dt} should be after last_run {last_run}"
+        assert interval_dt > last_run, f"Interval next {interval_dt} should be after last_run {last_run}"