nap-cli/scripts/nap_install.py

#!/usr/bin/env python3
"""Install the Napseer local command.

Usage:
  python3 nap_install.py
  nap project create
  nap project status
  nap gateway start
  nap gateway stop
  nap gateway configure
  nap gateway logs
  nap status
  nap agent list
  nap update
"""

import json
import hashlib
import os
import pathlib
import shutil
import signal
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request


BASE_URL = os.environ.get("NAPSEER_BASE_URL", "https://api.napseer.com").rstrip("/")
INSTALL_DIR = pathlib.Path(os.environ.get("NAPSEER_HOME", pathlib.Path.home() / ".local" / "share" / "napseer"))
BIN_DIR = pathlib.Path(os.environ.get("NAPSEER_BIN_DIR", pathlib.Path.home() / ".local" / "bin"))
SCRIPT_NAMES = (
    "napseer_mcp_server.py",
    "napseer_spake2.py",
    "terminal_init.py",
    "terminal_protocol.py",
    "terminal_pty_manager.py",
    "nap_install.py",
)
HTTP_TIMEOUT_SECONDS = int(os.environ.get("NAPSEER_HTTP_TIMEOUT_SECONDS", "30"))
LOCAL_HOST = "127.0.0.1"
DEFAULT_GATEWAY_PORT = os.environ.get("NAPSEER_GATEWAY_PORT")
RUNTIME_SCRIPT_NAMES = (
    "napseer_spake2.py",
    "terminal_init.py",
    "terminal_protocol.py",
    "terminal_pty_manager.py",
    "napseer_mcp_server.py",
)
MAX_SERVICE_LOG_BYTES = int(os.environ.get("NAPSEER_MAX_SERVICE_LOG_BYTES", str(5 * 1024 * 1024)))
CLI_DISTRIBUTION_CONTRACT_VERSION = "2026-05-19"
CLI_GENERATED_SOURCE_REPO = os.environ.get("NAPSEER_CLI_SOURCE_REPO", "https://github.com/napseer/nap-cli")
CLI_GENERATED_SOURCE_REVISION = os.environ.get("NAPSEER_CLI_SOURCE_REVISION", "unresolved")
CLI_GENERATED_SOURCE_REVISION_STATUS = os.environ.get("NAPSEER_CLI_SOURCE_REVISION_STATUS", "unresolved")


def chmod_best_effort(path, mode):
    try:
        path.chmod(mode)
    except OSError:
        pass


def normalize_proxy_url(value):
    if not value:
        return None
    proxy_url = str(value).strip()
    if not proxy_url:
        return None
    if "://" not in proxy_url:
        proxy_url = f"http://{proxy_url}"
    return urllib.parse.urlparse(proxy_url)


def api_proxy_url(parsed):
    host = parsed.hostname or ""
    if host and urllib.request.proxy_bypass(host):
        return None
    explicit = (
        os.environ.get("NAPSEER_API_PROXY_URL")
        or os.environ.get("NAPSEER_HTTP_PROXY_URL")
        or os.environ.get("NAPSEER_PROXY_URL")
        or os.environ.get("NAPSEER_RELAY_PROXY_URL")
        or os.environ.get("NAPSEER_GATEWAY_PROXY_URL")
    )
    if explicit:
        return explicit
    proxies = urllib.request.getproxies_environment()
    return proxies.get(parsed.scheme) or proxies.get("all")


def api_urlopen(request, timeout):
    parsed = urllib.parse.urlparse(request.full_url)
    proxy = normalize_proxy_url(api_proxy_url(parsed))
    if not proxy:
        return urllib.request.urlopen(request, timeout=timeout)
    proxy_url = urllib.parse.urlunparse(proxy)
    opener = urllib.request.build_opener(urllib.request.ProxyHandler({parsed.scheme: proxy_url}))
    return opener.open(request, timeout=timeout)


def read_url_bytes(url, user_agent, label):
    request = urllib.request.Request(url, headers={"User-Agent": user_agent}, method="GET")
    try:
        with api_urlopen(request, timeout=HTTP_TIMEOUT_SECONDS) as response:
            return response.read()
    except (TimeoutError, socket.timeout) as exc:
        raise RuntimeError(f"{label} timed out after {HTTP_TIMEOUT_SECONDS}s") from exc
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"{label} failed: HTTP {exc.code}: {body}") from exc


