From 71f4de3cd83f315388190e36bb5b8553fd9a8dda Mon Sep 17 00:00:00 2001 From: Burke Autrey Date: Fri, 17 Jul 2026 14:27:23 -0500 Subject: [PATCH] fix(gateway): supervise long-lived watcher tasks (task-level) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-applied against v0.18.2. Upstream now wraps each watcher's INNER loop in try/except (kept — not re-added), but the 9 long-lived watchers in start() are still bare asyncio.create_task: no handle, no exception logging, no restart. A raise in _platform_reconnect_watcher's OUTER while-loop / pre-try region still dies silently, permanently losing platform reconnection. Add _spawn_supervised (track + log + bounded-backoff restart; no respawn on clean return to avoid busy-spin) and route all 9 watchers through it. - tests: clean-return spawned exactly once; exception-restart bounded at ceiling+1. Co-Authored-By: Claude Opus 4.8 (1M context) --- gateway/run.py | 80 +++++++++++++++++++++--- tests/gateway/test_platform_reconnect.py | 61 ++++++++++++++++++ 2 files changed, 132 insertions(+), 9 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 6deee096c5aa..1fd57dead008 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -7617,7 +7617,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Process in batches of 100 with event-loop yield points to avoid # O(n^2) event-loop blocking when recovering thousands of watchers. for i, watcher in enumerate(watchers): - asyncio.create_task(self._run_process_watcher(watcher)) + self._spawn_supervised( + lambda w=watcher: self._run_process_watcher(w), + f"process_watcher:{watcher.get('session_id')}", + restart=False, + ) logger.info("Resumed watcher for recovered process %s", watcher.get("session_id")) if i % 100 == 99: await asyncio.sleep(0) @@ -7625,18 +7629,18 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew logger.error("Recovered watcher setup error: %s", e) # Start background session expiry watcher to finalize expired sessions - asyncio.create_task(self._session_expiry_watcher()) + self._spawn_supervised(self._session_expiry_watcher, "session_expiry_watcher") # Start background kanban notifier — delivers `completed`, `blocked`, # `spawn_auto_blocked`, and `crashed` events to gateway subscribers # so human-in-the-loop workflows hear back without polling. - asyncio.create_task(self._kanban_notifier_watcher()) + self._spawn_supervised(self._kanban_notifier_watcher, "kanban_notifier_watcher") # Start background kanban dispatcher — spawns workers for ready # tasks. Gated by `kanban.dispatch_in_gateway` (default True). # When false, users run `hermes kanban daemon` externally or # simply don't use kanban; this loop becomes a no-op. - asyncio.create_task(self._kanban_dispatcher_watcher()) + self._spawn_supervised(self._kanban_dispatcher_watcher, "kanban_dispatcher_watcher") # Start background reconnection watcher for platforms that failed at startup if self._failed_platforms: @@ -7645,19 +7649,19 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew len(self._failed_platforms), ", ".join(p.value for p in self._failed_platforms), ) - asyncio.create_task(self._platform_reconnect_watcher()) + self._spawn_supervised(self._platform_reconnect_watcher, "platform_reconnect_watcher") # Start background handoff watcher — picks up CLI sessions marked # handoff_state='pending' in state.db and re-binds them to the # destination platform's home channel, then forges a synthetic user # turn so the agent kicks off the new chat. - asyncio.create_task(self._handoff_watcher()) + self._spawn_supervised(self._handoff_watcher, "handoff_watcher") # Start background async-delegation watcher — drains completion events # from delegate_task(background=true) subagents and injects each # result back into its originating session as a new turn, covering the # idle case where the subagent finishes with no agent turn running. - asyncio.create_task(self._async_delegation_watcher()) + self._spawn_supervised(self._async_delegation_watcher, "async_delegation_watcher") # Start the scale-to-zero idle watcher ONLY when this instance is opted # in (the NAS "Labs" HERMES_SCALE_TO_ZERO stamp), messaging is @@ -7671,7 +7675,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "scale-to-zero: armed (idle timeout %.0fs) — watching for idle", self._scale_to_zero_idle_timeout_seconds(), ) - asyncio.create_task(self._scale_to_zero_watcher()) + self._spawn_supervised(self._scale_to_zero_watcher, "scale_to_zero_watcher") else: # Surface WHY an OPTED-IN instance didn't arm (a non-opted instance # not arming is normal — stay silent there). Without this, a failed @@ -7686,12 +7690,70 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # left behind by a prior instantiation (durable-volume restart, NS-570) # is ignored via its instantiation epoch; only a current-epoch marker # engages drain on the first tick. - asyncio.create_task(self._drain_control_watcher()) + self._spawn_supervised(self._drain_control_watcher, "drain_control_watcher") logger.info("Press Ctrl+C to stop") return True + _MAX_SUPERVISED_RESTARTS = 5 + + def _spawn_supervised(self, coro_factory, name, *, restart=True, _attempt=0): + """Launch a long-lived background task with task-level supervision. + + Complements upstream's per-iteration inner-loop try/except (which only + guards a single loop-body) by covering what that CANNOT: an exception + raised in the watcher's OUTER ``while self._running:`` loop or its + pre-try setup region, plus task-level death generally. A bare + ``asyncio.create_task`` drops such an exception on the floor — no log, + no restart, the watcher silently gone. This retains the handle in + ``self._background_tasks``, logs any crash, and restarts with capped + exponential backoff up to ``_MAX_SUPERVISED_RESTARTS`` consecutive + failures. + """ + if getattr(self, "_background_tasks", None) is None: + self._background_tasks = set() + + # Deliberately do NOT pass name= to create_task — some test doubles mock + # create_task with a signature that rejects the name kwarg. + task = asyncio.create_task(coro_factory()) + self._background_tasks.add(task) + + def _done(t): + self._background_tasks.discard(t) + if t.cancelled(): + return + exc = t.exception() + if exc is None: + # Clean return == deliberate shutdown or a self-disabling watcher + # (e.g. a gated no-op that returns synchronously). Respawning here + # would busy-spin such a watcher — so NEVER restart on clean exit. + return + logger.error("Supervised task %s died: %r", name, exc, exc_info=exc) + if restart and self._running: + if _attempt >= self._MAX_SUPERVISED_RESTARTS: + logger.error( + "Supervised task %s died %d times consecutively — giving up restarts", + name, + _attempt, + ) + return + backoff = min(60, 2 ** min(_attempt, 6)) + + async def _respawn(): + await asyncio.sleep(backoff) + if self._running: + self._spawn_supervised( + coro_factory, name, restart=restart, _attempt=_attempt + 1 + ) + + respawn_task = asyncio.create_task(_respawn()) + self._background_tasks.add(respawn_task) + respawn_task.add_done_callback(self._background_tasks.discard) + + task.add_done_callback(_done) + return task + async def _handoff_watcher(self, interval: float = 2.0) -> None: """Background task that processes pending CLI→gateway session handoffs. diff --git a/tests/gateway/test_platform_reconnect.py b/tests/gateway/test_platform_reconnect.py index 2f8a0c98943a..46c9a1e4590e 100644 --- a/tests/gateway/test_platform_reconnect.py +++ b/tests/gateway/test_platform_reconnect.py @@ -840,3 +840,64 @@ class TestPlatformSlashCommand: runner = _make_runner() out = await runner._handle_platform_command(self._make_event("/platform")) assert "Gateway platforms" in out + + +# --- Supervised task wrapper (_spawn_supervised) --- + +class TestSpawnSupervised: + """Verify the task-level supervision wrapper around watcher launches.""" + + @pytest.mark.asyncio + async def test_clean_synchronous_return_is_not_respawned(self): + # A supervised coro that returns immediately (clean exit) must be + # invoked EXACTLY ONCE — a clean return means deliberate shutdown or a + # gated no-op watcher; respawning it would busy-spin the event loop. + runner = _make_runner() + calls = {"n": 0} + + async def _coro(): + calls["n"] += 1 + return + + runner._spawn_supervised(lambda: _coro(), "clean_watcher") + + # Drive the loop so the done-callback fires; if it (wrongly) respawned, + # the count would keep climbing across these ticks. + for _ in range(50): + await asyncio.sleep(0) + + assert calls["n"] == 1 + + @pytest.mark.asyncio + async def test_exception_restart_bounded_by_ceiling(self, monkeypatch): + # A coro that always raises is restarted with backoff, but the restart + # chain is capped: initial launch + _MAX_SUPERVISED_RESTARTS respawns. + runner = _make_runner() + calls = {"n": 0} + + # Collapse the backoff sleeps to a single loop-yield so the restart + # chain converges fast. Bind the real sleep BEFORE patching so the + # replacement still yields control (and doesn't recurse into itself). + real_sleep = asyncio.sleep + + async def _instant_sleep(_delay): + await real_sleep(0) + + monkeypatch.setattr("gateway.run.asyncio.sleep", _instant_sleep) + + async def _coro(): + calls["n"] += 1 + raise RuntimeError("boom") + + runner._spawn_supervised(lambda: _coro(), "always_raises") + + expected = runner._MAX_SUPERVISED_RESTARTS + 1 + for _ in range(500): + await real_sleep(0) + if calls["n"] >= expected: + break + # A few extra ticks to prove the chain has stopped (no over-restart). + for _ in range(20): + await real_sleep(0) + + assert calls["n"] == expected