From 2314abcbb0effe03cddf8df6098e66513f756ea1 Mon Sep 17 00:00:00 2001 From: webtecnica Date: Sat, 1 Aug 2026 23:22:27 -0300 Subject: [PATCH] fix(cron): run job without blocking the calling turn (#76502) --- tests/tools/test_cronjob_run_immediate.py | 49 ++++++++++++++++++ tools/cronjob_tools.py | 61 ++++++++++++++++++++++- 2 files changed, 109 insertions(+), 1 deletion(-) diff --git a/tests/tools/test_cronjob_run_immediate.py b/tests/tools/test_cronjob_run_immediate.py index 0311c42daf741..a7ad1e07628e5 100644 --- a/tests/tools/test_cronjob_run_immediate.py +++ b/tests/tools/test_cronjob_run_immediate.py @@ -5,11 +5,18 @@ 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. + +#76502: the inline fire is synchronous, so while it runs it fires a heartbeat +into the calling agent's activity tracker — otherwise the gateway inactivity +watchdog kills the parent turn at ~1800s. """ import json +import threading +import time from unittest.mock import patch from tools.cronjob_tools import cronjob, _execute_job_now +from tools.environments.base import set_activity_callback _JOB = {"id": "job-run-1", "name": "manual run", "prompt": "hi", @@ -53,3 +60,45 @@ class TestCronjobRunExecutesImmediately: assert res["success"] is False assert "boom" in res["error"] m_mark.assert_called_once() + + def test_execute_job_now_heartbeats_while_job_runs(self): + """A manual run ticks the caller's activity tracker while the job + executes so the gateway inactivity watchdog doesn't kill the parent + turn (#76502).""" + touches = [] + set_activity_callback(lambda desc: touches.append(desc)) + try: + started = threading.Event() + + def slow_run(job): + started.set() + time.sleep(0.15) + return True + + with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ + patch("tools.cronjob_tools._CRON_RUN_HEARTBEAT_INTERVAL", 0.05), \ + patch("cron.scheduler.run_one_job", side_effect=slow_run) as m_run, \ + patch("tools.cronjob_tools.get_job", + return_value={"last_status": "ok", "last_error": None}): + res = _execute_job_now(dict(_JOB)) + + m_run.assert_called_once() + assert res["success"] is True + assert any("cronjob: running job" in t for t in touches), touches + finally: + set_activity_callback(None) + + def test_execute_job_now_without_callback_does_not_heartbeat(self): + """No activity callback registered (direct callers, tests) → the + heartbeat thread is never started and behavior is unchanged.""" + set_activity_callback(None) + try: + with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ + patch("cron.scheduler.run_one_job", return_value=True) as m_run, \ + patch("tools.cronjob_tools.get_job", + return_value={"last_status": "ok", "last_error": None}): + res = _execute_job_now(dict(_JOB)) + assert res["success"] is True + m_run.assert_called_once() + finally: + set_activity_callback(None) diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 1eb0bd58d0041..dd4ca1b0243f2 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -9,6 +9,8 @@ import json import logging import re import sys +import threading +import time from pathlib import Path from typing import Any, Dict, List, Optional, Union @@ -16,6 +18,13 @@ from hermes_constants import display_hermes_home logger = logging.getLogger(__name__) +# Cadence for the heartbeat that keeps the calling agent's inactivity watchdog +# at bay while a manual `cronjob(action="run")` executes the job synchronously +# in-process (#76502). Mirrors the 10s cadence used by +# tools/environments/base.py::touch_activity_if_due and delegate_task's +# heartbeat — comfortably below the 1800s default HERMES_AGENT_TIMEOUT. +_CRON_RUN_HEARTBEAT_INTERVAL = 10.0 + # Import from cron module (will be available when properly installed) sys.path.insert(0, str(Path(__file__).parent.parent)) @@ -608,7 +617,57 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: # 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) + # + # A manual `run` executes the job synchronously on the caller's thread, + # and a cron job is itself a full agent run that routinely takes + # minutes. The calling turn emits no tool activity for that entire + # window, so the gateway inactivity watchdog concludes the agent is + # hung and kills the parent turn (#76502). Fire a heartbeat into the + # caller's activity tracker (the same signal tool progress uses) while + # the job runs, so the watchdog sees a working tool instead of a + # silent one — mirrors the delegate_task heartbeat pattern. Best-effort: + # if no activity callback is registered (direct Python callers, tests), + # behavior is unchanged. + try: + from tools.environments.base import _get_activity_callback + + # Capture on THIS thread: the callback is thread-local (installed + # by the tool executor as the calling agent's _touch_activity), so + # a freshly spawned thread cannot read it back. + activity_cb = _get_activity_callback() + except Exception: + activity_cb = None + + _heartbeat_stop = threading.Event() + _heartbeat_thread = None + + if activity_cb is not None: + job_name = str(job.get("name") or job_id) + + def _heartbeat_loop() -> None: + started = time.monotonic() + while not _heartbeat_stop.wait(_CRON_RUN_HEARTBEAT_INTERVAL): + try: + elapsed = int(time.monotonic() - started) + activity_cb( + f"cronjob: running job '{job_name}' ({elapsed}s elapsed)" + ) + except Exception: + return # never break the job run + + _heartbeat_thread = threading.Thread( + target=_heartbeat_loop, + daemon=True, + name="cronjob-run-heartbeat", + ) + _heartbeat_thread.start() + + try: + processed = run_one_job(job) + finally: + _heartbeat_stop.set() + if _heartbeat_thread is not None: + _heartbeat_thread.join(timeout=_CRON_RUN_HEARTBEAT_INTERVAL + 1) refreshed = get_job(job_id) or {} ok = refreshed.get("last_status") == "ok" return {