Cron scheduler starvation and missed-job fixes

Fixes on record

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

• #13021 fix(cron): run due jobs in parallel to prevent serial tick starvation

  merged to main

  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

  Show the diff

  • cron/jobs.py

    @@ -9,6 +9,7 @@
     import json
     import logging
     import tempfile
    +import threading
     import os
     import re
     import uuid
    @@ -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
     
    @@ -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:
    @@ -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

    @@ -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
    @@ -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 ""
     
    @@ -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")
    @@ -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)
    @@ -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

    @@ -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,
    @@ -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

    @@ -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

    @@ -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), \
    @@ -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

    @@ -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,

• #39008 fix(cron): decouple tick() dispatch from job completion (#37312)

  merged to main

  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

  Show the diff

  • cron/scheduler.py

    @@ -9,6 +9,7 @@
     """
     
     import asyncio
    +import atexit
     import concurrent.futures
     import contextvars
     import json
    @@ -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
    @@ -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
     
    @@ -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.
         
    @@ -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"])
     
    @@ -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

    @@ -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

    @@ -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)
    @@ -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

    @@ -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)
    @@ -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

    @@ -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

  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)