From 8fd1a68106ccbed7f8d9dc85b1f5b2e69484e548 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sun, 2 Aug 2026 13:41:54 +0530 Subject: [PATCH] refactor(cron): harden the run heartbeat (review follow-ups) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - heartbeat loop continues past a raising activity callback instead of silently stopping (matches delegate_task / touch_activity_if_due swallow-and-continue semantics) — one transient error must not drop watchdog protection for the rest of a long job - hard 6h elapsed ceiling so a wedged job under HERMES_CRON_TIMEOUT=0 (unlimited child watchdog) cannot mask the gateway watchdog forever - public get_activity_callback() accessor in tools/environments/base.py instead of importing the private _get_activity_callback cross-module - tests: deterministic heartbeat test (event-gated, no timing sleep), no-callback test now asserts the thread is truly never created, new exception-survival guard; dead started event removed - fix comment: delegate_task heartbeat cadence is 30s, not 10s --- tests/tools/test_cronjob_run_immediate.py | 51 ++++++++++++++++++++--- tools/cronjob_tools.py | 32 ++++++++++---- tools/environments/base.py | 15 +++++-- 3 files changed, 80 insertions(+), 18 deletions(-) diff --git a/tests/tools/test_cronjob_run_immediate.py b/tests/tools/test_cronjob_run_immediate.py index a7ad1e07628e5..ad22c1eb07565 100644 --- a/tests/tools/test_cronjob_run_immediate.py +++ b/tests/tools/test_cronjob_run_immediate.py @@ -66,13 +66,18 @@ class TestCronjobRunExecutesImmediately: 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() + heartbeat_seen = threading.Event() + def record(desc): + touches.append(desc) + heartbeat_seen.set() + + set_activity_callback(record) + try: def slow_run(job): - started.set() - time.sleep(0.15) + # Deterministic: block until at least one heartbeat has fired + # (bounded so a broken heartbeat can't hang the test). + assert heartbeat_seen.wait(timeout=5.0), "no heartbeat within 5s" return True with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ @@ -96,9 +101,43 @@ class TestCronjobRunExecutesImmediately: 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}): + return_value={"last_status": "ok", "last_error": None}), \ + patch("tools.cronjob_tools.threading.Thread") as m_thread: res = _execute_job_now(dict(_JOB)) assert res["success"] is True m_run.assert_called_once() + m_thread.assert_not_called() # heartbeat thread truly never created + finally: + set_activity_callback(None) + + def test_heartbeat_survives_callback_exception(self): + """One raising callback must not silently kill watchdog protection + for the rest of a long job — the loop continues heartbeating.""" + calls = [] + second_beat = threading.Event() + + def flaky(desc): + calls.append(desc) + if len(calls) >= 2: + second_beat.set() + if len(calls) == 1: + raise RuntimeError("transient") + + set_activity_callback(flaky) + try: + def slow_run(job): + # Block until a heartbeat AFTER the raising one has fired. + assert second_beat.wait(timeout=5.0), \ + "heartbeat stopped after one callback exception" + 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), \ + 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 + assert len(calls) >= 2, calls finally: set_activity_callback(None) diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index dd4ca1b0243f2..bffe0cf233927 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -20,11 +20,20 @@ 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. +# in-process (#76502). Mirrors the 10s cadence of +# tools/environments/base.py::touch_activity_if_due (delegate_task's heartbeat +# uses 30s) — comfortably below the 1800s default HERMES_AGENT_TIMEOUT. _CRON_RUN_HEARTBEAT_INTERVAL = 10.0 +# Hard ceiling on how long the heartbeat keeps the parent watchdog at bay. +# The child cron run has its own inactivity watchdog (HERMES_CRON_TIMEOUT, +# default 600s) that bounds a wedged job, but with HERMES_CRON_TIMEOUT=0 +# (explicit "unlimited") a truly hung run_one_job would otherwise mask the +# gateway watchdog forever — pre-#76502 the parent was at least reaped at +# ~1800s. After this ceiling the heartbeat stops and the gateway watchdog +# regains authority over the turn. +_CRON_RUN_HEARTBEAT_CEILING = 6 * 3600.0 + # Import from cron module (will be available when properly installed) sys.path.insert(0, str(Path(__file__).parent.parent)) @@ -629,12 +638,12 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: # if no activity callback is registered (direct Python callers, tests), # behavior is unchanged. try: - from tools.environments.base import _get_activity_callback + 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() + activity_cb = get_activity_callback() except Exception: activity_cb = None @@ -647,13 +656,20 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: def _heartbeat_loop() -> None: started = time.monotonic() while not _heartbeat_stop.wait(_CRON_RUN_HEARTBEAT_INTERVAL): + elapsed = time.monotonic() - started + if elapsed > _CRON_RUN_HEARTBEAT_CEILING: + # Stop masking the gateway watchdog — a run this long + # with an unlimited child watchdog is likely wedged. + return try: - elapsed = int(time.monotonic() - started) activity_cb( - f"cronjob: running job '{job_name}' ({elapsed}s elapsed)" + f"cronjob: running job '{job_name}' ({int(elapsed)}s elapsed)" ) except Exception: - return # never break the job run + # Never break the job run; keep heartbeating — one + # transient callback error must not silently drop + # watchdog protection for the rest of a long job. + continue _heartbeat_thread = threading.Thread( target=_heartbeat_loop, diff --git a/tools/environments/base.py b/tools/environments/base.py index c541e52509195..1e5c60b39de4e 100644 --- a/tools/environments/base.py +++ b/tools/environments/base.py @@ -150,7 +150,14 @@ def set_activity_callback(cb: Callable[[str], None] | None) -> None: _activity_callback_local.callback = cb -def _get_activity_callback() -> Callable[[str], None] | None: +def get_activity_callback() -> Callable[[str], None] | None: + """Return the thread-local activity callback (see ``set_activity_callback``). + + Public accessor for callers outside this module that need to capture the + calling thread's callback before handing work to another thread (the + callback is thread-local, so a freshly spawned thread cannot read it + back) — e.g. the manual cron-run heartbeat (#76502). + """ return getattr(_activity_callback_local, "callback", None) @@ -172,7 +179,7 @@ def touch_activity_if_due( return state["last_touch"] = now try: - cb = _get_activity_callback() + cb = get_activity_callback() if cb: elapsed = int(now - state["start"]) cb(f"{label} ({elapsed}s elapsed)") @@ -997,7 +1004,7 @@ class BaseEnvironment(ABC): _iter_count = 0 _last_heartbeat = _now _last_interrupt_state = False - _cb_was_none = _get_activity_callback() is None + _cb_was_none = get_activity_callback() is None if _DEBUG_INTERRUPT: logger.info( "[interrupt-debug] _wait_for_process ENTER tid=%s pid=%s " @@ -1047,7 +1054,7 @@ class BaseEnvironment(ABC): # the activity-callback state (thread-local, can get clobbered # by nested tool calls or executor thread reuse). if _DEBUG_INTERRUPT and time.monotonic() - _last_heartbeat >= 30.0: - _cb_now_none = _get_activity_callback() is None + _cb_now_none = get_activity_callback() is None logger.info( "[interrupt-debug] _wait_for_process HEARTBEAT " "tid=%s pid=%s iter=%d elapsed=%.0fs "