From b138fcc05408abf0a07f01548d7d689d5dad0dca Mon Sep 17 00:00:00 2001 From: ciregenz Date: Fri, 3 Jul 2026 15:10:22 -0700 Subject: [PATCH] [eric] 9router: watchdog + detection-revive so the router self-heals; down only when the user quits --- .../agents/manager/configure_provider_env.py | 56 +++++--- backend/apps/nine_router/process.py | 66 +++++++++- backend/tests/test_router_watchdog.py | 122 ++++++++++++++++++ 3 files changed, 224 insertions(+), 20 deletions(-) create mode 100644 backend/tests/test_router_watchdog.py diff --git a/backend/apps/agents/manager/configure_provider_env.py b/backend/apps/agents/manager/configure_provider_env.py index ee5e3cec..1cd13b0f 100644 --- a/backend/apps/agents/manager/configure_provider_env.py +++ b/backend/apps/agents/manager/configure_provider_env.py @@ -15,6 +15,34 @@ from backend.auth import get_auth_token logger = __import__("logging").getLogger(__name__) +@typechecked +async def p_router_available(global_settings: AppSettings) -> bool: + """True when 9Router is up, reviving it first if it died. A dead router must never masquerade + as "no provider configured": detection now shares the dispatch path's lazy-start, so a crashed + or orphaned router self-heals on the very next send instead of erroring the turn. Revival is + gated on EVIDENCE of a provider (a settings key, proxy mode, or an active connection in the + router's on-disk db) so a zero-config user keeps the clean no-provider message instead of us + booting a router with nothing to route.""" + from backend.apps.nine_router import ensure_running as p_ensure, is_running as p_running + from backend.apps.nine_router.process import has_persisted_connections + if p_running(): + return True + p_evidence = any([ + getattr(global_settings, "anthropic_api_key", None), + getattr(global_settings, "openai_api_key", None), + getattr(global_settings, "google_api_key", None), + getattr(global_settings, "openrouter_api_key", None), + getattr(global_settings, "connection_mode", "own_key") in ("openswarm-pro", "free-trial"), + bool(getattr(global_settings, "custom_providers", None) or []), + has_persisted_connections(), + ]) + if not p_evidence: + return False + logger.info("[MCP-DEBUG] 9Router down at provider detection; reviving before concluding") + await p_ensure() + return p_running() + + @typechecked async def configure_provider_env( options_kwargs: Dict, @@ -153,7 +181,7 @@ async def configure_provider_env( elif api_type == "anthropic" and not resolved_is_9router and global_settings.anthropic_api_key: options_kwargs["env"] = {"ANTHROPIC_API_KEY": global_settings.anthropic_api_key} logger.info("[MCP-DEBUG] Using direct Anthropic API key") - elif nine_router_running(): + elif await p_router_available(global_settings): # Gemini-bound ids go through the local proxy for schema scrubbing; everything else hits 9Router directly. is_gemini_bound = ( isinstance(resolved_model, str) @@ -203,21 +231,11 @@ async def configure_provider_env( options_kwargs["env"] = env logger.info(f"[MCP-DEBUG] Using 9Router (api_type={api_type})") else: - if api_type != "anthropic": - from backend.apps.nine_router import ensure_running as nine_router_ensure - logger.info(f"[MCP-DEBUG] 9Router not running for non-Anthropic model {session.model}; waiting for startup") - await nine_router_ensure() - if nine_router_running(): - options_kwargs["env"] = { - "ANTHROPIC_API_KEY": "9router", - "ANTHROPIC_BASE_URL": "http://localhost:20128", - } - logger.info(f"[MCP-DEBUG] 9Router started; routing {session.model} via 9Router") - else: - raise ValueError( - f"9Router is not running; cannot use {session.model}. " - "Install Node.js and restart the app, or switch to a model " - "with a direct API key." - ) - else: - raise ValueError("No AI provider configured. Set an API key or connect a subscription.") + # p_router_available() above already attempted a revival; reaching here means it truly can't start. + if api_type != "anthropic" or resolved_is_9router: + raise ValueError( + f"9Router is not running; cannot use {session.model}. " + "Install Node.js and restart the app, or switch to a model " + "with a direct API key." + ) + raise ValueError("No AI provider configured. Set an API key or connect a subscription.") diff --git a/backend/apps/nine_router/process.py b/backend/apps/nine_router/process.py index 7b94d2c8..67c330c2 100644 --- a/backend/apps/nine_router/process.py +++ b/backend/apps/nine_router/process.py @@ -353,6 +353,66 @@ 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. + if is_running(): + start_watchdog() + + +def has_persisted_connections() -> bool: + """True when 9Router's on-disk db shows an active provider connection. Readable while the + router is DOWN, so revival logic can tell a sub-only user (revive!) from a zero-config one + (don't boot a router that has nothing to route). Fail-closed on any read problem.""" + try: + import json as p_json + with open(os.path.join(p_nine_router_data_dir(), "db.json"), encoding="utf-8") as f: + db = p_json.load(f) + return any( + isinstance(c, dict) and c.get("isActive") + for c in (db.get("providerConnections") or []) + ) + except Exception: + return False + + +# 20s pulse while healthy; after 3 straight failed revives (no node, broken install) back way off so a dead-end setup logs once per 5min instead of crash-looping. +P_WATCHDOG_INTERVAL_SECONDS = 20.0 +P_WATCHDOG_BACKOFF_SECONDS = 300.0 +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.""" + failures = 0 + while True: + await asyncio.sleep(P_WATCHDOG_BACKOFF_SECONDS if failures >= 3 else P_WATCHDOG_INTERVAL_SECONDS) + try: + if is_running(): + failures = 0 + continue + logger.warning("9Router watchdog: router is down; reviving") + await ensure_running() + if is_running(): + failures = 0 + logger.info("9Router watchdog: revived") + else: + failures += 1 + except asyncio.CancelledError: + raise + except Exception: + failures += 1 + logger.exception("9Router watchdog iteration failed") + + +def start_watchdog() -> None: + """Idempotent; armed by ensure_running() on success, cancelled by stop().""" + global p_watchdog_task + if p_watchdog_task is not None and not p_watchdog_task.done(): + return + try: + p_watchdog_task = asyncio.get_running_loop().create_task(p_watchdog_loop()) + except RuntimeError: + logger.warning("9Router watchdog: no running loop; not armed") async def p_ensure_running_impl(): @@ -481,7 +541,11 @@ async def p_ensure_running_impl(): def stop(): """Stop the 9Router subprocess.""" - global p_process + 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"). + if p_watchdog_task is not None: + p_watchdog_task.cancel() + p_watchdog_task = None if p_process: try: p_process.terminate() diff --git a/backend/tests/test_router_watchdog.py b/backend/tests/test_router_watchdog.py new file mode 100644 index 00000000..23bae15b --- /dev/null +++ b/backend/tests/test_router_watchdog.py @@ -0,0 +1,122 @@ +"""9Router resilience: the watchdog revives a dead router (backing off on repeated failure and +dying with stop()), and provider DETECTION revives before concluding "no provider" — gated on +evidence so a zero-config user never boots a router with nothing to route.""" + +import asyncio +import json +import os +from unittest.mock import patch + +import pytest + +import backend.apps.nine_router.process as proc +from backend.apps.settings.models import AppSettings + + +def test_watchdog_revives_then_backs_off(): + async def run(): + sleeps: list = [] + ensures: list = [] + + async def fake_sleep(d): + sleeps.append(d) + await real_sleep(0) + + async def fake_ensure(): + ensures.append(1) + + real_sleep = asyncio.sleep + with patch.object(proc, "is_running", return_value=False), \ + 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: + 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 sleeps[0] == proc.P_WATCHDOG_INTERVAL_SECONDS + assert sleeps[4] == proc.P_WATCHDOG_BACKOFF_SECONDS, "3 straight failures must back off" + + asyncio.run(run()) + + +def test_watchdog_healthy_router_never_spawns(): + async def run(): + sleeps: list = [] + ensures: list = [] + + async def fake_sleep(d): + sleeps.append(d) + await real_sleep(0) + + async def fake_ensure(): + ensures.append(1) + + real_sleep = asyncio.sleep + with patch.object(proc, "is_running", return_value=True), \ + 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) < 4: + await real_sleep(0) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + assert not ensures + assert all(d == proc.P_WATCHDOG_INTERVAL_SECONDS for d in sleeps) + + asyncio.run(run()) + + +def test_stop_cancels_watchdog(): + async def run(): + async def forever(): + while True: + await asyncio.sleep(3600) + + proc.p_watchdog_task = asyncio.get_running_loop().create_task(forever()) + proc.stop() + assert proc.p_watchdog_task is None + + asyncio.run(run()) + + +def test_has_persisted_connections(tmp_path, monkeypatch): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + assert proc.has_persisted_connections() is False # no db at all + (tmp_path / "db.json").write_text(json.dumps({"providerConnections": [{"provider": "claude", "isActive": False}]})) + assert proc.has_persisted_connections() is False # inactive only + (tmp_path / "db.json").write_text(json.dumps({"providerConnections": [{"provider": "claude", "isActive": True}]})) + assert proc.has_persisted_connections() is True + (tmp_path / "db.json").write_text("{corrupt") + assert proc.has_persisted_connections() is False # fail-closed + + +def test_detection_revival_gated_on_evidence(): + from backend.apps.agents.manager import configure_provider_env as cpe + + async def run(): + ensures: list = [] + + async def fake_ensure(): + ensures.append(1) + + import backend.apps.nine_router as nr_pkg + with patch.object(nr_pkg, "is_running", return_value=False), \ + patch.object(nr_pkg, "ensure_running", fake_ensure), \ + patch.object(proc, "has_persisted_connections", return_value=False): + # Zero-config: no keys, no proxy mode, no persisted connections -> no revival attempt. + assert await cpe.p_router_available(AppSettings()) is False + assert not ensures + # A persisted subscription connection alone IS evidence -> revival attempted. + with patch.object(proc, "has_persisted_connections", return_value=True): + assert await cpe.p_router_available(AppSettings()) is False # ensure failed (router stays down) + assert ensures, "sub-only users must get a revival attempt" + + asyncio.run(run())