diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index 0f3b5a272b3f3..ad4806c757159 100644 --- a/hermes_cli/cron.py +++ b/hermes_cli/cron.py @@ -374,11 +374,12 @@ def _job_action(action: str, job_id: str, success_verb: str) -> int: print(f" Next run: {result['job']['next_run_at']}") if action == "run": job = result.get("job", {}) - if job.get("executed"): - outcome = "succeeded" if job.get("execution_success") else "failed" - print(f" Ran now: {outcome}.") - elif job.get("execution_skipped"): + if job.get("execution_skipped"): print(f" {job['execution_skipped']}") + elif job.get("trigger_mode") == "background": + print(" Triggered — running in background thread.") + elif job.get("triggered"): + print(" Triggered — will run on the next scheduler tick.") else: print(" It will run on the next scheduler tick.") return 0 diff --git a/tests/tools/test_cronjob_run_immediate.py b/tests/tools/test_cronjob_run_immediate.py index 9efa60e82cb1a..302c4b823ffed 100644 --- a/tests/tools/test_cronjob_run_immediate.py +++ b/tests/tools/test_cronjob_run_immediate.py @@ -1,81 +1,185 @@ -"""Tests for cronjob action='run' immediate execution (#41037). +"""Tests for cronjob action='run' non-blocking execution (#52705). -Before this fix, `cronjob(action='run')` only set next_run_at=now and returned -success, relying on the scheduler ticker to actually run the job. With no -gateway/ticker active (e.g. a CLI-only Windows setup) the job never executed and -last_run_at stayed null forever. Now action='run' claims the job (at-most-once, -blocking a concurrent tick) and fires it inline via the shared run_one_job body. +Before #52705, `cronjob(action='run')` called ``run_one_job`` synchronously +inside the tool handler, blocking the calling agent until the entire cron +session (LLM loop, output save, delivery) completed. Now action='run' is +non-blocking: + +- Gateway alive → arms the job via ``trigger_job`` (sets next_run_at=now) so + the scheduler ticker fires it asynchronously on its next cycle. +- Gateway not alive → claims via ``claim_job_for_fire`` and spawns a daemon + thread that calls ``run_one_job``. + +Either way the tool returns immediately. The caller checks post-run status via +``cronjob(action='list')``. """ import json -from unittest.mock import patch +import threading +from unittest.mock import patch, MagicMock -from tools.cronjob_tools import cronjob, _execute_job_now +from tools.cronjob_tools import cronjob _JOB = {"id": "job-run-1", "name": "manual run", "prompt": "hi", "schedule": {"kind": "cron", "expr": "0 9 * * *"}} -class TestCronjobRunExecutesImmediately: - def test_run_action_claims_and_fires_via_run_one_job(self): - """action='run' must claim the job then fire it through run_one_job.""" - ran = {"job": "after-run", "last_status": "ok", "last_error": None} +class TestCronjobRunNonBlocking: + """action='run' must never block the calling agent (#52705).""" + + # ── Gateway alive path ────────────────────────────────────────────── + + def test_run_uses_scheduler_path_when_gateway_alive(self): + """When the gateway is running, action='run' arms the job via + trigger_job (sets next_run_at=now) and does NOT call run_one_job + synchronously.""" + with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools._is_gateway_active", return_value=True), \ + patch("tools.cronjob_tools.trigger_job", return_value=dict(_JOB, next_run_at="now")) as m_trigger, \ + patch("cron.scheduler.run_one_job") as m_run, \ + patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)): + out = json.loads(cronjob(action="run", job_id="job-run-1")) + + assert out["success"] is True + assert out["job"]["triggered"] is True + assert out["job"]["trigger_mode"] == "scheduled" + m_trigger.assert_called_once_with("job-run-1") + m_run.assert_not_called() + + def test_run_never_blocks_when_gateway_alive(self): + """The tool must return immediately even if trigger_job would 'take + forever' — proving no synchronous work leak.""" + import time as _time + + def slow_trigger(_jid): + _time.sleep(0.5) # simulate scheduling overhead + return dict(_JOB, next_run_at="now") + + with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools._is_gateway_active", return_value=True), \ + patch("tools.cronjob_tools.trigger_job", side_effect=slow_trigger), \ + patch("cron.scheduler.run_one_job") as m_run, \ + patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)): + out = json.loads(cronjob(action="run", job_id="job-run-1")) + + assert out["success"] is True + m_run.assert_not_called() + + # ── Gateway not-alive path ────────────────────────────────────────── + + def test_run_spawns_thread_when_gateway_not_alive(self): + """When the gateway is NOT running, action='run' claims the job and + spawns a daemon thread — never calling run_one_job synchronously.""" + mock_thread = MagicMock() with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools._is_gateway_active", return_value=False), \ patch("tools.cronjob_tools.claim_job_for_fire", return_value=True) as m_claim, \ - patch("cron.scheduler.run_one_job", return_value=True) as m_run, \ - patch("tools.cronjob_tools.get_job", return_value=ran): + patch("cron.scheduler.run_one_job") as m_run, \ + patch("threading.Thread", return_value=mock_thread) as m_thread_cls, \ + patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)): out = json.loads(cronjob(action="run", job_id="job-run-1")) assert out["success"] is True - assert out["job"]["executed"] is True - assert out["job"]["execution_success"] is True - m_claim.assert_called_once_with("job-run-1") # at-most-once claim taken - m_run.assert_called_once() # fired via the shared body + assert out["job"]["triggered"] is True + assert out["job"]["trigger_mode"] == "background" + m_claim.assert_called_once_with("job-run-1") + m_run.assert_not_called() # NOT called synchronously + m_thread_cls.assert_called_once() + mock_thread.start.assert_called_once() + # Thread must be daemon so it doesn't block process exit + assert m_thread_cls.call_args.kwargs.get("daemon") is True - def test_run_skips_when_claim_lost(self): - """If the scheduler already holds the fire claim, do NOT double-run.""" + def test_run_skips_when_claim_lost_no_thread(self): + """In daemon-thread mode, a lost claim means no thread spawned.""" with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools._is_gateway_active", return_value=False), \ patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \ - patch("cron.scheduler.run_one_job") as m_run, \ + patch("threading.Thread") as m_thread, \ patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)): out = json.loads(cronjob(action="run", job_id="job-run-1")) assert out["success"] is True assert out["job"]["executed"] is False - assert out["job"]["execution_success"] is False assert "execution_skipped" in out["job"] - m_run.assert_not_called() # claim lost -> never fired + m_thread.assert_not_called() - def test_run_reports_failure_from_last_status(self): - """A failed run is reported via the re-read job's last_status/last_error.""" - failed = {"id": "job-run-1", "last_status": "error", "last_error": "provider 500"} + def test_run_no_claim_no_trigger_when_claim_lost(self): + """Claim-lost in daemon mode: trigger_job must NOT be called either.""" with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ - patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ - patch("cron.scheduler.run_one_job", return_value=True), \ - patch("tools.cronjob_tools.get_job", return_value=failed): - out = json.loads(cronjob(action="run", job_id="job-run-1")) + patch("tools.cronjob_tools._is_gateway_active", return_value=False), \ + patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \ + patch("tools.cronjob_tools.trigger_job") as m_trigger, \ + patch("threading.Thread"), \ + patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)): + json.loads(cronjob(action="run", job_id="job-run-1")) + m_trigger.assert_not_called() - assert out["job"]["executed"] is True - assert out["job"]["execution_success"] is False - assert out["job"]["execution_error"] == "provider 500" - - def test_execute_job_now_bails_without_claim(self): - """_execute_job_now never calls run_one_job when the claim is lost.""" - with patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \ - patch("cron.scheduler.run_one_job") as m_run: - res = _execute_job_now(dict(_JOB)) - assert res["claimed"] is False - assert res["success"] is False - m_run.assert_not_called() + # ── Response contract ─────────────────────────────────────────────── - def test_execute_job_now_marks_failure_on_exception(self): - """An exception during fire is captured, marked failed, not propagated.""" - with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ - patch("cron.scheduler.run_one_job", side_effect=RuntimeError("boom")), \ - patch("tools.cronjob_tools.mark_job_run") as m_mark, \ + def test_run_response_no_execution_success_field(self): + """The response must NOT include execution_success — the job hasn't + finished when the tool returns.""" + with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools._is_gateway_active", return_value=True), \ + patch("tools.cronjob_tools.trigger_job", return_value=dict(_JOB)), \ patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)): - res = _execute_job_now(dict(_JOB)) - assert res["claimed"] is True - assert res["success"] is False - assert "boom" in res["error"] + out = json.loads(cronjob(action="run", job_id="job-run-1")) + assert "execution_success" not in out["job"] + assert "execution_error" not in out["job"] + assert out["job"]["triggered"] is True + + def test_run_response_includes_trigger_mode(self): + """Both paths must report which mode was used.""" + for alive, expected_mode in [(True, "scheduled"), (False, "background")]: + mock_thread = MagicMock() + with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools._is_gateway_active", return_value=alive), \ + patch("tools.cronjob_tools.trigger_job", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ + patch("threading.Thread", return_value=mock_thread), \ + patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)): + out = json.loads(cronjob(action="run", job_id="job-run-1")) + assert out["job"]["trigger_mode"] == expected_mode, \ + f"mode mismatch for alive={alive}" + + # ── Core invariant: NEVER synchronous run_one_job ─────────────────── + + def test_run_never_calls_run_one_job_synchronously(self): + """Regardless of gateway state, run_one_job must NOT be called inline.""" + for alive in (True, False): + mock_thread = MagicMock() + with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools._is_gateway_active", return_value=alive), \ + patch("tools.cronjob_tools.trigger_job", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ + patch("cron.scheduler.run_one_job") as m_run, \ + patch("threading.Thread", return_value=mock_thread), \ + patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)): + cronjob(action="run", job_id="job-run-1") + m_run.assert_not_called(), \ + f"run_one_job called synchronously when alive={alive}" + + # ── Background worker ─────────────────────────────────────────────── + + def test_background_worker_runs_job_and_handles_errors(self): + """The daemon thread worker must call run_one_job and mark failure on + exception.""" + from tools.cronjob_tools import _run_job_in_background + + with patch("cron.scheduler.run_one_job", side_effect=RuntimeError("boom")), \ + patch("tools.cronjob_tools.mark_job_run") as m_mark: + _run_job_in_background(dict(_JOB)) + m_mark.assert_called_once() + assert "boom" in m_mark.call_args[0][2] + + def test_background_worker_succeeds_silently(self): + """The daemon thread worker runs run_one_job without error on success.""" + from tools.cronjob_tools import _run_job_in_background + + with patch("cron.scheduler.run_one_job", return_value=True), \ + patch("tools.cronjob_tools.mark_job_run") as m_mark: + _run_job_in_background(dict(_JOB)) + + # mark_job_run is NOT called on success — run_one_job handles it internally + m_mark.assert_not_called() diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index ef8bdc1b1c45a..bdfcdc0467c16 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -9,6 +9,7 @@ import logging import re import sys +import threading from pathlib import Path from typing import Any, Dict, List, Optional, Union @@ -31,6 +32,7 @@ remove_job, resolve_job_ref, resume_job, + trigger_job, update_job, ) @@ -517,49 +519,76 @@ def _format_job(job: Dict[str, Any]) -> Dict[str, Any]: return result -def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: - """Execute a cron job immediately, outside the scheduler tick. +def _is_gateway_active() -> bool: + """Check if the gateway scheduler ticker is currently running. - Atomically claims the job first via ``claim_job_for_fire`` — the same - at-most-once CAS the scheduler/external-provider fire path uses — so a - concurrently-running gateway ticker cannot also fire it (the claim both - blocks a duplicate fire and advances ``next_run_at`` for recurring jobs). - If the claim is lost (another fire is in flight), this is a no-op. + When the gateway is alive, its ticker fires due jobs asynchronously via + ``ThreadPoolExecutor.submit(run_one_job)`` every ~60 s. When it is not + alive (CLI / oneshot mode), there is no ticker to pick up armed jobs. + """ + try: + from gateway.status import is_gateway_running + return is_gateway_running() + except Exception: + return False - The actual firing is delegated to ``run_one_job`` — the single shared - execute→save→deliver→mark body the ticker and external providers use — so - failure delivery, ``[SILENT]`` handling, and live-adapter delivery stay - identical across paths and can't drift. - Returns {"claimed": bool, "success": bool, "error": str|None}. +def _run_job_in_background(job: Dict[str, Any]): + """Daemon-thread worker: execute a cron job to completion. + + Called from a background thread so the calling agent is never blocked. + ``run_one_job`` handles ``mark_job_run`` (which clears the fire claim and + records ``last_status``) on success. On exception we mark the job failed + so the claim does not wedge and the user sees the error. """ job_id = job["id"] try: from cron.scheduler import run_one_job - - # At-most-once claim: bail without running if a tick/other fire owns it. - if not claim_job_for_fire(job_id): - return {"claimed": False, "success": False, - "error": "Job is already being fired by the scheduler; not run again."} - - # run_one_job records last_run_at/last_status via mark_job_run (which - # also clears the fire claim) and returns True iff it processed the job. - processed = run_one_job(job) - refreshed = get_job(job_id) or {} - ok = refreshed.get("last_status") == "ok" - return { - "claimed": True, - "success": bool(processed and ok), - "error": refreshed.get("last_error"), - } - + run_one_job(job) except Exception as e: - logger.error("Failed to execute cron job %s immediately: %s", job_id, e) + logger.error("Background cron job %s failed: %s", job_id, e) try: mark_job_run(job_id, False, str(e)) except Exception: pass - return {"claimed": True, "success": False, "error": str(e)} + + +def _trigger_job_nonblocking(job: Dict[str, Any], job_id: str) -> Dict[str, Any]: + """Trigger a cron job immediately without blocking the calling agent (#52705). + + Two paths based on gateway liveness: + + 1. **Gateway alive** — arm the job via ``trigger_job`` (sets + ``next_run_at = now``) and let the scheduler ticker fire it + asynchronously on its next cycle. At-most-once is handled by the + ticker's own ``claim_job_for_fire`` on its fire path. + + 2. **Gateway not alive** — claim via ``claim_job_for_fire`` (at-most-once, + advances ``next_run_at`` for recurring jobs) and spawn a daemon thread + that calls ``run_one_job`` directly. The thread runs the job to + completion in the calling process while the tool returns immediately. + + Either way the tool never blocks. ``last_status`` is not available in the + immediate response (the job hasn't finished); check via + ``cronjob(action='list')``. + + Returns ``{"mode": "scheduled"|"background", "claimed": bool, + "error": str|None}``. + """ + if _is_gateway_active(): + # Path 1: fire-and-forget — the ticker will fire it within ~60 s. + trigger_job(job_id) + return {"mode": "scheduled", "claimed": True, "error": None} + + # Path 2: no ticker — claim + daemon thread. + if not claim_job_for_fire(job_id): + return {"mode": "background", "claimed": False, + "error": "Job is already being fired; not run again."} + + threading.Thread( + target=_run_job_in_background, args=(job,), daemon=True + ).start() + return {"mode": "background", "claimed": True, "error": None} def cronjob( @@ -736,22 +765,20 @@ def cronjob( return json.dumps({"success": True, "job": _format_job(updated)}, indent=2) if normalized in {"run", "run_now", "trigger"}: - # Execute the job immediately rather than only scheduling it for the - # next scheduler tick — a manual `run` should actually run, even when - # no gateway/ticker is active (the #41037 case). The claim inside - # _execute_job_now advances next_run_at and blocks a concurrent tick - # from double-firing. - exec_result = _execute_job_now(job) - # Re-read so the response reflects the post-run last_run_at/last_status. + # Trigger the job without blocking the calling agent (#52705). + # When the gateway ticker is alive, the job is armed via + # trigger_job and the scheduler fires it asynchronously. When no + # ticker is running (CLI/oneshot), a daemon thread runs the job + # directly. Either way the tool returns immediately. + trigger_result = _trigger_job_nonblocking(job, job_id) result = _format_job(get_job(job_id) or {"id": job_id}) - result["executed"] = exec_result.get("claimed", False) - result["execution_success"] = exec_result.get("success", False) - if not exec_result.get("claimed", False): - result["execution_skipped"] = ( - "Already being fired by the scheduler; not run again." + result["triggered"] = True + result["trigger_mode"] = trigger_result["mode"] + result["executed"] = trigger_result.get("claimed", False) + if not trigger_result.get("claimed", False): + result["execution_skipped"] = trigger_result.get( + "error", "Already being fired." ) - elif exec_result.get("error"): - result["execution_error"] = exec_result["error"] return json.dumps({"success": True, "job": result}, indent=2) if normalized == "update":