diff --git a/README.md b/README.md index 64ad2b2..a9e6352 100644 --- a/README.md +++ b/README.md @@ -56,5 +56,19 @@ Authentication has one recovery verb: `nap auth repair`. Normal refresh is automatic. `nap gateway repair` creates the separate worker identity once; `--replace` is required to replace an existing identity. +Project-scoped MCP tools are zero-login by default. In a new workspace, the +first such tool enrolls an anonymous worker and creates its first project +automatically; `nap init` remains the explicit equivalent. Anonymous access +and refresh-token or SSH-key recovery continue without operator +authentication. Authentication through `nap project claim` is required only +after the anonymous account reaches its service limit (currently one project +or 100 stored nodes), or when its durable local enrollment identity is no +longer recoverable. + +In Bash, run commands as `nap update`, `nap auth login`, and so on. A leading +`!` is shell history expansion: `!nap update` replays the most recent command +whose text starts with `nap` and appends `update`; it does not invoke an +alternate Napseer update mode. + The direct wrapper remains the `nap` CLI implementation and an isolated protocol-debugging entrypoint. diff --git a/resources/scripts/nap_install.py b/resources/scripts/nap_install.py index 89c37b7..b8ebfb0 100644 --- a/resources/scripts/nap_install.py +++ b/resources/scripts/nap_install.py @@ -20,6 +20,7 @@ import ast import contextlib import json import hashlib +import http.client import os import pathlib import select @@ -47,6 +48,10 @@ SCRIPT_NAMES = ( "nap_install.py", ) HTTP_TIMEOUT_SECONDS = int(os.environ.get("NAPSEER_HTTP_TIMEOUT_SECONDS", "30")) +API_CONNECT_ATTEMPT_TIMEOUT_SECONDS = max( + 0.1, + float(os.environ.get("NAPSEER_CONNECT_ATTEMPT_TIMEOUT_SECONDS", "2")), +) LOCAL_HOST = "127.0.0.1" DEFAULT_GATEWAY_PORT = os.environ.get("NAPSEER_GATEWAY_PORT") RUNTIME_SCRIPT_NAMES = ( @@ -136,13 +141,85 @@ def api_proxy_url(parsed): return proxies.get(parsed.scheme) or proxies.get("all") +def connect_with_address_fallback( + address, + timeout=socket._GLOBAL_DEFAULT_TIMEOUT, + source_address=None, + *, + resolver=socket.getaddrinfo, + socket_factory=socket.socket, +): + """Connect without letting one unreachable DNS address consume the whole timeout.""" + host, port = address + addresses = list(resolver(host, port, 0, socket.SOCK_STREAM)) + addresses.sort(key=lambda item: 0 if item[0] == socket.AF_INET else 1) + if not addresses: + raise OSError("getaddrinfo returned no addresses") + + if timeout is socket._GLOBAL_DEFAULT_TIMEOUT or timeout is None: + attempt_timeout = API_CONNECT_ATTEMPT_TIMEOUT_SECONDS + established_timeout = None + else: + established_timeout = float(timeout) + attempt_timeout = min(established_timeout, API_CONNECT_ATTEMPT_TIMEOUT_SECONDS) + + last_error = None + for family, socktype, proto, _canonname, sockaddr in addresses: + sock = None + try: + sock = socket_factory(family, socktype, proto) + sock.settimeout(attempt_timeout) + if source_address: + sock.bind(source_address) + sock.connect(sockaddr) + sock.settimeout(established_timeout) + return sock + except OSError as exc: + last_error = exc + if sock is not None: + sock.close() + if last_error is not None: + raise last_error + raise OSError("connection failed without an address error") + + +class AddressFallbackHttpConnection(http.client.HTTPConnection): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._create_connection = connect_with_address_fallback + + +class AddressFallbackHttpsConnection(http.client.HTTPSConnection): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._create_connection = connect_with_address_fallback + + +class AddressFallbackHttpHandler(urllib.request.HTTPHandler): + def http_open(self, request): + return self.do_open(AddressFallbackHttpConnection, request) + + +class AddressFallbackHttpsHandler(urllib.request.HTTPSHandler): + def https_open(self, request): + return self.do_open( + AddressFallbackHttpsConnection, + request, + context=self._context, + ) + + 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})) + proxies = {} + if proxy: + proxies[parsed.scheme] = urllib.parse.urlunparse(proxy) + opener = urllib.request.build_opener( + urllib.request.ProxyHandler(proxies), + AddressFallbackHttpHandler(), + AddressFallbackHttpsHandler(), + ) return opener.open(request, timeout=timeout) @@ -748,6 +825,25 @@ def cli_option(args, *names, default=None): return default +def validate_cli_options(args, *, flags=(), value_options=(), command_label): + index = 0 + while index < len(args): + item = args[index] + if item in flags: + index += 1 + continue + if item in value_options: + if index + 1 >= len(args): + raise RuntimeError(f"{item} requires a value") + index += 2 + continue + raise RuntimeError( + f"unexpected argument {item!r} for `{command_label}`. " + "Bash expands `!nap`; if this followed `!nap update`, cancel it " + "and run `nap update` without `!`." + ) + + def start_service(kind, args): wrapper = runtime_asset_path("napseer_mcp_server.py") require_runtime_assets() @@ -1026,29 +1122,70 @@ def mcp_status(): worker = runtime_asset_path("napseer_mcp_server.py") supervisor = runtime_asset_path("napseer_mcp_supervisor.py") missing = runtime_assets_missing() - probe = mcp_runtime_probe() if not missing else {"status": "not_run", "transport": False, "read": False} + probe = mcp_runtime_probe() if not missing else { + "status": "not_run", + "transport": False, + "contract": False, + "local_state": False, + } + runtime_ready = bool( + not missing + and probe.get("transport") + and probe.get("contract") + and probe.get("local_state") + ) + lifecycle = mcp_lifecycle_summary(missing, probe) + if missing: + next_step = "nap mcp install" + elif not probe.get("transport") or not probe.get("contract"): + next_step = "nap mcp update; then restart the MCP client" + elif not runtime_ready: + next_step = "nap doctor" + else: + next_step = None return { - "status": "probe_ok" if not missing and probe.get("status") == "ok" else "repair_required", + "status": "probe_ok" if runtime_ready else "repair_required", "supervisor_installed": supervisor.is_file(), "worker_installed": worker.is_file(), "runtime_assets_missing": missing, "fresh_process_probe": probe, + "lifecycle": lifecycle, "client_connection": "not_observable", "message": ( - "A fresh MCP supervisor transport and authenticated read succeeded. Existing client connections are separate; restart the client if it reports Transport closed." - if probe.get("status") == "ok" - else "The installed MCP runtime did not pass a fresh-process probe." + "A fresh supervisor initialized, listed the contract, and read local status. Remote project access was not tested; existing client connections remain separate." + if runtime_ready + else "The installed MCP runtime did not pass its fresh local-process lifecycle probe." ), "serve_command": f"{sys.executable} {supervisor}", - "next": None if not missing else "nap mcp install", + "next": next_step, } -def mcp_runtime_probe(timeout_seconds=12): +def mcp_lifecycle_summary(missing, probe): + authentication = probe.get("authentication") or {} + project = probe.get("project") or {} + return { + "runtime": "missing" if missing else "installed", + "transport": "ready" if probe.get("transport") else "failed", + "contract": "ready" if probe.get("contract") else "failed", + "local_state": "ready" if probe.get("local_state") else "failed", + "authentication": "configured" if authentication.get("configured") else "unconfigured", + "project": "configured" if project.get("configured") else "unconfigured", + "remote_project_access": probe.get("remote_project_access") or "not_tested", + } + + +def mcp_runtime_probe(timeout_seconds=6): supervisor = runtime_asset_path("napseer_mcp_supervisor.py") worker = runtime_asset_path("napseer_mcp_server.py") if not supervisor.is_file() or not worker.is_file(): - return {"status": "not_installed", "transport": False, "read": False} + return { + "status": "not_installed", + "transport": False, + "contract": False, + "local_state": False, + } + deadline = time.monotonic() + max(0.1, float(timeout_seconds)) environment = os.environ.copy() environment["NAPSEER_PROJECT_ROOT"] = str(pathlib.Path.cwd()) process = subprocess.Popen( @@ -1060,50 +1197,115 @@ def mcp_runtime_probe(timeout_seconds=12): env=environment, ) + def send_message(message): + if process.stdin is None: + raise RuntimeError("MCP probe input is unavailable") + process.stdin.write(json.dumps(message) + "\n") + process.stdin.flush() + def request(request_id, method, params=None): message = {"jsonrpc": "2.0", "id": request_id, "method": method} if params is not None: message["params"] = params - process.stdin.write(json.dumps(message) + "\n") - process.stdin.flush() - ready, _, _ = select.select([process.stdout], [], [], timeout_seconds) + send_message(message) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("MCP probe total deadline exceeded") + ready, _, _ = select.select([process.stdout], [], [], remaining) if not ready: - raise TimeoutError("MCP probe timed out") + raise TimeoutError("MCP probe total deadline exceeded") return json.loads(process.stdout.readline()) + def tool_payload(response): + result = response.get("result") or {} + if result.get("isError"): + return None + structured = result.get("structuredContent") + if isinstance(structured, dict): + return structured + for item in result.get("content") or []: + if not isinstance(item, dict) or item.get("type") != "text": + continue + try: + parsed = json.loads(item.get("text") or "") + except (TypeError, ValueError): + continue + if isinstance(parsed, dict): + return parsed + return None + + stage = "initialize" + initialized_ok = False + contract_ok = False try: initialized = request(1, "initialize", { - "protocolVersion": "2025-03-26", + "protocolVersion": "2025-11-25", "capabilities": {}, "clientInfo": {"name": "nap-doctor", "version": CLI_RELEASE_VERSION}, }) - listed = request(2, "tools/list") - read = request(3, "tools/call", {"name": "nap_library_ls", "arguments": {}}) - tools = ((listed.get("result") or {}).get("tools") or []) - read_result = read.get("result") or {} - read_ok = not bool(read_result.get("isError")) initialized_ok = bool(initialized.get("result")) + stage = "initialized_notification" + send_message({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}) + stage = "tools_list" + listed = request(2, "tools/list") + tools = ((listed.get("result") or {}).get("tools") or []) + contract_ok = bool(tools) + stage = "local_status" + local_status = request(3, "tools/call", {"name": "nap_whoami", "arguments": {}}) + local_payload = tool_payload(local_status) + local_ok = isinstance(local_payload, dict) return { - "status": "ok" if initialized_ok and tools and read_ok else "failed", - "transport": bool(initialized_ok and tools), - "read": read_ok, + "status": "local_ready" if initialized_ok and contract_ok and local_ok else "failed", + "transport": initialized_ok, + "contract": contract_ok, + "local_state": local_ok, + "read": local_ok, + "read_scope": "local_status", "tool_count": len(tools), + "authentication": { + "configured": bool((local_payload or {}).get("has_token")), + "verified": False, + }, + "project": { + "configured": bool( + (local_payload or {}).get("project_configured") + or (local_payload or {}).get("project_id") + ), + "verified": False, + }, + "remote_project_access": "not_tested", } except Exception as exc: return { "status": "failed", - "transport": False, - "read": False, + "transport": initialized_ok, + "contract": contract_ok, + "local_state": False, "error_class": exc.__class__.__name__, + "failed_stage": stage, } finally: + if process.stdin is not None and not process.stdin.closed: + try: + process.stdin.close() + except OSError: + pass + if process.poll() is None: + shutdown_grace = max(0.0, min(0.25, deadline - time.monotonic())) + if shutdown_grace: + try: + process.wait(timeout=shutdown_grace) + except subprocess.TimeoutExpired: + process.terminate() + else: + process.terminate() if process.poll() is None: process.terminate() try: - process.wait(timeout=3) + process.wait(timeout=1) except subprocess.TimeoutExpired: process.kill() - process.wait(timeout=3) + process.wait(timeout=1) def handle_mcp(args): @@ -1140,25 +1342,46 @@ def doctor_status(): auth_error = "auth file is unreadable or invalid JSON" missing = runtime_assets_missing() issues = [] + setup = [] if auth_error: - issues.append({"code": "auth_file_invalid", "message": auth_error, "next": "nap auth login"}) + issues.append({"code": "auth_file_invalid", "message": auth_error, "next": "nap auth repair"}) elif not auth_path.exists(): - issues.append({"code": "project_not_initialized", "message": "current folder is not initialized", "next": "nap init"}) + setup.append({ + "code": "project_not_initialized", + "message": "current folder is not initialized; anonymous setup is available without login", + "next": "nap init or invoke a project-scoped MCP tool", + }) elif not auth.get("project_id"): - issues.append({"code": "project_not_attached", "message": "no project is attached", "next": "nap project attach"}) + setup.append({ + "code": "project_not_configured", + "message": "no project is configured for this folder; anonymous setup is available without login", + "next": "nap init", + }) 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"}) - probe = mcp_runtime_probe() if not missing else {"status": "not_run", "transport": False, "read": False} - if not missing and probe.get("status") != "ok": + probe = mcp_runtime_probe() if not missing else { + "status": "not_run", + "transport": False, + "contract": False, + "local_state": False, + } + probe_ok = bool(probe.get("transport") and probe.get("contract") and probe.get("local_state")) + if not missing and not probe_ok: issues.append({ - "code": "mcp_fresh_probe_failed", - "message": "fresh MCP transport or authenticated read failed", - "next": "nap mcp update; then restart the MCP client", + "code": "mcp_local_probe_failed", + "message": "fresh MCP transport, contract listing, or local status read failed", + "next": ( + "nap mcp update; then restart the MCP client" + if not probe.get("transport") + else "inspect the failed_stage field and local project binding" + ), }) + status = "repair_required" if issues else ("setup_available" if setup else "ok") return { - "status": "ok" if not issues else "repair_required", + "status": status, + "lifecycle": mcp_lifecycle_summary(missing, probe), "checks": { "state_directory": state["state_dir_exists"], "auth_file": auth_path.exists() and auth_error is None, @@ -1172,6 +1395,7 @@ def doctor_status(): "client_connection": "not_observable", }, "issues": issues, + "setup": setup, } @@ -1227,6 +1451,7 @@ def main(argv): print_json(version_status()) return 0 if command == "update": + validate_cli_options(args, command_label="nap update") result = install_assets() print_json({**result, "status": "updated", "message": "Napseer CLI and runtime scripts are updated."}) return 0 @@ -1241,6 +1466,20 @@ def main(argv): raise RuntimeError("unknown project command: bootstrap") return run_wrapper("project", args) if command == "auth": + if args and args[0] in {"login", "login-local", "operator-login"}: + validate_cli_options( + args[1:], + flags={"--no-browser"}, + value_options={ + "--frontend-url", + "--api-base-url", + "--base-url", + "--timeout", + "--timeout-seconds", + "--port", + }, + command_label="nap auth login", + ) if args and args[0] == "repair": return run_wrapper("auth", ["repair", *args[1:]]) if args and args[0] == "refresh": diff --git a/resources/scripts/napseer_mcp_server.py b/resources/scripts/napseer_mcp_server.py index a7ef87b..007ed3c 100755 --- a/resources/scripts/napseer_mcp_server.py +++ b/resources/scripts/napseer_mcp_server.py @@ -22,10 +22,12 @@ auth.json shape: import json import base64 +import concurrent.futures import getpass import hashlib import html import hmac +import http.client import os import pathlib import queue @@ -574,6 +576,12 @@ GATEWAY_LISTENER_ACTIVE_LANES = set() GATEWAY_LISTENER_UNSTABLE_UNTIL = 0.0 GATEWAY_LISTENER_STABILITY_LOCK = threading.RLock() AUTH_RENEW_LOCK = threading.Lock() +AUTH_STATE_LOCK = threading.RLock() +PROJECT_RESOLUTION_LOCK = threading.RLock() +MCP_MUTATION_LOCK = threading.RLock() +MCP_STDOUT_LOCK = threading.Lock() +MCP_REQUEST_LOCAL = threading.local() +MCP_MAX_WORKERS = max(2, int(os.environ.get("NAPSEER_MCP_MAX_WORKERS", "8"))) GATEWAY_LOG_LOCK = threading.Lock() GATEWAY_VAULT_SETUP_THREAD_STARTED = False GATEWAY_PERIODIC_LOG_STATE = {} @@ -800,26 +808,43 @@ def vault_exists(): def load_public_auth_file(): - if not AUTH_PATH.exists(): - return {} + with AUTH_STATE_LOCK: + if not AUTH_PATH.exists(): + return {} + try: + data = json.loads(AUTH_PATH.read_text(encoding="utf-8")) + except Exception as exc: + raise RuntimeError(f"failed to read {AUTH_PATH}: {exc}") from exc + if not isinstance(data, dict): + raise RuntimeError(f"{AUTH_PATH} must contain a JSON object") + return data + + +def atomic_write_auth_file(data): + AUTH_PATH.parent.mkdir(exist_ok=True) + chmod_best_effort(AUTH_PATH.parent, 0o700) + temporary = AUTH_PATH.with_name( + f".{AUTH_PATH.name}.{os.getpid()}.{threading.get_ident()}.tmp" + ) try: - data = json.loads(AUTH_PATH.read_text(encoding="utf-8")) - except Exception as exc: - raise RuntimeError(f"failed to read {AUTH_PATH}: {exc}") from exc - if not isinstance(data, dict): - raise RuntimeError(f"{AUTH_PATH} must contain a JSON object") - return data + temporary.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + chmod_best_effort(temporary, 0o600) + os.replace(temporary, AUTH_PATH) + chmod_best_effort(AUTH_PATH, 0o600) + finally: + try: + temporary.unlink() + except FileNotFoundError: + pass def write_public_auth(data): - public = {key: value for key, value in data.items() if key in AUTH_FILE_KEYS and value is not None} - if vault_exists() or data.get("gateway_auth_required"): - public["gateway_auth_required"] = True - public["vault_path"] = str(VAULT_PATH) - AUTH_PATH.parent.mkdir(exist_ok=True) - chmod_best_effort(AUTH_PATH.parent, 0o700) - AUTH_PATH.write_text(json.dumps(public, indent=2) + "\n", encoding="utf-8") - chmod_best_effort(AUTH_PATH, 0o600) + with AUTH_STATE_LOCK: + public = {key: value for key, value in data.items() if key in AUTH_FILE_KEYS and value is not None} + if vault_exists() or data.get("gateway_auth_required"): + public["gateway_auth_required"] = True + public["vault_path"] = str(VAULT_PATH) + atomic_write_auth_file(public) def read_vault(passphrase, return_key=False): @@ -2738,13 +2763,93 @@ def api_proxy_url(parsed): return proxies.get(parsed.scheme) or proxies.get("all") +def connect_with_address_fallback( + address, + timeout=socket._GLOBAL_DEFAULT_TIMEOUT, + source_address=None, + *, + resolver=socket.getaddrinfo, + socket_factory=socket.socket, + ipv4_only=None, +): + """Connect without letting one unreachable DNS address consume the whole timeout.""" + host, port = address + if ipv4_only is None: + ipv4_only = API_FORCE_IPV4 + resolver_family = socket.AF_INET if ipv4_only else 0 + addresses = list(resolver(host, port, resolver_family, socket.SOCK_STREAM)) + if ipv4_only: + addresses = [item for item in addresses if item[0] == socket.AF_INET] + else: + addresses.sort(key=lambda item: 0 if item[0] == socket.AF_INET else 1) + if not addresses: + suffix = " IPv4" if ipv4_only else "" + raise OSError(f"getaddrinfo returned no{suffix} addresses") + + if timeout is socket._GLOBAL_DEFAULT_TIMEOUT or timeout is None: + attempt_timeout = API_CONNECT_ATTEMPT_TIMEOUT_SECONDS + established_timeout = None + else: + established_timeout = float(timeout) + attempt_timeout = min(established_timeout, API_CONNECT_ATTEMPT_TIMEOUT_SECONDS) + + last_error = None + for family, socktype, proto, _canonname, sockaddr in addresses: + sock = None + try: + sock = socket_factory(family, socktype, proto) + sock.settimeout(attempt_timeout) + if source_address: + sock.bind(source_address) + sock.connect(sockaddr) + sock.settimeout(established_timeout) + return sock + except OSError as exc: + last_error = exc + if sock is not None: + sock.close() + if last_error is not None: + raise last_error + raise OSError("connection failed without an address error") + + +class AddressFallbackHttpConnection(http.client.HTTPConnection): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._create_connection = connect_with_address_fallback + + +class AddressFallbackHttpsConnection(http.client.HTTPSConnection): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._create_connection = connect_with_address_fallback + + +class AddressFallbackHttpHandler(urllib.request.HTTPHandler): + def http_open(self, request): + return self.do_open(AddressFallbackHttpConnection, request) + + +class AddressFallbackHttpsHandler(urllib.request.HTTPSHandler): + def https_open(self, request): + return self.do_open( + AddressFallbackHttpsConnection, + request, + context=self._context, + ) + + 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})) + proxies = {} + if proxy: + proxies[parsed.scheme] = urllib.parse.urlunparse(proxy) + opener = urllib.request.build_opener( + urllib.request.ProxyHandler(proxies), + AddressFallbackHttpHandler(), + AddressFallbackHttpsHandler(), + ) return opener.open(request, timeout=timeout) @@ -2775,7 +2880,9 @@ def connect_proxy_tunnel(proxy, target_host, target_port, timeout=20): proxy_port = proxy.port or (443 if proxy.scheme == "https" else 80) if proxy.scheme not in {"http", "https"}: raise RuntimeError("websocket proxy must use http:// or https://") - raw = configure_interactive_socket(socket.create_connection((proxy_host, proxy_port), timeout=timeout)) + raw = configure_interactive_socket( + connect_with_address_fallback((proxy_host, proxy_port), timeout=timeout) + ) sock = ssl.create_default_context().wrap_socket(raw, server_hostname=proxy_host) if proxy.scheme == "https" else raw target = f"{target_host}:{target_port}" request = ( @@ -2808,7 +2915,13 @@ def ws_connect(url, extra_headers=None): port = parsed.port or (443 if parsed.scheme == "wss" else 80) path = parsed.path + (f"?{parsed.query}" if parsed.query else "") proxy = normalize_proxy_url(websocket_proxy_url(parsed)) - raw = connect_proxy_tunnel(proxy, host, port, timeout=20) if proxy else configure_interactive_socket(socket.create_connection((host, port), timeout=20)) + raw = ( + connect_proxy_tunnel(proxy, host, port, timeout=20) + if proxy + else configure_interactive_socket( + connect_with_address_fallback((host, port), timeout=20) + ) + ) sock = ssl.create_default_context().wrap_socket(raw, server_hostname=host) if parsed.scheme == "wss" else raw configure_interactive_socket(sock) key = base64.b64encode(secrets.token_bytes(16)).decode("ascii") @@ -5385,6 +5498,16 @@ REFRESH_EXPIRES_AT = AUTH.get("refresh_expires_at") PROTOCOL_VERSION = "2025-11-25" SCHEDULE_SIGNATURE_NAMESPACE = "napseer-schedule-v1" HTTP_TIMEOUT_SECONDS = int(os.environ.get("NAPSEER_HTTP_TIMEOUT_SECONDS", "30")) +API_FORCE_IPV4 = os.environ.get("NAPSEER_FORCE_IPV4", "").strip().lower() in { + "1", + "true", + "yes", + "on", +} +API_CONNECT_ATTEMPT_TIMEOUT_SECONDS = max( + 0.1, + float(os.environ.get("NAPSEER_CONNECT_ATTEMPT_TIMEOUT_SECONDS", "2")), +) # Multi-line help for the `nap project` subcommand. Kept here (top- # level constant) so it stays in sync between the help branch and @@ -5438,8 +5561,9 @@ First-time flow: def send(message): - sys.stdout.write(json.dumps(message, separators=(",", ":")) + "\n") - sys.stdout.flush() + with MCP_STDOUT_LOCK: + sys.stdout.write(json.dumps(message, separators=(",", ":")) + "\n") + sys.stdout.flush() def rpc_result(request_id, result): @@ -5853,7 +5977,14 @@ def agent_scoped_folder(args): return f"{root}/{normalized.strip('/')}" +def raise_if_mcp_request_cancelled(): + cancelled = getattr(MCP_REQUEST_LOCAL, "cancelled", None) + if cancelled is not None and cancelled.is_set(): + raise SafeToolError("request_cancelled", "The MCP request was cancelled.") + + def request_json(method, path, payload=None, token_required=True, retry_auth=True, extra_headers=None): + raise_if_mcp_request_cancelled() if token_required: # `nap project attach` may run in a separate process while this MCP # server stays alive. Reload the project auth file before building @@ -5873,6 +6004,7 @@ def request_json(method, path, payload=None, token_required=True, retry_auth=Tru try: with api_urlopen(request, timeout=HTTP_TIMEOUT_SECONDS) as response: text = response.read().decode("utf-8") + raise_if_mcp_request_cancelled() return json.loads(text) if text else {} except (TimeoutError, socket.timeout) as exc: raise SafeToolError( @@ -5882,6 +6014,7 @@ def request_json(method, path, payload=None, token_required=True, retry_auth=Tru except urllib.error.HTTPError as exc: error_body = exc.read(65536).decode("utf-8", errors="replace") exc.close() + raise_if_mcp_request_cancelled() if exc.code == 401 and token_required and retry_auth: try: # The attach flow may have completed after this request was @@ -5897,7 +6030,7 @@ def request_json(method, path, payload=None, token_required=True, retry_auth=Tru retry_auth=False, extra_headers=extra_headers, ) - renew_auth() + renew_auth(stale_token=request_token) except Exception as renew_exc: raise safe_auth_refresh_error() from renew_exc return request_json(method, path, payload, token_required=True, retry_auth=False, extra_headers=extra_headers) @@ -5926,6 +6059,7 @@ def rest_query_string(params): def request_form_json(method, path, payload, token_required=False, extra_headers=None): + raise_if_mcp_request_cancelled() if token_required and not TOKEN: raise RuntimeError("NAPSEER_TOKEN is required") body = urllib.parse.urlencode(payload).encode("utf-8") @@ -5938,6 +6072,7 @@ def request_form_json(method, path, payload, token_required=False, extra_headers try: with api_urlopen(request, timeout=HTTP_TIMEOUT_SECONDS) as response: text = response.read().decode("utf-8") + raise_if_mcp_request_cancelled() return json.loads(text) if text else {} except (TimeoutError, socket.timeout) as exc: raise SafeToolError( @@ -6854,8 +6989,15 @@ def sign_with_auth_key(namespace, text): return sign_text(namespace, pathlib.Path(AUTH["ssh_key_path"]).expanduser(), text) +def has_existing_auth_signing_key(): + if AUTH.get("ssh_private_key") and AUTH.get("ssh_public_key"): + return True + key_path = pathlib.Path(AUTH["ssh_key_path"]).expanduser() + return key_path.is_file() and pathlib.Path(str(key_path) + ".pub").is_file() + + def renew_anonymous_auth_with_ssh(): - """Recover an anonymous worker using its durable local signing key.""" + """Recover a previously enrolled worker using its durable local signing key.""" public_key = auth_public_key() challenge = request_json( "POST", @@ -6957,10 +7099,15 @@ def _renew_auth_locked(): token_required=False, ) except RuntimeError as enrollment_refresh_error: - if oauth_refresh_error is not None: - raise safe_auth_refresh_error() from enrollment_refresh_error if AUTH.get("account_mode") in {None, "anonymous"}: return renew_anonymous_auth_with_ssh() + if has_existing_auth_signing_key(): + try: + return renew_anonymous_auth_with_ssh() + except Exception as ssh_recovery_error: + raise safe_auth_refresh_error() from ssh_recovery_error + if oauth_refresh_error is not None: + raise safe_auth_refresh_error() from enrollment_refresh_error raise token = refreshed["token"] worker = refreshed.get("worker") or {} @@ -6992,12 +7139,22 @@ def _renew_auth_locked(): return renew_anonymous_auth_with_ssh() -def renew_auth(): +def renew_auth(stale_token=None): # OAuth providers may rotate refresh tokens. Relay lanes run concurrently, # so serialize renewal and reload the latest token written by another lane - # before attempting a refresh. + # before attempting a refresh. If another lane already replaced the token + # that received a 401, reuse that state instead of rotating credentials a + # second time. with AUTH_RENEW_LOCK: refresh_public_auth_state() + if stale_token is not None and TOKEN and TOKEN != stale_token: + return { + "status": "already_renewed", + "method": "concurrent_refresh", + "token_expires_at": TOKEN_EXPIRES_AT, + "refresh_expires_at": REFRESH_EXPIRES_AT, + "project_id": DEFAULT_PROJECT_ID, + } return _renew_auth_locked() @@ -7236,6 +7393,11 @@ def operator_project_attach(args=None): def save_auth_file_credentials(updates): + with AUTH_STATE_LOCK: + return _save_auth_file_credentials_locked(updates) + + +def _save_auth_file_credentials_locked(updates): global AUTH, BASE_URL, TOKEN, DEFAULT_PROJECT_ID, TOKEN_EXPIRES_AT, REFRESH_TOKEN, REFRESH_EXPIRES_AT current_public = load_public_auth_file() current_public.update({key: value for key, value in updates.items() if value is not None}) @@ -7250,6 +7412,11 @@ def save_auth_file_credentials(updates): def save_auth(updates): + with AUTH_STATE_LOCK: + return _save_auth_locked(updates) + + +def _save_auth_locked(updates): global AUTH, BASE_URL, TOKEN, DEFAULT_PROJECT_ID, TOKEN_EXPIRES_AT, REFRESH_TOKEN, REFRESH_EXPIRES_AT, VAULT_SECRETS current_public = load_public_auth_file() cleaned = {key: value for key, value in updates.items() if value is not None} @@ -7265,10 +7432,7 @@ def save_auth(updates): AUTH = load_auth(public_override=current_public, secret_override=VAULT_SECRETS if gateway_is_unlocked() else {}) else: current_public.update(cleaned) - AUTH_PATH.parent.mkdir(exist_ok=True) - chmod_best_effort(AUTH_PATH.parent, 0o700) - AUTH_PATH.write_text(json.dumps(current_public, indent=2) + "\n", encoding="utf-8") - chmod_best_effort(AUTH_PATH, 0o600) + atomic_write_auth_file(current_public) AUTH = load_auth() BASE_URL = AUTH["base_url"].rstrip("/") TOKEN = AUTH["token"] @@ -7279,6 +7443,11 @@ def save_auth(updates): def refresh_public_auth_state(): + with AUTH_STATE_LOCK: + return _refresh_public_auth_state_locked() + + +def _refresh_public_auth_state_locked(): global AUTH, BASE_URL, TOKEN, DEFAULT_PROJECT_ID, TOKEN_EXPIRES_AT, REFRESH_TOKEN, REFRESH_EXPIRES_AT public = load_public_auth_file() AUTH = load_auth(public_override=public, secret_override=VAULT_SECRETS if gateway_is_unlocked() else {}) @@ -7659,20 +7828,41 @@ def chmod_best_effort(path, mode): def resolve_project_id(args): - # Project-scoped tools resolve their target before issuing an HTTP - # request. Reload auth here as well as in request_json so a long-running - # MCP worker observes `nap project attach` even when it started before a - # project was selected. - if AUTH_PATH.exists() or not DEFAULT_PROJECT_ID: - refresh_public_auth_state() - project_id = DEFAULT_PROJECT_ID - if not project_id: - raise SafeToolError( - "project_not_configured", - "No project is selected in this directory. Run `nap project attach` " - "for an existing project or `nap project create` for a new one.", - ) - return project_id + # Concurrent read calls can all arrive before a new workspace has an auth + # file. Serialize only the resolution/bootstrap transition; normal reads + # run concurrently once a project is selected. + with PROJECT_RESOLUTION_LOCK: + # Reload here as well as in request_json so a long-running MCP worker + # observes `nap project attach` performed by another process. + if AUTH_PATH.exists() or not DEFAULT_PROJECT_ID: + refresh_public_auth_state() + project_id = DEFAULT_PROJECT_ID + if not project_id: + slug = default_project_slug() + try: + initialized = bootstrap_project( + { + "slug": slug, + "name": default_project_name(slug), + "description": "Created automatically by the local Napseer MCP runtime.", + "encryption": "standard", + } + ) + except SafeToolError: + raise + except Exception as exc: + raise SafeToolError( + "project_bootstrap_failed", + "This directory has no Napseer project and automatic anonymous setup failed. " + "Run `nap doctor`, then retry; account login is not required unless an anonymous limit was reached.", + ) from exc + project_id = initialized.get("project_id") or DEFAULT_PROJECT_ID + if not project_id: + raise SafeToolError( + "project_bootstrap_failed", + "Automatic anonymous setup completed without selecting a project. Run `nap doctor` and retry.", + ) + return project_id def current_project_id(): @@ -7856,6 +8046,25 @@ def cli_flag(args, *names): return any(name in args for name in names) +def validate_cli_options(args, *, flags=(), value_options=(), command_label): + index = 0 + while index < len(args): + item = args[index] + if item in flags: + index += 1 + continue + if item in value_options: + if index + 1 >= len(args): + raise RuntimeError(f"{item} requires a value") + index += 2 + continue + raise RuntimeError( + f"unexpected argument {item!r} for `{command_label}`. " + "Bash expands `!nap`; if this followed `!nap update`, cancel it " + "and run `nap update` without `!`." + ) + + def cli_positionals(args): values = [] skip_next = False @@ -8944,43 +9153,19 @@ def list_project_nodes(args): if filters["active_only"]: base_query["include_archived"] = False base_query["archived_only"] = False - raw_q = str(base_query.get("q") or "").strip() - if raw_q and not base_query.get("cursor"): - query_analysis = analyze_search_query(raw_q) - collected = [] - diagnostics = {"server_queries": []} - for server_query in server_search_queries(query_analysis): - if len(unique_nodes(collected)) >= limit: - break - query = dict(base_query) - query["q"] = server_query - query["limit"] = max(1, limit - len(unique_nodes(collected))) - suffix = f"?{rest_query_string(query)}" - page = request_json("GET", f"/v1/projects/{project_id}/nodes{suffix}") - items = decrypt_nodes_for_return(filter_project_nodes(page.get("items", [])), project_id=project_id) - diagnostics["server_queries"].append({"q": server_query, "matched": len(items)}) - for node in items: - node = dict(node) - node["matched_query"] = server_query - collected.append(node) - items = apply_status_filters(unique_nodes(collected), filters)[:limit] - return { - **discovery_envelope( - project_id, - items, - {**args, "limit": limit}, - view=view, - diagnostics=diagnostics, - query_analysis=query_analysis, - warnings=view_warnings + filters["warnings"], - ), - } - query = rest_query_string(base_query) suffix = f"?{query}" if query else "" result = request_json("GET", f"/v1/projects/{project_id}/nodes{suffix}") items = decrypt_nodes_for_return(filter_project_nodes(result.get("items", [])), project_id=project_id) items = apply_status_filters(items, filters) + raw_q = str(base_query.get("q") or "").strip() + query_analysis = result.get("query_analysis") + diagnostics = result.get("diagnostics") + if raw_q and not base_query.get("cursor"): + query_analysis = query_analysis or analyze_search_query(raw_q) + diagnostics = diagnostics or { + "server_queries": [{"q": raw_q, "matched": len(items)}], + } return discovery_envelope( project_id, items[:limit], @@ -8988,7 +9173,8 @@ def list_project_nodes(args): view=view, next_cursor=result.get("next_cursor"), has_more=bool(result.get("next_cursor")), - query_analysis=result.get("query_analysis"), + diagnostics=diagnostics, + query_analysis=query_analysis, warnings=view_warnings + filters["warnings"], ) @@ -10571,16 +10757,6 @@ def unique_nodes(nodes): return items -def server_search_queries(query_analysis): - values = [query_analysis["raw"]] - if query_analysis["simplified_query"] and query_analysis["simplified_query"] != query_analysis["raw"]: - values.append(query_analysis["simplified_query"]) - values.extend(query_analysis["phrases"]) - if query_analysis["terms"]: - values.extend(query_analysis["terms"][:5]) - return unique_preserve_order([value for value in values if value])[:6] - - def search_memory(args): args = normalize_discovery_ranges(args) project_id = resolve_project_id(args) @@ -10629,21 +10805,20 @@ def search_memory(args): diagnostics["local_index"]["skipped_reason"] = "index_missing" if mode in {"hybrid", "server"} and (mode == "server" or encryption_state == "plaintext" or len(collected) < limit): - remaining = limit - for server_query in server_search_queries(query_analysis): - if len(unique_nodes(collected)) >= limit: - break - server_args = {**args, "q": server_query, "limit": remaining, "view": view} - server = list_project_nodes(server_args).get("items", []) - diagnostics["server_queries"].append({"q": server_query, "matched": len(server)}) - for node in server: - node = summarize_node(node, view=view) - node["search_source"] = "server" - node["matched_query"] = server_query - collected.append(node) - if server and "server" not in sources: - sources.append("server") - remaining = max(1, limit - len(unique_nodes(collected))) + # The backend already analyzes the raw query into important terms and + # ranks matches. Expanding query variants here (and again inside + # list_project_nodes) multiplied one discovery into many serial HTTP + # calls, especially on a new/sparse local index. + remaining = max(1, limit - len(unique_nodes(collected))) + server_result = list_project_nodes({**args, "q": q, "limit": remaining, "view": view}) + server = server_result.get("items", []) + diagnostics["server_queries"].append({"q": q, "matched": len(server)}) + for node in server: + node = summarize_node(node, view=view) + node["search_source"] = "server" + collected.append(node) + if server and "server" not in sources: + sources.append("server") items = [node for node in unique_nodes(collected) if node_in_updated_range(node, args)] items = apply_status_filters(items, filters)[:limit] @@ -14325,6 +14500,21 @@ def safe_http_error(exc, *, operation="request", body_text=""): service_code = error except (TypeError, ValueError): service_code = "" + anonymous_limit_messages = { + "anonymous_space_limit_reached": ( + "Anonymous space limit reached. Run `nap project claim` to authenticate and continue." + ), + "anonymous_node_limit_reached": ( + "Anonymous node limit reached. Run `nap project claim` to authenticate and continue." + ), + } + if service_code in anonymous_limit_messages: + return SafeToolError( + service_code, + anonymous_limit_messages[service_code], + status=status or None, + service_code=service_code, + ) return SafeToolError( f"http_{status}" if status else "http_error", f"Napseer {operation} failed with HTTP {status or 'error'}.", @@ -14406,19 +14596,20 @@ WORKER_BINDING_AUTH_KEYS = { def replace_public_auth_state(updates, *, clear_keys=()): - current_public = load_public_auth_file() - for key in clear_keys: - current_public.pop(key, None) - current_public.update( - { - key: value - for key, value in (updates or {}).items() - if value is not None and key in AUTH_FILE_KEYS - } - ) - write_public_auth(current_public) - refresh_public_auth_state() - return AUTH + with AUTH_STATE_LOCK: + current_public = load_public_auth_file() + for key in clear_keys: + current_public.pop(key, None) + current_public.update( + { + key: value + for key, value in (updates or {}).items() + if value is not None and key in AUTH_FILE_KEYS + } + ) + write_public_auth(current_public) + refresh_public_auth_state() + return AUTH def tool_category(name): @@ -15340,7 +15531,14 @@ def call_tool(name, args): started_at = time.time() tool_name = telemetry_safe_text(name, 96) try: - result = call_tool_impl(name, args) + if tool_side_effect(name) == "none": + result = call_tool_impl(name, args) + else: + # Remote writes and local lifecycle mutations share process-wide + # auth/index state. Keep their execution order explicit while + # allowing independent reads to proceed concurrently. + with MCP_MUTATION_LOCK: + result = call_tool_impl(name, args) send_telemetry_event_async( "mcp_tool_call", "mcp", @@ -15395,9 +15593,107 @@ def handle(message): return rpc_result(request_id, tool_result(result)) except Exception as exc: return rpc_result(request_id, tool_result({"error": safe_exception_payload(exc)}, is_error=True)) + if "id" not in message: + # JSON-RPC notifications are one-way, including extension methods the + # current worker does not recognize. + return None return rpc_error(request_id, -32601, "Method not found", method) +class McpRequestState: + def __init__(self, message): + self.message = message + self.cancelled = threading.Event() + self.future = None + + +class McpRequestRuntime: + """Bounded concurrent request dispatcher for the stdio worker.""" + + def __init__(self, handler=handle, sender=send, max_workers=MCP_MAX_WORKERS): + self.handler = handler + self.sender = sender + self._lock = threading.Lock() + self._requests = {} + self._closed = False + self._executor = concurrent.futures.ThreadPoolExecutor( + max_workers=max(2, int(max_workers)), + thread_name_prefix="napseer-mcp-tool", + ) + + @staticmethod + def _key(value): + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + + def _execute(self, key, state): + MCP_REQUEST_LOCAL.cancelled = state.cancelled + try: + response = self.handler(state.message) + except Exception: + response = rpc_error(state.message.get("id"), -32603, "Internal error") + finally: + with self._lock: + if self._requests.get(key) is state: + self._requests.pop(key, None) + try: + del MCP_REQUEST_LOCAL.cancelled + except AttributeError: + pass + if response is not None and not state.cancelled.is_set(): + self.sender(response) + + def cancel(self, request_id): + key = self._key(request_id) + with self._lock: + state = self._requests.get(key) + if state is None or state.message.get("method") == "initialize": + return False + state.cancelled.set() + future = state.future + if future is not None and future.cancel(): + self._requests.pop(key, None) + return True + + def dispatch(self, message): + if not isinstance(message, dict): + self.sender(rpc_error(None, -32600, "Invalid Request")) + return None + if "id" not in message: + if message.get("method") == "notifications/cancelled": + params = message.get("params") or {} + if isinstance(params, dict) and "requestId" in params: + self.cancel(params.get("requestId")) + return None + # Notifications are ordered on stdin and are intentionally + # handled before the next request is accepted. Any accidental + # handler return value is discarded because notifications never + # receive responses. + self.handler(message) + return None + + key = self._key(message.get("id")) + state = McpRequestState(message) + with self._lock: + if self._closed: + return None + if key in self._requests: + self.sender(rpc_error(message.get("id"), -32600, "Duplicate in-flight request id")) + return None + self._requests[key] = state + state.future = self._executor.submit(self._execute, key, state) + return state.future + + def close(self): + with self._lock: + self._closed = True + states = list(self._requests.values()) + for state in states: + state.cancelled.set() + if state.future is not None: + state.future.cancel() + self._executor.shutdown(wait=True, cancel_futures=True) + + def cli_main(argv): command = argv[1] if len(argv) > 1 else "" if command == "configure": @@ -15441,6 +15737,19 @@ Refresh is automatic during authenticated calls; use `repair` only for recovery. `login-local` remains a hidden compatibility alias for `login`.""") return if subcommand in {"login", "login-local", "operator-login"}: + validate_cli_options( + args, + flags={"--no-browser"}, + value_options={ + "--frontend-url", + "--api-base-url", + "--base-url", + "--timeout", + "--timeout-seconds", + "--port", + }, + command_label="nap auth login", + ) print(json.dumps(operator_account_login({ "frontend_url": cli_option(args, "--frontend-url", default=None), "api_base_url": cli_option(args, "--api-base-url", "--base-url", default=None), @@ -15882,15 +16191,19 @@ Refresh is automatic during authenticated calls; use `repair` only for recovery. def main(): - for line in sys.stdin: - if not line.strip(): - continue - try: - response = handle(json.loads(line)) - except Exception as exc: - response = rpc_error(None, -32700, "Parse error") - if response is not None: - send(response) + runtime = McpRequestRuntime() + try: + for line in sys.stdin: + if not line.strip(): + continue + try: + message = json.loads(line) + except (TypeError, ValueError): + send(rpc_error(None, -32700, "Parse error")) + continue + runtime.dispatch(message) + finally: + runtime.close() if __name__ == "__main__": diff --git a/resources/scripts/napseer_mcp_supervisor.py b/resources/scripts/napseer_mcp_supervisor.py index 1ce4e8f..6163968 100644 --- a/resources/scripts/napseer_mcp_supervisor.py +++ b/resources/scripts/napseer_mcp_supervisor.py @@ -1,31 +1,36 @@ #!/usr/bin/env python3 -"""Stable stdio supervisor for the generated Napseer MCP worker. +"""Stable, multiplexed stdio supervisor for the Napseer MCP worker. Codex keeps the stdio process it launches for the lifetime of a client -session. If that process exits, Codex keeps a closed transport and does not -respawn it. This supervisor owns the client-facing pipes and treats +session. This supervisor owns those client-facing pipes while treating ``napseer_mcp_server.py`` as a replaceable worker: -- a worker that exited between requests is restarted before the next request; -- a wrapper file replaced by ``nap update`` is reloaded before the next request; -- a worker that exits during a tool call is restarted, while the interrupted - call receives a bounded JSON-RPC error and is never replayed automatically. +- valid requests are forwarded without head-of-line blocking and responses + are correlated by JSON-RPC request id; +- notifications never create client-facing responses; +- cancellation releases the matching pending request and forwards the + notification to the worker; +- a worker exit fails every affected in-flight request once and never replays + it automatically; +- a replaced worker is reloaded at the next quiescent request boundary. The supervisor never logs MCP payloads, tool results, auth state, or child -stderr. Protocol stdout contains JSON-RPC responses only. +stderr. Protocol stdout contains JSON-RPC messages only. """ from __future__ import annotations +import concurrent.futures import json import os import pathlib -import selectors +import queue import signal import subprocess import sys import threading -from typing import BinaryIO +from dataclasses import dataclass, field +from typing import BinaryIO, Callable WORKER_PATH = pathlib.Path( @@ -36,9 +41,14 @@ WORKER_PATH = pathlib.Path( ).expanduser() RESPONSE_TIMEOUT_SECONDS = max( 1, - int(os.environ.get("NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS", "300")), + int(os.environ.get("NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS", "60")), +) +MAX_IN_FLIGHT_REQUESTS = max( + 2, + int(os.environ.get("NAPSEER_MCP_MAX_IN_FLIGHT_REQUESTS", "32")), ) WORKER_RESTARTED_ERROR = -32098 +_CANCELLED = object() class WorkerUnavailable(RuntimeError): @@ -50,22 +60,41 @@ def source_identity(path: pathlib.Path) -> tuple[int, int, int]: return stat.st_ino, stat.st_size, stat.st_mtime_ns -def request_id(message: bytes): +def parse_message(message: bytes): try: payload = json.loads(message) except (TypeError, ValueError): return None - if not isinstance(payload, dict): - return None - return payload.get("id") + return payload if isinstance(payload, dict) else None + + +def request_id(message: bytes): + payload = parse_message(message) + return payload.get("id") if payload is not None else None + + +def request_key(value) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + + +def request_method(message: bytes) -> str: + payload = parse_message(message) + return str(payload.get("method") or "") if payload is not None else "" def is_notification(message: bytes) -> bool: - try: - payload = json.loads(message) - except (TypeError, ValueError): - return False - return isinstance(payload, dict) and "id" not in payload + payload = parse_message(message) + return payload is not None and "id" not in payload + + +def cancelled_request_id(message: bytes): + payload = parse_message(message) + if payload is None or payload.get("method") != "notifications/cancelled": + return False, None + params = payload.get("params") or {} + if not isinstance(params, dict) or "requestId" not in params: + return False, None + return True, params.get("requestId") def worker_error(message: bytes) -> bytes: @@ -74,17 +103,35 @@ def worker_error(message: bytes) -> bytes: "id": request_id(message), "error": { "code": WORKER_RESTARTED_ERROR, - "message": "Napseer MCP worker restarted; retry this tool call.", + "message": ( + "Napseer MCP could not complete this request. The outcome may be " + "uncertain; inspect state before retrying a mutation." + ), }, } return json.dumps(payload, separators=(",", ":")).encode("utf-8") + b"\n" +@dataclass +class PendingRequest: + message: bytes + key: str + generation: int + response: queue.Queue = field(default_factory=lambda: queue.Queue(maxsize=1)) + + class Worker: - def __init__(self, path: pathlib.Path): + def __init__(self, path: pathlib.Path, unsolicited: Callable[[bytes], None] | None = None): self.path = path self.process: subprocess.Popen[bytes] | None = None self.identity: tuple[int, int, int] | None = None + self.generation = 0 + self.unsolicited = unsolicited + self._lifecycle_lock = threading.RLock() + self._stdin_lock = threading.Lock() + self._pending_lock = threading.Lock() + self._pending: dict[str, PendingRequest] = {} + self._reload_pending = False def _discard_stderr(self, stream: BinaryIO) -> None: try: @@ -93,35 +140,82 @@ class Worker: finally: stream.close() - def start(self) -> None: - self.stop() - try: - identity = source_identity(self.path) - process = subprocess.Popen( - [sys.executable, "-u", str(self.path)], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - bufsize=0, - ) - except (OSError, subprocess.SubprocessError) as exc: - raise WorkerUnavailable("unable to start Napseer MCP worker") from exc - if process.stdin is None or process.stdout is None or process.stderr is None: - process.kill() - process.wait() - raise WorkerUnavailable("Napseer MCP worker pipes are unavailable") - self.process = process - self.identity = identity - threading.Thread( - target=self._discard_stderr, - args=(process.stderr,), - daemon=True, - ).start() + def _pending_count(self, generation: int | None = None) -> int: + with self._pending_lock: + if generation is None: + return len(self._pending) + return sum(1 for item in self._pending.values() if item.generation == generation) - def stop(self) -> None: + def _signal(self, pending: PendingRequest, value) -> None: + try: + pending.response.put_nowait(value) + except queue.Full: + pass + + def _fail_generation(self, generation: int, error: WorkerUnavailable) -> None: + failed = [] + with self._pending_lock: + for key, pending in list(self._pending.items()): + if pending.generation != generation: + continue + self._pending.pop(key, None) + failed.append(pending) + for pending in failed: + self._signal(pending, error) + + def _deliver_response(self, generation: int, response: bytes) -> bool: + payload = parse_message(response) + if payload is None: + return False + normalized = response if response.endswith(b"\n") else response + b"\n" + if "id" in payload: + key = request_key(payload.get("id")) + with self._pending_lock: + pending = self._pending.get(key) + if pending is not None and pending.generation == generation: + self._pending.pop(key, None) + else: + pending = None + if pending is not None: + self._signal(pending, normalized) + return True + if "method" in payload and self.unsolicited is not None: + self.unsolicited(normalized) + return True + + def _read_stdout(self, process: subprocess.Popen[bytes], generation: int) -> None: + stream = process.stdout + if stream is None: + self._fail_generation(generation, WorkerUnavailable("Napseer MCP worker output is unavailable")) + return + invalid_response = False + try: + for response in stream: + if not self._deliver_response(generation, response): + invalid_response = True + break + except (OSError, ValueError): + pass + if invalid_response and process.poll() is None: + try: + process.terminate() + except OSError: + pass + self._fail_generation( + generation, + WorkerUnavailable( + "Napseer MCP worker returned invalid JSON" + if invalid_response + else "Napseer MCP worker exited before responding" + ), + ) + + def _stop_locked(self) -> None: process = self.process + generation = self.generation self.process = None self.identity = None + self._reload_pending = False if process is None: return if process.poll() is None: @@ -133,83 +227,187 @@ class Worker: process.wait() for stream in (process.stdin, process.stdout): if stream is not None: - stream.close() + try: + stream.close() + except OSError: + pass + self._fail_generation(generation, WorkerUnavailable("Napseer MCP worker stopped")) + + def start(self) -> None: + with self._lifecycle_lock: + self._stop_locked() + try: + identity = source_identity(self.path) + process = subprocess.Popen( + [sys.executable, "-u", str(self.path)], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + bufsize=0, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise WorkerUnavailable("unable to start Napseer MCP worker") from exc + if process.stdin is None or process.stdout is None or process.stderr is None: + process.kill() + process.wait() + raise WorkerUnavailable("Napseer MCP worker pipes are unavailable") + self.generation += 1 + generation = self.generation + self.process = process + self.identity = identity + threading.Thread( + target=self._discard_stderr, + args=(process.stderr,), + name=f"napseer-mcp-stderr-{generation}", + daemon=True, + ).start() + threading.Thread( + target=self._read_stdout, + args=(process, generation), + name=f"napseer-mcp-stdout-{generation}", + daemon=True, + ).start() + + def stop(self) -> None: + with self._lifecycle_lock: + self._stop_locked() def ensure_current(self) -> None: - try: - current_identity = source_identity(self.path) - except OSError as exc: - self.stop() - raise WorkerUnavailable("Napseer MCP worker is missing") from exc - if ( - self.process is None - or self.process.poll() is not None - or self.identity != current_identity - ): - self.start() + with self._lifecycle_lock: + try: + current_identity = source_identity(self.path) + except OSError as exc: + self._stop_locked() + raise WorkerUnavailable("Napseer MCP worker is missing") from exc + process_dead = self.process is None or self.process.poll() is not None + source_changed = self.identity is not None and self.identity != current_identity + if source_changed and not process_dead and self._pending_count(self.generation): + self._reload_pending = True + return + if process_dead or self.identity is None or source_changed or self._reload_pending: + self.start() - def _read_response(self) -> bytes: - process = self.process - if process is None or process.stdout is None: - raise WorkerUnavailable("Napseer MCP worker is not running") - selector = selectors.DefaultSelector() - try: - selector.register(process.stdout, selectors.EVENT_READ) - if not selector.select(RESPONSE_TIMEOUT_SECONDS): - raise WorkerUnavailable("Napseer MCP worker response timed out") - response = process.stdout.readline() - finally: - selector.close() - if not response: - raise WorkerUnavailable("Napseer MCP worker exited before responding") - try: - json.loads(response) - except ValueError as exc: - raise WorkerUnavailable("Napseer MCP worker returned invalid JSON") from exc - return response if response.endswith(b"\n") else response + b"\n" + def _write(self, process: subprocess.Popen[bytes], message: bytes) -> None: + with self._stdin_lock: + if self.process is not process or process.poll() is not None or process.stdin is None: + raise WorkerUnavailable("Napseer MCP worker is not running") + try: + process.stdin.write(message if message.endswith(b"\n") else message + b"\n") + process.stdin.flush() + except (BrokenPipeError, OSError) as exc: + raise WorkerUnavailable("Napseer MCP worker input closed") from exc - def exchange(self, message: bytes) -> bytes | None: + def submit(self, message: bytes) -> PendingRequest | None: self.ensure_current() - process = self.process - if process is None or process.stdin is None: + with self._lifecycle_lock: + process = self.process + generation = self.generation + if process is None: raise WorkerUnavailable("Napseer MCP worker is not running") - try: - process.stdin.write(message if message.endswith(b"\n") else message + b"\n") - process.stdin.flush() - except (BrokenPipeError, OSError) as exc: - raise WorkerUnavailable("Napseer MCP worker input closed") from exc if is_notification(message): + self._write(process, message) return None - return self._read_response() + + key = request_key(request_id(message)) + pending = PendingRequest(message=message, key=key, generation=generation) + with self._pending_lock: + if key in self._pending: + raise WorkerUnavailable("duplicate in-flight JSON-RPC request id") + if len(self._pending) >= MAX_IN_FLIGHT_REQUESTS: + raise WorkerUnavailable("too many in-flight Napseer MCP requests") + self._pending[key] = pending + try: + self._write(process, message) + except WorkerUnavailable: + with self._pending_lock: + if self._pending.get(key) is pending: + self._pending.pop(key, None) + raise + return pending + + def cancel(self, value) -> bool: + key = request_key(value) + with self._pending_lock: + pending = self._pending.get(key) + if pending is None or request_method(pending.message) == "initialize": + return False + self._pending.pop(key, None) + self._signal(pending, _CANCELLED) + return True + + def wait(self, pending: PendingRequest) -> bytes | None: + try: + result = pending.response.get(timeout=RESPONSE_TIMEOUT_SECONDS) + except queue.Empty as exc: + with self._pending_lock: + if self._pending.get(pending.key) is pending: + self._pending.pop(pending.key, None) + raise WorkerUnavailable("Napseer MCP worker response timed out") from exc + if result is _CANCELLED: + return None + if isinstance(result, WorkerUnavailable): + raise result + return result def supervise() -> int: - worker = Worker(WORKER_PATH) + output_lock = threading.Lock() + closing = threading.Event() + + def emit(response: bytes) -> None: + if closing.is_set(): + return + with output_lock: + sys.stdout.buffer.write(response if response.endswith(b"\n") else response + b"\n") + sys.stdout.buffer.flush() + + worker = Worker(WORKER_PATH, unsolicited=emit) + waiters = concurrent.futures.ThreadPoolExecutor( + max_workers=MAX_IN_FLIGHT_REQUESTS, + thread_name_prefix="napseer-mcp-request", + ) + + def finish(message: bytes, pending: PendingRequest) -> None: + try: + response = worker.wait(pending) + except WorkerUnavailable: + response = worker_error(message) + if response is not None: + emit(response) def stop_worker(_signum=None, _frame=None): + closing.set() worker.stop() + waiters.shutdown(wait=False, cancel_futures=True) raise SystemExit(0) signal.signal(signal.SIGTERM, stop_worker) signal.signal(signal.SIGINT, stop_worker) - for message in sys.stdin.buffer: - if not message.strip(): - continue - try: - response = worker.exchange(message) - except WorkerUnavailable: - worker.stop() + try: + for message in sys.stdin.buffer: + if not message.strip(): + continue + cancelled, value = cancelled_request_id(message) + if cancelled: + worker.cancel(value) + try: + worker.submit(message) + except WorkerUnavailable: + pass + continue try: - worker.start() + pending = worker.submit(message) except WorkerUnavailable: - pass - response = None if is_notification(message) else worker_error(message) - if response is not None: - sys.stdout.buffer.write(response) - sys.stdout.buffer.flush() - - worker.stop() + if not is_notification(message): + emit(worker_error(message)) + continue + if pending is not None: + waiters.submit(finish, message, pending) + finally: + closing.set() + worker.stop() + waiters.shutdown(wait=True, cancel_futures=True) return 0 diff --git a/tests/test_anonymous_auth_recovery.py b/tests/test_anonymous_auth_recovery.py index 8371577..fd16e50 100644 --- a/tests/test_anonymous_auth_recovery.py +++ b/tests/test_anonymous_auth_recovery.py @@ -69,6 +69,7 @@ def test_failed_anonymous_refresh_uses_ssh_recovery(): mod.DEFAULT_PROJECT_ID = "project" mod.REFRESH_TOKEN = "stale-refresh" mod.REFRESH_EXPIRES_AT = None + mod.refresh_public_auth_state = lambda: mod.AUTH mod.auth_public_key = lambda: "ssh-ed25519 test-public-key" mod.sign_with_auth_key = lambda *args: "test-signature" saved = [] @@ -106,7 +107,65 @@ def test_failed_anonymous_refresh_uses_ssh_recovery(): assert saved[0]["refresh_token"] == "recovered-refresh" +def test_failed_mixed_state_refresh_uses_ssh_recovery(): + mod = load_module() + mod.AUTH = { + **anonymous_auth(mod), + "account_mode": "operator_project", + "oauth_client_id": "nap-cli", + "project_id": "project", + "ssh_private_key": "test-private-key", + "ssh_public_key": "ssh-ed25519 test-public-key", + } + mod.BASE_URL = mod.AUTH["base_url"] + mod.DEFAULT_PROJECT_ID = "project" + mod.REFRESH_TOKEN = "stale-refresh" + mod.REFRESH_EXPIRES_AT = None + mod.refresh_public_auth_state = lambda: mod.AUTH + mod.auth_public_key = lambda: "ssh-ed25519 test-public-key" + mod.sign_with_auth_key = lambda *args: "test-signature" + saved = [] + mod.save_auth = lambda payload: saved.append(dict(payload)) + + def request_form(method, path, payload=None, token_required=False): + assert (method, path, token_required) == ("POST", "/v1/oauth/token", False) + raise RuntimeError("HTTP 401") + + def request(method, path, payload=None, token_required=False): + if path == "/v1/enrollment/refresh": + raise RuntimeError("HTTP 401") + if path == "/v1/enrollment/challenges": + return {"challenge_id": "challenge", "challenge_text": "sign-me"} + if path == "/v1/enrollment/verify": + return { + "token": { + "access_token": "recovered-access", + "refresh_token": "recovered-refresh", + "expires_at": "2099-01-01T00:00:00Z", + "refresh_expires_at": "2099-12-31T00:00:00Z", + }, + "worker": { + "id": "worker", + "agent_id": "agent-id", + "name": "agent", + "device_fingerprint": "device", + "root_path": "/repo", + "project_id": "project", + }, + } + raise AssertionError(path) + + mod.request_form_json = request_form + mod.request_json = request + result = mod.renew_auth() + + assert result["method"] == "anonymous_ssh_recovery" + assert result["project_id"] == "project" + assert saved[0]["refresh_token"] == "recovered-refresh" + + if __name__ == "__main__": test_initial_enrollment_persists_refresh_credential() test_failed_anonymous_refresh_uses_ssh_recovery() + test_failed_mixed_state_refresh_uses_ssh_recovery() print("ok: anonymous auth recovery tests passed") diff --git a/tests/test_cli_command_system.py b/tests/test_cli_command_system.py index c60f853..5e44b93 100644 --- a/tests/test_cli_command_system.py +++ b/tests/test_cli_command_system.py @@ -2,6 +2,8 @@ import importlib.util import hashlib import json import pathlib +import time +import pytest SCRIPT_PATH = ( @@ -114,6 +116,18 @@ def test_update_does_not_overwrite_repository_source(monkeypatch, capsys): assert json.loads(capsys.readouterr().out)["status"] == "updated" +def test_update_and_expanded_auth_tail_fail_before_side_effects(monkeypatch): + module = load_installer() + monkeypatch.setattr(module, "install_assets", fail_if_called) + monkeypatch.setattr(module, "run_wrapper", fail_if_called) + + with pytest.raises(RuntimeError, match="unexpected argument 'extra' for `nap update`"): + module.main(["nap", "update", "extra"]) + + with pytest.raises(RuntimeError, match="Bash expands `!nap`"): + module.main(["nap", "auth", "login", "project", "attach", "update"]) + + def test_doctor_reports_presence_without_exposing_values(tmp_path, monkeypatch): module = load_installer() state_dir = tmp_path / ".napseer" @@ -136,7 +150,18 @@ def test_doctor_reports_presence_without_exposing_values(tmp_path, monkeypatch): (install_dir / "napseer_mcp_supervisor.py").write_text("# supervisor\n", encoding="utf-8") monkeypatch.setattr(module, "INSTALL_DIR", install_dir) monkeypatch.setattr(module, "runtime_assets_missing", lambda: []) - monkeypatch.setattr(module, "mcp_runtime_probe", lambda: {"status": "ok", "transport": True, "read": True, "tool_count": 56}) + monkeypatch.setattr( + module, + "mcp_runtime_probe", + lambda: { + "status": "local_ready", + "transport": True, + "contract": True, + "local_state": True, + "read": True, + "tool_count": 56, + }, + ) monkeypatch.setattr( module, "state_dir_status", @@ -174,6 +199,153 @@ def test_doctor_does_not_create_state_directory(tmp_path, monkeypatch): assert not state_dir.exists() +def test_doctor_treats_missing_project_as_setup_not_runtime_repair(tmp_path, monkeypatch): + module = load_installer() + install_dir = tmp_path / "install" + install_dir.mkdir() + (install_dir / "napseer_mcp_server.py").write_text("# worker\n", encoding="utf-8") + (install_dir / "napseer_mcp_supervisor.py").write_text("# supervisor\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(module, "INSTALL_DIR", install_dir) + monkeypatch.setattr(module, "runtime_assets_missing", lambda: []) + monkeypatch.setattr( + module, + "mcp_runtime_probe", + lambda: { + "status": "local_ready", + "transport": True, + "contract": True, + "local_state": True, + "authentication": {"configured": False, "verified": False}, + "project": {"configured": False, "verified": False}, + }, + ) + + result = module.doctor_status() + + assert result["status"] == "setup_available" + assert result["issues"] == [] + assert result["setup"][0]["code"] == "project_not_initialized" + assert result["lifecycle"]["transport"] == "ready" + assert not (tmp_path / ".napseer").exists() + + +def test_mcp_status_separates_runtime_health_from_project_readiness(tmp_path, monkeypatch): + module = load_installer() + install_dir = tmp_path / "install" + install_dir.mkdir() + (install_dir / "napseer_mcp_server.py").write_text("# worker\n", encoding="utf-8") + (install_dir / "napseer_mcp_supervisor.py").write_text("# supervisor\n", encoding="utf-8") + monkeypatch.setattr(module, "INSTALL_DIR", install_dir) + monkeypatch.setattr(module, "runtime_assets_missing", lambda: []) + monkeypatch.setattr( + module, + "mcp_runtime_probe", + lambda: { + "status": "local_ready", + "transport": True, + "contract": True, + "local_state": True, + "tool_count": 56, + "authentication": {"configured": False, "verified": False}, + "project": {"configured": False, "verified": False}, + }, + ) + + result = module.mcp_status() + + assert result["status"] == "probe_ok" + assert result["lifecycle"]["runtime"] == "installed" + assert result["lifecycle"]["transport"] == "ready" + assert result["lifecycle"]["authentication"] == "unconfigured" + assert result["lifecycle"]["project"] == "unconfigured" + assert "not tested" in result["message"] + + +def write_probe_server(path, trace_path, delay=0): + path.write_text( + f"""\ +import json +import pathlib +import sys +import time + +TRACE = pathlib.Path({str(trace_path)!r}) +DELAY = {delay!r} + +for line in sys.stdin: + message = json.loads(line) + params = message.get("params") or {{}} + record = {{"method": message.get("method")}} + if message.get("method") == "initialize": + record["protocol_version"] = params.get("protocolVersion") + if message.get("method") == "tools/call": + record["tool"] = params.get("name") + with TRACE.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(record) + "\\n") + if "id" not in message: + continue + time.sleep(DELAY) + if message.get("method") == "initialize": + result = {{"protocolVersion": "2025-11-25", "capabilities": {{}}, "serverInfo": {{"name": "fake", "version": "1"}}}} + elif message.get("method") == "tools/list": + result = {{"tools": [{{"name": "nap_whoami"}}]}} + else: + payload = {{"project_configured": False, "has_token": False}} + result = {{"structuredContent": payload, "content": [{{"type": "text", "text": json.dumps(payload)}}], "isError": False}} + print(json.dumps({{"jsonrpc": "2.0", "id": message["id"], "result": result}}), flush=True) +""", + encoding="utf-8", + ) + + +def test_mcp_runtime_probe_uses_local_status_without_project_bootstrap(tmp_path, monkeypatch): + module = load_installer() + supervisor = tmp_path / "probe.py" + worker = tmp_path / "worker.py" + trace = tmp_path / "trace.jsonl" + write_probe_server(supervisor, trace) + worker.write_text("# exists\n", encoding="utf-8") + paths = { + "napseer_mcp_supervisor.py": supervisor, + "napseer_mcp_server.py": worker, + } + monkeypatch.setattr(module, "runtime_asset_path", lambda name: paths[name]) + + result = module.mcp_runtime_probe(timeout_seconds=2) + records = [json.loads(line) for line in trace.read_text(encoding="utf-8").splitlines()] + + assert result["status"] == "local_ready" + assert result["transport"] is True + assert result["local_state"] is True + assert records[0]["protocol_version"] == "2025-11-25" + assert [record.get("tool") for record in records if record.get("tool")] == ["nap_whoami"] + assert any(record["method"] == "notifications/initialized" for record in records) + + +def test_mcp_runtime_probe_timeout_is_one_total_deadline(tmp_path, monkeypatch): + module = load_installer() + supervisor = tmp_path / "slow-probe.py" + worker = tmp_path / "worker.py" + trace = tmp_path / "trace.jsonl" + write_probe_server(supervisor, trace, delay=0.15) + worker.write_text("# exists\n", encoding="utf-8") + paths = { + "napseer_mcp_supervisor.py": supervisor, + "napseer_mcp_server.py": worker, + } + monkeypatch.setattr(module, "runtime_asset_path", lambda name: paths[name]) + started = time.monotonic() + + result = module.mcp_runtime_probe(timeout_seconds=0.2) + + assert result["status"] == "failed" + assert result["error_class"] == "TimeoutError" + assert result["transport"] is True + assert result["contract"] is False + assert time.monotonic() - started < 0.35 + + def test_wrapper_dispatch_never_installs_missing_runtime(monkeypatch): module = load_installer() monkeypatch.setattr(module, "runtime_assets_missing", lambda: ["napseer_mcp_server.py"]) diff --git a/tests/test_cli_dispatch_consolidation.py b/tests/test_cli_dispatch_consolidation.py index edcf5a1..39291fd 100755 --- a/tests/test_cli_dispatch_consolidation.py +++ b/tests/test_cli_dispatch_consolidation.py @@ -36,6 +36,18 @@ def expect_unknown(mod, argv, expected_fragment): raise AssertionError(f"expected RuntimeError for {argv}") +def test_auth_login_rejects_history_expansion_tail_before_browser(): + mod = load_module() + mod.operator_account_login = lambda _args: (_ for _ in ()).throw( + AssertionError("malformed auth login must fail before browser login") + ) + expect_unknown( + mod, + ["napseer_mcp_server.py", "auth", "login", "project", "attach", "update"], + "Bash expands `!nap`", + ) + + def test_chat_secret_aliases_rejected(): mod = load_module() for args in [ @@ -214,6 +226,7 @@ def test_project_encryption_status_aliases_are_read_only_compatibility(): if __name__ == "__main__": + test_auth_login_rejects_history_expansion_tail_before_browser() test_chat_secret_aliases_rejected() test_lineage_aliases_rejected() test_plan_aliases_rejected() diff --git a/tests/test_connection_address_fallback.py b/tests/test_connection_address_fallback.py new file mode 100644 index 0000000..d23cebb --- /dev/null +++ b/tests/test_connection_address_fallback.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 +"""Regression tests for bounded DNS address fallback in the local wrapper.""" + +import importlib.util +import pathlib +import socket +import sys + + +def load_module(): + script_path = ( + pathlib.Path(__file__).resolve().parents[1] + / "resources" + / "scripts" + / "napseer_mcp_server.py" + ) + sys.path.insert(0, str(script_path.parent)) + spec = importlib.util.spec_from_file_location( + "napseer_mcp_server_address_fallback_test", script_path + ) + module = importlib.util.module_from_spec(spec) + assert spec and spec.loader + spec.loader.exec_module(module) + return module + + +class FakeSocket: + def __init__(self, family, attempts, failing_addresses): + self.family = family + self.attempts = attempts + self.failing_addresses = failing_addresses + self.timeouts = [] + self.closed = False + + def settimeout(self, value): + self.timeouts.append(value) + + def bind(self, _address): + return None + + def connect(self, address): + self.attempts.append((self.family, address[0])) + if address[0] in self.failing_addresses: + raise TimeoutError("unreachable test address") + + def close(self): + self.closed = True + + +def test_ipv4_is_preferred_and_failed_addresses_are_bounded(): + mod = load_module() + attempts = [] + sockets = [] + + def resolver(_host, port, _family, _socktype): + return [ + (socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("2001:db8::1", port, 0, 0)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.1", port)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.2", port)), + ] + + def factory(family, _socktype, _proto): + created = FakeSocket(family, attempts, {"192.0.2.1"}) + sockets.append(created) + return created + + connected = mod.connect_with_address_fallback( + ("api.example.test", 443), + timeout=30, + resolver=resolver, + socket_factory=factory, + ) + + assert connected is sockets[1] + assert attempts == [ + (socket.AF_INET, "192.0.2.1"), + (socket.AF_INET, "192.0.2.2"), + ] + assert sockets[0].timeouts == [mod.API_CONNECT_ATTEMPT_TIMEOUT_SECONDS] + assert sockets[0].closed + assert sockets[1].timeouts == [mod.API_CONNECT_ATTEMPT_TIMEOUT_SECONDS, 30.0] + + +def test_ipv4_only_never_attempts_ipv6(): + mod = load_module() + attempts = [] + resolver_families = [] + + def resolver(_host, port, family, _socktype): + resolver_families.append(family) + return [ + (socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("2001:db8::1", port, 0, 0)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.1", port)), + ] + + def factory(family, _socktype, _proto): + return FakeSocket(family, attempts, {"192.0.2.1"}) + + try: + mod.connect_with_address_fallback( + ("api.example.test", 443), + timeout=30, + resolver=resolver, + socket_factory=factory, + ipv4_only=True, + ) + except TimeoutError: + pass + else: + raise AssertionError("IPv4-only connection unexpectedly succeeded") + + assert resolver_families == [socket.AF_INET] + assert attempts == [(socket.AF_INET, "192.0.2.1")] + + +if __name__ == "__main__": + test_ipv4_is_preferred_and_failed_addresses_are_bounded() + test_ipv4_only_never_attempts_ipv6() + print("ok: connection address fallback tests passed") diff --git a/tests/test_installer_connection_address_fallback.py b/tests/test_installer_connection_address_fallback.py new file mode 100644 index 0000000..86b6d2b --- /dev/null +++ b/tests/test_installer_connection_address_fallback.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Regression tests for bounded DNS address fallback in the updater transport.""" + +import importlib.util +import pathlib +import socket + + +def load_module(): + script_path = ( + pathlib.Path(__file__).resolve().parents[1] + / "resources" + / "scripts" + / "nap_install.py" + ) + spec = importlib.util.spec_from_file_location( + "nap_install_address_fallback_test", script_path + ) + module = importlib.util.module_from_spec(spec) + assert spec and spec.loader + spec.loader.exec_module(module) + return module + + +class FakeSocket: + def __init__(self, family, attempts, failing_addresses): + self.family = family + self.attempts = attempts + self.failing_addresses = failing_addresses + self.timeouts = [] + self.closed = False + + def settimeout(self, value): + self.timeouts.append(value) + + def bind(self, _address): + return None + + def connect(self, address): + self.attempts.append((self.family, address[0])) + if address[0] in self.failing_addresses: + raise TimeoutError("unreachable test address") + + def close(self): + self.closed = True + + +def test_updater_prefers_ipv4_and_bounds_failed_addresses(): + mod = load_module() + attempts = [] + sockets = [] + + def resolver(_host, port, _family, _socktype): + return [ + (socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("2001:db8::1", port, 0, 0)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.1", port)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.2", port)), + ] + + def factory(family, _socktype, _proto): + created = FakeSocket(family, attempts, {"192.0.2.1"}) + sockets.append(created) + return created + + connected = mod.connect_with_address_fallback( + ("api.example.test", 443), + timeout=30, + resolver=resolver, + socket_factory=factory, + ) + + assert connected is sockets[1] + assert attempts == [ + (socket.AF_INET, "192.0.2.1"), + (socket.AF_INET, "192.0.2.2"), + ] + assert sockets[0].timeouts == [mod.API_CONNECT_ATTEMPT_TIMEOUT_SECONDS] + assert sockets[0].closed + assert sockets[1].timeouts == [mod.API_CONNECT_ATTEMPT_TIMEOUT_SECONDS, 30.0] + + +if __name__ == "__main__": + test_updater_prefers_ipv4_and_bounds_failed_addresses() + print("ok: updater connection address fallback tests passed") diff --git a/tests/test_live_auth_state_reload.py b/tests/test_live_auth_state_reload.py index b0672bd..9832f3e 100644 --- a/tests/test_live_auth_state_reload.py +++ b/tests/test_live_auth_state_reload.py @@ -55,13 +55,44 @@ def run(): current = {} mod.DEFAULT_PROJECT_ID = "stale-project" + bootstrap_calls = [] + + def bootstrap_project(args): + bootstrap_calls.append(dict(args)) + mod.DEFAULT_PROJECT_ID = "anonymous-project" + return {"status": "created", "project_id": "anonymous-project"} + + mod.bootstrap_project = bootstrap_project + assert mod.resolve_project_id({}) == "anonymous-project" + assert bootstrap_calls == [ + { + "slug": mod.default_project_slug(), + "name": mod.default_project_name(mod.default_project_slug()), + "description": "Created automatically by the local Napseer MCP runtime.", + "encryption": "standard", + } + ] + + current = {} + mod.DEFAULT_PROJECT_ID = None + + def limited_project(_args): + raise mod.SafeToolError( + "anonymous_space_limit_reached", + "Anonymous space limit reached. Run `nap project claim` to authenticate and continue.", + status=403, + service_code="anonymous_space_limit_reached", + ) + + mod.bootstrap_project = limited_project try: mod.resolve_project_id({}) except mod.SafeToolError as exc: - assert exc.code == "project_not_configured" - assert "nap project attach" in exc.safe_message + assert exc.code == "anonymous_space_limit_reached" + assert exc.service_code == "anonymous_space_limit_reached" + assert "nap project claim" in exc.safe_message else: - raise AssertionError("missing project selection must return an actionable safe error") + raise AssertionError("anonymous space limit must remain actionable") def open_success(request, timeout=None): requests.append(request) diff --git a/tests/test_mcp_search_latency.py b/tests/test_mcp_search_latency.py new file mode 100644 index 0000000..864e80e --- /dev/null +++ b/tests/test_mcp_search_latency.py @@ -0,0 +1,76 @@ +import importlib.util +import pathlib +import sys +import time + + +def load_module(): + script_path = ( + pathlib.Path(__file__).resolve().parents[1] + / "resources" + / "scripts" + / "napseer_mcp_server.py" + ) + sys.path.insert(0, str(script_path.parent)) + spec = importlib.util.spec_from_file_location( + "napseer_mcp_server_search_latency_test", script_path + ) + module = importlib.util.module_from_spec(spec) + assert spec and spec.loader + spec.loader.exec_module(module) + return module + + +def test_natural_language_discovery_uses_one_backend_search_request(tmp_path): + mod = load_module() + calls = [] + mod.INDEX_PATH = tmp_path / "missing-index.sqlite" + mod.resolve_project_id = lambda _args: "project-1" + mod.project_encryption_state = lambda _project_id: "standard" + + def request(method, path, payload=None, **_kwargs): + calls.append((method, path, payload)) + return {"items": [], "next_cursor": None, "query_analysis": {"mode": "terms"}} + + mod.request_json = request + + result = mod.discover_memory( + { + "q": "SIARA signer facturacion certificate API XML signature", + "limit": 20, + "view": "summary", + } + ) + + assert result["ok"] is True + assert len(calls) == 1 + assert calls[0][0] == "GET" + assert "/nodes?" in calls[0][1] + + +def test_sparse_index_search_latency_does_not_scale_with_query_variants(tmp_path): + mod = load_module() + calls = [] + mod.INDEX_PATH = tmp_path / "missing-index.sqlite" + mod.resolve_project_id = lambda _args: "project-1" + mod.project_encryption_state = lambda _project_id: "standard" + + def request(_method, _path, _payload=None, **_kwargs): + calls.append(time.monotonic()) + time.sleep(0.03) + return {"items": [], "next_cursor": None} + + mod.request_json = request + started = time.monotonic() + + mod.search_memory( + { + "q": "one two three four five six seven eight", + "limit": 20, + "view": "summary", + } + ) + + elapsed = time.monotonic() - started + assert len(calls) == 1 + assert elapsed < 0.15 diff --git a/tests/test_mcp_supervisor.py b/tests/test_mcp_supervisor.py index 713c45f..f2ee079 100644 --- a/tests/test_mcp_supervisor.py +++ b/tests/test_mcp_supervisor.py @@ -10,6 +10,7 @@ import time ROOT = pathlib.Path(__file__).resolve().parents[1] SUPERVISOR = ROOT / "resources" / "scripts" / "napseer_mcp_supervisor.py" +REAL_WORKER = ROOT / "resources" / "scripts" / "napseer_mcp_server.py" def write_worker(path: pathlib.Path, version: int) -> None: @@ -18,31 +19,64 @@ def write_worker(path: pathlib.Path, version: int) -> None: import json import os import sys +import threading +import time VERSION = {version} +WRITE_LOCK = threading.Lock() +CANCELLED = set() -for line in sys.stdin: - message = json.loads(line) - if message.get("method") == "notifications/initialized": - continue - if message.get("method") == "tools/call": + +def emit(payload): + with WRITE_LOCK: + print(json.dumps(payload), flush=True) + + +def handle(message): + method = message.get("method") + if method == "notifications/initialized": + return + if method == "notifications/cancelled": + CANCELLED.add((message.get("params") or {{}}).get("requestId")) + return + if method == "notifications/noisy": + emit({{"jsonrpc": "2.0", "id": None, "error": {{"code": -32601, "message": "noise"}}}}) + return + if "id" not in message: + return + if method == "tools/call": name = (message.get("params") or {{}}).get("name") + if name == "slow": + time.sleep(1) if name == "crash": + time.sleep(0.1) os._exit(17) result = {{"pid": os.getpid(), "version": VERSION, "name": name}} else: result = {{"version": VERSION}} - print(json.dumps({{"jsonrpc": "2.0", "id": message.get("id"), "result": result}}), flush=True) + if message.get("id") in CANCELLED: + return + emit({{"jsonrpc": "2.0", "id": message.get("id"), "result": result}}) + +for line in sys.stdin: + message = json.loads(line) + if "id" not in message: + handle(message) + continue + threading.Thread(target=handle, args=(message,), daemon=True).start() """, encoding="utf-8", ) class SupervisorClient: - def __init__(self, worker_path: pathlib.Path): + def __init__(self, worker_path: pathlib.Path, project_root: pathlib.Path | None = None): environment = os.environ.copy() environment["NAPSEER_MCP_WORKER_PATH"] = str(worker_path) environment["NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS"] = "5" + environment["NAPSEER_TELEMETRY"] = "0" + if project_root is not None: + environment["NAPSEER_PROJECT_ROOT"] = str(project_root) self.process = subprocess.Popen( [sys.executable, str(SUPERVISOR)], stdin=subprocess.PIPE, @@ -52,21 +86,40 @@ class SupervisorClient: ) def request(self, request_id: int, method: str, params=None): + self.send(request_id, method, params) + return self.receive() + + def send(self, request_id: int, method: str, params=None): assert self.process.stdin is not None - assert self.process.stdout is not None message = {"jsonrpc": "2.0", "id": request_id, "method": method} if params is not None: message["params"] = params self.process.stdin.write(json.dumps(message).encode("utf-8") + b"\n") self.process.stdin.flush() - readable, _, _ = select.select([self.process.stdout], [], [], 8) + + def notify(self, method: str, params=None): + assert self.process.stdin is not None + message = {"jsonrpc": "2.0", "method": method} + if params is not None: + message["params"] = params + self.process.stdin.write(json.dumps(message).encode("utf-8") + b"\n") + self.process.stdin.flush() + + def receive(self, timeout=8): + assert self.process.stdout is not None + readable, _, _ = select.select([self.process.stdout], [], [], timeout) assert readable, "supervisor did not return a response" return json.loads(self.process.stdout.readline()) def close(self): if self.process.poll() is None: - self.process.terminate() - self.process.wait(timeout=5) + assert self.process.stdin is not None + self.process.stdin.close() + try: + self.process.wait(timeout=5) + except subprocess.TimeoutExpired: + self.process.terminate() + self.process.wait(timeout=5) def test_restarts_worker_killed_between_requests(tmp_path): @@ -86,6 +139,124 @@ def test_restarts_worker_killed_between_requests(tmp_path): client.close() +def test_concurrent_requests_are_correlated_without_head_of_line_blocking(tmp_path): + worker_path = tmp_path / "worker.py" + write_worker(worker_path, 1) + client = SupervisorClient(worker_path) + try: + client.send(1, "tools/call", {"name": "slow", "arguments": {}}) + time.sleep(0.05) + client.send(2, "tools/call", {"name": "fast", "arguments": {}}) + + first = client.receive(timeout=0.75) + second = client.receive(timeout=2) + + assert first["id"] == 2 + assert first["result"]["name"] == "fast" + assert second["id"] == 1 + assert second["result"]["name"] == "slow" + finally: + client.close() + + +def test_notification_response_cannot_poison_the_next_request(tmp_path): + worker_path = tmp_path / "worker.py" + write_worker(worker_path, 1) + client = SupervisorClient(worker_path) + try: + client.notify("notifications/noisy") + time.sleep(0.05) + + response = client.request(7, "tools/list") + + assert response["id"] == 7 + assert response["result"]["version"] == 1 + finally: + client.close() + + +def test_cancellation_drops_the_cancelled_and_late_worker_response(tmp_path): + worker_path = tmp_path / "worker.py" + write_worker(worker_path, 1) + client = SupervisorClient(worker_path) + try: + client.send(1, "tools/call", {"name": "slow", "arguments": {}}) + time.sleep(0.05) + client.notify( + "notifications/cancelled", + {"requestId": 1, "reason": "test cancellation"}, + ) + client.send(2, "tools/call", {"name": "fast", "arguments": {}}) + + response = client.receive(timeout=0.75) + assert response["id"] == 2 + + time.sleep(1.05) + follow_up = client.request(3, "tools/list") + assert follow_up["id"] == 3 + finally: + client.close() + + +def test_worker_crash_fails_every_in_flight_request_once_then_recovers(tmp_path): + worker_path = tmp_path / "worker.py" + write_worker(worker_path, 1) + client = SupervisorClient(worker_path) + try: + client.send(1, "tools/call", {"name": "slow", "arguments": {}}) + client.send(2, "tools/call", {"name": "crash", "arguments": {}}) + + failed = [client.receive(timeout=2), client.receive(timeout=2)] + + assert {item["id"] for item in failed} == {1, 2} + assert all(item["error"]["code"] == -32098 for item in failed) + + recovered = client.request(3, "tools/list") + assert recovered["id"] == 3 + assert recovered["result"]["version"] == 1 + assert client.process.poll() is None + finally: + client.close() + + +def test_real_supervisor_worker_protocol_lifecycle(tmp_path): + client = SupervisorClient(REAL_WORKER, project_root=tmp_path) + try: + initialized = client.request( + 1, + "initialize", + { + "protocolVersion": "2025-11-25", + "capabilities": {}, + "clientInfo": {"name": "protocol-test", "version": "1"}, + }, + ) + client.notify("notifications/initialized", {}) + client.send(2, "tools/list") + client.send(3, "tools/call", {"name": "nap_whoami", "arguments": {}}) + + responses = {item["id"]: item for item in (client.receive(), client.receive())} + + assert initialized["result"]["protocolVersion"] == "2025-11-25" + tool_names = {tool["name"] for tool in responses[2]["result"]["tools"]} + assert "nap_whoami" in tool_names + assert responses[3]["result"]["isError"] is False + + client.notify("notifications/unknown", {"proof": "one-way"}) + follow_up = client.request( + 4, + "tools/call", + {"name": "nap_whoami", "arguments": {}}, + ) + assert follow_up["id"] == 4 + assert follow_up["result"]["isError"] is False + finally: + client.close() + + assert client.process.returncode == 0 + assert not (tmp_path / ".napseer").exists() + + def test_reloads_replaced_worker_between_requests(tmp_path): worker_path = tmp_path / "worker.py" write_worker(worker_path, 1) diff --git a/tests/test_mcp_worker_lifecycle.py b/tests/test_mcp_worker_lifecycle.py new file mode 100644 index 0000000..64a16e6 --- /dev/null +++ b/tests/test_mcp_worker_lifecycle.py @@ -0,0 +1,244 @@ +import importlib.util +import pathlib +import sys +import threading +import time + + +def load_module(): + script_path = ( + pathlib.Path(__file__).resolve().parents[1] + / "resources" + / "scripts" + / "napseer_mcp_server.py" + ) + sys.path.insert(0, str(script_path.parent)) + spec = importlib.util.spec_from_file_location( + "napseer_mcp_server_lifecycle_test", script_path + ) + module = importlib.util.module_from_spec(spec) + assert spec and spec.loader + spec.loader.exec_module(module) + return module + + +def wait_until(predicate, timeout=2): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(0.01) + raise AssertionError("condition was not reached before timeout") + + +def test_worker_runtime_completes_fast_read_before_slow_read(): + mod = load_module() + emitted = [] + emitted_lock = threading.Lock() + + def handler(message): + if message["id"] == 1: + time.sleep(0.4) + return mod.rpc_result(message["id"], {"done": True}) + + def emit(response): + with emitted_lock: + emitted.append(response) + + runtime = mod.McpRequestRuntime(handler=handler, sender=emit, max_workers=4) + try: + runtime.dispatch({"jsonrpc": "2.0", "id": 1, "method": "tools/call"}) + runtime.dispatch({"jsonrpc": "2.0", "id": 2, "method": "tools/call"}) + + wait_until(lambda: len(emitted) == 2) + + assert [response["id"] for response in emitted] == [2, 1] + finally: + runtime.close() + + +def test_worker_runtime_cancellation_suppresses_unused_result(): + mod = load_module() + emitted = [] + + def handler(message): + if message["id"] == 1: + time.sleep(0.3) + return mod.rpc_result(message["id"], {"done": True}) + + runtime = mod.McpRequestRuntime(handler=handler, sender=emitted.append, max_workers=2) + try: + runtime.dispatch({"jsonrpc": "2.0", "id": 1, "method": "tools/call"}) + runtime.dispatch( + { + "jsonrpc": "2.0", + "method": "notifications/cancelled", + "params": {"requestId": 1, "reason": "unused"}, + } + ) + runtime.dispatch({"jsonrpc": "2.0", "id": 2, "method": "tools/call"}) + + wait_until(lambda: any(response["id"] == 2 for response in emitted)) + time.sleep(0.35) + + assert [response["id"] for response in emitted] == [2] + finally: + runtime.close() + + +def test_initialize_cannot_be_cancelled(): + mod = load_module() + emitted = [] + + def handler(message): + time.sleep(0.1) + return mod.rpc_result(message["id"], {"initialized": True}) + + runtime = mod.McpRequestRuntime(handler=handler, sender=emitted.append, max_workers=2) + try: + runtime.dispatch({"jsonrpc": "2.0", "id": 1, "method": "initialize"}) + runtime.dispatch( + { + "jsonrpc": "2.0", + "method": "notifications/cancelled", + "params": {"requestId": 1}, + } + ) + + wait_until(lambda: len(emitted) == 1) + + assert emitted[0]["id"] == 1 + finally: + runtime.close() + + +def test_unknown_notification_never_returns_a_response(): + mod = load_module() + + response = mod.handle( + {"jsonrpc": "2.0", "method": "notifications/unknown", "params": {}} + ) + + assert response is None + + +def test_mutating_tools_are_serialized_while_reads_can_overlap(monkeypatch): + mod = load_module() + active = 0 + max_active = 0 + state_lock = threading.Lock() + + def fake_call(_name, _args): + nonlocal active, max_active + with state_lock: + active += 1 + max_active = max(max_active, active) + time.sleep(0.1) + with state_lock: + active -= 1 + return {"ok": True} + + monkeypatch.setattr(mod, "call_tool_impl", fake_call) + + read_threads = [ + threading.Thread(target=mod.call_tool, args=("nap_whoami", {})) + for _ in range(2) + ] + for thread in read_threads: + thread.start() + for thread in read_threads: + thread.join() + assert max_active == 2 + + max_active = 0 + write_threads = [ + threading.Thread(target=mod.call_tool, args=("nap_create_node", {})) + for _ in range(2) + ] + for thread in write_threads: + thread.start() + for thread in write_threads: + thread.join() + assert max_active == 1 + + +def test_concurrent_first_reads_share_one_project_bootstrap(tmp_path, monkeypatch): + mod = load_module() + mod.AUTH_PATH = tmp_path / "missing-auth.json" + mod.DEFAULT_PROJECT_ID = None + monkeypatch.setattr(mod, "refresh_public_auth_state", lambda: None) + calls = [] + + def bootstrap(_args): + calls.append("bootstrap") + time.sleep(0.1) + mod.DEFAULT_PROJECT_ID = "project-1" + return {"project_id": "project-1"} + + monkeypatch.setattr(mod, "bootstrap_project", bootstrap) + results = [] + threads = [ + threading.Thread(target=lambda: results.append(mod.resolve_project_id({}))) + for _ in range(3) + ] + + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert results == ["project-1"] * 3 + assert calls == ["bootstrap"] + + +def test_concurrent_auth_updates_are_atomic_and_merge_fields(tmp_path): + mod = load_module() + mod.AUTH_DIR = tmp_path / ".napseer" + mod.AUTH_PATH = mod.AUTH_DIR / "auth.json" + mod.VAULT_PATH = mod.AUTH_DIR / "vault.json" + mod.AUTH_DIR.mkdir() + mod.write_public_auth({"base_url": "https://api.example.test"}) + barrier = threading.Barrier(3) + + def update(values): + barrier.wait() + mod.save_auth(values) + + threads = [ + threading.Thread(target=update, args=({"project_slug": "demo"},)), + threading.Thread(target=update, args=({"project_name": "Demo"},)), + ] + for thread in threads: + thread.start() + barrier.wait() + for thread in threads: + thread.join() + + saved = mod.load_public_auth_file() + assert saved["project_slug"] == "demo" + assert saved["project_name"] == "Demo" + assert not list(mod.AUTH_DIR.glob(".auth.json.*.tmp")) + + +def test_auth_renewal_reuses_token_refreshed_by_another_request(monkeypatch): + mod = load_module() + mod.TOKEN = "fresh-access" + mod.TOKEN_EXPIRES_AT = "future" + mod.REFRESH_EXPIRES_AT = "later" + mod.DEFAULT_PROJECT_ID = "project-1" + monkeypatch.setattr(mod, "refresh_public_auth_state", lambda: None) + monkeypatch.setattr( + mod, + "_renew_auth_locked", + lambda: (_ for _ in ()).throw(AssertionError("must not rotate again")), + ) + + result = mod.renew_auth(stale_token="stale-access") + + assert result == { + "status": "already_renewed", + "method": "concurrent_refresh", + "token_expires_at": "future", + "refresh_expires_at": "later", + "project_id": "project-1", + } diff --git a/tests/test_project_create_oauth.py b/tests/test_project_create_oauth.py index e98641b..116212e 100644 --- a/tests/test_project_create_oauth.py +++ b/tests/test_project_create_oauth.py @@ -2,6 +2,7 @@ """Regression tests for account-login -> project-create authorization.""" import importlib.util +import json import pathlib import sys @@ -125,9 +126,40 @@ def test_server_scope_denial_is_actionable(): raise AssertionError("scope denial should return an actionable error") +def test_anonymous_limits_are_actionable_without_exposing_response_details(): + mod = load_module() + + class HttpError: + code = 403 + + for code, label in ( + ("anonymous_space_limit_reached", "space"), + ("anonymous_node_limit_reached", "node"), + ): + error = mod.safe_http_error( + HttpError(), + operation="POST request", + body_text=json.dumps( + { + "error": { + "code": code, + "message": "server detail must not be forwarded", + } + } + ), + ) + assert error.code == code + assert error.status == 403 + assert error.service_code == code + assert label in error.safe_message + assert "nap project claim" in error.safe_message + assert "server detail" not in error.safe_message + + if __name__ == "__main__": test_account_login_requests_create_and_runtime_scopes() test_legacy_account_scope_fails_before_http_with_recovery() test_successful_create_transitions_local_state_to_project_mode() test_server_scope_denial_is_actionable() + test_anonymous_limits_are_actionable_without_exposing_response_details() print("ok: project create OAuth lifecycle passed")