[eric] 9router: event-driven death-watcher heals in ~0.2s + two-strike watchdog (busy-probe false negatives)

backend/apps/nine_router/process.py

@@ -353,9 +353,10 @@ async def ensure_running():
         p_start_lock = asyncio.Lock()
     async with p_start_lock:
         await p_ensure_running_impl()
-    # Arm the watchdog the moment the router becomes a live dependency; users who never route through it never spawn it.
+    # Arm both healers the moment the router becomes a live dependency; users who never route through it never spawn them.
     if is_running():
         start_watchdog()
+        start_death_watcher()
 
 
 def has_persisted_connections() -> bool:
@@ -381,8 +382,10 @@ p_watchdog_task: "asyncio.Task | None" = None
 
 
 async def p_watchdog_loop() -> None:
-    """Revive 9Router whenever it dies mid-session (OOM, crash, orphaned by a hard kill) so a
-    running app never sits on a dead router; only quitting OpenSwarm (stop()) ends it."""
+    """Backstop healer for routers we DIDN'T spawn (adopted port-holders have no handle for the
+    death-watcher). Two-strike confirmation before reviving: the sync is_running probe can
+    false-negative while a busy router streams, and acting on one bad probe would rotate a LIVE
+    router's request log and burn a duplicate spawn attempt."""
     failures = 0
     while True:
         await asyncio.sleep(P_WATCHDOG_BACKOFF_SECONDS if failures >= 3 else P_WATCHDOG_INTERVAL_SECONDS)
@@ -390,7 +393,11 @@ async def p_watchdog_loop() -> None:
             if is_running():
                 failures = 0
                 continue
-            logger.warning("9Router watchdog: router is down; reviving")
+            await asyncio.sleep(2)
+            if is_running():
+                failures = 0
+                continue
+            logger.warning("9Router watchdog: router is down (confirmed twice); reviving")
             await ensure_running()
             if is_running():
                 failures = 0
@@ -404,6 +411,49 @@ async def p_watchdog_loop() -> None:
             logger.exception("9Router watchdog iteration failed")
 
 
+# Instant healer for the process WE spawned: its exit wakes us the moment it happens (no polling,
+# no false positives), so total heal time = just the respawn. Crash-loop guard: 3 deaths inside
+# 60s defers to the backed-off watchdog instead of hot-spinning a broken install.
+p_death_watcher_task: "asyncio.Task | None" = None
+p_recent_death_monos: "list[float]" = []
+
+
+async def p_death_watch(proc_handle: "subprocess.Popen[Any]") -> None:
+    global p_is_running_last_ok
+    loop = asyncio.get_running_loop()
+    try:
+        await loop.run_in_executor(None, proc_handle.wait)
+    except asyncio.CancelledError:
+        raise
+    except Exception:
+        return
+    # stop() nulls p_process before this continuation can run (it blocks the loop through wait), so a deliberate quit or a superseded handle never triggers a revive.
+    if proc_handle is not p_process:
+        return
+    now = time.monotonic()
+    p_recent_death_monos.append(now)
+    del p_recent_death_monos[:-3]
+    if len(p_recent_death_monos) == 3 and now - p_recent_death_monos[0] < 60:
+        logger.warning("9Router died 3x in 60s; leaving revival to the backed-off watchdog")
+        return
+    logger.warning("9Router process died; instant revive")
+    p_is_running_last_ok = 0.0
+    await ensure_running()
+
+
+def start_death_watcher() -> None:
+    """Idempotent per spawned handle; no-op for adopted routers (no handle to wait on)."""
+    global p_death_watcher_task
+    if p_process is None or p_process.poll() is not None:
+        return
+    if p_death_watcher_task is not None and not p_death_watcher_task.done():
+        return
+    try:
+        p_death_watcher_task = asyncio.get_running_loop().create_task(p_death_watch(p_process))
+    except RuntimeError:
+        logger.warning("9Router death-watcher: no running loop; not armed")
+
+
 def start_watchdog() -> None:
     """Idempotent; armed by ensure_running() on success, cancelled by stop()."""
     global p_watchdog_task
@@ -541,11 +591,14 @@ async def p_ensure_running_impl():
 
 def stop():
     """Stop the 9Router subprocess."""
-    global p_process, p_watchdog_task
-    # Cancel the watchdog FIRST or it would revive the router we're about to kill (shutdown = the one sanctioned "down").
+    global p_process, p_watchdog_task, p_death_watcher_task
+    # Cancel the healers FIRST or they would revive the router we're about to kill (shutdown = the one sanctioned "down").
     if p_watchdog_task is not None:
         p_watchdog_task.cancel()
         p_watchdog_task = None
