From 5f864eab06e8ffacd061ed10463a026255ea2d69 Mon Sep 17 00:00:00 2001 From: HumbleTurtle Date: Thu, 30 Jul 2026 00:27:22 -0300 Subject: [PATCH] Remove duplicate bootstrap installers --- README.md | 1 - resources/scripts/napseer_setup.py | 409 ---------- scripts/nap_install.py | 1153 ---------------------------- 3 files changed, 1563 deletions(-) delete mode 100644 resources/scripts/napseer_setup.py delete mode 100755 scripts/nap_install.py diff --git a/README.md b/README.md index e3e11f2..77bd2df 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,6 @@ Current state: - `resources/scripts/napseer_mcp_supervisor.py` is the recommended Codex stdio entrypoint. It keeps the client transport alive while restarting or reloading the generated `napseer_mcp_server.py` worker between requests. -- `scripts/nap_install.py` is kept as a convenient top-level command source copy during the initial import. - Gateway service compatibility remains in the worker while the standalone gateway image is built from the public gateway repository. diff --git a/resources/scripts/napseer_setup.py b/resources/scripts/napseer_setup.py deleted file mode 100644 index 93e57da..0000000 --- a/resources/scripts/napseer_setup.py +++ /dev/null @@ -1,409 +0,0 @@ -#!/usr/bin/env python3 -"""One-command Napseer project setup. - -Run from a project root: - python3 napseer_setup.py - -What it does: - 1. Creates or reuses the active local state directory - 2. Creates .gitignore inside it so local secrets are not committed - 3. Generates id_ed25519 there if missing - 4. Enrolls this project copy in Napseer - 5. Creates or reuses a Napseer project - 6. Writes ./.napseer/auth.json for local MCP wrappers - -Optional environment: - NAPSEER_BASE_URL=https://api.napseer.com - NAPSEER_PROJECT_SLUG= - NAPSEER_PROJECT_NAME= - NAPSEER_WORKER_NAME= -""" - -import json -import getpass -import base64 -import os -import pathlib -import queue -import runpy -import secrets -import socket -import subprocess -import sys -import tempfile -import threading -import time -import urllib.error -import urllib.parse -import urllib.request -import uuid -import webbrowser -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer - - -ROOT = pathlib.Path.cwd() - - -def local_state_dir(): - preferred = ROOT / ".napseer" - legacy = ROOT / "napseer" - if preferred.exists() or not legacy.exists(): - return preferred - return legacy - - -NAPSEER_DIR = local_state_dir() -KEY_PATH = NAPSEER_DIR / "id_ed25519" -AUTH_PATH = NAPSEER_DIR / "auth.json" -VAULT_PATH = NAPSEER_DIR / "vault.json" -BASE_URL = os.environ.get("NAPSEER_BASE_URL", "https://api.napseer.com").rstrip("/") - - -def post_json(path, payload, token=None, idempotency_key=None): - data = json.dumps(payload).encode("utf-8") - headers = {"Content-Type": "application/json", "User-Agent": "napseer-setup-python/0.1"} - if token: - headers["Authorization"] = f"Bearer {token}" - if idempotency_key: - headers["Idempotency-Key"] = idempotency_key - request = urllib.request.Request(f"{BASE_URL}{path}", data=data, headers=headers, method="POST") - try: - with urllib.request.urlopen(request) as response: - return json.loads(response.read().decode("utf-8")) - except urllib.error.HTTPError as exc: - body = exc.read().decode("utf-8") - raise RuntimeError(f"POST {path} failed: HTTP {exc.code}: {body}") from exc - - -def put_json(path, payload, token=None): - data = json.dumps(payload).encode("utf-8") - headers = {"Content-Type": "application/json", "User-Agent": "napseer-setup-python/0.1"} - if token: - headers["Authorization"] = f"Bearer {token}" - request = urllib.request.Request(f"{BASE_URL}{path}", data=data, headers=headers, method="PUT") - try: - with urllib.request.urlopen(request) as response: - return json.loads(response.read().decode("utf-8")) - except urllib.error.HTTPError as exc: - body = exc.read().decode("utf-8") - raise RuntimeError(f"PUT {path} failed: HTTP {exc.code}: {body}") from exc - - -def get_json(path): - request = urllib.request.Request( - f"{BASE_URL}{path}", - headers={"User-Agent": "napseer-setup-python/0.1"}, - method="GET", - ) - try: - with urllib.request.urlopen(request) as response: - return json.loads(response.read().decode("utf-8")) - except urllib.error.HTTPError as exc: - body = exc.read().decode("utf-8") - raise RuntimeError(f"GET {path} failed: HTTP {exc.code}: {body}") from exc - - -def ensure_local_files(): - NAPSEER_DIR.mkdir(exist_ok=True) - chmod_best_effort(NAPSEER_DIR, 0o700) - gitignore = NAPSEER_DIR / ".gitignore" - gitignore.write_text("*\n!.gitignore\n", encoding="utf-8") - if not KEY_PATH.exists(): - subprocess.run( - ["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(KEY_PATH)], - check=True, - ) - chmod_best_effort(KEY_PATH, 0o600) - chmod_best_effort(pathlib.Path(str(KEY_PATH) + ".pub"), 0o644) - - -def read_gateway_passphrase(): - value = os.environ.get("NAPSEER_GATEWAY_PASSPHRASE") or os.environ.get("NAPSEER_MASTER_PASSPHRASE") - if value: - return value - if not sys.stdin.isatty(): - raise RuntimeError("NAPSEER_GATEWAY_PASSPHRASE is required in non-interactive mode") - first = getpass.getpass("Create gateway relay/storage passphrase: ") - second = getpass.getpass("Confirm gateway relay/storage passphrase: ") - if first != second: - raise RuntimeError("gateway relay/storage passphrases did not match") - return first - - -def load_gateway_vault_helpers(): - local_wrapper = pathlib.Path(__file__).resolve().with_name("napseer_mcp_server.py") - local_spake2 = pathlib.Path(__file__).resolve().with_name("napseer_spake2.py") - namespace = None - if not local_spake2.exists(): - remote_spake2 = get_json("/v1/scripts/napseer_spake2.py") - spake2_content = remote_spake2.get("content") - if isinstance(spake2_content, str) and spake2_content.strip(): - local_spake2.write_text(spake2_content, encoding="utf-8") - if local_wrapper.exists(): - namespace = runpy.run_path(str(local_wrapper)) - else: - remote = get_json("/v1/scripts/napseer_mcp_server.py") - content = remote.get("content") - if not isinstance(content, str) or not content.strip(): - raise RuntimeError("could not load gateway vault helper") - namespace = {"__file__": str(ROOT / "napseer_mcp_server.py"), "__name__": "napseer_mcp_server_setup_helper"} - exec(compile(content, "napseer_mcp_server.py", "exec"), namespace) - for name in ("write_vault", "write_public_auth"): - if name not in namespace: - raise RuntimeError(f"gateway vault helper is missing {name}") - return namespace - - -def write_gateway_vault(passphrase, public_auth, secret_auth): - helpers = load_gateway_vault_helpers() - helpers["write_vault"]( - passphrase, - { - "version": 1, - "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), - "secrets": {key: value for key, value in secret_auth.items() if value is not None}, - }, - ) - helpers["write_public_auth"]({**public_auth, "gateway_auth_required": True, "vault_path": str(VAULT_PATH)}) - - -def remove_plain_private_key(): - try: - if KEY_PATH.exists(): - KEY_PATH.unlink() - except OSError as exc: - raise RuntimeError(f"failed to remove plaintext private key at {KEY_PATH}: {exc}") from exc - - -def chmod_best_effort(path, mode): - try: - path.chmod(mode) - except OSError: - # Windows and some mounted filesystems do not support POSIX modes. - pass - - -def sign_text(namespace, key_path, text): - with tempfile.TemporaryDirectory() as tmp: - data_path = pathlib.Path(tmp) / "data.txt" - data_path.write_text(text, encoding="utf-8") - subprocess.run( - ["ssh-keygen", "-Y", "sign", "-q", "-n", namespace, "-f", str(key_path), str(data_path)], - check=True, - stdout=subprocess.DEVNULL, - ) - return pathlib.Path(str(data_path) + ".sig").read_text(encoding="utf-8") - - -def default_slug(): - raw = ROOT.name.lower() - slug = "".join(char if char.isalnum() else "-" for char in raw).strip("-") - return slug or "napseer-project" - - -def choose_account_mode(): - mode = os.environ.get("NAPSEER_ACCOUNT_MODE", "").strip().lower() - if mode in {"anonymous", "login"}: - return mode - if not sys.stdin.isatty(): - return "anonymous" - - print("Napseer account mode:") - print(" 1. Continue anonymous. Fastest, but recovery depends on local SSH keys.") - print(" 2. Login/claim account. Lets you recover and rotate SSH keys after web authentication.") - choice = input("Choose [1/2] (default 1): ").strip() - return "login" if choice == "2" else "anonymous" - - -def start_claim_receiver(): - result_queue = queue.Queue(maxsize=1) - - class Handler(BaseHTTPRequestHandler): - def log_message(self, format, *args): - return - - def do_GET(self): - parsed = urllib.parse.urlparse(self.path) - if parsed.path != "/napseer/claim-callback": - self.send_response(404) - self.end_headers() - return - - params = urllib.parse.parse_qs(parsed.query) - result = { - "status": params.get("status", [""])[0], - "account_id": params.get("account_id", [""])[0], - "claim_result_token": params.get("claim_result_token", [""])[0], - } - try: - result_queue.put_nowait(result) - except queue.Full: - pass - - self.send_response(200) - self.send_header("Content-Type", "text/html; charset=utf-8") - self.end_headers() - self.wfile.write( - b"

