# NousResearch/hermes-agent · Cron scheduler starvation and missed-job fixes ## Open now - Nothing open on GitHub. These items address cron scheduler issues where holding locks during job execution causes starvation and missed runs, and where jobs fast-forwarded past a missed-grace window are skipped instead of executed. Multiple approaches to decoupling tick dispatch from job completion and releasing locks earlier are represented. ## TL;DR - 22 member(s): 0 open issue(s), 6 closed issue(s), 0 open PR(s), 3 merged PR(s), 13 PR(s) closed without merging. - A fix was merged to main in PR #13021. Verify it addresses your case. - **Recommended:** read PR #13021 (and any competing PRs below) before filing. Confirm the merged change covers your symptom. ## Maintainer resolution verdict - Internal-tracker verdict over this cluster's issue members: `partially_resolved` · deciding PR(s): `pr-13021,pr-39008,pr-50062`. This is distinct from the fix status above. Weigh it yourself. ## How this cluster was grouped - Link evidence (`STRUCTURAL_PARTIAL`): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text. - Members outside that structural core may belong to a different underlying problem. - Pairwise link density `0.08`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked. ## Closed / duplicate issues - #3752: Long-running cron jobs block the scheduler tick loop - #9086: # \[Bug\]: Serial cron tick execution causes scheduled jobs to be silently skipped when preceding jobs run long - #27485: cron: tick lock held for full job duration causes scheduler starvation and missed runs on long-running jobs - #33315: bug(cron): recurring jobs perpetually deferred after long execution — missed-grace fast-forward skips instead of running - #37312: bug(cron): tick() blocks on as_completed() — decoupling dispatch from completion needed - #38779: Cron ticker waits for all due jobs, so one long active job can starve later schedules ## Prior & competing PRs - **#3764: fix(cron): release scheduler lock before running jobs**. closed without merge · fix tier `suggested` - maintainer direction: category `uncategorized`; implication `needs_llm` - **#13021: fix(cron): run due jobs in parallel to prevent serial tick starvation**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `cron/jobs.py`, `cron/scheduler.py`, `gateway/session_context.py`, `hermes_cli/config.py`, `tests/cron/test_scheduler.py`, `tools/send_message_tool.py` - fix hunks: - `cron/jobs.py` ```diff @@ -9,6 +9,7 @@ import json import logging import tempfile +import threading import os import re import uuid ``` ```diff @@ -34,6 +35,11 @@ HERMES_DIR = get_hermes_home().resolve() CRON_DIR = HERMES_DIR / "cron" JOBS_FILE = CRON_DIR / "jobs.json" + +# In-process lock protecting load_jobs→modify→save_jobs cycles. +# Required when tick() runs jobs in parallel threads — without this, +# concurrent mark_job_run / advance_next_run calls can clobber each other. +_jobs_file_lock = threading.Lock() OUTPUT_DIR = CRON_DIR / "output" ONESHOT_GRACE_SECONDS = 120 ``` ```diff @@ -594,43 +600,44 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None, ``delivery_error`` is tracked separately from the agent error — a job can succeed (agent produced output) but fail delivery (platform down). """ - jobs = load_jobs() - for i, job in enumerate(jobs): - if job["id"] == job_id: - now = _hermes_now().isoformat() - job["last_run_at"] = now - job["last_status"] = "ok" if success else "error" - job["last_error"] = error if not success else None - # Track delivery failures separately — cleared on successful delivery - job["last_delivery_error"] = delivery_error - - # Increment completed count - if job.get("repeat"): - job["repeat"]["completed"] = job["repeat"].get("completed", 0) + 1 + with _jobs_file_lock: + jobs = load_jobs() + for i, job in enumerate(jobs): + if job["id"] == job_id: + now = _hermes_now().isoformat() + job["last_run_at"] = now + job["last_status"] = "ok" if success else "error" + job["last_error"] = error if not success else None + # Track delivery failures separately — cleared on successful delivery + job["last_delivery_error"] = delivery_error - # Check if we've hit the repeat limit - times = job["repeat"].get("times") - completed = job["repeat"]["completed"] - if times is not None and times > 0 and completed >= times: - # Remove the job (limit reached) - jobs.pop(i) - save_jobs(jobs) - return - - # Compute next run - job["next_run_at"] = compute_next_run(job["schedule"], now) + # Increment completed count + if job.get("repeat"): + job["repeat"]["completed"] = job["repeat"].get("completed", 0) + 1 + + # Check if we've hit the repeat limit + times = job["repeat"].get("times") + completed = job["repeat"]["completed"] + if times is not None and times > 0 and completed >= times: + # Remove the job (limit reached) + jobs.pop(i) + save_jobs(jobs) + return + + # Compute next run + job["next_run_at"] = compute_next_run(job["schedule"], now) - # If no next run (one-shot completed), disable - if job["next_run_at"] is None: - job["enabled"] = False - job["state"] = "completed" - elif job.get("state") != "paused": - job["state"] = "scheduled" + # If no next run (one-shot completed), disable + if job["next_run_at"] is None: + job["enabled"] = False + job["state"] = "completed" + elif job.get("state") != "paused": + job["state"] = "scheduled" - save_jobs(jobs) - return + save_jobs(jobs) + return - logger.warning("mark_job_run: job_id %s not found, skipping save", job_id) + logger.warning("mark_job_run: job_id %s not found, skipping save", job_id) def advance_next_run(job_id: str) -> bool: ``` ```diff @@ -645,20 +652,21 @@ def advance_next_run(job_id: str) -> bool: Returns True if next_run_at was advanced, False otherwise. """ - jobs = load_jobs() - for job in jobs: - if job["id"] == job_id: - kind = job.get("schedule", {}).get("kind") - if kind not in ("cron", "interval"): + with _jobs_file_lock: + jobs = load_jobs() + for job in jobs: + if job["id"] == job_id: + kind = job.get("schedule", {}).get("kind") + if kind not in ("cron", "interval"): + return False + now = _hermes_now().isoformat() + new_next = compute_next_run(job["schedule"], now) + if new_next and new_next != job.get("next_run_at"): + job["next_run_at"] = new_next + save_jobs(jobs) + return True return False - now = _hermes_now().isoformat() - new_next = compute_next_run(job["schedule"], now) - if new_next and new_next != job.get("next_run_at"): - job["next_run_at"] = new_next - save_jobs(jobs) - return True - return False - return False + return False def get_due_jobs() -> List[Dict[str, Any]]: ``` - `cron/scheduler.py` ```diff @@ -747,14 +747,17 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: # scheduler process — every job this process runs is a cron job. os.environ["HERMES_CRON_SESSION"] = "1" + # Use ContextVars for per-job session/delivery state so parallel jobs + # don't clobber each other's targets (os.environ is process-global). + from gateway.session_context import set_session_vars, clear_session_vars, _VAR_MAP + + _ctx_tokens = [redacted:secret-assignment] + platform=origin["platform"] if origin else "", + chat_id=str(origin["chat_id"]) if origin else "", + chat_name=origin.get("chat_name", "") if origin else "", + ) + try: - # Inject origin context so the agent's send_message tool knows the chat. - # Must be INSIDE the try block so the finally cleanup always runs. - if origin: - os.environ["HERMES_SESSION_PLATFORM"] = origin["platform"] - os.environ["HERMES_SESSION_CHAT_ID"] = str(origin["chat_id"]) - if origin.get("chat_name"): - os.environ["HERMES_SESSION_CHAT_NAME"] = origin["chat_name"] # Re-read .env and config.yaml fresh every run so provider/key # changes take effect without a gateway restart. from dotenv import load_dotenv ``` ```diff @@ -765,10 +768,10 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: delivery_target = _resolve_delivery_target(job) if delivery_target: - os.environ["HERMES_CRON_AUTO_DELIVER_PLATFORM"] = delivery_target["platform"] - os.environ["HERMES_CRON_AUTO_DELIVER_CHAT_ID"] = str(delivery_target["chat_id"]) + _VAR_MAP["HERMES_CRON_AUTO_DELIVER_PLATFORM"].set(delivery_target["platform"]) + _VAR_MAP["HERMES_CRON_AUTO_DELIVER_CHAT_ID"].set(str(delivery_target["chat_id"])) if delivery_target.get("thread_id") is not None: - os.environ["HERMES_CRON_AUTO_DELIVER_THREAD_ID"] = str(delivery_target["thread_id"]) + _VAR_MAP["HERMES_CRON_AUTO_DELIVER_THREAD_ID"].set(str(delivery_target["thread_id"])) model = job.get("model") or os.getenv("HERMES_MODEL") or "" ``` ```diff @@ -1012,16 +1015,8 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: return False, output, "", error_msg finally: - # Clean up injected env vars so they don't leak to other jobs - for key in ( - "HERMES_SESSION_PLATFORM", - "HERMES_SESSION_CHAT_ID", - "HERMES_SESSION_CHAT_NAME", - "HERMES_CRON_AUTO_DELIVER_PLATFORM", - "HERMES_CRON_AUTO_DELIVER_CHAT_ID", - "HERMES_CRON_AUTO_DELIVER_THREAD_ID", - ): - os.environ.pop(key, None) + # Clean up ContextVar session/delivery state for this job. + clear_session_vars(_ctx_tokens) if _session_db: try: _session_db.end_session(_cron_session_id, "cron_complete") ``` ```diff @@ -1074,15 +1069,42 @@ def tick(verbose: bool = True, adapters=None, loop=None) -> int: if verbose: logger.info("%s - %s job(s) due", _hermes_now().strftime('%H:%M:%S'), len(due_jobs)) - executed = 0 + # Advance next_run_at for all recurring jobs FIRST, under the file lock, + # before any execution begins. This preserves at-most-once semantics. for job in due_jobs: + advance_next_run(job["id"]) + + # Resolve max parallel workers: env var > config.yaml > unbounded. + # Set HERMES_CRON_MAX_PARALLEL=1 to restore old serial behaviour. + _max_workers: Optional[int] = None + try: + _env_par = os.getenv("HERMES_CRON_MAX_PARALLEL", "").strip() + if _env_par: + _max_workers = int(_env_par) or None + except (ValueError, TypeError): + logger.warning("Invalid HERMES_CRON_MAX_PARALLEL value; defaulting to unbounded") + if _max_workers is None: try: - # For recurring jobs (cron/interval), advance next_run_at to the - # next future occurrence BEFORE execution. This way, if the - # process crashes mid-run, the job won't re-fire on restart. - # One-shot jobs are left alone so they can retry on restart. - advance_next_run(job["id"]) + from hermes_cli.config import load_config + _ucfg = load_config() or {} + _cfg_par = ( + _ucfg.get("cron", {}) if isinstance(_ucfg, dict) else {} + ).get("max_parallel_jobs") + if _cfg_par is not None: + _max_workers = int(_cfg_par) or None + except Exception: + pass + if verbose: + logger.info( + "Running %d job(s) in parallel (max_workers=%s)", + len(due_jobs), + _max_workers if _max_workers else "unbounded", + ) + + def _process_job(job: dict) -> bool: + """Run one due job end-to-end: execute, save, deliver, mark.""" + try: success, output, final_response, error = run_job(job) output_file = save_job_output(job["id"], output) ``` ```diff @@ -1114,13 +1136,23 @@ def tick(verbose: bool = True, adapters=None, loop=None) -> int: error = "Agent completed but produced empty response (model error, timeout, or misconfiguration)" mark_job_run(job["id"], success, error, delivery_error=delivery_error) - executed += 1 + return True except Exception as e: logger.error("Error processing job %s: %s", job['id'], e) mark_job_run(job["id"], False, str(e)) - - return executed + return False + + # Run all due jobs concurrently, each in its own ContextVar copy + # so session/delivery state stays isolated per-thread. + with concurrent.futures.ThreadPoolExecutor(max_workers=_max_workers) as _tick_pool: + _futures = [] + for job in due_jobs: + _ctx = contextvars.copy_context() + _futures.append(_tick_pool.submit(_ctx.run, _process_job, job)) + _results = [f.result() for f in _futures] + + return sum(_results) finally: if fcntl: fcntl.flock(lock_fd, fcntl.LOCK_UN) ``` - `gateway/session_context.py` ```diff @@ -56,6 +56,12 @@ _SESSION_USER_NAME: ContextVar = ContextVar("HERMES_SESSION_USER_NAME", default=_UNSET) _SESSION_KEY: ContextVar = ContextVar("HERMES_SESSION_KEY", default=_UNSET) +# Cron auto-delivery vars — set per-job in run_job() so concurrent jobs +# don't clobber each other's delivery targets. +_CRON_AUTO_DELIVER_PLATFORM: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_PLATFORM", default=_UNSET) +_CRON_AUTO_DELIVER_CHAT_ID: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_CHAT_ID", default=_UNSET) +_CRON_AUTO_DELIVER_THREAD_ID: ContextVar = ContextVar("HERMES_CRON_AUTO_DELIVER_THREAD_ID", default=_UNSET) + _VAR_MAP = { "HERMES_SESSION_PLATFORM": _SESSION_PLATFORM, "HERMES_SESSION_CHAT_ID": _SESSION_CHAT_ID, ``` ```diff @@ -64,6 +70,9 @@ "HERMES_SESSION_USER_ID": _SESSION_USER_ID, "HERMES_SESSION_USER_NAME": _SESSION_USER_NAME, "HERMES_SESSION_KEY": _SESSION_KEY, + "HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM, + "HERMES_CRON_AUTO_DELIVER_CHAT_ID": _CRON_AUTO_DELIVER_CHAT_ID, + "HERMES_CRON_AUTO_DELIVER_THREAD_ID": _CRON_AUTO_DELIVER_THREAD_ID, } ``` - `hermes_cli/config.py` ```diff @@ -796,6 +796,11 @@ def _ensure_hermes_home_managed(home: Path): # Wrap delivered cron responses with a header (task name) and footer # ("The agent cannot see this message"). Set to false for clean output. "wrap_response": True, + # Maximum number of due jobs to run in parallel per tick. + # null/0 = unbounded (limited only by thread count). + # 1 = serial (pre-v0.9 behaviour). + # Also overridable via HERMES_CRON_MAX_PARALLEL env var. + "max_parallel_jobs": None, }, # execute_code settings — controls the tool used for programmatic tool calls. ``` - `tests/cron/test_scheduler.py` ```diff @@ -772,9 +772,10 @@ def __init__(self, *args, **kwargs): pass def run_conversation(self, *args, **kwargs): - seen["platform"] = os.getenv("HERMES_CRON_AUTO_DELIVER_PLATFORM") - seen["chat_id"] = os.getenv("HERMES_CRON_AUTO_DELIVER_CHAT_ID") - seen["thread_id"] = os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID") + from gateway.session_context import get_session_env + seen["platform"] = get_session_env("HERMES_CRON_AUTO_DELIVER_PLATFORM") or None + seen["chat_id"] = get_session_env("HERMES_CRON_AUTO_DELIVER_CHAT_ID") or None + seen["thread_id"] = get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID") or None return {"final_response": "ok"} with patch("cron.scheduler._hermes_home", tmp_path), \ ``` ```diff @@ -1457,3 +1458,125 @@ def test_multiple_media_files_all_delivered(self): self._run_with_loop(adapter, "123", media_files, None, {"id": "j3"}) adapter.send_voice.assert_called_once() adapter.send_image_file.assert_called_once() + + +class TestParallelTick: + """Verify that tick() runs due jobs concurrently and isolates ContextVars.""" + + @pytest.fixture(autouse=True) + def _isolate_tick_lock(self, tmp_path): + """Point the tick file lock at a per-test temp dir to avoid xdist contention.""" + lock_dir = tmp_path / "cron" + lock_dir.mkdir() + with patch("cron.scheduler._LOCK_DIR", lock_dir), \ + patch("cron.scheduler._LOCK_FILE", lock_dir / ".tick.lock"): + yield + + def test_parallel_jobs_run_concurrently(self): + """Two jobs launched in the same tick should overlap in time.""" + import threading + import time + + barrier = threading.Barrier(2, timeout=5) + call_order = [] + + def mock_run_job(job): + """Each job hits a barrier — both must be active simultaneously.""" + call_order.append(("start", job["id"])) + barrier.wait() # blocks until both threads reach here + call_order.append(("end", job["id"])) + return (True, "output", "response", None) + + jobs = [ + {"id": "job-a", "name": "a", "deliver": "local"}, + {"id": "job-b", "name": "b", "deliver": "local"}, + ] + + with patch("cron.scheduler.get_due_jobs", return_value=jobs), \ + patch("cron.scheduler.advance_next_run"), \ + patch("cron.scheduler.run_job", side_effect=mock_run_job), \ + patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ + patch("cron.scheduler._deliver_result", return_value=None), \ + patch("cron.scheduler.mark_job_run"): + from cron.scheduler import tick + result = tick(verbose=False) + + assert result == 2 + # Both starts happened before both ends — proof of concurrency + starts = [i for i, (action, _) in enumerate(call_order) if action == "start"] + ends = [i for i, (action, _) in enumerate(call_order) if action == "end"] + assert len(starts) == 2 + assert len(ends) == 2 + assert max(starts) < min(ends), f"Jobs not concurrent: {call_order}" + + def test_parallel_jobs_isolated_contextvars(self): + """Each job's ContextVars must be isolated — no cross-contamination.""" + from gateway.session_context import get_session_env + seen = {} + + def mock_run_job(job): + origin = job.get("origin", {}) + # run_job sets ContextVars — verify each job sees its own + from gateway.session_context import set_session_vars, clear_session_vars + tokens = [redacted:secret-assignment] + platform=origin.get("platform", ""), + chat_id=str(origin.get("chat_id", "")), + ) + import time + time.sleep(0.05) # give other thread time to set its vars + platform = get_session_env("HERMES_SESSION_PLATFORM") + chat_id = get_session_env("HERMES_SESSION_CHAT_ID") + seen[job["id"]] = {"platform": platform, "chat_id": chat_id} + clear_session_vars(tokens) + return (True, "output", "response", None) + + jobs = [ + {"id": "tg-job", "name": "tg", "deliver": "local", + "origin": {"platform": "telegram", "chat_id": "111"}}, + {"id": "dc-job", "name": "dc", "deliver": "local", + "origin": {"platform": "discord", "chat_id": "222"}}, + ] + + with patch("cron.scheduler.get_due_jobs", return_value=jobs), \ + patch("cron.scheduler.advance_next_run"), \ + patch("cron.scheduler.run_job", side_effect=mock_run_job), \ + patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ + patch("cron.scheduler._deliver_result", return_value=None), \ + patch("cron.scheduler.mark_job_run"): + from cron.scheduler import tick + tick(verbose=False) + + assert seen["tg-job"] == {"platform": "telegram", "chat_id": "111"} + assert seen["dc-job"] == {"platform": "discord", "chat_id": "222"} + + def test_max_parallel_env_var(self, monkeypatch): + """HERMES_CRON_MAX_PARALLEL=1 should restore serial behaviour.""" + monkeypatch.setenv("HERMES_CRON_MAX_PARALLEL", "1") + call_times = [] + + def mock_run_job(job): + import time + call_times.append(("start", job["id"], time.monotonic())) + time.sleep(0.05) + call_times.append(("end", job["id"], time.monotonic())) + return (True, "output", "response", None) + + jobs = [ + {"id": "s1", "name": "s1", "deliver": "local"}, + {"id": "s2", "name": "s2", "deliver": "local"}, + ] + + with patch("cron.scheduler.get_due_jobs", return_value=jobs), \ + patch("cron.scheduler.advance_next_run"), \ + patch("cron.scheduler.run_job", side_effect=mock_run_job), \ + patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ + patch("cron.scheduler._deliver_result", return_value=None), \ + patch("cron.scheduler.mark_job_run"): + from cron.scheduler import tick + result = tick(verbose=False) + + assert result == 2 + # With max_workers=1, second job starts after first ends + end_s1 = [t for action, jid, t in call_times if action == "end" and jid == "s1"][0] + start_s2 = [t for action, jid, t in call_times if action == "start" and jid == "s2"][0] + assert start_s2 >= end_s1, "Jobs ran concurrently despite max_parallel=1" ``` - `tools/send_message_tool.py` ```diff @@ -359,11 +359,12 @@ def _describe_media_for_mirror(media_files): def _get_cron_auto_delivery_target(): """Return the cron scheduler's auto-delivery target for the current run, if any.""" - platform = os.getenv("HERMES_CRON_AUTO_DELIVER_PLATFORM", "").strip().lower() - chat_id = os.getenv("HERMES_CRON_AUTO_DELIVER_CHAT_ID", "").strip() + from gateway.session_context import get_session_env + platform = get_session_env("HERMES_CRON_AUTO_DELIVER_PLATFORM", "").strip().lower() + chat_id = get_session_env("HERMES_CRON_AUTO_DELIVER_CHAT_ID", "").strip() if not platform or not chat_id: return None - thread_id = os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID", "").strip() or None + thread_id = get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID", "").strip() or None return { "platform": platform, "chat_id": chat_id, ``` - **#27492: fix(cron): release tick file lock before executing due jobs**. closed without merge · fix tier `suggested` - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#27485` - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._ - **#33318: fix(cron): execute missed-grace jobs once instead of deferring**. closed without merge · fix tier `suggested` - maintainer direction: category `uncategorized`; implication `needs_llm`; refs `#50062` - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._ - **#33559: fix(cron): prevent perpetual-defer loop for slow recurring jobs (fixes #33315)**. closed without merge · fix tier `suggested` - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#33318` - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._ - **#38624: fix(cron): release tick lock before job execution to prevent scheduler starvation**. closed without merge · fix tier `suggested` - maintainer direction: category `uncategorized`; implication `needs_llm`; refs `eb9cde734642cc2a8ecd142ba7c63d3964ca835b`, `9fbfeb31b9c611c9f06f84ca95ea6b4fbad6c3e1`, `#27485` - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._ - **#38718: fix(cron): NameError get_fallback_chain + silent ticker death + tick() blocks (#36734 #36854 #37312)**. closed without merge · fix tier `suggested` - maintainer direction: category `partial_fix`; implication `needs_llm`; refs `#37312`, `#39008`, `#36734`, `#36854` - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._ - **#39008: fix(cron): decouple tick() dispatch from job completion (#37312)**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `cron/scheduler.py`, `gateway/run.py`, `tests/cron/test_cron_profile.py`, `tests/cron/test_cron_workdir.py`, `tests/cron/test_parallel_pool.py` - fix hunks: - `cron/scheduler.py` ```diff @@ -9,6 +9,7 @@ """ import asyncio +import atexit import concurrent.futures import contextvars import json ``` ```diff @@ -17,6 +18,7 @@ import shutil import subprocess import sys +import threading from contextlib import contextmanager # fcntl is Unix-only; on Windows use msvcrt for file locking ``` ```diff @@ -154,6 +156,69 @@ def _resolve_cron_enabled_toolsets(job: dict, cfg: dict) -> list[str] | None: # locally for audit. SILENT_MARKER = "[SILENT]" +# --------------------------------------------------------------------------- +# Persistent thread pool for parallel cron jobs. +# The tick function submits jobs here and returns immediately so the ticker +# thread is never blocked by long-running jobs (e.g. the fixer running 15+ min). +# --------------------------------------------------------------------------- +_parallel_pool: Optional[concurrent.futures.ThreadPoolExecutor] = None +_parallel_pool_max_workers: Optional[int] = None +_running_job_ids: set = set() +_running_lock = threading.Lock() + +# Sequential (env/context-mutating) cron jobs — workdir/profile jobs that touch +# process-global runtime state — must run one at a time, but must NOT block the +# ticker thread. A persistent single-thread executor preserves ordering across +# ticks while keeping dispatch fire-and-forget, the same as the parallel pool. +_sequential_pool: Optional[concurrent.futures.ThreadPoolExecutor] = None + + +def _get_parallel_pool(max_workers: Optional[int]) -> concurrent.futures.ThreadPoolExecutor: + """Return (or create) the persistent parallel pool.""" + global _parallel_pool, _parallel_pool_max_workers + if _parallel_pool is None or _parallel_pool_max_workers != max_workers: + if _parallel_pool is not None: + _parallel_pool.shutdown(wait=False, cancel_futures=False) + _parallel_pool = concurrent.futures.ThreadPoolExecutor( + max_workers=max_workers, + thread_name_prefix="cron-parallel", + ) + _parallel_pool_max_workers = max_workers + return _parallel_pool + + +def _get_sequential_pool() -> concurrent.futures.ThreadPoolExecutor: + """Return (or create) the persistent single-thread sequential pool. + + A single worker guarantees env/context-mutating jobs never overlap, even + across ticks: a job queued by a newer tick waits for the previous tick's + sequential jobs to finish rather than corrupting their os.environ / + profile state. + """ + global _sequential_pool + if _sequential_pool is None: + _sequential_pool = concurrent.futures.ThreadPoolExecutor( + max_workers=1, + thread_name_prefix="cron-seq", + ) + return _sequential_pool + + +def _shutdown_parallel_pool() -> None: + """Shut down the persistent pools on process exit.""" + global _parallel_pool, _parallel_pool_max_workers, _sequential_pool + if _parallel_pool is not None: + _parallel_pool.shutdown(wait=True, cancel_futures=False) + _parallel_pool = None + _parallel_pool_max_workers = None + if _sequential_pool is not None: + _sequential_pool.shutdown(wait=True, cancel_futures=False) + _sequential_pool = None + + +atexit.register(_shutdown_parallel_pool) + + # Backward-compatible module override used by tests and emergency monkeypatches. _hermes_home: Path | None = None ``` ```diff @@ -1895,7 +1960,7 @@ def _run_job_impl(job: dict) -> tuple[bool, str, str, Optional[str]]: logger.debug("Job '%s': failed to reap stale auxiliary clients: %s", job_id, e) -def tick(verbose: bool = True, adapters=None, loop=None) -> int: +def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> int: """ Check and run all due jobs. ``` ```diff @@ -1939,6 +2004,9 @@ def tick(verbose: bool = True, adapters=None, loop=None) -> int: # Advance next_run_at for all recurring jobs FIRST, under the file lock, # before any execution begins. This preserves at-most-once semantics. + # For parallel jobs that are already running, advance_next_run keeps + # bumping next_run_at forward so the grace window never expires. + # mark_job_run() overwrites next_run_at on completion. for job in due_jobs: advance_next_run(job["id"]) ``` ```diff @@ -2030,36 +2098,103 @@ def _process_job(job: dict) -> bool: ] _results: list = [] - - # Sequential pass for env/context-mutating jobs. - for job in sequential_jobs: + _all_futures: list = [] + + def _submit_with_guard(job: dict, pool: concurrent.futures.ThreadPoolExecutor): + """Submit a job fire-and-forget with the in-flight dedup guard. + + Returns the future, or None if the job was skipped because a prior + tick's run of the same job is still in flight. The running-set + membership is released in the worker's finally block. + """ + job_id = job["id"] + with _running_lock: + if job_id in _running_job_ids: + logger.info("Job '%s' already running — skipping", job.get("name", job_id)) + return None + _running_job_ids.add(job_id) _ctx = contextvars.copy_context() - _results.append(_ctx.run(_process_job, job)) - # Parallel pass for the rest — same behaviour as before. + def _run_and_release(j=job, ctx=_ctx): + try: + return ctx.run(_process_job, j) + finally: + with _running_lock: + _running_job_ids.discard(j["id"]) + + return pool.submit(_run_and_release) + + # Sequential pass for env/context-mutating (workdir/profile) jobs. + # Queued to a persistent single-thread pool so they run one at a time + # WITHOUT blocking the ticker thread — a long workdir/profile job no + # longer starves the rest of the schedule (same fix as the parallel + # pass, just serialized). The in-flight guard prevents a still-running + # job from being re-queued on the next tick. + if sequential_jobs: + seq_pool = _get_sequential_pool() + for job in sequential_jobs: + fut = _submit_with_guard(job, seq_pool) + if fut is None: + continue + _all_futures.append(fut) + if not sync: + _results.append(True) # optimistically counted + + # Parallel pass — persistent pool, non-blocking dispatch. + # Jobs that are already running (from a previous tick) are skipped. + # mark_job_run() updates next_run_at on completion, so the next tick + # after completion finds the job due again naturally. No catch-up + # queue needed. if parallel_jobs: - with concurrent.futures.ThreadPoolExecutor(max_workers=_max_workers) as _tick_pool: - _futures = [] - for job in parallel_jobs: - _ctx = contextvars.copy_context() - _futures.append(_tick_pool.submit(_ctx.run, _process_job, job)) - for f in concurrent.futures.as_completed(_futures, timeout=600): - try: - _results.append(f.result()) - except Exception as exc: - logger.error("Parallel cron job future failed: %s", exc) - _results.append(False) + pool = _get_parallel_pool(_max_workers) + for job in parallel_jobs: + fut = _submit_with_guard(job, pool) + if fut is None: + continue + _all_futures.append(fut) + if not sync: + _results.append(True) # optimistically counted # Best-effort sweep of MCP stdio subprocesses that survived their - # session teardown during this tick. Runs AFTER every job has - # finished so active sessions (including live user chats) are - # never touched — only PIDs explicitly detected as orphans in - # tools.mcp_tool._run_stdio's finally block are reaped. - try: - from tools.mcp_tool import _kill_orphaned_mcp_children - _kill_orphaned_mcp_children() - except Exception as _e: - logger.debug("Post-tick MCP orphan cleanup failed: %s", _e) + # session teardown. Must run AFTER jobs finish so active sessions + # (including live user chats) are never touched — only PIDs explicitly + # detected as orphans in tools.mcp_tool._run_stdio's finally block are + # reaped. + def _sweep_mcp_orphans() -> None: + try: + from tools.mcp_tool import _kill_orphaned_mcp_children + _kill_orphaned_mcp_children() + except Exception as _e: + logger.debug("Post-tick MCP orphan cleanup failed: %s", _e) + + if sync: + # Sync mode (tests / manual ticks): wait for all dispatched jobs, + # collect results, then sweep once. + for f in concurrent.futures.as_completed(_all_futures): + try: + _results.append(f.result()) + except Exception as exc: + logger.error("Cron job future failed: %s", exc) + _results.append(False) + _sweep_mcp_orphans() + return sum(_results) + + # Async (gateway ticker) mode: don't block. Sweep orphans via a + # done-callback fired after the LAST dispatched job completes, so the + # sweep still happens after jobs finish without stalling the tick. + if _all_futures: + _remaining = [len(_all_futures)] + + def _on_done(_f: concurrent.futures.Future) -> None: + _remaining[0] -= 1 + if _remaining[0] <= 0: + _sweep_mcp_orphans() + + for _f in _all_futures: + _f.add_done_callback(_on_done) + else: + # Nothing dispatched (all skipped / no due jobs) — sweep inline. + _sweep_mcp_orphans() return sum(_results) finally: ``` - `gateway/run.py` ```diff @@ -19215,7 +19215,7 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in tick_count = 0 while not stop_event.is_set(): try: - cron_tick(verbose=False, adapters=adapters, loop=loop) + cron_tick(verbose=False, adapters=adapters, loop=loop, sync=False) except Exception as e: logger.debug("Cron tick error: %s", e) ``` - `tests/cron/test_cron_profile.py` ```diff @@ -410,16 +410,20 @@ def test_profile_jobs_run_sequentially(self, isolated_cron_profile_home, monkeyp import threading import cron.scheduler as sched - profile_job = {"id": "a", "name": "A", "profile": "default"} - parallel_job = {"id": "b", "name": "B", "profile": None} + # Two profile jobs (both sequential) + one parallel job. + profile_a = {"id": "a", "name": "A", "profile": "default"} + profile_b = {"id": "b", "name": "B", "profile": "default"} + parallel_job = {"id": "c", "name": "C", "profile": None} - monkeypatch.setattr(sched, "get_due_jobs", lambda: [profile_job, parallel_job]) + monkeypatch.setattr(sched, "get_due_jobs", lambda: [profile_a, profile_b, parallel_job]) monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) calls: list[tuple[str, str]] = [] + order_lock = threading.Lock() def fake_run_job(job): - calls.append((job["id"], threading.current_thread().name)) + with order_lock: + calls.append((job["id"], threading.current_thread().name)) return True, "output", "response", None monkeypatch.setattr(sched, "run_job", fake_run_job) ``` ```diff @@ -429,9 +433,17 @@ def fake_run_job(job): n = sched.tick(verbose=False) - assert n == 2 + assert n == 3 ids = [job_id for job_id, _thread_name in calls] + # Sequential profile jobs preserve submission order relative to each + # other (single-thread pool). assert ids.index("a") < ids.index("b") - main_thread_name = threading.current_thread().name - profile_thread_name = next(thread for job_id, thread in calls if job_id == "a") - assert profile_thread_name == main_thread_name + # Sequential (profile) jobs run on the persistent single-thread + # cron-seq pool — NOT the main thread — so a long profile job never + # blocks the ticker. Parallel jobs run on the cron-parallel pool. + for jid in ("a", "b"): + seq_thread = next(t for job_id, t in calls if job_id == jid) + assert seq_thread != threading.current_thread().name + assert seq_thread.startswith("cron-seq"), seq_thread + par_thread = next(t for job_id, t in calls if job_id == "c") + assert par_thread.startswith("cron-parallel"), par_thread ``` - `tests/cron/test_cron_workdir.py` ```diff @@ -207,20 +207,23 @@ class TestTickWorkdirPartition: def test_workdir_jobs_run_sequentially(self, tmp_path, monkeypatch): import cron.scheduler as sched - # Two "jobs" — one with workdir, one without. get_due_jobs returns both. - workdir_job = {"id": "a", "name": "A", "workdir": str(tmp_path)} - parallel_job = {"id": "b", "name": "B", "workdir": None} + # Two workdir jobs (both sequential) + one parallel job. + workdir_a = {"id": "a", "name": "A", "workdir": str(tmp_path)} + workdir_b = {"id": "b", "name": "B", "workdir": str(tmp_path)} + parallel_job = {"id": "c", "name": "C", "workdir": None} - monkeypatch.setattr(sched, "get_due_jobs", lambda: [workdir_job, parallel_job]) + monkeypatch.setattr(sched, "get_due_jobs", lambda: [workdir_a, workdir_b, parallel_job]) monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) # Record call order / thread context. import threading - calls: list[tuple[str, bool]] = [] + calls: list[tuple[str, str]] = [] + order_lock = threading.Lock() def fake_run_job(job): # Return a minimal tuple matching run_job's signature. - calls.append((job["id"], threading.current_thread().name)) + with order_lock: + calls.append((job["id"], threading.current_thread().name)) return True, "output", "response", None monkeypatch.setattr(sched, "run_job", fake_run_job) ``` ```diff @@ -231,16 +234,22 @@ def fake_run_job(job): ) n = sched.tick(verbose=False) - assert n == 2 + assert n == 3 ids = [c[0] for c in calls] - # Workdir jobs always come before parallel jobs. + # Sequential workdir jobs preserve submission order relative to each + # other (single-thread pool). assert ids.index("a") < ids.index("b") - # The workdir job must run on the main thread (sequential pass). + # Workdir jobs run on the persistent single-thread cron-seq pool — + # NOT the main thread — so a long workdir job never blocks the ticker. main_thread_name = threading.current_thread().name - workdir_thread_name = next(t for jid, t in calls if jid == "a") - assert workdir_thread_name == main_thread_name + for jid in ("a", "b"): + workdir_thread_name = next(t for j, t in calls if j == jid) + assert workdir_thread_name != main_thread_name + assert workdir_thread_name.startswith("cron-seq"), workdir_thread_name + par_thread_name = next(t for j, t in calls if j == "c") + assert par_thread_name.startswith("cron-parallel"), par_thread_name # --------------------------------------------------------------------------- ``` - `tests/cron/test_parallel_pool.py` ```diff @@ -0,0 +1,274 @@ +"""Tests for the persistent parallel pool and running-job guard in cron/scheduler.py. + +These verify the fix for the tick-blocking issue where as_completed(timeout=600) +prevented the ticker thread from firing, causing all other jobs to be fast-forwarded. +""" + +import concurrent.futures +import threading +import time +from unittest.mock import patch + +import pytest + + +class TestPersistentPool: + """_get_parallel_pool returns a persistent ThreadPoolExecutor.""" + + def test_pool_is_reused(self, monkeypatch): + """Same pool instance returned when max_workers doesn't change.""" + import cron.scheduler as sched + + # Reset module state. + sched._parallel_pool = None + sched._parallel_pool_max_workers = None + + pool1 = sched._get_parallel_pool(4) + pool2 = sched._get_parallel_pool(4) + assert pool1 is pool2 + + # Cleanup. + sched._shutdown_parallel_pool() + + def test_pool_is_recreated_on_worker_change(self, monkeypatch): + """New pool when max_workers changes.""" + import cron.scheduler as sched + + sched._parallel_pool = None + sched._parallel_pool_max_workers = None + + pool1 = sched._get_parallel_pool(2) + pool2 = sched._get_parallel_pool(4) + assert pool1 is not pool2 + + sched._shutdown_parallel_pool() + + def test_shutdown_clears_pool(self, monkeypatch): + """_shutdown_parallel_pool resets state.""" + import cron.scheduler as sched + + sched._parallel_pool = None + sched._parallel_pool_max_workers = None + sched._get_parallel_pool(2) + + sched._shutdown_parallel_pool() + assert sched._parallel_pool is None + assert sched._parallel_pool_max_workers is None + + +class TestRunningJobGuard: + """_running_job_ids prevents double-dispatch of active jobs.""" + + def test_running_set_prevents_double_dispatch(self, tmp_path, monkeypatch): + """A job already in _running_job_ids is skipped on the next tick.""" + import cron.scheduler as sched + + # Reset state. + sched._parallel_pool = None + sched._parallel_pool_max_workers = None + sched._running_job_ids.clear() + + job = { + "id": "guard-job", + "name": "guard-test", + "prompt": "test", + "schedule": "every 5m", + "enabled": True, + "next_run_at": "2020-01-01T00:00:00", + "deliver": "local", + } + + # Simulate the job already running. + sched._running_job_ids.add("guard-job") + + dispatched = [] + monkeypatch.setattr(sched, "get_due_jobs", lambda: [job]) + monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "run_job", lambda j: dispatched.append(j["id"]) or (True, "out", "resp", None)) + monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) + + n = sched.tick(verbose=False) + assert n == 0 # skipped, not dispatched + assert dispatched == [] + + sched._running_job_ids.discard("guard-job") + sched._shutdown_parallel_pool() + + +class TestSyncMode: + """tick() blocks by default (sync=True); tick(sync=False) returns immediately.""" + + def test_sync_true_blocks_and_returns_correct_count(self, tmp_path, monkeypatch): + """sync=True waits for jobs and returns actual results.""" + import cron.scheduler as sched + + sched._parallel_pool = None + sched._parallel_pool_max_workers = None + sched._running_job_ids.clear() + + jobs = [ + {"id": f"job-{i}", "name": f"Job {i}", "prompt": "test", + "schedule": "every 5m", "enabled": True, + "next_run_at": "2020-01-01T00:00:00", "deliver": "local"} + for i in range(3) + ] + + monkeypatch.setattr(sched, "get_due_jobs", lambda: jobs) + monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "run_job", lambda j: (True, "out", "resp", None)) + monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: "/tmp/out") + monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) + + n = sched.tick(verbose=False) + assert n == 3 + + sched._shutdown_parallel_pool() + + def test_sync_false_returns_immediately(self, tmp_path, monkeypatch): + """sync=False returns before parallel jobs finish (optimistic count).""" + import cron.scheduler as sched + + sched._parallel_pool = None + sched._parallel_pool_max_workers = None + sched._running_job_ids.clear() + + job = { + "id": "slow-job", + "name": "slow", + "prompt": "test", + "schedule": "every 5m", + "enabled": True, + "next_run_at": "2020-01-01T00:00:00", + "deliver": "local", + } + + barrier = threading.Barrier(2, timeout=5) + + def slow_run(j): + barrier.wait() # blocks until test thread also waits + return True, "out", "resp", None + + monkeypatch.setattr(sched, "get_due_jobs", lambda: [job]) + monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "run_job", slow_run) + monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: "/tmp/out") + monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) + + start = time.monotonic() + n = sched.tick(verbose=False, sync=False) # opt-in: non-blocking + elapsed = time.monotonic() - start + + assert n == 1 # optimistic count + assert elapsed < 1.0 # returned immediately, didn't wait for slow_run + + # Let the job finish so cleanup works. + barrier.wait() + time.sleep(0.1) + sched._shutdown_parallel_pool() + + +class TestSequentialPool: + """Sequential (workdir/profile) jobs use the persistent cron-seq pool. + + Verifies the follow-up fix: env/context-mutating jobs no longer run inline + in the ticker thread, so a long workdir/profile job can't starve the + schedule the same way the parallel path used to. + """ + + def test_sequential_job_does_not_block_ticker(self, tmp_path, monkeypatch): + """sync=False returns immediately even when a workdir job is slow.""" + import cron.scheduler as sched + + sched._parallel_pool = None + sched._parallel_pool_max_workers = None + sched._sequential_pool = None + sched._running_job_ids.clear() + + job = { + "id": "slow-workdir", + "name": "slow-workdir", + "prompt": "test", + "schedule": "every 5m", + "enabled": True, + "next_run_at": "2020-01-01T00:00:00", + "deliver": "local", + "workdir": str(tmp_path), # makes it sequential + } + + barrier = threading.Barrier(2, timeout=5) + + def slow_run(j): + barrier.wait() + return True, "out", "resp", None + + monkeypatch.setattr(sched, "get_due_jobs", lambda: [job]) + monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "run_job", slow_run) + monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: "/tmp/out") + monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) + + start = time.monotonic() + n = sched.tick(verbose=False, sync=False) + elapsed = time.monotonic() - start + + assert n == 1 # optimistic count + assert elapsed < 1.0 # did NOT block on the slow workdir job + + barrier.wait() + time.sleep(0.1) + sched._shutdown_parallel_pool() + + def test_sequential_running_guard_prevents_double_dispatch(self, tmp_path, monkeypatch): + """A workdir job already in _running_job_ids is skipped on next tick.""" + import cron.scheduler as sched + + sched._parallel_pool = None + sched._parallel_pool_max_workers = None + sched._sequential_pool = None + sched._running_job_ids.clear() + + job = { + "id": "guard-seq", + "name": "guard-seq", + "prompt": "test", + "schedule": "every 5m", + "enabled": True, + "next_run_at": "2020-01-01T00:00:00", + "deliver": "local", + "workdir": str(tmp_path), + } + + # Simulate the job already running. + sched._running_job_ids.add("guard-seq") + + dispatched = [] + monkeypatch.setattr(sched, "get_due_jobs", lambda: [job]) + monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "run_job", lambda j: dispatched.append(j["id"]) or (True, "out", "resp", None)) + monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) + monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) + + n = sched.tick(verbose=False) + assert n == 0 # skipped, not dispatched + assert dispatched == [] + + sched._running_job_ids.discard("guard-seq") + sched._shutdown_parallel_pool() + + def test_get_sequential_pool_is_persistent(self): + """_get_sequential_pool returns the same single-thread pool.""" + import cron.scheduler as sched + + sched._sequential_pool = None + pool1 = sched._get_sequential_pool() + pool2 = sched._get_sequential_pool() + assert pool1 is pool2 + + sched._shutdown_parallel_pool() + assert sched._sequential_pool is None ``` - **#50062: fix(cron): run missed-grace jobs once instead of deferring forever**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `cron/jobs.py`, `tests/cron/test_jobs.py` - fix hunks: - `cron/jobs.py` ```diff @@ -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() ``` ```diff @@ -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` ```diff @@ -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) ``` ```diff @@ -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() ``` ```diff @@ -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) ``` - **#7158: feat(cron): run independent cron jobs in parallel with safe non-overlap**. closed without merge - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `eb9cde734`, `#3752` - **#9097: fix(cron): #9086 differentiate scheduler-blocking from gateway downti…**. closed without merge - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `c86915024`, `#9086`, `#13021`, `#9086`, `#9086` - **#9169: fix(cron): run due jobs in parallel to prevent serial tick starvation**. closed without merge - maintainer direction: category `merged_via`; implication `superseded`; refs `#13021` - **#33361: fix(cron): execute job once after missed-grace fast-forward instead of skipping**. closed without merge - maintainer direction: category `merged_via`; implication `superseded`; refs `#50062`, `#50062`, `#33318` - **#37313: fix(cron): decouple job dispatch from completion in tick()**. closed without merge - maintainer direction: category `merged_via`; implication `superseded`; refs `eb9cde734`, `#39008` - **#37355: fix(cron): decouple tick() dispatch from job completion #37312**. closed without merge - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#37312`, `#39008`, `#37313`, `#37313` - **#37668: fix(cron): decouple parallel dispatch from completion so slow jobs can't stall the ticker (fixes #37312)**. closed without merge - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#37312`, `#39008`, `#37313` ## Fix status - Merged to main in PR #13021 (cited above). No release data is tracked. Verify downstream. - Fix last verified (merged) `2026-06-22`. - Subsystem: `cron`