+    if p_death_watcher_task is not None:
+        p_death_watcher_task.cancel()
+        p_death_watcher_task = None
     if p_process:
         try:
             p_process.terminate()

backend/tests/test_router_watchdog.py

@@ -30,16 +30,102 @@ def test_watchdog_revives_then_backs_off():
              patch.object(proc, "ensure_running", fake_ensure), \
              patch.object(proc.asyncio, "sleep", fake_sleep):
             task = asyncio.get_running_loop().create_task(proc.p_watchdog_loop())
-            while len(sleeps) < 6:
+            while len(sleeps) < 9:
                 await real_sleep(0)
             task.cancel()
             try:
                 await task
             except asyncio.CancelledError:
                 pass
-        assert len(ensures) >= 3, "a down router must be revived on every pulse"
+        assert len(ensures) >= 2, "a confirmed-down router must be revived"
         assert sleeps[0] == proc.P_WATCHDOG_INTERVAL_SECONDS
-        assert sleeps[4] == proc.P_WATCHDOG_BACKOFF_SECONDS, "3 straight failures must back off"
+        assert sleeps[1] == 2, "two-strike: a single failed probe must be re-confirmed before reviving"
+        assert proc.P_WATCHDOG_BACKOFF_SECONDS in sleeps, "3 straight failures must back off"
+
+    asyncio.run(run())
+
+
+def test_watchdog_single_false_negative_never_revives():
+    async def run():
+        sleeps: list = []
+        ensures: list = []
+        probes: list = []
+
+        async def fake_sleep(d):
+            sleeps.append(d)
+            await real_sleep(0)
+
+        async def fake_ensure():
+            ensures.append(1)
+
+        def flaky_is_running():
+            # First probe of each pulse fails (busy-router false negative); the confirm succeeds.
+            probes.append(1)
+            return len(probes) % 2 == 0
+
+        real_sleep = asyncio.sleep
+        with patch.object(proc, "is_running", flaky_is_running), \
+             patch.object(proc, "ensure_running", fake_ensure), \
+             patch.object(proc.asyncio, "sleep", fake_sleep):
+            task = asyncio.get_running_loop().create_task(proc.p_watchdog_loop())
+            while len(probes) < 8:
+                await real_sleep(0)
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+        assert not ensures, "a transient probe failure must never trigger a revive"
+
+    asyncio.run(run())
+
+
+def test_death_watcher_revives_instantly_and_guards_loops():
+    async def run():
+        ensures: list = []
+
+        async def fake_ensure():
+            ensures.append(1)
+
+        class FakeProc:
+            def __init__(self):
+                self.dead = False
+            def wait(self):
+                while not self.dead:
+                    pass
+            def poll(self):
+                return 1 if self.dead else None
+
+        fp = FakeProc()
+        proc.p_recent_death_monos.clear()
+        with patch.object(proc, "ensure_running", fake_ensure), \
+             patch.object(proc, "p_process", fp):
+            task = asyncio.get_running_loop().create_task(proc.p_death_watch(fp))
+            await asyncio.sleep(0.05)
+            assert not ensures, "no revive while the process lives"
+            fp.dead = True
+            for _ in range(200):
+                if ensures:
+                    break
+                await asyncio.sleep(0.01)
+            assert ensures, "process death must trigger an instant revive"
+            await task
+        # Crash-loop guard: a 3rd death inside 60s defers to the watchdog.
+        ensures.clear()
+        proc.p_recent_death_monos[:] = [proc.time.monotonic() - 5, proc.time.monotonic() - 3]
+        fp2 = FakeProc(); fp2.dead = True
+        with patch.object(proc, "ensure_running", fake_ensure), \
+             patch.object(proc, "p_process", fp2):
+            await proc.p_death_watch(fp2)
+        assert not ensures, "3 deaths in 60s must defer to the backed-off watchdog"
+        # A superseded/stopped handle never revives.
+        ensures.clear()
+        proc.p_recent_death_monos.clear()
+        fp3 = FakeProc(); fp3.dead = True
+        with patch.object(proc, "ensure_running", fake_ensure), \
+             patch.object(proc, "p_process", None):
+            await proc.p_death_watch(fp3)
+        assert not ensures, "a deliberately stopped router must stay down"
 
     asyncio.run(run())