fix(cron): run job without blocking the calling turn (#76502)

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)

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 {