fix(cron): prevent silent ticker stall (17h no-heartbeat incident)

cron/scheduler.py

@@ -17,6 +17,8 @@
 import shutil
 import subprocess
 import sys
+import threading
+import time
 from contextlib import contextmanager
 
 # fcntl is Unix-only; on Windows use msvcrt for file locking
@@ -44,6 +46,100 @@
 logger = logging.getLogger(__name__)
 
 
+# ---------------------------------------------------------------------------
+# Ticker liveness heartbeat + self-healing support.
+#
+# The gateway drives cron from a single background thread (see
+# gateway.run._start_cron_ticker). Historically that loop called tick()
+# inline with no liveness signal and no watchdog: if a tick blocked forever
+# (a hung delivery send, a stuck job, a pool that never drained) the thread
+# wedged and cron silently stopped firing while the gateway itself stayed up.
+# One incident ran ~17h with no jobs firing and no error surfaced.
+#
+# record_ticker_heartbeat() is called at the TOP of every loop iteration,
+# BEFORE any job work, so the heartbeat measures the loop's own liveness —
+# not whether individual jobs succeed. A supervisor (gateway side) watches
+# the heartbeat age and restarts the ticker thread if it goes stale.
+# ---------------------------------------------------------------------------
+_ticker_heartbeat_lock = threading.Lock()
+_ticker_heartbeat = {"monotonic": None, "wall": None, "count": 0}
+
+
+def record_ticker_heartbeat() -> None:
+    """Record that the cron ticker loop is alive (called at the top of each tick).
+
+    Uses a monotonic clock for age math (immune to wall-clock jumps) and also
+    stashes a human-readable wall timestamp for logs/diagnostics. Best-effort
+    file mirror lets operators inspect ticker liveness without attaching to the
+    process.
+    """
+    with _ticker_heartbeat_lock:
+        _ticker_heartbeat["monotonic"] = time.monotonic()
+        _ticker_heartbeat["wall"] = _hermes_now().isoformat()
+        _ticker_heartbeat["count"] += 1
+        count = _ticker_heartbeat["count"]
+        wall = _ticker_heartbeat["wall"]
+
+    # Best-effort persisted mirror — never let heartbeat bookkeeping raise into
+    # the ticker loop.
+    try:
+        lock_dir, _ = _get_lock_paths()
+        lock_dir.mkdir(parents=True, exist_ok=True)
+        (lock_dir / ".ticker_heartbeat").write_text(
+            json.dumps({"wall": wall, "count": count}), encoding="utf-8"
+        )
+    except Exception:
+        pass
+
+
+def get_ticker_heartbeat_age() -> Optional[float]:
+    """Return seconds since the last recorded ticker heartbeat, or None if never set."""
+    with _ticker_heartbeat_lock:
+        mono = _ticker_heartbeat["monotonic"]
+    if mono is None:
+        return None
+    return time.monotonic() - mono
+
+
+def reset_ticker_heartbeat() -> None:
+    """Clear the heartbeat state (used by tests to start from a known state)."""
+    with _ticker_heartbeat_lock:
+        _ticker_heartbeat["monotonic"] = None
+        _ticker_heartbeat["wall"] = None
+        _ticker_heartbeat["count"] = 0
+
+
+def ticker_heartbeat_is_stale(interval: float, multiplier: float = 5.0) -> bool:
+    """Whether the ticker heartbeat is older than ``multiplier`` tick intervals.
+
+    Returns False when no heartbeat has been recorded yet (the ticker may not
+    have started) so a supervisor doesn't thrash before the first tick.
+    """
+    age = get_ticker_heartbeat_age()
+    if age is None:
+        return False
+    return age > max(interval, 0.0) * multiplier
+
+
+def run_cron_tick_once(verbose: bool = False, adapters=None, loop=None) -> int:
+    """Run a single cron tick with heartbeat + exception isolation.
+
+    This is the loop body the gateway ticker thread should call each interval.
+    The heartbeat is recorded FIRST, before any job work, so a hung or failing
+    job cannot make the loop look dead. Any exception escaping ``tick()`` is
+    logged and swallowed so a single bad tick can never kill the ticker thread.
+    """
+    record_ticker_heartbeat()
+    try:
+        return tick(verbose=verbose, adapters=adapters, loop=loop)
+    except Exception:
+        # Escaped exceptions used to be logged at DEBUG and effectively lost.
+        # Log with traceback at ERROR so a recurring tick failure is visible,
+        # then continue — the next tick still fires.
+        logger.exception("Cron tick raised; loop continues to next interval")
+        return 0
+
+
 class CronPromptInjectionBlocked(Exception):
     """Raised by _build_job_prompt when the fully-assembled prompt trips the
     injection scanner. Caught in run_job so the operator sees a clean
@@ -511,6 +607,12 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
 _VIDEO_EXTS = frozenset({'.mp4', '.mov', '.avi', '.mkv', '.webm', '.3gp'})
 _IMAGE_EXTS = frozenset({'.jpg', '.jpeg', '.png', '.webp', '.gif'})
 
+# Upper bound (seconds) on the standalone delivery send. This send runs inline
+# on the ticker's tick path, so an unbounded platform HTTP call could wedge the
+# whole cron ticker — a delivery that never returns must fail as an error, not
+# hang forever.
+_DELIVERY_TIMEOUT = 60
+
 
 def _send_media_via_adapter(
     adapter,
@@ -731,8 +833,20 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 )
 
         if not delivered:
-            # Standalone path: run the async send in a fresh event loop (safe from any thread)
-            coro = _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)
+            # Standalone path: run the async send in a fresh event loop (safe from any thread).
+            # Bound the send with a timeout — this runs inline on the ticker's tick
+            # path, so a platform whose HTTP send never returns would otherwise wedge
+            # the whole cron ticker (the 17h silent-stall failure mode).
+            def _bounded_send_coro():
+                return asyncio.wait_for(
+                    _send_to_platform(
+                        platform, pconfig, chat_id, cleaned_delivery_content,
+                        thread_id=thread_id, media_files=media_files,
+                    ),
+                    timeout=_DELIVERY_TIMEOUT,
+                )
+
+            coro = _bounded_send_coro()
             try:
                 result = asyncio.run(coro)
             except RuntimeError:
@@ -742,8 +856,8 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 # fresh thread that has no running loop.
                 coro.close()
                 with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
-                    future = pool.submit(asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files))
-                    result = future.result(timeout=30)
+                    future = pool.submit(lambda: asyncio.run(_bounded_send_coro()))
+                    result = future.result(timeout=_DELIVERY_TIMEOUT + 5)
             except Exception as e:
                 msg = f"delivery to {platform_name}:{chat_id} failed: {e}"
                 logger.error("Job '%s': %s", job["id"], msg)

gateway/run.py

@@ -17741,7 +17741,7 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in
     image/audio/document cache + expired ``hermes debug share`` pastes
     once per hour.
     """