def fetch_script(name):
    try:
        body = read_url_bytes(
            f"{BASE_URL}/v1/scripts/{name}",
            "nap-install-python/0.1",
            f"GET /v1/scripts/{name}",
        )
        payload = json.loads(body.decode("utf-8"))
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"{name} response was not valid JSON") from exc
    content = payload.get("content")
    if not isinstance(content, str) or not content.strip():
        raise RuntimeError(f"{name} response did not include script content")
    expected_sha256 = payload.get("sha256")
    if isinstance(expected_sha256, str) and expected_sha256.strip():
        actual_sha256 = hashlib.sha256(content.encode("utf-8")).hexdigest()
        if actual_sha256 != expected_sha256.lower():
            raise RuntimeError(
                f"{name} backend script sha256 mismatch: expected {expected_sha256.lower()}, got {actual_sha256}"
            )
    return payload, content


def write_file(path, content, executable=False):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")
    chmod_best_effort(path, 0o755 if executable else 0o644)


def path_warning():
    path_entries = os.environ.get("PATH", "").split(os.pathsep)
    if str(BIN_DIR) not in path_entries:
        return f"Add {BIN_DIR} to PATH to run nap from any directory."
    return None


def source_revision_status_for(value):
    if not isinstance(value, str):
        return "unresolved"
    revision = value.strip()
    if not revision or revision in ("unresolved", "unknown", "null"):
        return "unresolved"
    return "resolved"


def install_assets(update_project=False):
    scripts = {}
    fetched = {}
    for name in SCRIPT_NAMES:
        payload, content = fetch_script(name)
        fetched[name] = (payload, content)

    install_order = []
    for name in (*RUNTIME_SCRIPT_NAMES, "nap_install.py", *SCRIPT_NAMES):
        if name not in install_order:
            install_order.append(name)
    for name in install_order:
        payload, content = fetched[name]
        target = INSTALL_DIR / str(payload.get("filename") or name)
        write_file(target, content, executable=name.endswith(".py"))
        scripts[name] = {
            "path": str(target),
            "version": payload.get("version"),
            "build_commit": payload.get("build_commit"),
            "sha256": payload.get("sha256"),
            "source_repo": payload.get("source_repo"),
            "source_revision": payload.get("source_revision"),
            "source_revision_status": payload.get("source_revision_status")
            or source_revision_status_for(payload.get("source_revision")),
            "generated_at": payload.get("generated_at"),
            "generator_version": payload.get("generator_version"),
            "contract_version": payload.get("contract_version"),
            "supported_backend_api_versions": payload.get("supported_backend_api_versions"),
            "supported_gateway_contract_versions": payload.get("supported_gateway_contract_versions"),
        }

    installer_content = (INSTALL_DIR / "nap_install.py").read_text(encoding="utf-8")
    nap_path = BIN_DIR / "nap"
    write_file(nap_path, installer_content, executable=True)
    if os.name == "nt":
        write_file(BIN_DIR / "nap.cmd", f'@echo off\r\n"{sys.executable}" "{nap_path}" %*\r\n', executable=False)

    updated_project_paths = []
    if update_project:
        for script_name in ("napseer_mcp_server.py", "napseer_spake2.py"):
            content = (INSTALL_DIR / script_name).read_text(encoding="utf-8")
            for candidate in (
                pathlib.Path.cwd() / script_name,
                pathlib.Path.cwd() / "resources" / "scripts" / script_name,
            ):
                if candidate.exists():
                    write_file(candidate, content, executable=script_name.endswith(".py"))
                    updated_project_paths.append(str(candidate))

    installer_payload = fetched["nap_install.py"][0]
    cli_source_revision = installer_payload.get("source_revision") or CLI_GENERATED_SOURCE_REVISION
    cli_source_revision_status = (
        installer_payload.get("source_revision_status")
        or CLI_GENERATED_SOURCE_REVISION_STATUS
        or source_revision_status_for(cli_source_revision)
    )
    result = {
        "status": "installed",
        "message": "Napseer CLI and runtime scripts are installed.",
        "bin": str(nap_path),
        "install_dir": str(INSTALL_DIR),
        "scripts": scripts,
        "cli_distribution": {
            "contract_version": CLI_DISTRIBUTION_CONTRACT_VERSION,
            "source_repo": installer_payload.get("source_repo") or CLI_GENERATED_SOURCE_REPO,
            "source_revision": None
            if source_revision_status_for(cli_source_revision) == "unresolved"
            else cli_source_revision,
            "source_revision_status": cli_source_revision_status,
        },
        "updated_project_paths": updated_project_paths,
        "path_warning": path_warning(),
        "next": ["nap project create", "nap gateway configure", "nap gateway start", "nap gateway logs", "nap update"],
    }
    return result


