Cognitive_Agent/tools/start_n7web.py

#!/usr/bin/env python3
"""Start N7Web and its read-only LAN mode-card service together."""

from __future__ import annotations

import argparse
import json
import os
import shlex
import shutil
import socket
import subprocess
import sys
import time
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path


REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CATALOG_PATH = REPOSITORY_ROOT.parent / "scenses" / "demo-scenes.json"
PYTHON_SOURCE_PATHS = (
    REPOSITORY_ROOT / "src",
    REPOSITORY_ROOT / "domains" / "seat" / "src",
)
WEB_CONSOLE_ROOT = REPOSITORY_ROOT / "apps" / "web-console"


@dataclass(frozen=True)
class ChildProcess:
    name: str
    process: subprocess.Popen[bytes]


def _port(value: str) -> int:
    try:
        port = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("port must be an integer") from exc
    if not 1 <= port <= 65_535:
        raise argparse.ArgumentTypeError("port must be between 1 and 65535")
    return port


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "Start the N7Web application and the read-only LAN mode-card service. "
            "Mode-card synchronization is event-driven inside the main service."
        )
    )
    parser.add_argument(
        "--host",
        default="127.0.0.1",
        help="main N7Web bind address (use 0.0.0.0 for trusted-LAN access)",
    )
    parser.add_argument("--port", type=_port, default=8000, help="main N7Web port")
    parser.add_argument(
        "--scene-bind",
        default="0.0.0.0",
        help="read-only mode-card HTTP/UDP bind address",
    )
    parser.add_argument(
        "--scene-port",
        type=_port,
        default=8090,
        help="read-only mode-card HTTP/UDP port",
    )
    parser.add_argument(
        "--catalog",
        type=Path,
        default=DEFAULT_CATALOG_PATH,
        help="synchronized n7-scenes JSON path",
    )
    parser.add_argument(
        "--server-id",
        default="n7-cognitive-agent",
        help="server_id written to and served with the catalog",
    )
    parser.add_argument(
        "--python",
        type=Path,
        help="child Python executable (defaults to the repository .venv)",
    )
    parser.add_argument(
        "--catalog-ready-timeout",
        type=float,
        default=30.0,
        help="seconds to wait for the main service to create a valid catalog",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="print the resolved commands without starting processes",
    )
    parser.add_argument(
        "--skip-web-build",
        action="store_true",
        help="reuse the existing Vite dist output instead of rebuilding it",
    )
    return parser


def _resolve_path(path: Path) -> Path:
    expanded = path.expanduser()
    if not expanded.is_absolute():
        expanded = REPOSITORY_ROOT / expanded
    return expanded.resolve()


def _resolve_python(configured: Path | None) -> Path:
    if configured is not None:
        executable = _resolve_path(configured)
        if not executable.is_file():
            raise ValueError(f"Python executable does not exist: {executable}")
        return executable

    candidates = (
        REPOSITORY_ROOT / ".venv" / "Scripts" / "python.exe",
        REPOSITORY_ROOT / ".venv" / "bin" / "python",
    )
    for candidate in candidates:
        if candidate.is_file():
            return candidate.resolve()
    return Path(sys.executable).resolve()


def _child_environment(base: Mapping[str, str] | None = None) -> dict[str, str]:
    environment = dict(os.environ if base is None else base)
    existing = [
        entry
        for entry in environment.get("PYTHONPATH", "").split(os.pathsep)
        if entry
    ]
    combined = [str(path.resolve()) for path in PYTHON_SOURCE_PATHS]
    seen = {os.path.normcase(os.path.abspath(entry)) for entry in combined}
    for entry in existing:
        normalized = os.path.normcase(os.path.abspath(entry))
        if normalized not in seen:
            combined.append(entry)
            seen.add(normalized)
    environment["PYTHONPATH"] = os.pathsep.join(combined)
    environment.setdefault("PYTHONUNBUFFERED", "1")
    return environment


def _commands(
    args: argparse.Namespace,
    python: Path,
    catalog: Path,
) -> tuple[list[str], list[str]]:
    main_command = [
        str(python),
        "-m",
        "cognitive_agent.server",
        "--host",
        args.host,
        "--port",
        str(args.port),
        "--scene-catalog",
        str(catalog),
        "--scene-server-id",
        args.server_id,
    ]
    scene_command = [
        str(python),
        "-m",
        "cognitive_agent.lan_scene_server",
        "--bind",
        args.scene_bind,
        "--port",
        str(args.scene_port),
        "--catalog",
        str(catalog),
        "--server-id",
        args.server_id,
    ]
    return main_command, scene_command


def _display_command(command: Sequence[str]) -> str:
    if os.name == "nt":
        return subprocess.list2cmdline(command)
    return shlex.join(command)


def _probe_host(bind_address: str) -> str:
    address = str(bind_address).strip()
    if address in {"", "0.0.0.0"}:
        return "127.0.0.1"
    if address == "::":
        return "::1"
    return address


def _tcp_port_is_listening(bind_address: str, port: int) -> bool:
    try:
        with socket.create_connection((_probe_host(bind_address), port), timeout=0.25):
            return True
    except OSError:
        return False


def _ensure_service_ports_available(args: argparse.Namespace) -> None:
    endpoints = (
        ("main N7Web service", args.host, args.port),
        ("LAN mode-card service", args.scene_bind, args.scene_port),
    )
    occupied = [
        f"{name} ({_probe_host(host)}:{port})"
        for name, host, port in endpoints
        if _tcp_port_is_listening(host, port)
    ]
    if occupied:
        joined = ", ".join(occupied)
        raise RuntimeError(
            f"service port already in use: {joined}; stop the previous N7Web "
            "instance before starting another"
        )