-    from cron.scheduler import tick as cron_tick
+    from cron.scheduler import run_cron_tick_once
     from gateway.platforms.base import cleanup_image_cache, cleanup_document_cache
     from hermes_cli.debug import _sweep_expired_pastes
 
@@ -17753,10 +17753,10 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in
     logger.info("Cron ticker started (interval=%ds)", interval)
     tick_count = 0
     while not stop_event.is_set():
-        try:
-            cron_tick(verbose=False, adapters=adapters, loop=loop)
-        except Exception as e:
-            logger.debug("Cron tick error: %s", e)
+        # run_cron_tick_once records the liveness heartbeat at the top of the
+        # tick (before any job work) and swallows/logs any exception, so a
+        # single failing tick can never kill this loop or make it look dead.
+        run_cron_tick_once(verbose=False, adapters=adapters, loop=loop)
 
         tick_count += 1
 
@@ -17822,6 +17822,51 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in
     logger.info("Cron ticker stopped")
 
 
+def _cron_ticker_supervisor(
+    stop_event: threading.Event,
+    restart_ticker,
+    interval: int = 60,
+    stale_multiplier: float = 5.0,
+    check_interval: int = 60,
+):
+    """Watchdog that restarts the cron ticker if its heartbeat goes stale.
+
+    The ticker records a liveness heartbeat at the top of every tick
+    (cron.scheduler.record_ticker_heartbeat). If that heartbeat is older than
+    ``stale_multiplier`` tick intervals, the ticker thread has almost certainly
+    wedged (a hung delivery, a stuck job, a pool that never drained). We log a
+    warning and call ``restart_ticker`` to spin up a fresh ticker thread; the
+    file lock inside tick() prevents the old and new threads from overlapping.
+
+    This closes the gap behind the 17h silent-stall incident: previously a hung
+    ticker was never detected and cron simply stopped firing while the gateway
+    stayed up.
+    """
+    from cron.scheduler import get_ticker_heartbeat_age, ticker_heartbeat_is_stale
+
+    logger.info(
+        "Cron ticker supervisor started (stale after %.0fs)",
+        interval * stale_multiplier,
+    )
+    while not stop_event.is_set():
+        stop_event.wait(timeout=check_interval)
+        if stop_event.is_set():
+            break
+        try:
+            if ticker_heartbeat_is_stale(interval, stale_multiplier):
+                age = get_ticker_heartbeat_age()
+                logger.warning(
+                    "Cron ticker heartbeat stale (%.0fs old, > %.0fs threshold) — "
+                    "restarting ticker thread",
+                    age if age is not None else -1.0,
+                    interval * stale_multiplier,
+                )
+                restart_ticker()
+        except Exception as e:
+            logger.debug("Cron ticker supervisor error: %s", e)
+    logger.info("Cron ticker supervisor stopped")
+
+
 async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = False, verbosity: Optional[int] = 0) -> bool:
     """
     Start the gateway and run until interrupted.
@@ -18173,26 +18218,48 @@ def restart_signal_handler():
     # Start background cron ticker so scheduled jobs fire automatically.
     # Pass the event loop so cron delivery can use live adapters (E2EE support).
     cron_stop = threading.Event()
-    cron_thread = threading.Thread(
-        target=_start_cron_ticker,
-        args=(cron_stop,),
-        kwargs={"adapters": runner.adapters, "loop": asyncio.get_running_loop()},
+    _cron_loop = asyncio.get_running_loop()
+    # Mutable holder so the supervisor can swap in a fresh ticker thread if the
+    # current one wedges. daemon=True: a hung old thread is abandoned, not joined.
+    _cron_ticker_holder = {"thread": None}
+
+    def _spawn_cron_ticker() -> None:
+        t = threading.Thread(
+            target=_start_cron_ticker,
+            args=(cron_stop,),
+            kwargs={"adapters": runner.adapters, "loop": _cron_loop},
+            daemon=True,
+            name="cron-ticker",
+        )
+        _cron_ticker_holder["thread"] = t
+        t.start()
+
+    _spawn_cron_ticker()
+
+    # Self-healing watchdog: restarts the ticker if its heartbeat goes stale
+    # (the 17h silent-stall failure mode). Runs in its own daemon thread.
+    cron_supervisor_thread = threading.Thread(
+        target=_cron_ticker_supervisor,
+        args=(cron_stop, _spawn_cron_ticker),
         daemon=True,
-        name="cron-ticker",
+        name="cron-ticker-supervisor",
     )
-    cron_thread.start()
-    
+    cron_supervisor_thread.start()
+
     # Wait for shutdown
     await runner.wait_for_shutdown()
 
     if runner.should_exit_with_failure:
         if runner.exit_reason:
             logger.error("Gateway exiting with failure: %s", runner.exit_reason)
         return False
-    
-    # Stop cron ticker cleanly
+
+    # Stop cron ticker + supervisor cleanly
     cron_stop.set()
-    cron_thread.join(timeout=5)
+    _current_ticker = _cron_ticker_holder.get("thread")
+    if _current_ticker is not None:
+        _current_ticker.join(timeout=5)
+    cron_supervisor_thread.join(timeout=5)
 
     # Close MCP server connections
     try: