fix(gateway): supervise long-lived watcher tasks (task-level)

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) <noreply@anthropic.com>

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.
 

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