def runtime_assets_missing():
    missing = []
    for name in RUNTIME_SCRIPT_NAMES:
        path = INSTALL_DIR / {
            "terminal_init.py": "terminal/__init__.py",
            "terminal_protocol.py": "terminal/protocol.py",
            "terminal_pty_manager.py": "terminal/pty_manager.py",
        }.get(name, name)
        if not path.exists() or path.stat().st_size == 0:
            missing.append(name)
    return missing


def ensure_runtime_assets():
    missing = runtime_assets_missing()
    if missing:
        install_assets(update_project=False)
        missing = runtime_assets_missing()
    if missing:
        raise RuntimeError(f"missing installed Napseer runtime scripts after repair: {', '.join(missing)}")


def run_wrapper(command_name, args):
    wrapper = INSTALL_DIR / "napseer_mcp_server.py"
    ensure_runtime_assets()
    return subprocess.call([sys.executable, str(wrapper), command_name, *args])


def preferred_state_dir():
    return pathlib.Path.cwd() / ".napseer"


def cwd_state_dir():
    preferred = preferred_state_dir()
    preferred.mkdir(exist_ok=True)
    chmod_best_effort(preferred, 0o700)
    return preferred


def state_dir_status():
    preferred = preferred_state_dir()
    active = cwd_state_dir()
    return {
        "active_state_dir": str(active),
        "active_auth_path": str(active / "auth.json"),
        "state_dir": str(preferred),
        "auth_path": str(preferred / "auth.json"),
        "state_dir_exists": preferred.exists(),
        "state_dir_message": ".napseer directory is active.",
    }


def service_state_path(kind):
    return cwd_state_dir() / f"{kind}.json"


def service_log_path(kind):
    return cwd_state_dir() / f"{kind}.log"


def read_service_state(kind):
    path = service_state_path(kind)
    if not path.exists():
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return None


def write_service_state(kind, state):
    path = service_state_path(kind)
    write_file(path, json.dumps(state, indent=2), executable=False)
    chmod_best_effort(path, 0o600)


def pid_running(pid):
    if not pid:
        return False
    proc_stat = pathlib.Path("/proc") / str(pid) / "stat"
    try:
        fields = proc_stat.read_text(encoding="utf-8").split()
        if len(fields) > 2 and fields[2] == "Z":
            return False
    except Exception:
        pass
    try:
        os.kill(int(pid), 0)
        return True
    except OSError:
        return False


def find_free_port():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind((LOCAL_HOST, 0))
        return int(sock.getsockname()[1])


