refactor(cron): log the heartbeat ceiling stop + test it

- logger.warning when the 6h ceiling stops the heartbeat (matches the
  delegate_task stale-stop precedent) so the eventual watchdog reap is
  explainable from logs instead of silent
- new mutation-checked test: past the ceiling the heartbeat stops while
  the job still completes
- clearer assertion messages (surface res on failure)

tests/tools/test_cronjob_run_immediate.py

@@ -88,7 +88,7 @@ class TestCronjobRunExecutesImmediately:
                 res = _execute_job_now(dict(_JOB))
 
             m_run.assert_called_once()
-            assert res["success"] is True
+            assert res["success"] is True, res
             assert any("cronjob: running job" in t for t in touches), touches
         finally:
             set_activity_callback(None)
@@ -110,6 +110,37 @@ class TestCronjobRunExecutesImmediately:
         finally:
             set_activity_callback(None)
 
+    def test_heartbeat_stops_at_ceiling_but_job_completes(self):
+        """Past _CRON_RUN_HEARTBEAT_CEILING the heartbeat stops (so the
+        gateway watchdog regains authority over a wedged run) while the job
+        itself keeps running to completion."""
+        touches = []
+        first_beat = threading.Event()
+
+        def record(desc):
+            touches.append(desc)
+            first_beat.set()
+
+        set_activity_callback(record)
+        try:
+            def slow_run(job):
+                # Ceiling=0 → the very first wake stops the loop without
+                # touching. Give it a couple of cycles to prove silence.
+                time.sleep(0.2)
+                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("tools.cronjob_tools._CRON_RUN_HEARTBEAT_CEILING", 0.0), \
+                 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, res
+            assert not first_beat.is_set(), touches   # heartbeat never fired
+        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."""
@@ -137,7 +168,7 @@ class TestCronjobRunExecutesImmediately:
                  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 res["success"] is True, res
             assert len(calls) >= 2, calls
         finally:
             set_activity_callback(None)

tools/cronjob_tools.py

@@ -660,6 +660,12 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]:
                     if elapsed > _CRON_RUN_HEARTBEAT_CEILING:
                         # Stop masking the gateway watchdog — a run this long
                         # with an unlimited child watchdog is likely wedged.
+                        logger.warning(
+                            "cronjob run heartbeat ceiling reached for job "
+                            "'%s' (%.0fs) — stopping heartbeat; gateway "
+                            "watchdog regains authority",
+                            job_name, elapsed,
+                        )
                         return
                     try:
                         activity_cb(