Cron job execution bugs and scheduler fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #50025 fix(cron): execute job immediately on action=run

  merged to main

  files: hermes_cli/cron.py, tests/tools/test_cronjob_run_immediate.py, tools/cronjob_tools.py

  Show the diff

  • hermes_cli/cron.py

    @@ -313,7 +313,14 @@ def _job_action(action: str, job_id: str, success_verb: str) -> int:
         if action in {"resume", "run"} and result.get("job", {}).get("next_run_at"):
             print(f"  Next run: {result['job']['next_run_at']}")
         if action == "run":
    -        print("  It will run on the next scheduler tick.")
    +        job = result.get("job", {})
    +        if job.get("executed"):
    +            outcome = "succeeded" if job.get("execution_success") else "failed"
    +            print(f"  Ran now: {outcome}.")
    +        elif job.get("execution_skipped"):
    +            print(f"  {job['execution_skipped']}")
    +        else:
    +            print("  It will run on the next scheduler tick.")
         return 0
     
     

  • tests/tools/test_cronjob_run_immediate.py

    @@ -0,0 +1,81 @@
    +"""Tests for cronjob action='run' immediate execution (#41037).
    +
    +Before this fix, `cronjob(action='run')` only set next_run_at=now and returned
    +success, relying on the scheduler ticker to actually run the job. With no
    +gateway/ticker active (e.g. a CLI-only Windows setup) the job never executed and
    +last_run_at stayed null forever. Now action='run' claims the job (at-most-once,
    +blocking a concurrent tick) and fires it inline via the shared run_one_job body.
    +"""
    +import json
    +from unittest.mock import patch
    +
    +from tools.cronjob_tools import cronjob, _execute_job_now
    +
    +
    +_JOB = {"id": "job-run-1", "name": "manual run", "prompt": "hi",
    +        "schedule": {"kind": "cron", "expr": "0 9 * * *"}}
    +
    +
    +class TestCronjobRunExecutesImmediately:
    +    def test_run_action_claims_and_fires_via_run_one_job(self):
    +        """action='run' must claim the job then fire it through run_one_job."""
    +        ran = {"job": "after-run", "last_status": "ok", "last_error": None}
    +        with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \
    +             patch("tools.cronjob_tools.claim_job_for_fire", return_value=True) as m_claim, \
    +             patch("cron.scheduler.run_one_job", return_value=True) as m_run, \
    +             patch("tools.cronjob_tools.get_job", return_value=ran):
    +            out = json.loads(cronjob(action="run", job_id="job-run-1"))
    +
    +        assert out["success"] is True
    +        assert out["job"]["executed"] is True
    +        assert out["job"]["execution_success"] is True
    +        m_claim.assert_called_once_with("job-run-1")   # at-most-once claim taken
    +        m_run.assert_called_once()                       # fired via the shared body
    +
    +    def test_run_skips_when_claim_lost(self):
    +        """If the scheduler already holds the fire claim, do NOT double-run."""
    +        with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \
    +             patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \
    +             patch("cron.scheduler.run_one_job") as m_run, \
    +             patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)):
    +            out = json.loads(cronjob(action="run", job_id="job-run-1"))
    +
    +        assert out["success"] is True
    +        assert out["job"]["executed"] is False
    +        assert out["job"]["execution_success"] is False
    +        assert "execution_skipped" in out["job"]
    +        m_run.assert_not_called()  # claim lost -> never fired
    +
    +    def test_run_reports_failure_from_last_status(self):
    +        """A failed run is reported via the re-read job's last_status/last_error."""
    +        failed = {"id": "job-run-1", "last_status": "error", "last_error": "provider 500"}
    +        with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \
    +             patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
    +             patch("cron.scheduler.run_one_job", return_value=True), \
    +             patch("tools.cronjob_tools.get_job", return_value=failed):
    +            out = json.loads(cronjob(action="run", job_id="job-run-1"))
    +
    +        assert out["job"]["executed"] is True
    +        assert out["job"]["execution_success"] is False
    +        assert out["job"]["execution_error"] == "provider 500"
    +
    +    def test_execute_job_now_bails_without_claim(self):
    +        """_execute_job_now never calls run_one_job when the claim is lost."""
    +        with patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \
    +             patch("cron.scheduler.run_one_job") as m_run:
    +            res = _execute_job_now(dict(_JOB))
    +        assert res["claimed"] is False
    +        assert res["success"] is False
    +        m_run.assert_not_called()
    +
    +    def test_execute_job_now_marks_failure_on_exception(self):
    +        """An exception during fire is captured, marked failed, not propagated."""
    +        with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
    +             patch("cron.scheduler.run_one_job", side_effect=RuntimeError("boom")), \
    +             patch("tools.cronjob_tools.mark_job_run") as m_mark, \
    +             patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)):
    +            res = _execute_job_now(dict(_JOB))
    +        assert res["claimed"] is True
    +        assert res["success"] is False
    +        assert "boom" in res["error"]
    +        m_mark.assert_called_once()

  • tools/cronjob_tools.py

    @@ -21,14 +21,16 @@
     
     from cron.jobs import (
         AmbiguousJobReference,
    +    claim_job_for_fire,
         create_job,
    +    get_job,
         list_jobs,
    +    mark_job_run,
         parse_schedule,
         pause_job,
         remove_job,
         resolve_job_ref,
         resume_job,
    -    trigger_job,
         update_job,
     )
     
    @@ -472,6 +474,51 @@ def _format_job(job: Dict[str, Any]) -> Dict[str, Any]:
         return result
     
     
    +def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]:
    +    """Execute a cron job immediately, outside the scheduler tick.
    +
    +    Atomically claims the job first via ``claim_job_for_fire`` — the same
    +    at-most-once CAS the scheduler/external-provider fire path uses — so a
    +    concurrently-running gateway ticker cannot also fire it (the claim both
    +    blocks a duplicate fire and advances ``next_run_at`` for recurring jobs).
    +    If the claim is lost (another fire is in flight), this is a no-op.
    +
    +    The actual firing is delegated to ``run_one_job`` — the single shared
    +    execute→save→deliver→mark body the ticker and external providers use — so
    +    failure delivery, ``[SILENT]`` handling, and live-adapter delivery stay
    +    identical across paths and can't drift.
    +
    +    Returns {"claimed": bool, "success": bool, "error": str|None}.
    +    """
    +    job_id = job["id"]
    +    try:
    +        from cron.scheduler import run_one_job
    +
    +        # At-most-once claim: bail without running if a tick/other fire owns it.
    +        if not claim_job_for_fire(job_id):
    +            return {"claimed": False, "success": False,
    +                    "error": "Job is already being fired by the scheduler; not run again."}
    +
    +        # run_one_job records last_run_at/last_status via mark_job_run (which
    +        # also clears the fire claim) and returns True iff it processed the job.
    +        processed = run_one_job(job)
    +        refreshed = get_job(job_id) or {}
    +        ok = refreshed.get("last_status") == "ok"
    +        return {
    +            "claimed": True,
    +            "success": bool(processed and ok),
    +            "error": refreshed.get("last_error"),
    +        }
    +
    +    except Exception as e:
    +        logger.error("Failed to execute cron job %s immediately: %s", job_id, e)
    +        try:
    +            mark_job_run(job_id, False, str(e))
    +        except Exception:
    +            pass
    +        return {"claimed": True, "success": False, "error": str(e)}
    +
    +
     def cronjob(
         action: str,
         job_id: Optional[str] = None,
    @@ -640,8 +687,23 @@ def cronjob(
                 return json.dumps({"success": True, "job": _format_job(updated)}, indent=2)
     
             if normalized in {"run", "run_now", "trigger"}:
    -            updated = trigger_job(job_id)
    -            return json.dumps({"success": True, "job": _format_job(updated)}, indent=2)
    +            # Execute the job immediately rather than only scheduling it for the
    +            # next scheduler tick — a manual `run` should actually run, even when
    +            # no gateway/ticker is active (the #41037 case). The claim inside
    +            # _execute_job_now advances next_run_at and blocks a concurrent tick
    +            # from double-firing.
    +            exec_result = _execute_job_now(job)
    +            # Re-read so the response reflects the post-run last_run_at/last_status.
    +            result = _format_job(get_job(job_id) or {"id": job_id})
    +            result["executed"] = exec_result.get("claimed", False)
    +            result["execution_success"] = exec_result.get("success", False)
    +            if not exec_result.get("claimed", False):
    +                result["execution_skipped"] = (
    +                    "Already being fired by the scheduler; not run again."
    +                )
    +            elif exec_result.get("error"):
    +                result["execution_error"] = exec_result["error"]
    +            return json.dumps({"success": True, "job": result}, indent=2)
     
             if normalized == "update":
                 updates: Dict[str, Any] = {}

• #50062 fix(cron): run missed-grace jobs once instead of deferring forever

  merged to main

  files: cron/jobs.py, tests/cron/test_jobs.py

  Show the diff

  • cron/jobs.py

    @@ -1240,10 +1240,16 @@ def claim_job_for_fire(job_id: str, *, claim_ttl_seconds: int = 300) -> bool:
     def get_due_jobs() -> List[Dict[str, Any]]:
         """Get all jobs that are due to run now.
     
    -    For recurring jobs (cron/interval), if the scheduled time is stale
    -    (more than one period in the past, e.g. because the gateway was down),
    -    the job is fast-forwarded to the next future run instead of firing
    -    immediately.  This prevents a burst of missed jobs on gateway restart.
    +    For recurring jobs (cron/interval), if the scheduled time is stale (more
    +    than one period in the past, e.g. because the gateway was down OR because a
    +    long-running previous execution overran the interval), the accumulated
    +    missed runs are collapsed — ``next_run_at`` is fast-forwarded to the next
    +    future occurrence so a backlog does NOT burst-fire on restart — but the job
    +    still fires ONCE now. This prevents the perpetual-defer loop (#33315) where
    +    a job whose runtime exceeds ``interval + grace`` would be skipped forever.
    +
    +    Note: firing once on catch-up flows through ``mark_job_run``, so a job with
    +    a ``repeat.times`` limit consumes one of its runs on that catch-up fire.
         """
         with _jobs_lock():
             return _get_due_jobs_locked()
    @@ -1351,25 +1357,34 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
                 # the next future occurrence instead of firing a stale run.
                 grace = _compute_grace_seconds(schedule)
                 if kind in {"cron", "interval"} and (now - next_run_dt).total_seconds() > grace:
    -                # Job is past its catch-up grace window — this is a stale missed run.
    -                # Grace scales with schedule period: daily=2h, hourly=30m, 10min=5m.
    +                # Job is past its catch-up grace window — skip accumulated
    +                # missed runs but still execute once now to avoid deferring
    +                # indefinitely (e.g. a long-running job just finished).
                     new_next = compute_next_run(schedule, now.isoformat())
                     if new_next:
                         logger.info(
                             "Job '%s' missed its scheduled time (%s, grace=%ds). "
    -                        "Fast-forwarding to next run: %s",
    +                        "Running now; next run provisionally set to: %s "
    +                        "(re-anchored on completion)",
                             job.get("name", job["id"]),
                             next_run,
                             grace,
                             new_next,
                         )
    -                    # Update the job in storage
    +                    # Persist the fast-forward to storage now (skip accumulated
    +                    # slots). In the built-in ticker path this is shortly
    +                    # overwritten by advance_next_run + mark_job_run, but it is
    +                    # NOT redundant: it (a) protects the crash window between
    +                    # here and mark_job_run, and (b) covers the external
    +                    # fire_due provider path, which does not call
    +                    # advance_next_run. mark_job_run re-anchors next_run_at off
    +                    # the actual completion time, so this value is provisional.
                         for rj in raw_jobs:
                             if rj["id"] == job["id"]:
                                 rj["next_run_at"] = new_next
                                 needs_save = True
                                 break
    -                    continue  # Skip this run
    +                    # Fall through to due.append(job) — execute once now
     
                 due.append(job)
     

  • tests/cron/test_jobs.py

    @@ -685,10 +685,11 @@ def test_past_due_within_window_returned(self, tmp_cron_dir):
             assert len(due) == 1
             assert due[0]["id"] == job["id"]
     
    -    def test_stale_past_due_skipped(self, tmp_cron_dir):
    -        """Recurring jobs past their dynamic grace window are fast-forwarded, not fired.
    +    def test_stale_past_due_runs_once_and_fast_forwards(self, tmp_cron_dir):
    +        """Recurring jobs past their grace window run once now and fast-forward next_run_at.
     
             For an hourly job, grace = 30 min. Setting 35 min late exceeds the window.
    +        The job should be returned as due (execute once) with next_run_at in the future.
             """
             job = create_job(prompt="Stale", schedule="every 1h")
             # Force next_run_at to 35 minutes ago (beyond the 30-min grace for hourly)
    @@ -697,13 +698,62 @@ def test_stale_past_due_skipped(self, tmp_cron_dir):
             save_jobs(jobs)
     
             due = get_due_jobs()
    -        assert len(due) == 0
    -        # next_run_at should be fast-forwarded to the future
    +        # Job is returned as due — execute once now instead of skipping
    +        assert len(due) == 1
    +        assert due[0]["id"] == job["id"]
    +        # next_run_at should be fast-forwarded to the future (accumulated slots skipped)
             updated = get_job(job["id"])
             from cron.jobs import _ensure_aware, _hermes_now
             next_dt = _ensure_aware(datetime.fromisoformat(updated["next_run_at"]))
             assert next_dt > _hermes_now()
     
    +
    +    def test_long_execution_does_not_perpetually_defer(self, tmp_cron_dir, monkeypatch):
    +        """#33315: a recurring job whose runtime exceeds interval+grace must still
    +        run once when the tick comes back, not skip forever.
    +
    +        Reproduces the production loop: a 5-min interval job whose previous run
    +        overran the interval, leaving next_run_at ~11 min in the past — beyond
    +        the 150s grace for a 5m interval. The job must be returned as due (run
    +        once) AND have next_run_at fast-forwarded (so accumulated missed slots
    +        don't all fire)."""
    +        from cron.jobs import _ensure_aware, _hermes_now
    +        job = create_job(prompt="Long job", schedule="every 5m")
    +        jobs = load_jobs()
    +        # 11 minutes ago: > grace (150s for a 5m interval) — the "still running" miss.
    +        stale = (_hermes_now() - timedelta(minutes=11)).isoformat()
    +        jobs[0]["next_run_at"] = stale
    +        jobs[0]["last_run_at"] = (_hermes_now() - timedelta(minutes=1)).isoformat()
    +        save_jobs(jobs)
    +
    +        due = get_due_jobs()
    +        assert [j["id"] for j in due] == [job["id"]], "long-execution job was skipped (perpetual-defer bug)"
    +        # next_run_at fast-forwarded into the future (no burst of missed slots).
    +        nxt = _ensure_aware(datetime.fromisoformat(get_job(job["id"])["next_run_at"]))
    +        assert nxt > _hermes_now()
    +
    +
    +    def test_stale_repeat_limited_job_consumes_one_run_on_catchup(self, tmp_cron_dir, monkeypatch):
    +        """#33315 behavior note: a stale recurring job with a repeat.times limit
    +        fires ONCE on catch-up and consumes one of its runs (it is no longer
    +        silently skipped). Pins the documented repeat-count interaction so it
    +        isn't changed accidentally."""
    +        from cron.jobs import _hermes_now
    +        job = create_job(prompt="Limited", schedule="every 5m", repeat=3)
    +        jobs = load_jobs()
    +        jobs[0]["next_run_at"] = (_hermes_now() - timedelta(minutes=11)).isoformat()
    +        jobs[0]["last_run_at"] = (_hermes_now() - timedelta(minutes=11)).isoformat()
    +        save_jobs(jobs)
    +
    +        # The stale job is returned to fire once (not skipped).
    +        due = get_due_jobs()
    +        assert [j["id"] for j in due] == [job["id"]]
    +        # Simulate the run completing: mark_job_run increments completed.
    +        mark_job_run(job["id"], True)
    +        survived = get_job(job["id"])
    +        assert survived is not None, "job should survive (3 > 1 completed)"
    +        assert survived["repeat"]["completed"] == 1
    +
         def test_future_not_returned(self, tmp_cron_dir):
             create_job(prompt="Not yet", schedule="every 1h")
             due = get_due_jobs()
    @@ -911,10 +961,15 @@ def test_cron_offset_migration_does_not_repair_already_passed_wall_time(self, tm
                 }]
             )
     
    -        # The wall-clock time has already passed, so this follows the existing
    -        # stale-run fast-forward behavior instead of the timezone-migration
    -        # repair path for future wall-clock runs.
    -        assert get_due_jobs() == []
    +        # The wall-clock time has already passed, so this does NOT take the
    +        # timezone-migration repair path (which is for still-future wall-clock
    +        # runs). It falls through to the stale-grace path, which — since #33315
    +        # — runs the job once now and fast-forwards next_run_at (rather than
    +        # skipping). The key assertion for THIS test is that the repaired
    +        # next_run_at is the normal next cron occurrence, not the migration
    +        # path's same-day rebase.
    +        due = get_due_jobs()
    +        assert [j["id"] for j in due] == ["cron-tz-missed"]  # runs once now (#33315)
             repaired = datetime.fromisoformat(get_job("cron-tz-missed")["next_run_at"])
             assert repaired == datetime(2026, 5, 26, 9, 0, 0, tzinfo=current_tz)
     

• #51695 fix(cron): anchor naive schedule timestamps to configured timezone

  merged to main

  files: cron/jobs.py, tests/cron/test_jobs.py

  Show the diff

  • cron/jobs.py

    @@ -359,8 +359,19 @@ def parse_schedule(schedule: str) -> Dict[str, Any]:
                 dt = datetime.fromisoformat(schedule.replace('Z', '+00:00'))
                 # Make naive timestamps timezone-aware at parse time so the stored
                 # value doesn't depend on the system timezone matching at check time.
    +            #
    +            # Anchor to the CONFIGURED Hermes timezone, not the server's local
    +            # timezone. The due-check (`get_due_jobs`) compares `next_run_at`
    +            # against `hermes_time.now()`, which uses the configured zone. If a
    +            # naive "20:07" were interpreted as server-local (e.g. UTC) while
    +            # now() runs in Asia/Kolkata, the stored instant would land hours
    +            # off from the user's wall-clock intent — far enough that one-shots
    +            # never become due and recurring jobs fire at the wrong time. Using
    +            # the configured zone makes "20:07" mean 20:07 on the same clock the
    +            # scheduler checks against (#51021).
                 if dt.tzinfo is None:
    -                dt = dt.astimezone()  # Interpret as local timezone
    +                hermes_tz = _hermes_now().tzinfo
    +                dt = dt.replace(tzinfo=hermes_tz)
                 return {
                     "kind": "once",
                     "run_at": dt.isoformat(),

  • tests/cron/test_jobs.py

    @@ -110,6 +110,57 @@ def test_invalid_cron_raises(self):
             with pytest.raises(ValueError):
                 parse_schedule("99 99 99 99 99")
     
    +    def test_naive_iso_anchors_to_configured_tz_not_server_local(self, monkeypatch):
    +        """A naive ISO timestamp must be interpreted in the CONFIGURED Hermes
    +        timezone, NOT the server's local timezone (#51021).
    +
    +        Regression: when the configured zone differs from the server's local
    +        zone (common on cloud hosts running UTC), parse_schedule used
    +        ``dt.astimezone()`` (server-local), baking in the wrong offset. The
    +        due-check compares against ``_hermes_now()`` (configured zone), so the
    +        stored instant landed hours off the user's wall-clock intent — far
    +        enough that one-shots never became due. This asserts the parsed offset
    +        matches the configured-now offset, the invariant that keeps the stored
    +        instant on the same clock the scheduler checks against.
    +        """
    +        configured_now = datetime(2026, 6, 22, 20, 0, 0, tzinfo=timezone(timedelta(hours=5, minutes=30)))
    +        monkeypatch.setattr("cron.jobs._hermes_now", lambda: configured_now)
    +
    +        result = parse_schedule("2026-06-22T20:07:00")  # naive, user wall-clock
    +
    +        assert result["kind"] == "once"
    +        parsed = datetime.fromisoformat(result["run_at"])
    +        assert parsed.utcoffset() == configured_now.utcoffset()
    +        # Same wall-clock the user typed, on the configured clock.
    +        assert parsed.replace(tzinfo=None) == datetime(2026, 6, 22, 20, 7, 0)
    +
    +
    +# =========================================================================
    +# Timezone-divergence regression (#51021)
    +# =========================================================================
    +
    +class TestNaiveScheduleTimezoneDivergence:
    +    """End-to-end: a one-shot created with a naive recent-past timestamp must
    +    become due even when the configured Hermes timezone differs from the
    +    server's local timezone. Before #51021 the naive value was anchored to
    +    server-local, so the job never fired."""
    +
    +    def test_recent_past_oneshot_is_due_under_diverging_tz(self, tmp_cron_dir, monkeypatch):
    +        # Configured zone: a fixed +05:30 offset. The server's actual local
    +        # zone is irrelevant to the parse now — that is the whole point.
    +        configured = timezone(timedelta(hours=5, minutes=30))
    +        now = datetime(2026, 6, 22, 20, 7, 30, tzinfo=configured)
    +        monkeypatch.setattr("cron.jobs._hermes_now", lambda: now)
    +
    +        # 30s ago in the configured wall clock, supplied as a NAIVE string.
    +        naive_str = (now - timedelta(seconds=30)).replace(tzinfo=None).isoformat()
    +        job = create_job(prompt="test message", schedule=naive_str, deliver="local")
    +
    +        due = get_due_jobs()
    +        assert any(d["id"] == job["id"] for d in due), (
    +            f"one-shot should be due; next_run_at={job['next_run_at']}"
    +        )
    +
     
     # =========================================================================
     # compute_next_run

• #80807 fix: dispatch cronjob(action='run') to the background like delegate_task

  merged to main

  files: cron/scheduler.py, tests/tools/test_cronjob_run_background.py, tools/cronjob_tools.py