fix(gateway): recover watchdog after transient loop stalls

gateway/systemd_notify.py

@@ -60,6 +60,8 @@ def watchdog_interval_seconds() -> Optional[float]:
 class SystemdWatchdog:
     """Feed systemd while the asyncio event loop continues to make progress."""
 
+    _RECOVERY_TICKS = 2
+
     def __init__(
         self,
         *,
@@ -71,6 +73,7 @@ def __init__(
         self._lag_tolerance_seconds = lag_tolerance_seconds
         self._task: Optional[asyncio.Task[None]] = None
         self._unhealthy = False
+        self._timely_ticks = 0
         self._stopping = False
         self._stopping_notified = False
 
@@ -111,6 +114,7 @@ def start(self) -> bool:
             return False
         self._stopping = False
         self._unhealthy = False
+        self._timely_ticks = 0
         self._stopping_notified = False
         self._task = asyncio.create_task(self._run(), name="hermes-systemd-watchdog")
         return True
@@ -123,17 +127,35 @@ def ready(self, status: str = "Gateway running") -> bool:
         return notify(f"READY=1\nSTATUS={safe_status}")
 
     def record_tick(self, *, scheduled_at: float, now: float) -> bool:
-        """Feed systemd only when the event loop woke within its lag budget."""
-        if not self.enabled or self._stopping or self._unhealthy:
+        """Feed systemd whenever the event loop is making progress again.
+
+        A late wake-up is evidence of starvation, but the fact that this method
+        is running means the loop recovered before systemd killed the process.
+        Keep feeding the external watchdog while reporting a degraded status;
+        otherwise one transient delay permanently guarantees a later restart.
+        """
+        if not self.enabled or self._stopping:
             return False
         try:
             lag = float(now) - float(scheduled_at)
         except (TypeError, ValueError):
             lag = float("inf")
         if not math.isfinite(lag) or lag > self._lag_tolerance():
             self._unhealthy = True
-            notify("STATUS=watchdog unhealthy: event loop progress is late")
-            return False
+            self._timely_ticks = 0
+            notify("WATCHDOG=1\nSTATUS=watchdog degraded: event loop progress was late")
+            return True
+
+        if self._unhealthy:
+            self._timely_ticks += 1
+            if self._timely_ticks >= self._RECOVERY_TICKS:
+                self._unhealthy = False
+                self._timely_ticks = 0
+                notify(
+                    "WATCHDOG=1\nSTATUS=watchdog healthy: event loop progress recovered"
+                )
+                return True
+
         notify("WATCHDOG=1")
         return True
 
@@ -145,7 +167,7 @@ async def _run(self) -> None:
         loop = asyncio.get_running_loop()
         scheduled_at = loop.time() + cadence
         try:
-            while not self._stopping and not self._unhealthy:
+            while not self._stopping:
                 await asyncio.sleep(max(0.0, scheduled_at - loop.time()))
                 now = loop.time()
                 if not self.record_tick(scheduled_at=scheduled_at, now=now):

tests/gateway/test_systemd_notify.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import socket
+import time
 
 import pytest
 
@@ -55,6 +56,70 @@ def send(self, payload):
     assert calls[0] == ("setblocking", False)
 
 
+def test_watchdog_recovers_after_loop_progress_is_late(monkeypatch):
+    calls: list[str] = []
+    monkeypatch.setenv("NOTIFY_SOCKET", "/tmp/hermes-test-notify")
+    monkeypatch.setenv("WATCHDOG_USEC", "1000000")
+
+    import gateway.systemd_notify as notify_mod
+
+    monkeypatch.setattr(
+        notify_mod, "notify", lambda message: calls.append(message) or True
+    )
+    watchdog = notify_mod.SystemdWatchdog(lag_tolerance_seconds=0.1)
+
+    assert watchdog.record_tick(scheduled_at=10.0, now=10.05) is True
+    assert calls == ["WATCHDOG=1"]
+    assert watchdog.record_tick(scheduled_at=10.0, now=10.2) is True
+    assert watchdog.unhealthy is True
+    assert "WATCHDOG=1" in calls[-1]
+    assert "STATUS=watchdog degraded" in calls[-1]
+
+    # A recovered event loop must keep feeding systemd.  Require two timely
+    # samples before clearing the degraded status so one lucky wake-up does not
+    # hide recurring starvation.
+    assert watchdog.record_tick(scheduled_at=10.3, now=10.35) is True
+    assert watchdog.unhealthy is True
+    assert watchdog.record_tick(scheduled_at=10.4, now=10.45) is True
+    assert watchdog.unhealthy is False
+    assert "STATUS=watchdog healthy" in calls[-1]
+
+
+@pytest.mark.asyncio
+async def test_watchdog_task_survives_a_transient_event_loop_stall(monkeypatch):
+    calls: list[str] = []
+    degraded = asyncio.Event()
+    healthy = asyncio.Event()
+    monkeypatch.setenv("NOTIFY_SOCKET", "/tmp/hermes-test-notify")
+    monkeypatch.setenv("WATCHDOG_USEC", "100000")
+
+    import gateway.systemd_notify as notify_mod
+
+    def _capture(message: str) -> bool:
+        calls.append(message)
+        if "STATUS=watchdog degraded" in message:
+            degraded.set()
+        if "STATUS=watchdog healthy" in message:
+            healthy.set()
+        return True
+
+    monkeypatch.setattr(notify_mod, "notify", _capture)
+    watchdog = notify_mod.SystemdWatchdog(lag_tolerance_seconds=0.01)
+
+    assert watchdog.start() is True
+    try:
+        await asyncio.sleep(0)  # Let the watchdog establish its first deadline.
+        time.sleep(0.08)  # Fault injection: delay the loop, but not past WatchdogSec.
+
+        await asyncio.wait_for(degraded.wait(), timeout=2.0)
+        await asyncio.wait_for(healthy.wait(), timeout=2.0)
+        assert watchdog.task is not None
+        assert not watchdog.task.done()
+        assert calls.count("WATCHDOG=1") >= 1
+    finally:
+        await watchdog.stop()
+
+
 @pytest.mark.asyncio
 async def test_watchdog_sends_ready_heartbeat_and_stopping(monkeypatch):
     calls: list[str] = []
@@ -77,5 +142,3 @@ async def test_watchdog_sends_ready_heartbeat_and_stopping(monkeypatch):
     assert "WATCHDOG=1" in calls
     assert calls[-1] == "STOPPING=1"
     assert watchdog.unhealthy is False
-
-

website/docs/user-guide/messaging/index.md

@@ -180,11 +180,14 @@ hermes gateway install --force
 ```
 
 A positive value makes the generated unit use `Type=notify`,
-`NotifyAccess=main`, and the matching `WatchdogSec`. Hermes sends heartbeats
-only while its event loop is making timely progress; systemd restarts the
-process when they stop. The default `0` keeps the existing `Type=simple`
-behavior. This setting is Linux/systemd-only and does not treat an ordinary
-platform network disconnect as an event-loop failure.
+`NotifyAccess=main`, and the matching `WatchdogSec`. Hermes normally sends
+heartbeats while its event loop makes timely progress. A late callback shows
+that the loop has resumed, so Hermes renews the watchdog lease with a degraded
+status and reports healthy again after two consecutive timely samples. A full
+stall prevents the callback from running, so no heartbeat is sent and systemd
+restarts the process when `WatchdogSec` expires. The default `0` keeps the
+existing `Type=simple` behavior. This setting is Linux/systemd-only and does
+not treat an ordinary platform network disconnect as an event-loop failure.
 
 ## Chat Commands (Inside Messaging)