refactor(cron): harden the run heartbeat (review follow-ups)

- 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

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

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,

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 "