openswarm/backend/apps/subscriptions/NineRouter/helpers/NineRouterProcess/NineRouterProcess.py

"""Subprocess lifecycle management for the 9Router process.

9Router is a local Node.js proxy that lets users connect their
Claude/ChatGPT/Gemini subscriptions to OpenSwarm without API keys.
"""

import asyncio
import os
import shutil
import subprocess
import threading
from typing import Optional

import httpx
from pydantic import Field, BaseModel, InstanceOf
from typeguard import typechecked

from backend.ports import NINE_ROUTER_PORT
from backend.apps.subscriptions.NineRouter.helpers.constants import NINE_ROUTER_V1, NINE_ROUTER_URL
from backend.apps.subscriptions.NineRouter.helpers.NineRouterProcess.helpers.forward_output import forward_output
from backend.apps.subscriptions.NineRouter.helpers.NineRouterProcess.helpers.find_9router_dir import find_9router_dir
from backend.apps.subscriptions.NineRouter.helpers.NineRouterProcess.helpers.find_node import find_node

P_THIS_DIR: str = os.path.dirname(os.path.abspath(__file__))


class NineRouterProcess(BaseModel):
    p_process: Optional[InstanceOf[subprocess.Popen]] = Field(default=None)

    @typechecked
    def is_running(self) -> bool:
        try:
            r = httpx.get(f"{NINE_ROUTER_V1}/models", timeout=2.0)
            return r.status_code == 200
        except Exception:
            return False

    @typechecked
    async def ensure_running(self) -> None:
        """Start 9Router if not already running."""
        _is_packaged: bool = os.environ.get("OPENSWARM_PACKAGED") == "1"

        if self.is_running():
            if not _is_packaged:
                try:
                    result = subprocess.run(
                        ["pgrep", "-f", "next-server"],
                        capture_output=True, text=True, timeout=3,
                    )
                    if result.stdout.strip():
                        print("9Router: killing stale standalone to use next dev", flush=True)
                        subprocess.run(["pkill", "-f", "next-server"], timeout=5)
                        await asyncio.sleep(2)
                    else:
                        return
                except Exception:
                    return
            else:
                return

        p_9router_dir: Optional[str] = find_9router_dir(P_THIS_DIR)
        cmd: list[str]
        cwd: str | None
        env: dict[str, str]

        if _is_packaged and p_9router_dir:
            standalone_server: str = os.path.join(p_9router_dir, "server.js")
            if not os.path.exists(standalone_server):
                standalone_server = os.path.join(p_9router_dir, ".next", "standalone", "server.js")
            if not os.path.exists(standalone_server):
                print("9Router: standalone build not found in", p_9router_dir, flush=True)
                return

            node: Optional[str] = find_node()
            if not node:
                print("9Router: Node.js not found, cannot start in packaged mode", flush=True)
                return

            print(f"9Router: starting (production) on port {NINE_ROUTER_PORT}...", flush=True)
            cmd = [node, standalone_server]
            cwd = os.path.dirname(standalone_server)
            env = {
                **os.environ,
                "PORT": str(NINE_ROUTER_PORT),
                "NEXT_PUBLIC_BASE_URL": NINE_ROUTER_URL,
                "NODE_ENV": "production",
            }
            if node == os.environ.get("OPENSWARM_ELECTRON_PATH"):
                env["ELECTRON_RUN_AS_NODE"] = "1"

        elif p_9router_dir:
            npx: str | None = shutil.which("npx")
            if not npx:
                print("9Router: npx not found, cannot auto-start", flush=True)
                return

            if not os.path.isdir(os.path.join(p_9router_dir, "node_modules")):
                print("9Router: installing dependencies...", flush=True)
                npm: str | None = shutil.which("npm")
                if npm:
                    subprocess.run(
                        [npm, "install"], cwd=p_9router_dir,
                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=120,
                    )

            print(f"9Router: starting (dev) on port {NINE_ROUTER_PORT}...", flush=True)
            cmd = [npx, "next", "dev", "--webpack", "-p", str(NINE_ROUTER_PORT)]
            cwd = p_9router_dir
            env = {
                **os.environ,
                "PORT": str(NINE_ROUTER_PORT),
                "NEXT_PUBLIC_BASE_URL": NINE_ROUTER_URL,
            }

        else:
            npx = shutil.which("npx")
            if not npx:
                print("9Router: npx not found and no bundled 9router directory", flush=True)
                return
            print(f"9Router: starting (npx) on port {NINE_ROUTER_PORT}...", flush=True)
            cmd = [npx, "9router", "--port", str(NINE_ROUTER_PORT), "--no-browser", "--skip-update"]
            cwd = None
            env = {
                **os.environ,
                "PORT": str(NINE_ROUTER_PORT),
                "NEXT_PUBLIC_BASE_URL": NINE_ROUTER_URL,
            }

        try:
            self.p_process = subprocess.Popen(
                cmd, cwd=cwd,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                env=env,
            )
            threading.Thread(target=forward_output, args=(self.p_process.stdout,), daemon=True).start()

            timeout: int = 20 if _is_packaged else 30
            for _ in range(timeout * 2):
                await asyncio.sleep(0.5)
                if self.is_running():
                    print("9Router: started successfully", flush=True)
                    return

            print(f"9Router: did not start within {timeout}s", flush=True)
        except Exception as e:
            print(f"9Router: failed to start: {e}", flush=True)

    @typechecked
    def stop(self) -> None:
        if self.p_process:
            try:
                self.p_process.terminate()
                self.p_process.wait(timeout=5)
            except Exception:
                try:
                    self.p_process.kill()
                except Exception:
                    pass
            self.p_process = None
            print("9Router stopped")