def _resolve_npm() -> Path:
    executable = shutil.which("npm.cmd" if os.name == "nt" else "npm")
    if not executable:
        raise ValueError(
            "npm is required to build the Vue 3 + Vite web console; "
            "install Node.js or pass --skip-web-build to reuse an existing dist"
        )
    return Path(executable).resolve()


def _web_build_commands(npm: Path) -> list[list[str]]:
    commands: list[list[str]] = []
    vite_binary = WEB_CONSOLE_ROOT / "node_modules" / ".bin" / (
        "vite.cmd" if os.name == "nt" else "vite"
    )
    if not vite_binary.is_file():
        commands.append([str(npm), "ci"])
    commands.append([str(npm), "run", "build"])
    return commands


def _build_web_console(npm: Path) -> None:
    for command in _web_build_commands(npm):
        print(f"Web build:  {_display_command(command)}", flush=True)
        completed = subprocess.run(command, cwd=WEB_CONSOLE_ROOT, check=False)
        if completed.returncode != 0:
            raise RuntimeError(
                f"Vue 3 + Vite web build failed with exit code {completed.returncode}"
            )


def _catalog_is_ready(path: Path) -> bool:
    try:
        document = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return False
    return bool(
        isinstance(document, dict)
        and document.get("service") == "n7-scenes"
        and document.get("schema_version") == 1
        and isinstance(document.get("scenes"), list)
    )


def _wait_for_catalog(
    path: Path,
    main_process: subprocess.Popen[bytes],
    timeout: float,
) -> None:
    if timeout <= 0:
        raise ValueError("catalog-ready-timeout must be greater than zero")
    deadline = time.monotonic() + timeout
    while not _catalog_is_ready(path):
        exit_code = main_process.poll()
        if exit_code is not None:
            raise RuntimeError(
                f"main N7Web service exited before the catalog was ready "
                f"(exit code {exit_code})"
            )
        if time.monotonic() >= deadline:
            raise RuntimeError(
                f"timed out waiting for the synchronized catalog: {path}"
            )
        time.sleep(0.1)


def _wait_for_exit(children: Sequence[ChildProcess]) -> tuple[str, int]:
    while True:
        for child in children:
            exit_code = child.process.poll()
            if exit_code is not None:
                return child.name, exit_code
        time.sleep(0.25)


def _stop_children(
    children: Sequence[ChildProcess],
    *,
    interrupt_was_broadcast: bool,
) -> None:
    # Console Ctrl+C is delivered to the parent and both children.  Give the
    # services time to run their own finally blocks before escalating.
    if interrupt_was_broadcast:
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            if all(child.process.poll() is not None for child in children):
                return
            time.sleep(0.1)

    running = [child for child in children if child.process.poll() is None]
    for child in running:
        child.process.terminate()

    deadline = time.monotonic() + 5.0
    while time.monotonic() < deadline:
        if all(child.process.poll() is not None for child in running):
            return
        time.sleep(0.1)
    for child in running:
        if child.process.poll() is None:
            child.process.kill()


def run(argv: Sequence[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    if not args.server_id.strip():
        print("error: server-id must not be empty", file=sys.stderr)
        return 2

    try:
        python = _resolve_python(args.python)
        catalog = _resolve_path(args.catalog)
        npm = None if args.skip_web_build else _resolve_npm()
        if args.catalog_ready_timeout <= 0:
            raise ValueError("catalog-ready-timeout must be greater than zero")
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    environment = _child_environment()
    main_command, scene_command = _commands(args, python, catalog)
    print(f"Repository: {REPOSITORY_ROOT}", flush=True)
    print(f"Catalog:    {catalog}", flush=True)
    print(f"Main:       {_display_command(main_command)}", flush=True)
    print(f"Cards:      {_display_command(scene_command)}", flush=True)
    if npm is not None:
        for command in _web_build_commands(npm):
            print(f"Web build:  {_display_command(command)}", flush=True)
    if args.dry_run:
        return 0

    try:
        _ensure_service_ports_available(args)
        if npm is not None:
            _build_web_console(npm)
    except RuntimeError as exc:
        print(f"startup failed: {exc}", file=sys.stderr, flush=True)
        return 1

    children: list[ChildProcess] = []
    interrupted = False
    try:
        main_process = subprocess.Popen(
            main_command,
            cwd=REPOSITORY_ROOT,
            env=environment,
        )
        children.append(ChildProcess("main N7Web service", main_process))
        _wait_for_catalog(catalog, main_process, args.catalog_ready_timeout)

        scene_process = subprocess.Popen(
            scene_command,
            cwd=REPOSITORY_ROOT,
            env=environment,
        )
        children.append(ChildProcess("LAN mode-card service", scene_process))
        print(
            "Both services are running; press Ctrl+C once to stop both.",
            flush=True,
        )
        name, exit_code = _wait_for_exit(children)
        print(f"{name} exited (exit code {exit_code}); stopping the other service.")
        return exit_code if exit_code != 0 else 1
    except KeyboardInterrupt:
        interrupted = True
        print("\nStopping N7Web services...", flush=True)
        return 0
    except (OSError, RuntimeError, ValueError) as exc:
        print(f"startup failed: {exc}", file=sys.stderr, flush=True)
        return 1
    finally:
        _stop_children(children, interrupt_was_broadcast=interrupted)


if __name__ == "__main__":
    raise SystemExit(run())