Napseer account connected.

You can return to the terminal.

" - ) - - server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) - port = server.server_address[1] - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - - return { - "server": server, - "queue": result_queue, - "return_url": f"http://127.0.0.1:{port}/napseer/claim-callback", - } - - -def wait_for_claim_result(receiver, timeout_seconds): - try: - result = receiver["queue"].get(timeout=timeout_seconds) - except queue.Empty: - return None - finally: - receiver["server"].shutdown() - - token = result.get("claim_result_token") - if not token: - return result - - verified = get_json(f"/v1/account/claim-results/{urllib.parse.quote(token)}") - result["verified"] = verified - return result - - -def main(): - ensure_local_files() - gateway_passphrase = read_gateway_passphrase() - public_key = pathlib.Path(str(KEY_PATH) + ".pub").read_text(encoding="utf-8").strip() - private_key = KEY_PATH.read_text(encoding="utf-8") - project_slug = os.environ.get("NAPSEER_PROJECT_SLUG", default_slug()) - project_name = os.environ.get("NAPSEER_PROJECT_NAME", ROOT.name or "Napseer Project") - worker_name = os.environ.get("NAPSEER_WORKER_NAME", f"{project_slug}-agent") - device_fingerprint = os.environ.get("NAPSEER_DEVICE_FINGERPRINT", socket.gethostname()) - - challenge = post_json( - "/v1/enrollment/challenges", - { - "public_key": public_key, - "worker_name": worker_name, - "device_fingerprint": device_fingerprint, - "root_path": str(ROOT), - "worker_capabilities": {"local_mcp": True, "gateway": True, "setup_script": "napseer_setup.py"}, - }, - ) - signature = sign_text("napseer", KEY_PATH, challenge["challenge_text"]) - verified = post_json( - "/v1/enrollment/verify", - {"challenge_id": challenge["challenge_id"], "signature": signature}, - ) - token = verified["token"]["access_token"] - - project = post_json( - "/v1/projects", - { - "slug": project_slug, - "name": project_name, - "description": f"Project initialized from {ROOT}", - }, - token=token, - idempotency_key=f"napseer-setup-{project_slug}", - ) - signing_key = put_json( - f"/v1/projects/{project['id']}/signing-key", - {"public_key": public_key, "label": "local-project-key"}, - token=token, - ) - account_mode = choose_account_mode() - account_claim = None - claim_receiver = None - if account_mode == "login": - configured_return_url = os.environ.get("NAPSEER_CLAIM_RETURN_URL") - claim_return_url = configured_return_url - if not claim_return_url and sys.stdin.isatty(): - claim_receiver = start_claim_receiver() - claim_return_url = claim_receiver["return_url"] - account_claim = post_json( - "/v1/account/claim-links", - {"return_url": claim_return_url}, - token=token, - ) - - public_auth = { - "base_url": BASE_URL, - "project_id": project["id"], - "project_slug": project["slug"], - "project_name": project["name"], - "project_signing_key_fingerprint": signing_key["fingerprint"], - "worker_id": verified["worker"]["id"], - "agent_id": verified["worker"]["agent_id"], - "worker_name": worker_name, - "device_fingerprint": device_fingerprint, - "root_path": str(ROOT), - "worker_capabilities": {"local_mcp": True, "gateway": True, "setup_script": "napseer_setup.py"}, - "account_mode": "claim_pending" if account_claim else "anonymous", - } - secret_auth = { - "token": token, - "token_expires_at": verified["token"].get("expires_at"), - "ssh_private_key": private_key, - "ssh_public_key": public_key, - "gateway_encryption_key": base64.b64encode(secrets.token_bytes(32)).decode("ascii"), - } - if account_claim: - secret_auth["account_claim_url"] = account_claim["claim_url"] - secret_auth["account_claim_expires_at"] = account_claim["expires_at"] - write_gateway_vault(gateway_passphrase, public_auth, secret_auth) - remove_plain_private_key() - - output = { - "status": "ok", - "auth_path": str(AUTH_PATH), - "vault_path": str(VAULT_PATH), - "project_id": project["id"], - "agent_id": verified["worker"]["agent_id"], - "project_signing_key_fingerprint": signing_key["fingerprint"], - "next": [ - "Run the local MCP wrapper from this project root.", - "python3 napseer_mcp_server.py", - ], - } - if account_claim: - output["account_claim"] = { - "claim_url": account_claim["claim_url"], - "expires_at": account_claim["expires_at"], - "next": "Open this URL in a browser to login and attach this anonymous account.", - } - print(json.dumps(output, indent=2)) - - if account_claim and claim_receiver: - print("Opening browser for Napseer account claim...", file=sys.stderr) - webbrowser.open(account_claim["claim_url"]) - timeout = int(os.environ.get("NAPSEER_CLAIM_WAIT_SECONDS", "180")) - claim_result = wait_for_claim_result(claim_receiver, timeout) - if claim_result: - public_auth["account_mode"] = "claimed" - public_auth["claimed_account_id"] = claim_result.get("account_id") - public_auth["claimed_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - secret_auth["claim_result"] = claim_result.get("verified", claim_result) - write_gateway_vault(master_passphrase, public_auth, secret_auth) - print(json.dumps({"status": "claimed", "account_id": public_auth["claimed_account_id"]}, indent=2)) - else: - print( - json.dumps( - { - "status": "claim_wait_timeout", - "claim_url": account_claim["claim_url"], - "next": "Open the claim URL manually; the local receiver has stopped.", - }, - indent=2, - ) - ) - - -if __name__ == "__main__": - main() diff --git a/scripts/nap_install.py b/scripts/nap_install.py deleted file mode 100755 index bf9c969..0000000 --- a/scripts/nap_install.py +++ /dev/null @@ -1,1153 +0,0 @@ -#!/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 ast -import contextlib -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_mcp_supervisor.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", - "napseer_mcp_supervisor.py", -) -MAX_SERVICE_LOG_BYTES = int(os.environ.get("NAPSEER_MAX_SERVICE_LOG_BYTES", str(5 * 1024 * 1024))) -CLI_RELEASE_VERSION = "0.2.0" -CLI_DISTRIBUTION_CONTRACT_VERSION = "2026-07-29" -CLI_MINIMUM_CONTRACT_VERSION = "2026-05-19" -CLI_BUNDLE_SCHEMA_VERSION = "napseer.cli.bundle.v1" -INSTALL_PATHS = { - "nap_install.py": "nap_install.py", - "napseer_mcp_server.py": "napseer_mcp_server.py", - "napseer_mcp_supervisor.py": "napseer_mcp_supervisor.py", - "napseer_spake2.py": "napseer_spake2.py", - "terminal_init.py": "terminal/__init__.py", - "terminal_protocol.py": "terminal/protocol.py", - "terminal_pty_manager.py": "terminal/pty_manager.py", -} -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") -HELP_TOKENS = {"help", "-h", "--help"} -CANONICAL_COMMANDS = ( - ("init", "Initialize Napseer in the current folder."), - ("status", "Show concise account, project, and runtime state."), - ("doctor", "Diagnose local setup without changing it."), - ("auth", "Authenticate an account or repair credentials."), - ("project", "Create, attach, claim, or inspect a project."), - ("mcp", "Install, update, inspect, or serve the local MCP runtime."), - ("gateway", "Set up and operate the local gateway."), - ("update", "Update the installed CLI and runtime bundle."), - ("version", "Show version, source revision, and contract identity."), - ("help", "Show this help or help for one command."), -) -COMMAND_METADATA = { - "init": {"mutates": True, "visible": True}, - "status": {"mutates": False, "visible": True}, - "doctor": {"mutates": False, "visible": True}, - "auth": {"mutates": True, "visible": True}, - "project": {"mutates": True, "visible": True}, - "mcp": {"mutates": True, "visible": True}, - "gateway": {"mutates": True, "visible": True}, - "update": {"mutates": True, "visible": True}, - "version": {"mutates": False, "visible": True}, - "help": {"mutates": False, "visible": True}, - "chat": {"mutates": True, "visible": False}, - "plan": {"mutates": True, "visible": False}, - "lineage": {"mutates": False, "visible": False}, - "agent": {"mutates": True, "visible": False}, - "feedback": {"mutates": True, "visible": False}, - "where": {"mutates": False, "visible": False}, - "install": {"mutates": True, "visible": False}, -} - - -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_json(path, label): - try: - body = read_url_bytes( - f"{BASE_URL}{path}", - "nap-install-python/0.1", - f"GET {path}", - ) - payload = json.loads(body.decode("utf-8")) - except json.JSONDecodeError as exc: - raise RuntimeError(f"{label} response was not valid JSON") from exc - if not isinstance(payload, dict): - raise RuntimeError(f"{label} response must be a JSON object") - return payload - - -def fetch_script(name): - payload = fetch_json(f"/v1/scripts/{name}", name) - 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 bundle_manifest(): - return fetch_json("/v1/scripts", "CLI bundle manifest") - - -def validated_manifest_items(manifest): - if manifest.get("schema_version") != CLI_BUNDLE_SCHEMA_VERSION: - raise RuntimeError( - f"unsupported CLI bundle schema: {manifest.get('schema_version')!r}" - ) - contract = manifest.get("contract") - if not isinstance(contract, dict): - raise RuntimeError("CLI bundle manifest is missing contract metadata") - if contract.get("current") != CLI_DISTRIBUTION_CONTRACT_VERSION: - raise RuntimeError("CLI bundle contract does not match this installer") - if contract.get("minimum_supported") != CLI_MINIMUM_CONTRACT_VERSION: - raise RuntimeError("CLI bundle minimum contract metadata is inconsistent") - if manifest.get("release_version") != CLI_RELEASE_VERSION: - raise RuntimeError("CLI bundle release version does not match this installer") - source = manifest.get("source") - if ( - not isinstance(source, dict) - or source_revision_status_for(source.get("revision")) != "resolved" - ): - raise RuntimeError("CLI bundle source revision must be resolved") - items = manifest.get("items") - if not isinstance(items, list): - raise RuntimeError("CLI bundle manifest items must be an array") - by_name = {} - for item in items: - if not isinstance(item, dict) or not isinstance(item.get("name"), str): - raise RuntimeError("CLI bundle manifest contains an invalid item") - name = item["name"] - if name in by_name: - raise RuntimeError(f"CLI bundle manifest contains duplicate asset {name}") - by_name[name] = item - missing = [name for name in SCRIPT_NAMES if name not in by_name] - if missing: - raise RuntimeError(f"CLI bundle manifest is missing assets: {', '.join(missing)}") - for name in SCRIPT_NAMES: - item = by_name[name] - expected_path = INSTALL_PATHS[name] - path = pathlib.PurePosixPath(str(item.get("install_path") or "")) - if str(path) != expected_path or path.is_absolute() or ".." in path.parts: - raise RuntimeError(f"CLI bundle asset {name} has unsafe install_path") - if item.get("mode") != "0755": - raise RuntimeError(f"CLI bundle asset {name} has unsupported mode") - if item.get("source_repo") != source.get("repo"): - raise RuntimeError(f"CLI bundle asset {name} has inconsistent source repo") - if item.get("source_revision") != source.get("revision"): - raise RuntimeError(f"CLI bundle asset {name} has inconsistent source revision") - if item.get("contract_version") != contract.get("current"): - raise RuntimeError(f"CLI bundle asset {name} has inconsistent contract version") - if item.get("minimum_contract_version") != contract.get("minimum_supported"): - raise RuntimeError(f"CLI bundle asset {name} has inconsistent minimum contract") - if not isinstance(item.get("bytes"), int) or item["bytes"] <= 0: - raise RuntimeError(f"CLI bundle asset {name} has invalid byte length") - digest = str(item.get("sha256") or "") - if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest): - raise RuntimeError(f"CLI bundle asset {name} has invalid sha256") - return {name: by_name[name] for name in SCRIPT_NAMES} - - -def verify_asset_payload(name, manifest_item, payload, content): - actual = content.encode("utf-8") - if len(actual) != manifest_item["bytes"]: - raise RuntimeError(f"{name} byte length does not match bundle manifest") - if hashlib.sha256(actual).hexdigest() != manifest_item["sha256"]: - raise RuntimeError(f"{name} sha256 does not match bundle manifest") - for field in ( - "version", - "source_repo", - "source_revision", - "contract_version", - "minimum_contract_version", - "install_path", - "mode", - ): - if payload.get(field) != manifest_item.get(field): - raise RuntimeError(f"{name} {field} does not match bundle manifest") - - -def release_name(manifest): - revision = manifest["source"]["revision"] - identity = str(manifest.get("bundle_id") or "") - suffix = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:8] - return f"{manifest['release_version']}-{revision[:12]}-{suffix}" - - -def validate_staged_python(release_dir): - for install_path in INSTALL_PATHS.values(): - path = release_dir / install_path - if path.suffix == ".py": - try: - ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) - except (OSError, SyntaxError) as exc: - raise RuntimeError( - f"staged Python asset failed validation: {install_path}" - ) from exc - - -def validate_release_hashes(release_dir, items): - for name, item in items.items(): - path = release_dir / INSTALL_PATHS[name] - try: - content = path.read_bytes() - except OSError as exc: - raise RuntimeError(f"release asset is missing: {name}") from exc - if len(content) != item["bytes"]: - raise RuntimeError(f"release asset byte length is invalid: {name}") - if hashlib.sha256(content).hexdigest() != item["sha256"]: - raise RuntimeError(f"release asset sha256 is invalid: {name}") - - -def pid_exists(pid): - try: - os.kill(int(pid), 0) - return True - except (OSError, TypeError, ValueError): - return False - - -@contextlib.contextmanager -def installation_lock(): - INSTALL_DIR.mkdir(parents=True, exist_ok=True) - lock_path = INSTALL_DIR / ".install.lock" - for attempt in range(2): - try: - descriptor = os.open( - str(lock_path), - os.O_CREAT | os.O_EXCL | os.O_WRONLY, - 0o600, - ) - with os.fdopen(descriptor, "w", encoding="utf-8") as handle: - handle.write(str(os.getpid())) - break - except FileExistsError: - try: - owner = int(lock_path.read_text(encoding="utf-8").strip()) - except (OSError, ValueError): - owner = None - if attempt == 0 and not pid_exists(owner): - lock_path.unlink(missing_ok=True) - continue - raise RuntimeError("another Napseer install or update is already running") - try: - yield - finally: - lock_path.unlink(missing_ok=True) - - -def atomic_symlink(path, target): - temporary = path.with_name(f".{path.name}.next-{os.getpid()}") - temporary.unlink(missing_ok=True) - os.symlink(target, temporary) - os.replace(temporary, path) - - -def preserve_legacy_path(path): - if path.is_symlink() or not path.exists(): - return - backup = path.with_name(f"{path.name}.pre-bundle") - if backup.exists() or backup.is_symlink(): - raise RuntimeError( - f"cannot preserve legacy install path because backup exists: {backup}" - ) - os.replace(path, backup) - - -def activate_release(release_dir): - current = INSTALL_DIR / "current" - if os.name == "nt": - for install_path in INSTALL_PATHS.values(): - target = INSTALL_DIR / install_path - content = (release_dir / install_path).read_text(encoding="utf-8") - write_file(target, content, executable=True) - nap_path = BIN_DIR / "nap" - write_file( - nap_path, - (release_dir / "nap_install.py").read_text(encoding="utf-8"), - executable=True, - ) - write_file( - BIN_DIR / "nap.cmd", - f'@echo off\r\n"{sys.executable}" "{nap_path}" %*\r\n', - executable=False, - ) - return nap_path - - for filename in ( - "nap_install.py", - "napseer_mcp_server.py", - "napseer_mcp_supervisor.py", - "napseer_spake2.py", - ): - path = INSTALL_DIR / filename - preserve_legacy_path(path) - atomic_symlink(path, f"current/{filename}") - terminal_path = INSTALL_DIR / "terminal" - preserve_legacy_path(terminal_path) - atomic_symlink(terminal_path, "current/terminal") - - BIN_DIR.mkdir(parents=True, exist_ok=True) - nap_path = BIN_DIR / "nap" - preserve_legacy_path(nap_path) - atomic_symlink( - nap_path, - os.path.relpath(INSTALL_DIR / "nap_install.py", BIN_DIR), - ) - atomic_symlink(current, os.path.relpath(release_dir, INSTALL_DIR)) - return nap_path - - -def install_assets(): - with installation_lock(): - manifest = bundle_manifest() - items = validated_manifest_items(manifest) - releases_dir = INSTALL_DIR / "releases" - releases_dir.mkdir(parents=True, exist_ok=True) - final_release = releases_dir / release_name(manifest) - staging = releases_dir / f".staging-{os.getpid()}-{time.time_ns()}" - staging.mkdir(mode=0o700) - scripts = {} - try: - for name in SCRIPT_NAMES: - manifest_item = items[name] - payload, content = fetch_script(name) - verify_asset_payload(name, manifest_item, payload, content) - target = staging / INSTALL_PATHS[name] - write_file(target, content, executable=True) - scripts[name] = { - "path": str(INSTALL_DIR / "current" / INSTALL_PATHS[name]), - "version": payload.get("version"), - "sha256": payload.get("sha256"), - "source_repo": payload.get("source_repo"), - "source_revision": payload.get("source_revision"), - "contract_version": payload.get("contract_version"), - } - validate_staged_python(staging) - write_file( - staging / "bundle-manifest.json", - json.dumps(manifest, indent=2, sort_keys=True) + "\n", - executable=False, - ) - if final_release.exists(): - shutil.rmtree(staging) - else: - os.replace(staging, final_release) - validate_release_hashes(final_release, items) - nap_path = activate_release(final_release) - except Exception: - if staging.exists(): - shutil.rmtree(staging) - raise - - return { - "status": "installed", - "message": "Napseer CLI and runtime bundle installed atomically.", - "bin": str(nap_path), - "install_dir": str(INSTALL_DIR), - "release_dir": str(final_release), - "bundle_id": manifest.get("bundle_id"), - "scripts": scripts, - "cli_distribution": { - "release_version": manifest.get("release_version"), - "contract_version": manifest["contract"]["current"], - "minimum_contract_version": manifest["contract"]["minimum_supported"], - "source_repo": manifest["source"]["repo"], - "source_revision": manifest["source"]["revision"], - "source_revision_status": manifest["source"]["revision_status"], - }, - "updated_project_paths": [], - "path_warning": path_warning(), - "next": ["nap init", "nap mcp status", "nap gateway setup"], - } - - -def runtime_assets_missing(): - missing = [] - for name in RUNTIME_SCRIPT_NAMES: - path = runtime_asset_path(name) - if not path.exists() or path.stat().st_size == 0: - missing.append(name) - return missing - - -def active_install_root(): - current = INSTALL_DIR / "current" - return current if current.exists() else INSTALL_DIR - - -def runtime_asset_path(name): - return active_install_root() / INSTALL_PATHS.get(name, name) - - -def require_runtime_assets(): - missing = runtime_assets_missing() - if missing: - raise RuntimeError( - "missing installed Napseer runtime scripts: " - f"{', '.join(missing)}. Run `nap mcp install` to repair them." - ) - - -def run_wrapper(command_name, args): - wrapper = runtime_asset_path("napseer_mcp_server.py") - require_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() - return { - "active_state_dir": str(preferred), - "active_auth_path": str(preferred / "auth.json"), - "state_dir": str(preferred), - "auth_path": str(preferred / "auth.json"), - "state_dir_exists": preferred.exists(), - "state_dir_message": ( - ".napseer directory is active." - if preferred.exists() - else ".napseer directory is not initialized." - ), - } - - -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 = runtime_asset_path("napseer_mcp_server.py") - require_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 with the vault/content passphrase.") - print(" nap gateway vault rotate-secret --kind memory") - print(" nap chat secret setup Configure chat secret state with the vault/content passphrase.") - 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] [--vault-passphrase TEXT]") - 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; setup uses the same vault/content passphrase.") - 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 print_root_help(): - print("Usage: nap [arguments]") - print() - print("Commands:") - width = max(len(name) for name, _summary in CANONICAL_COMMANDS) - for name, summary in CANONICAL_COMMANDS: - print(f" {name.ljust(width)} {summary}") - print() - print("Run `nap help ` for command-specific help.") - - -def print_command_help(command): - help_text = { - "init": ( - "Usage: nap init [--slug SLUG] [--name NAME] [--description TEXT]\n" - "Initialize the current folder and create its first Napseer project." - ), - "status": ( - "Usage: nap status\n" - "Show the current folder's account, project, and runtime state." - ), - "doctor": ( - "Usage: nap doctor\n" - "Run read-only checks and print exact repair commands." - ), - "auth": ( - "Usage: nap auth [status|login|repair]\n" - " login Authenticate an account; does not select a project.\n" - " status Show credential state without credential values.\n" - " repair Recover failed automatic credential renewal." - ), - "project": ( - "Usage: nap project [init|create|attach|claim|status|encryption]\n" - " init Initialize a fresh folder (same workflow as `nap init`).\n" - " create Create and bind a new project.\n" - " attach Select an existing project owned by the logged-in account.\n" - " claim Move an anonymous project into an authenticated account.\n" - " status Show the current folder binding." - ), - "mcp": ( - "Usage: nap mcp [status|install|update|serve]\n" - " status Show supervisor and worker installation state.\n" - " install Install or repair local MCP assets.\n" - " update Update local MCP assets from verified published sources.\n" - " serve Run the stable stdio supervisor." - ), - "gateway": ( - "Usage: nap gateway [setup|start|stop|restart|status|logs|vault|terminal|schedule|service]\n" - "Common lifecycle commands are setup, start, stop, restart, status, and logs.\n" - "Vault, terminal, schedule, and service are advanced capability groups." - ), - "update": ( - "Usage: nap update\n" - "Update the installed CLI and runtime bundle. Help never performs an update." - ), - "version": ( - "Usage: nap version\n" - "Show distribution contract and canonical source revision." - ), - "install": "Compatibility command. Use `nap mcp install`.", - "where": "Compatibility command. Use `nap doctor` or `nap mcp status`.", - "chat": "Compatibility command. Chat-secret operations are an advanced runtime workflow.", - "plan": "Compatibility command. Agent planning belongs to the authenticated MCP surface.", - "lineage": "Compatibility command. Agent lineage checks belong to the authenticated MCP surface.", - "agent": "Compatibility command. Agent workspace operations belong to the authenticated MCP surface.", - "feedback": "Compatibility command. Feedback administration is not a primary operator workflow.", - } - if command in {"help", "-h", "--help", ""}: - print_root_help() - return 0 - text = help_text.get(command) - if text is None: - raise RuntimeError(f"unknown help topic: {command}") - print(text) - return 0 - - -def mcp_status(): - worker = runtime_asset_path("napseer_mcp_server.py") - supervisor = runtime_asset_path("napseer_mcp_supervisor.py") - missing = runtime_assets_missing() - return { - "status": "ok" if not missing else "repair_required", - "supervisor_installed": supervisor.is_file(), - "worker_installed": worker.is_file(), - "runtime_assets_missing": missing, - "serve_command": f"{sys.executable} {supervisor}", - "next": None if not missing else "nap mcp install", - } - - -def handle_mcp(args): - subcommand = args[0] if args else "status" - rest = args[1:] - if subcommand == "status": - print_json(mcp_status()) - return 0 - if subcommand == "install": - print_json(install_assets()) - return 0 - if subcommand == "update": - result = install_assets() - print_json({**result, "status": "updated", "message": "Napseer MCP runtime is updated."}) - return 0 - if subcommand == "serve": - if rest: - raise RuntimeError("nap mcp serve does not accept arguments") - require_runtime_assets() - supervisor = runtime_asset_path("napseer_mcp_supervisor.py") - os.execv(sys.executable, [sys.executable, str(supervisor)]) - raise RuntimeError("unknown MCP command; use nap mcp status|install|update|serve") - - -def doctor_status(): - state = state_dir_status() - auth_path = pathlib.Path(state["active_auth_path"]) - auth = {} - auth_error = None - if auth_path.exists(): - try: - auth = json.loads(auth_path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - auth_error = "auth file is unreadable or invalid JSON" - missing = runtime_assets_missing() - issues = [] - if auth_error: - issues.append({"code": "auth_file_invalid", "message": auth_error, "next": "nap auth login"}) - elif not auth_path.exists(): - issues.append({"code": "project_not_initialized", "message": "current folder is not initialized", "next": "nap init"}) - elif not auth.get("project_id"): - issues.append({"code": "project_not_attached", "message": "no project is attached", "next": "nap project attach"}) - if missing: - issues.append({"code": "runtime_assets_missing", "message": "local MCP runtime is incomplete", "next": "nap mcp install"}) - if not runtime_asset_path("napseer_mcp_supervisor.py").is_file(): - issues.append({"code": "supervisor_missing", "message": "stable MCP supervisor is not installed", "next": "nap mcp install"}) - return { - "status": "ok" if not issues else "repair_required", - "checks": { - "state_directory": state["state_dir_exists"], - "auth_file": auth_path.exists() and auth_error is None, - "account_mode": auth.get("account_mode"), - "token_present": bool(auth.get("token")), - "refresh_present": bool(auth.get("refresh_token")), - "project_attached": bool(auth.get("project_id")), - "mcp_supervisor": runtime_asset_path("napseer_mcp_supervisor.py").is_file(), - "mcp_worker": runtime_asset_path("napseer_mcp_server.py").is_file(), - }, - "issues": issues, - } - - -def version_status(): - manifest_path = active_install_root() / "bundle-manifest.json" - if manifest_path.is_file(): - try: - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - source = manifest.get("source") or {} - contract = manifest.get("contract") or {} - return { - "release_version": manifest.get("release_version"), - "contract_version": contract.get("current"), - "minimum_contract_version": contract.get("minimum_supported"), - "source_repo": source.get("repo"), - "source_revision": source.get("revision"), - "source_revision_status": source.get("revision_status"), - "bundle_id": manifest.get("bundle_id"), - } - except (OSError, json.JSONDecodeError): - pass - return { - "release_version": CLI_RELEASE_VERSION, - "contract_version": CLI_DISTRIBUTION_CONTRACT_VERSION, - "minimum_contract_version": CLI_MINIMUM_CONTRACT_VERSION, - "source_repo": CLI_GENERATED_SOURCE_REPO, - "source_revision": None - if source_revision_status_for(CLI_GENERATED_SOURCE_REVISION) == "unresolved" - else CLI_GENERATED_SOURCE_REVISION, - "source_revision_status": source_revision_status_for(CLI_GENERATED_SOURCE_REVISION), - } - - -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 in HELP_TOKENS: - return print_command_help(args[0] if args else "") - if any(item in HELP_TOKENS for item in args): - return print_command_help(command) - if command == "init": - return run_wrapper("project", ["init", *args]) - if command == "doctor": - print_json(doctor_status()) - return 0 - if command == "mcp": - return handle_mcp(args) - if command == "version": - print_json(version_status()) - return 0 - if command == "install": - print("`nap install` is deprecated; use `nap mcp install`.", file=sys.stderr) - print_json(install_assets()) - return 0 - if command == "update": - result = install_assets() - 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": - if args and args[0] == "repair": - return run_wrapper("auth", ["refresh", *args[1:]]) - if args and args[0] == "refresh": - print("`nap auth refresh` is deprecated; use `nap auth repair`.", file=sys.stderr) - 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(runtime_asset_path("napseer_mcp_server.py")), - "supervisor": str(runtime_asset_path("napseer_mcp_supervisor.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 - 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)