def port_accepts_connections(port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.settimeout(0.2)
        return sock.connect_ex((LOCAL_HOST, int(port))) == 0


def port_can_bind(port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        try:
            sock.bind((LOCAL_HOST, int(port)))
            return True
        except OSError:
            return False


def local_gateway_status(port):
    request = urllib.request.Request(
        f"http://{LOCAL_HOST}:{int(port)}/gateway/status",
        headers={"User-Agent": "nap-cli-python/0.1"},
        method="GET",
    )
    try:
        with urllib.request.urlopen(request, timeout=1) as response:
            return json.loads(response.read().decode("utf-8"))
    except Exception:
        return None


def log_tail(path, max_chars=2000):
    try:
        text = pathlib.Path(path).read_text(encoding="utf-8", errors="replace")
    except Exception:
        return ""
    return text[-max_chars:]


def log_tail_lines(path, max_lines=120):
    try:
        lines = pathlib.Path(path).read_text(encoding="utf-8", errors="replace").splitlines()
    except FileNotFoundError:
        return []
    except Exception as exc:
        return [f"failed to read log: {exc}"]
    return lines[-max(1, max_lines):]


def compact_log_file(path, max_bytes=MAX_SERVICE_LOG_BYTES):
    if max_bytes <= 0:
        return
    file_path = pathlib.Path(path)
    try:
        size = file_path.stat().st_size
    except FileNotFoundError:
        return
    except OSError:
        return
    if size <= max_bytes:
        return
    try:
        with file_path.open("rb") as handle:
            handle.seek(-max_bytes, os.SEEK_END)
            chunk = handle.read(max_bytes)
    except OSError:
        return
    # Try to start on a newline boundary so the log stays line-oriented.
    newline_index = chunk.find(b"\n")
    if newline_index >= 0 and newline_index + 1 < len(chunk):
        chunk = chunk[newline_index + 1 :]
    tmp_path = file_path.with_suffix(file_path.suffix + ".tmp")
    try:
        tmp_path.write_bytes(chunk)
        tmp_path.replace(file_path)
    except OSError:
        try:
            tmp_path.unlink(missing_ok=True)
        except OSError:
            pass


def ensure_port_available(kind, port):
    if not port_accepts_connections(port) and port_can_bind(port):
        return
    status = local_gateway_status(port)
    if status:
        cwd = status.get("cwd") or "unknown cwd"
        project_id = status.get("project_id") or "no project"
        raise RuntimeError(
            f"{LOCAL_HOST}:{port} is already serving a Napseer gateway for {cwd} "
            f"(project {project_id}). Stop that gateway before starting this one."
        )
    raise RuntimeError(f"{LOCAL_HOST}:{port} is already in use or still being released")


def choose_service_port(kind, args, existing=None):
    requested = cli_option(args, "--port", default=None)
    if requested is not None:
        return int(requested)
    if kind == "gateway":
        if DEFAULT_GATEWAY_PORT:
            return int(DEFAULT_GATEWAY_PORT)
        existing_port = (existing or {}).get("port")
        if existing_port and port_can_bind(existing_port):
            return int(existing_port)
        return find_free_port()
    return find_free_port()


def wait_for_service(kind, process, port, log_path):
    deadline = time.time() + 3
    while time.time() < deadline:
        if process.poll() is not None:
            tail = log_tail(log_path)
            detail = f": {tail.strip()}" if tail.strip() else ""
            raise RuntimeError(f"{kind} service exited during startup{detail}")
        if kind == "gateway" and local_gateway_status(port):
            return
        if kind != "gateway" and port_accepts_connections(port):
            return
        time.sleep(0.1)
    raise RuntimeError(f"{kind} service did not start listening on {LOCAL_HOST}:{port}")


def cli_option(args, *names, default=None):
    for name in names:
        if name in args:
            index = args.index(name)
            if index + 1 >= len(args):
                raise RuntimeError(f"{name} requires a value")
            return args[index + 1]
    return default


def start_service(kind, args):
    wrapper = INSTALL_DIR / "napseer_mcp_server.py"
    ensure_runtime_assets()
    existing = read_service_state(kind)
    if existing and pid_running(existing.get("pid")):
        return {
            **existing,
            "status": "running",
            "message": f"{kind} service is already running.",
        }
    port = choose_service_port(kind, args, existing)
    ensure_port_available(kind, port)
    log_path = service_log_path(kind)
    compact_log_file(log_path)
    command = [sys.executable, str(wrapper), "open-ui", "--port", str(port), "--no-browser"]
    if kind == "ui" and "--no-browser" not in args:
        command.remove("--no-browser")
    with log_path.open("ab") as log:
        kwargs = {"stdout": log, "stderr": subprocess.STDOUT, "cwd": str(pathlib.Path.cwd())}
        if os.name != "nt":
            kwargs["start_new_session"] = True
        else:
            kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
        process = subprocess.Popen(command, **kwargs)
    wait_for_service(kind, process, port, log_path)
    state = {
        "status": "running",
        "kind": kind,
        "pid": process.pid,
        "host": LOCAL_HOST,
        "port": port,
        "url": f"http://{LOCAL_HOST}:{port}/",
        "cwd": str(pathlib.Path.cwd()),
        "log_path": str(log_path),
        "started_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    }
    write_service_state(kind, state)
    return {**state, "message": f"{kind} service started."}


def stop_service(kind):
    state = read_service_state(kind)
    if not state:
        return {"status": "stopped", "kind": kind}
    pid = state.get("pid")
    if pid_running(pid):
        if os.name == "nt":
            subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        else:
            try:
                os.killpg(int(pid), signal.SIGTERM)
            except OSError:
                try:
                    os.kill(int(pid), signal.SIGTERM)
                except OSError:
                    pass
        deadline = time.time() + 5
        while time.time() < deadline and pid_running(pid):
            time.sleep(0.1)
        if pid_running(pid):
            if os.name == "nt":
                subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            else:
                try:
                    os.killpg(int(pid), signal.SIGKILL)
                except OSError:
                    try:
                        os.kill(int(pid), signal.SIGKILL)
                    except OSError:
                        pass
            deadline = time.time() + 2
            while time.time() < deadline and pid_running(pid):
                time.sleep(0.1)
    port = state.get("port")
    if port:
        deadline = time.time() + 2
        while time.time() < deadline and not port_can_bind(port):
            time.sleep(0.1)
    stopped = not pid_running(pid) and (not port or port_can_bind(port))
    state = {**state, "status": "stopped" if stopped else "stopping"}
    write_service_state(kind, state)
    if not stopped:
        state["warning"] = f"{kind} process or port {port} is still being released"
        state["message"] = state["warning"]
    else:
        state["message"] = f"{kind} service stopped."
    return state


def list_service(kind):
    state = read_service_state(kind) or {
        "kind": kind,
        "host": LOCAL_HOST,
        "port": None,
        "url": None,
        "cwd": str(pathlib.Path.cwd()),
    }
    state["status"] = "running" if pid_running(state.get("pid")) else "stopped"
    state["message"] = f"{kind} service is {state['status']}."
    return state


def service_logs(kind, args):
    max_lines = int(cli_option(args, "--lines", "--tail", default="120"))
    path = service_log_path(kind)
    compact_log_file(path)
    lines = log_tail_lines(path, max_lines=max_lines)
    status = "ok" if path.exists() else "missing"
    return {
        "status": status,
        "message": f"{kind} log {'read' if path.exists() else 'does not exist yet'}.",
        "kind": kind,
        "log_path": str(path),
        "line_count": len(lines),
        "lines": lines,
        "text": "\n".join(lines),
    }


def handle_gateway(args):
    subcommand = args[0] if args else "ls"
    rest = args[1:] if args else []
    if subcommand in {"help", "-h", "--help"}:
        print("Usage: nap gateway [start|stop|status|logs|configure|setup|restart|kill|vault|terminal|schedule|service]")
        print("  nap gateway start [--port PORT]       Start the local gateway service.")
        print("  nap gateway status                    Show the managed gateway process.")
        print("  nap gateway configure [--command CMD] Configure the local gateway command.")
        print("  nap gateway setup                     Create or migrate local gateway storage.")
        print("  nap gateway vault                     Show gateway vault status and pending setup requests.")
        print("  nap gateway vault process             Complete pending setup requests.")
        print("  nap gateway vault rotate-secret --kind memory")
        print("  nap chat secret setup                 Configure chat secret state through the gateway.")
        return 0
    if subcommand == "vault" and rest and rest[0] in {"help", "-h", "--help"}:
        print("Usage: nap gateway vault [status|list|process|rotate-secret] [--kind memory] [--all] [--project-id ID]")
        print("  status        Show local gateway state and pending setup requests.")
        print("  process       Upload opaque client-wrapped key bundle records for backend-owned HashiCorp storage.")
        print("  rotate-secret Rotate the memory project secret from the gateway worker path.")
        print("  Chat secrets use: nap chat secret setup|status|rotate")
        return 0
    if subcommand == "process":
        return run_wrapper("gateway", ["vault", "process", *rest])
    if subcommand == "start":
        return print_json(start_service("gateway", ["--no-browser", *rest])) or 0
    if subcommand == "stop":
        return print_json(stop_service("gateway")) or 0
    if subcommand in {"ls", "list", "status"}:
        return print_json(list_service("gateway")) or 0
    if subcommand in {"logs", "log"}:
        return print_json(service_logs("gateway", rest)) or 0
    if subcommand == "configure":
        return run_wrapper("gateway", ["configure", *rest])
    if subcommand in {"setup", "restart", "kill", "vault", "terminal", "schedule", "service"}:
        return run_wrapper("gateway", [subcommand, *rest])
    raise RuntimeError("unknown gateway command; use nap gateway start|stop|configure|setup|restart|kill|logs|status|vault|terminal|schedule|service")


def print_json(value):
    print(json.dumps(value, indent=2))


def main(argv):
    invoked_as_nap = pathlib.Path(argv[0]).name in {"nap", "nap.cmd"}
    command = argv[1] if len(argv) > 1 else ("help" if invoked_as_nap else "install")
    args = argv[2:]
    if command == "install":
        print_json(install_assets(update_project=False))
        return 0
    if command == "update":
        result = install_assets(update_project=True)
        print_json({**result, "status": "updated", "message": "Napseer CLI and runtime scripts are updated."})
        return 0
    if command == "gateway":
        return handle_gateway(args)
    if command == "chat":
        return run_wrapper("chat", args or ["secret", "status"])
    if command == "status":
        return run_wrapper("configure", args)
    if command == "project":
        if not args:
            return run_wrapper("project", ["status"])
        if args[0] == "bootstrap":
            raise RuntimeError("unknown project command: bootstrap")
        return run_wrapper("project", args)
    if command == "lineage":
        return run_wrapper("lineage", args or ["status"])
    if command == "plan":
        return run_wrapper("plan", args or ["list"])
    if command == "agent":
        return run_wrapper("agent", args)
    if command == "auth":
        return run_wrapper("auth", args)
    if command == "feedback":
        return run_wrapper("feedback", args)
    if command == "where":
        state = state_dir_status()
        print_json(
            {
                "bin": shutil.which("nap") or str(BIN_DIR / "nap"),
                "install_dir": str(INSTALL_DIR),
                "wrapper": str(INSTALL_DIR / "napseer_mcp_server.py"),
                "cwd": str(pathlib.Path.cwd()),
                "cwd_auth_path": state["active_auth_path"],
                "cwd_auth_configured": pathlib.Path(state["active_auth_path"]).exists(),
                **state,
                "path_warning": path_warning(),
                "status": "ok",
                "message": f"Napseer install and cwd paths resolved. {state['state_dir_message']}",
            }
        )
        return 0
    if command in {"help", "-h", "--help"}:
        print("Usage: nap [auth|project|chat|plan|lineage|gateway|agent|feedback|status|update|where|install]")
        print("  auth        Auth commands: login-local, status.")
        print("              Example: nap auth login-local")
        print("  project     Project commands: create, status, encryption.")
        print("              Examples: nap project create, nap project status, nap project encryption status")
        print("              Encrypted project setup is completed by the local gateway: nap gateway vault process")
        print("  chat        Chat secret commands: secret status, secret setup, secret rotate.")
        print("              Examples: nap chat secret setup, nap chat secret status")
        print("  plan        Plan commands: list, to-kanban.")
        print("              Example: nap plan to-kanban /plans/example --column todo")
        print("  lineage     Generic source-record -> plan -> kanban lineage checks.")
        print("              Example: nap lineage status")
        print("  gateway     Gateway commands: start, stop, configure, setup, restart, kill, logs, status, vault.")
        print("              Start auto-selects a local port; use --port to pin one.")
        print("              Examples: nap gateway start, nap gateway vault, nap gateway vault process --all")
        print("  agent       Agent commands: list, workspaces, create, show, edit.")
        print("              list shows registered project agents; workspaces shows /agents folders.")
        print("  feedback    Feedback commands: list, global, status, resolve.")
        print("              Examples: nap feedback list, nap feedback global --all, nap feedback resolve <id> --notes TEXT")
        print("  status      Show cwd Napseer configuration and next setup commands.")
        print("  update      Refresh the globally installed nap command and wrapper.")
        print("  where       Show installed nap paths and cwd auth location.")
        print("  install     Reinstall/repair nap from the hosted script directory.")
        return 0
    raise RuntimeError(f"unknown command: {command}")


if __name__ == "__main__":
    try:
        raise SystemExit(main(sys.argv))
    except Exception as exc:
        print(json.dumps({"error": str(exc)}, indent=2), file=sys.stderr)
        raise SystemExit(1)