docs: align agent discovery and project help with configured access

README.md

@@ -25,9 +25,16 @@ Current state:
   `tools/list`, and authenticated read probe. They deliberately report an
   existing client connection as `not_observable`; a successful fresh probe
   does not claim that a previously opened Codex transport is connected.
-- Operator OAuth credentials remain in `.napseer/auth.json`. A repaired local
+- `nap auth login` stores the shared account session in the user data folder
-  gateway uses a separate `.napseer/gateway-auth.json` worker identity, so
+  (`~/.local/share/napseer/credentials/default.json` on Linux; platform user data directories
-  relay token renewal cannot rotate or overwrite the operator/MCP session.
+  on macOS/Windows). `NAPSEER_USER_DATA_DIR` overrides that location.
+  `nap auth login --project` selects a local override. Legacy repository
+  credentials remain an explicit override until `nap auth migrate` moves them.
+  Gateway identities remain separate in `.napseer/gateway-auth.json`.
+- `.napseer/project.json` is the commit-safe project locator. It contains only
+  the schema, API origin, project UUID, and slug; it never contains tokens,
+  account or worker identity, encryption state, keys, passphrases, or claim
+  links. A fresh clone uses it to attach the intended project.
 - CLI terminal and schedule operations call the running gateway's
   loopback-only, CSRF-protected control API. PTYs therefore live in the daemon
   instead of disappearing when a CLI subprocess exits.
@@ -46,24 +53,63 @@ directory, add a trusted project `.codex/config.toml` override with both
 `cwd` and `NAPSEER_PROJECT_ROOT`. MCP processes inherit their launch directory;
 they cannot infer a later client workspace switch over stdio.
 
+If `NAPSEER_MCP_WORKER_PATH` points to a stable launcher that loads a separate,
+replaceable runtime, set `NAPSEER_MCP_WORKER_WATCH_PATHS` to the
+`os.pathsep`-separated runtime paths. The supervisor then reloads the launcher
+after any watched runtime is atomically replaced, preserving the same bounded,
+non-replaying request lifecycle used for direct workers.
+
+MCP admits at most 32 unfinished requests by default, including cancelled work
+that is still cleaning up. `-32097` means a request was rejected before
+execution; retry after capacity is available. `-32098` means the result
+is uncertain: inspect current state before retrying a mutation. Cancellation
+prevents queued mutations from starting and attempts to release acquired locks.
+It cannot undo a mutation already accepted by the API.
+
+The supervisor waits up to 60 seconds for a response, then requests cancellation.
+After 10 seconds without a cleanup acknowledgement, it terminates the worker and
+fails affected requests without replay. Worker input writes have a five-second
+deadline. Override these limits with `NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS`,
+`NAPSEER_MCP_CANCEL_GRACE_SECONDS`, `NAPSEER_MCP_WRITE_TIMEOUT_SECONDS`, and
+`NAPSEER_MCP_MAX_IN_FLIGHT_REQUESTS`. New source loads only after unfinished work
+has drained. `nap_contract` exposes the runtime's request lifecycle contract.
+These process recovery guarantees require the supervisor; the direct worker is
+for isolated protocol debugging. An already-running older supervisor needs a
+client reconnect after updating.
+
 The public operator surface is intentionally small:
 
 ```text
-nap init | status | doctor | auth | project | mcp | gateway | update | version | help
+nap init | status | doctor | auth | project | files | export | mcp | gateway | update | version | help
 ```
 
 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
+Agents start only after the user configures an account login or a delegated API
-first such tool enrolls an anonymous worker and creates its first project
+key. With an account login, `nap init` creates a project owned by that account.
-automatically; `nap init` remains the explicit equivalent. Anonymous access
+A committed `.napseer/project.json` always selects the existing project; it
-and refresh-token or SSH-key recovery continue without operator
+never causes a replacement to be created. Concurrent refreshes share a bounded
-authentication. Authentication through `nap project claim` is required only
+credential lock, while project overrides remain independent. Network failures
-after the anonymous account reaches its service limit (currently one project
+preserve credentials and offer recovery through `nap auth repair`.
-or 100 stored nodes), or when its durable local enrollment identity is no
+
-longer recoverable.
+`nap auth keys --help` describes scoped API key issuance;
+`nap auth api-key --env VARIABLE_NAME` installs a delegated key privately.
+Keys can expire or be revoked and cannot approve agent decisions. Use separate
+keys when agents need separate identities or permissions.
+
+Upgrading from 0.2.x to 0.3.0 requires the current bootstrap once because old
+installers have a fixed list of runtime files. `nap update` on those versions
+rejects the new bundle before changing the active installation. Run:
+
+```sh
+curl -fsSL https://api.napseer.com/install | python3 -
+```
+
+The bootstrap verifies every asset and switches the runtime atomically without
+changing credentials or project locators. Restart existing MCP client sessions
+after this upgrade; subsequent releases use `nap update` normally.
 
 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

resources/scripts/nap_install.py

@@ -6,6 +6,7 @@ Usage:
   nap init
   nap status
   nap doctor
+  nap reindex
   nap auth login
   nap project create
   nap project attach
@@ -42,6 +43,7 @@ SCRIPT_NAMES = (
     "napseer_mcp_server.py",
     "napseer_mcp_supervisor.py",
     "napseer_spake2.py",
+    "napseer_credentials.py",
     "terminal_init.py",
     "terminal_protocol.py",
     "terminal_pty_manager.py",
@@ -56,6 +58,7 @@ LOCAL_HOST = "127.0.0.1"
 DEFAULT_GATEWAY_PORT = os.environ.get("NAPSEER_GATEWAY_PORT")
 RUNTIME_SCRIPT_NAMES = (
     "napseer_spake2.py",
+    "napseer_credentials.py",
     "terminal_init.py",
     "terminal_protocol.py",
     "terminal_pty_manager.py",
@@ -63,15 +66,18 @@ RUNTIME_SCRIPT_NAMES = (
     "napseer_mcp_supervisor.py",
 )
 MAX_SERVICE_LOG_BYTES = int(os.environ.get("NAPSEER_MAX_SERVICE_LOG_BYTES", str(5 * 1024 * 1024)))
-CLI_RELEASE_VERSION = "0.2.2"
+CLI_RELEASE_VERSION = "0.3.0"
-CLI_DISTRIBUTION_CONTRACT_VERSION = "2026-08-01"
+CLI_DISTRIBUTION_CONTRACT_VERSION = "2026-09-10"
-CLI_MINIMUM_CONTRACT_VERSION = "2026-05-19"
+# Older installers have a fixed asset list and would omit the credentials module.
+# Reject them before activation; the published bootstrap performs the upgrade.
+CLI_MINIMUM_CONTRACT_VERSION = "2026-09-10"
 CLI_BUNDLE_SCHEMA_VERSION = "napseer.cli.bundle.v1"
 INSTALL_PATHS = {
     "nap_install.py": "nap_install.py",
     "napseer_mcp_server.py": "napseer_mcp_server.py",
     "napseer_mcp_supervisor.py": "napseer_mcp_supervisor.py",
     "napseer_spake2.py": "napseer_spake2.py",
+    "napseer_credentials.py": "napseer_credentials.py",
     "terminal_init.py": "terminal/__init__.py",
     "terminal_protocol.py": "terminal/protocol.py",
     "terminal_pty_manager.py": "terminal/pty_manager.py",
@@ -84,8 +90,11 @@ CANONICAL_COMMANDS = (
     ("init", "Initialize Napseer in the current folder."),
     ("status", "Show concise account, project, and runtime state."),
     ("doctor", "Diagnose local setup without changing it."),
+    ("reindex", "Rebuild the current project's local memory index."),
     ("auth", "Authenticate an account or repair credentials."),
     ("project", "Create, attach, claim, or inspect a project."),
+    ("files", "List, upload, download, or archive project files."),
+    ("export", "Export project documentation and files to a new ZIP archive."),
     ("mcp", "Install, update, inspect, or serve the local MCP runtime."),
     ("gateway", "Set up and operate the local gateway."),
     ("update", "Update the installed CLI and runtime bundle."),
@@ -96,8 +105,11 @@ COMMAND_METADATA = {
     "init": {"mutates": True, "visible": True},
     "status": {"mutates": False, "visible": True},
     "doctor": {"mutates": False, "visible": True},
+    "reindex": {"mutates": True, "visible": True},
     "auth": {"mutates": True, "visible": True},
     "project": {"mutates": True, "visible": True},
+    "files": {"mutates": True, "visible": True},
+    "export": {"mutates": True, "visible": True},
     "mcp": {"mutates": True, "visible": True},
     "gateway": {"mutates": True, "visible": True},
     "update": {"mutates": True, "visible": True},
@@ -500,6 +512,7 @@ def activate_release(release_dir):
         "napseer_mcp_server.py",
         "napseer_mcp_supervisor.py",
         "napseer_spake2.py",
+        "napseer_credentials.py",
     ):
         path = INSTALL_DIR / filename
         preserve_legacy_path(path)
@@ -629,11 +642,15 @@ def cwd_state_dir():
 
 def state_dir_status():
     preferred = preferred_state_dir()
+    from napseer_credentials import CredentialStore
+    try: auth_path, credential_source = CredentialStore(preferred).selected()
+    except RuntimeError: auth_path, credential_source = CredentialStore(preferred).user, "invalid_selection"
     return {
+        "credential_source": credential_source,
         "active_state_dir": str(preferred),
-        "active_auth_path": str(preferred / "auth.json"),
+        "active_auth_path": str(auth_path),
         "state_dir": str(preferred),
-        "auth_path": str(preferred / "auth.json"),
+        "auth_path": str(auth_path),
         "state_dir_exists": preferred.exists(),
         "state_dir_message": (
             ".napseer directory is active."
@@ -1066,12 +1083,25 @@ def print_command_help(command):
             "Usage: nap doctor\n"
             "Run read-only checks and print exact repair commands."
         ),
+        "reindex": (
+            "Usage: nap reindex [--limit N]\n"
+            "Rebuild the current project's local search and graph index."
+        ),
         "auth": (
             "Usage: nap auth [status|login|repair]\n"
             "  login   Authenticate an account; does not select a project.\n"
             "  status  Show credential state without credential values.\n"
-            "  repair  Recover failed automatic credential renewal."
+            "  repair  Recover failed automatic credential renewal.\n"
+            "  logout [--project]  Revoke the selected login.\n"
+            "  login --project    Store a separate project login in user data.\n"
+            "  migrate            Move legacy project credentials into user data.\n"
+            "  use-user           Explicitly select the general login for this folder.\n"
+            "  api-key --env NAME Configure project access from a named environment variable.\n"
+            "  keys list|create|revoke  Manage named, expiring delegated access."
+
         ),
+        "files": "Usage: nap files list|upload|download|archive [--path FILE] [--id UUID] [--node-id UUID] [--cursor CURSOR] [--limit N]",
+        "export": "Usage: nap export --path NEW_ARCHIVE.zip\nExports active project nodes and files; refuses changes during export or an existing destination.",
         "project": (
             "Usage: nap project [init|create|attach|claim|status|encryption]\n"
             "  init    Initialize a fresh folder (same workflow as `nap init`).\n"
@@ -1335,11 +1365,14 @@ def doctor_status():
     auth_path = pathlib.Path(state["active_auth_path"])
     auth = {}
     auth_error = None
-    if auth_path.exists():
     try:
-            auth = json.loads(auth_path.read_text(encoding="utf-8"))
+        from napseer_credentials import CredentialStore, read_json
-        except (OSError, json.JSONDecodeError):
+        store=CredentialStore(preferred_state_dir())
-            auth_error = "auth file is unreadable or invalid JSON"
+        auth=store.read()
+        locator=read_json(preferred_state_dir()/"project.json")
+        if not auth.get("project_id"): auth["project_id"]=locator.get("project_id")
+    except RuntimeError:
+        auth_error="credential state is unreadable; run nap auth login or inspect the selected override"
     missing = runtime_assets_missing()
     issues = []
     setup = []
@@ -1348,13 +1381,13 @@ def doctor_status():
     elif not auth_path.exists():
         setup.append({
             "code": "project_not_initialized",
-            "message": "current folder is not initialized; anonymous setup is available without login",
+            "message": "current folder is not initialized; configure a user login with nap auth login",
-            "next": "nap init or invoke a project-scoped MCP tool",
+            "next": "nap auth login",
         })
     elif not auth.get("project_id"):
         setup.append({
             "code": "project_not_configured",
-            "message": "no project is configured for this folder; anonymous setup is available without login",
+            "message": "no project is configured for this folder; nap init uses the configured user login",
             "next": "nap init",
         })
     if missing:
@@ -1431,6 +1464,10 @@ def version_status():
 
 def main(argv):
     invoked_as_nap = pathlib.Path(argv[0]).name in {"nap", "nap.cmd"}
+    # The published /install endpoint is executed as `python3 -`. Keep that
+    # bootstrap separate from the removed operator-facing `nap install` alias.
+    if len(argv) == 1 and not invoked_as_nap:
+        return handle_mcp(["install"])
     command = argv[1] if len(argv) > 1 else ("help" if invoked_as_nap else "install")
     args = argv[2:]
     if command in HELP_TOKENS:
@@ -1445,6 +1482,8 @@ def main(argv):
     if command == "doctor":
         print_json(doctor_status())
         return 0
+    if command == "reindex":
+        return run_wrapper("reindex", args)
     if command == "mcp":
         return handle_mcp(args)
     if command == "version":
@@ -1457,6 +1496,8 @@ def main(argv):
         return 0
     if command == "gateway":
         return handle_gateway(args)
+    if command in {"files", "export"}:
+        return run_wrapper(command, args)
     if command == "status":
         return run_wrapper("configure", args)
     if command == "project":
@@ -1469,7 +1510,7 @@ def main(argv):
         if args and args[0] in {"login", "login-local", "operator-login"}:
             validate_cli_options(
                 args[1:],
-                flags={"--no-browser"},
+                flags={"--no-browser", "--project"},
                 value_options={
                     "--frontend-url",
                     "--api-base-url",

resources/scripts/napseer_credentials.py

@@ -0,0 +1,240 @@
+"""User-owned credentials, separate from repository identity and gateway state.
+
+No network, enrollment, or credential output belongs in this module.
+"""
+from contextlib import contextmanager
+import hashlib
+import json
+import os
+from pathlib import Path
+import stat
+import sys
+import tempfile
+import threading
+import time
+
+
+class CredentialError(RuntimeError):
+    pass
+
+
+CREDENTIAL_KEYS = {
+    "base_url", "account_id", "token", "access_token", "refresh_token",
+    "token_expires_at", "refresh_expires_at", "account_mode", "oauth_client_id",
+    "oauth_scope", "logged_out", "credential_kind",
+}
+SELECTION_KEYS = {
+    "project_id", "project_slug", "project_name", "project_encryption_state",
+    "project_signing_key_fingerprint", "credential_profile",
+}
+_registry_lock = threading.Lock()
+_thread_locks = {}
+_held = threading.local()
+
+
+def user_data_dir():
+    if os.environ.get("NAPSEER_USER_DATA_DIR"):
+        return Path(os.environ["NAPSEER_USER_DATA_DIR"]).expanduser().absolute()
+    if sys.platform == "win32":
+        return Path(os.environ.get("LOCALAPPDATA", Path.home() / "AppData" / "Local")) / "napseer"
+    if sys.platform == "darwin":
+        return Path.home() / "Library" / "Application Support" / "napseer"
+    return Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share")) / "napseer"
+
+
+def read_json(path):
+    try:
+        fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+    except FileNotFoundError:
+        return {}
+    except OSError:
+        raise CredentialError("Credential state cannot be opened safely.") from None
+    try:
+        if not stat.S_ISREG(os.fstat(fd).st_mode):
+            raise CredentialError("Credential state must be a regular file.")
+        with os.fdopen(fd, "r", encoding="utf-8") as stream:
+            fd = None
+            raw = stream.read(65537)
+        if len(raw) > 65536:
+            raise CredentialError("Credential state exceeds the size limit.")
+        result = json.loads(raw)
+        if not isinstance(result, dict):
+            raise ValueError()
+        return result
+    except (ValueError, UnicodeError, OSError):
+        raise CredentialError("Credential state is unreadable; run nap auth login to repair it.") from None
+    finally:
+        if fd is not None:
+            os.close(fd)
+
+
+def atomic_json(path, data):
+    path = Path(path)
+    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    if path.is_symlink():
+        raise CredentialError("Credential state must not be a symbolic link.")
+    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as stream:
+            os.fchmod(stream.fileno(), 0o600)
+            json.dump(data, stream, indent=2)
+            stream.write("\n")
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(temporary, path)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+@contextmanager
+def credential_lock(path, timeout=30):
+    """Bounded process lock; nested calls in the same thread reuse the lock."""
+    path = Path(path).absolute()
+    with _registry_lock:
+        thread_lock = _thread_locks.setdefault(str(path), threading.RLock())
+    if not thread_lock.acquire(timeout=timeout):
+        raise CredentialError("Credential update is busy; retry after the other login or refresh finishes.")
+    try:
+        held = getattr(_held, "paths", {})
+        if str(path) in held:
+            yield
+            return
+        path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+        try:
+            fd = os.open(str(path) + ".lock", os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0), 0o600)
+        except OSError:
+            raise CredentialError("Credential lock cannot be opened safely.") from None
+        try:
+            if not stat.S_ISREG(os.fstat(fd).st_mode):
+                raise CredentialError("Credential lock must be a regular file.")
+            deadline = time.monotonic() + timeout
+            while True:
+                try:
+                    if os.name == "nt":
+                        import msvcrt
+                        os.lseek(fd, 0, os.SEEK_SET)
+                        msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
+                    else:
+                        import fcntl
+                        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                    break
+                except (BlockingIOError, OSError):
+                    if time.monotonic() >= deadline:
+                        raise CredentialError("Credential update is busy; retry after the other login or refresh finishes.") from None
+                    time.sleep(0.025)
+            held[str(path)] = fd
+            _held.paths = held
+            try:
+                yield
+            finally:
+                held.pop(str(path), None)
+        finally:
+            os.close(fd)
+    finally:
+        thread_lock.release()
+
+
+class CredentialStore:
+    def __init__(self, state_dir, legacy_path=None):
+        self.state_dir = Path(state_dir)
+        self.legacy = Path(legacy_path or self.state_dir / "auth.json")
+        self.user = user_data_dir() / "credentials" / "default.json"
+        project_key = hashlib.sha256(str(self.state_dir.parent.resolve()).encode()).hexdigest()
+        self.project = user_data_dir() / "credentials" / "projects" / f"{project_key}.json"
+        self.selection = self.state_dir / "state.json"
+
+    def selected(self):
+        bound = getattr(_held,"bindings",{}).get(str(self.state_dir.absolute()))
+        if bound: return bound
+        explicit = os.environ.get("NAPSEER_AUTH_FILE")
+        if explicit:
+            return Path(explicit).expanduser().absolute(), "explicit"
+        if self.legacy != self.state_dir / "auth.json":
+            return self.legacy, "explicit"
+        if read_json(self.selection).get("credential_profile") == "user":
+            return self.user, "user"
+        if self.project.exists() or self.project.is_symlink():
+            return self.project, "project"
+        if self.legacy.exists() or self.legacy.is_symlink():
+            return self.legacy, "legacy_project"
+        return self.user, "user"
+
+    @contextmanager
+    def transaction(self):
+        """Pin one credential source through read/refresh/write despite new overrides."""
+        identity = str(self.state_dir.absolute())
+        chosen = self.selected()
+        with credential_lock(chosen[0]):
+            bindings = getattr(_held,"bindings",{})
+            previous = bindings.get(identity)
+            bindings[identity] = chosen; _held.bindings = bindings
+            try: yield
+            finally:
+                if previous: bindings[identity] = previous
+                else: bindings.pop(identity,None)
+
+    def use_user(self):
+        current = read_json(self.user)
+        if not current.get("token") or current.get("logged_out"):
+            raise CredentialError("Run nap auth login before selecting the general user login.")
+        with credential_lock(self.selection):
+            selection = read_json(self.selection)
+            selection["credential_profile"] = "user"
+            atomic_json(self.selection, selection)
+        return self.user
+
+    def migrate(self):
+        with credential_lock(self.legacy), credential_lock(self.project):
+            legacy = read_json(self.legacy)
+            if not legacy: raise CredentialError("There is no legacy project credential to migrate.")
+            if self.project.exists(): raise CredentialError("A project credential already exists; migration will not overwrite it.")
+            atomic_json(self.project, legacy)
+            if read_json(self.project) != legacy: raise CredentialError("Migration verification failed; original credentials were preserved.")
+            self.legacy.unlink()
+            return self.project
+
+    def read(self):
+        path, source = self.selected()
+        credentials = read_json(path)
+        selection = {k: v for k, v in read_json(self.selection).items() if k in SELECTION_KEYS}
+        # Project-bound grants retain their issued project. General grants only
+        # consume repository selection; selection cannot confer authority.
+        return {**selection, **credentials}
+
+    def write(self, data):
+        path, source = self.selected()
+        with credential_lock(path):
+            if source in {"legacy_project", "explicit"}:
+                atomic_json(path, data)
+                return
+            credentials = {k: v for k, v in data.items() if k in CREDENTIAL_KEYS}
+            if source == "project":
+                credentials.update({k: v for k, v in data.items() if k in SELECTION_KEYS})
+            current = read_json(path)
+            # Reading project metadata must not churn or overwrite shared
+            # credentials, including after another process logged out.
+            if credentials != current:
+                atomic_json(path, credentials)
+            selection = {**read_json(self.selection), **{k: v for k, v in data.items() if k in SELECTION_KEYS}}
+            if selection != read_json(self.selection):
+                atomic_json(self.selection, selection)
+
+    def login(self, credentials, project=False):
+        target = self.project if project else self.user
+        with credential_lock(target):
+            atomic_json(target, credentials)
+        if project:
+            with credential_lock(self.selection):
+                selection = read_json(self.selection); selection["credential_profile"] = "project"
+                atomic_json(self.selection, selection)
+        return target
+
+    def logout(self, project=False):
+        target = self.selected()[0] if project else self.user
+        with credential_lock(target):
+            previous = read_json(target)
+            # Keep a tombstone: removing an override would silently restore the
+            # more privileged user session on the next request.
+            atomic_json(target, {"base_url": previous.get("base_url"), "logged_out": True})
+        return target

resources/scripts/napseer_mcp_supervisor.py

@@ -8,11 +8,12 @@ session. This supervisor owns those client-facing pipes while treating
 - 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
+- cancellation stops waiting for a result but retains unfinished work until
-  notification to the worker;
+  the worker acknowledges cleanup, with bounded grace before termination;
 - 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.
+- a replaced worker, or an explicitly configured runtime dependency behind a
+  stable launcher, 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 messages only.
@@ -25,10 +26,12 @@ import json
 import os
 import pathlib
 import queue
+import select
 import signal
 import subprocess
 import sys
 import threading
+import time
 from dataclasses import dataclass, field
 from typing import BinaryIO, Callable
 
@@ -39,6 +42,11 @@ WORKER_PATH = pathlib.Path(
         pathlib.Path(__file__).resolve().with_name("napseer_mcp_server.py"),
     )
 ).expanduser()
+WORKER_WATCH_PATHS = tuple(
+    pathlib.Path(value).expanduser()
+    for value in os.environ.get("NAPSEER_MCP_WORKER_WATCH_PATHS", "").split(os.pathsep)
+    if value.strip()
+)
 RESPONSE_TIMEOUT_SECONDS = max(
     1,
     int(os.environ.get("NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS", "60")),
@@ -48,6 +56,10 @@ MAX_IN_FLIGHT_REQUESTS = max(
     int(os.environ.get("NAPSEER_MCP_MAX_IN_FLIGHT_REQUESTS", "32")),
 )
 WORKER_RESTARTED_ERROR = -32098
+REQUEST_REJECTED_ERROR = -32097
+CANCEL_GRACE_SECONDS = max(1, int(os.environ.get("NAPSEER_MCP_CANCEL_GRACE_SECONDS", "10")))
+WRITE_TIMEOUT_SECONDS = max(1, int(os.environ.get("NAPSEER_MCP_WRITE_TIMEOUT_SECONDS", "5")))
+REQUEST_FINISHED = "notifications/napseer/requestFinished"
 _CANCELLED = object()
 
 
@@ -55,11 +67,24 @@ class WorkerUnavailable(RuntimeError):
     """The replaceable MCP worker could not complete a request."""
 
 
+class RequestRejected(WorkerUnavailable):
+    """The request was not sent to the worker and cannot have executed."""
+
+
 def source_identity(path: pathlib.Path) -> tuple[int, int, int]:
     stat = path.stat()
     return stat.st_ino, stat.st_size, stat.st_mtime_ns
 
 
+def runtime_identity(
+    worker_path: pathlib.Path,
+    watch_paths: tuple[pathlib.Path, ...] = (),
+) -> tuple[tuple[int, int, int], ...]:
+    """Identify every source whose replacement changes the worker runtime."""
+
+    return tuple(source_identity(path) for path in (worker_path, *watch_paths))
+
+
 def parse_message(message: bytes):
     try:
         payload = json.loads(message)
@@ -97,16 +122,16 @@ def cancelled_request_id(message: bytes):
     return True, params.get("requestId")
 
 
-def worker_error(message: bytes) -> bytes:
+def worker_error(message: bytes, *, rejected=False) -> bytes:
     payload = {
         "jsonrpc": "2.0",
         "id": request_id(message),
         "error": {
-            "code": WORKER_RESTARTED_ERROR,
+            "code": REQUEST_REJECTED_ERROR if rejected else WORKER_RESTARTED_ERROR,
             "message": (
                 "Napseer MCP could not complete this request. The outcome may be "
                 "uncertain; inspect state before retrying a mutation."
-            ),
+            ) if not rejected else "Napseer MCP is busy or this request id is still active. This request was not started; retry later.",
         },
     }
     return json.dumps(payload, separators=(",", ":")).encode("utf-8") + b"\n"
@@ -118,13 +143,21 @@ class PendingRequest:
     key: str
     generation: int
     response: queue.Queue = field(default_factory=lambda: queue.Queue(maxsize=1))
+    cancelled: bool = False
+    cancel_timer: threading.Timer | None = None
 
 
 class Worker:
-    def __init__(self, path: pathlib.Path, unsolicited: Callable[[bytes], None] | None = None):
+    def __init__(
+        self,
+        path: pathlib.Path,
+        unsolicited: Callable[[bytes], None] | None = None,
+        watch_paths: tuple[pathlib.Path, ...] = (),
+    ):
         self.path = path
+        self.watch_paths = tuple(watch_paths)
         self.process: subprocess.Popen[bytes] | None = None
-        self.identity: tuple[int, int, int] | None = None
+        self.identity: tuple[tuple[int, int, int], ...] | None = None
         self.generation = 0
         self.unsolicited = unsolicited
         self._lifecycle_lock = threading.RLock()
@@ -135,7 +168,7 @@ class Worker:
 
     def _discard_stderr(self, stream: BinaryIO) -> None:
         try:
-            for _line in stream:
+            while stream.read(4096):
                 pass
         finally:
             stream.close()
@@ -159,6 +192,8 @@ class Worker:
                 if pending.generation != generation:
                     continue
                 self._pending.pop(key, None)
+                if pending.cancel_timer is not None:
+                    pending.cancel_timer.cancel()
                 failed.append(pending)
         for pending in failed:
             self._signal(pending, error)
@@ -168,18 +203,32 @@ class Worker:
         if payload is None:
             return False
         normalized = response if response.endswith(b"\n") else response + b"\n"
+        if payload.get("method") == REQUEST_FINISHED:
+            params = payload.get("params") or {}
+            if isinstance(params, dict) and "requestId" in params:
+                key = request_key(params["requestId"])
+                with self._pending_lock:
+                    pending = self._pending.get(key)
+                    if pending is not None and pending.generation == generation and pending.cancelled:
+                        self._pending.pop(key, None)
+                        if pending.cancel_timer is not None:
+                            pending.cancel_timer.cancel()
+            return True
         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)
+                    if pending.cancel_timer is not None:
+                        pending.cancel_timer.cancel()
                 else:
                     pending = None
             if pending is not None:
+                if not pending.cancelled:
                     self._signal(pending, normalized)
                 return True
-        if "method" in payload and self.unsolicited is not None:
+        if generation == self.generation and "method" in payload and self.unsolicited is not None:
             self.unsolicited(normalized)
         return True
 
@@ -219,12 +268,15 @@ class Worker:
         if process is None:
             return
         if process.poll() is None:
+            try:
                 process.terminate()
+            except ProcessLookupError:
+                pass
             try:
                 process.wait(timeout=2)
             except subprocess.TimeoutExpired:
                 process.kill()
-                process.wait()
+                process.wait(timeout=2)
         for stream in (process.stdin, process.stdout):
             if stream is not None:
                 try:
@@ -237,13 +289,14 @@ class Worker:
         with self._lifecycle_lock:
             self._stop_locked()
             try:
-                identity = source_identity(self.path)
+                identity = runtime_identity(self.path, self.watch_paths)
                 process = subprocess.Popen(
                     [sys.executable, "-u", str(self.path)],
                     stdin=subprocess.PIPE,
                     stdout=subprocess.PIPE,
                     stderr=subprocess.PIPE,
                     bufsize=0,
+                    env={**os.environ, "NAPSEER_MCP_SUPERVISED": "1"},
                 )
             except (OSError, subprocess.SubprocessError) as exc:
                 raise WorkerUnavailable("unable to start Napseer MCP worker") from exc
@@ -254,6 +307,7 @@ class Worker:
             self.generation += 1
             generation = self.generation
             self.process = process
+            os.set_blocking(process.stdin.fileno(), False)
             self.identity = identity
             threading.Thread(
                 target=self._discard_stderr,
@@ -275,7 +329,7 @@ class Worker:
     def ensure_current(self) -> None:
         with self._lifecycle_lock:
             try:
-                current_identity = source_identity(self.path)
+                current_identity = runtime_identity(self.path, self.watch_paths)
             except OSError as exc:
                 self._stop_locked()
                 raise WorkerUnavailable("Napseer MCP worker is missing") from exc
@@ -288,14 +342,30 @@ class Worker:
                 self.start()
 
     def _write(self, process: subprocess.Popen[bytes], message: bytes) -> None:
-        with self._stdin_lock:
+        deadline = time.monotonic() + WRITE_TIMEOUT_SECONDS
+        if not self._stdin_lock.acquire(timeout=WRITE_TIMEOUT_SECONDS):
+            raise WorkerUnavailable("Napseer MCP worker input timed out")
+        try:
             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")
+                remaining = memoryview(message if message.endswith(b"\n") else message + b"\n")
-                process.stdin.flush()
+                while remaining:
-            except (BrokenPipeError, OSError) as exc:
+                    timeout = deadline - time.monotonic()
+                    if timeout <= 0 or self.process is not process:
+                        raise WorkerUnavailable("Napseer MCP worker input timed out")
+                    fd = process.stdin.fileno()
+                    if not select.select([], [fd], [], timeout)[1]:
+                        raise WorkerUnavailable("Napseer MCP worker input timed out")
+                    try:
+                        count = os.write(fd, remaining)
+                        remaining = remaining[count:]
+                    except BlockingIOError:
+                        continue
+            except (BrokenPipeError, OSError, ValueError) as exc:
                 raise WorkerUnavailable("Napseer MCP worker input closed") from exc
+        finally:
+            self._stdin_lock.release()
 
     def submit(self, message: bytes) -> PendingRequest | None:
         self.ensure_current()
@@ -305,16 +375,20 @@ class Worker:
         if process is None:
             raise WorkerUnavailable("Napseer MCP worker is not running")
         if is_notification(message):
+            try:
                 self._write(process, message)
+            except WorkerUnavailable:
+                self.stop()
+                raise
             return None
 
         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")
+                raise RequestRejected("duplicate in-flight JSON-RPC request id")
             if len(self._pending) >= MAX_IN_FLIGHT_REQUESTS:
-                raise WorkerUnavailable("too many in-flight Napseer MCP requests")
+                raise RequestRejected("too many unfinished Napseer MCP requests")
             self._pending[key] = pending
         try:
             self._write(process, message)
@@ -322,26 +396,54 @@ class Worker:
             with self._pending_lock:
                 if self._pending.get(key) is pending:
                     self._pending.pop(key, None)
+            # A partial write has an uncertain outcome and corrupts framing.
+            # Stop this generation rather than appending another request to it.
+            self.stop()
             raise
         return pending
 
-    def cancel(self, value) -> bool:
+    def cancel(self, value, *, expected: PendingRequest | None = None) -> bool:
         key = request_key(value)
         with self._pending_lock:
             pending = self._pending.get(key)
+            if expected is not None and pending is not expected:
+                return False
             if pending is None or request_method(pending.message) == "initialize":
                 return False
-            self._pending.pop(key, None)
+            if pending.cancelled:
-        self._signal(pending, _CANCELLED)
                 return True
+            pending.cancelled = True
+            pending.cancel_timer = threading.Timer(CANCEL_GRACE_SECONDS, self._expire_cancelled, args=(pending,))
+            pending.cancel_timer.daemon = True
+            timer = pending.cancel_timer
+        self._signal(pending, _CANCELLED)
+        timer.start()
+        process = self.process
+        if process is not None and pending.generation == self.generation:
+            try:
+                self._write(process, json.dumps({"jsonrpc": "2.0", "method": "notifications/cancelled",
+                                                "params": {"requestId": value}}).encode("utf-8"))
+            except WorkerUnavailable:
+                self._expire_cancelled(pending)
+        return True
+
+    def _expire_cancelled(self, pending: PendingRequest) -> None:
+        with self._lifecycle_lock:
+            with self._pending_lock:
+                unfinished = self._pending.get(pending.key) is pending
+            if unfinished and pending.generation == self.generation:
+                self._stop_locked()
 
     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 request_method(pending.message) == "initialize":
-                if self._pending.get(pending.key) is pending:
+                # Initialize is not cancellable, but a stuck initialization
-                    self._pending.pop(pending.key, None)
+                # must not permanently consume the worker generation.
+                self._expire_cancelled(pending)
+            else:
+                self.cancel(request_id(pending.message), expected=pending)
             raise WorkerUnavailable("Napseer MCP worker response timed out") from exc
         if result is _CANCELLED:
             return None
@@ -361,7 +463,11 @@ def supervise() -> int:
             sys.stdout.buffer.write(response if response.endswith(b"\n") else response + b"\n")
             sys.stdout.buffer.flush()
 
-    worker = Worker(WORKER_PATH, unsolicited=emit)
+    worker = Worker(
+        WORKER_PATH,
+        unsolicited=emit,
+        watch_paths=WORKER_WATCH_PATHS,
+    )
     waiters = concurrent.futures.ThreadPoolExecutor(
         max_workers=MAX_IN_FLIGHT_REQUESTS,
         thread_name_prefix="napseer-mcp-request",
@@ -391,16 +497,12 @@ def supervise() -> int:
             cancelled, value = cancelled_request_id(message)
             if cancelled:
                 worker.cancel(value)
-                try:
-                    worker.submit(message)
-                except WorkerUnavailable:
-                    pass
                 continue
             try:
                 pending = worker.submit(message)
-            except WorkerUnavailable:
+            except WorkerUnavailable as exc:
                 if not is_notification(message):
-                    emit(worker_error(message))
+                    emit(worker_error(message, rejected=isinstance(exc, RequestRejected)))
                 continue
             if pending is not None:
                 waiters.submit(finish, message, pending)

tests/conftest.py

@@ -0,0 +1,10 @@
+"""All tests use disposable user credential state, including legacy test scripts."""
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def isolated_user_credentials(monkeypatch, tmp_path):
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("NAPSEER_USER_DATA_DIR", str(tmp_path / "user-data"))
+    for name in ("NAPSEER_API_KEY", "NAPSEER_TOKEN", "NAPSEER_REFRESH_TOKEN", "NAPSEER_AUTH_FILE", "NAPSEER_PROJECT_ROOT", "NAPSEER_PROJECT_ID", "NAPSEER_BASE_URL"):
+        monkeypatch.delenv(name, raising=False)

tests/test_agent_files.py

@@ -0,0 +1,127 @@
+"""Public worker/CLI entry points against synthetic binary HTTP storage."""
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import threading
+import uuid
+import zipfile
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import urlsplit, parse_qs
+
+import pytest
+from test_user_credentials import worker, outcome, SCRIPT
+from napseer_credentials import CredentialStore, atomic_json
+
+PROJECT="11111111-1111-4111-8111-111111111111"
+NODE="22222222-2222-4222-8222-222222222222"
+
+@pytest.fixture
+def storage(tmp_path):
+    state={"files":{},"corrupt":False,"keys":[],"projects":[]}
+    node={"id":NODE,"name":"Guide","full_path":"/services/guide","type":"documentation","updated_at":"2026-09-09T00:00:00Z","content_text":"Recovery procedure","metadata":{},"links":[]}
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self,*_args): pass
+        def reply(self,status,value,binary=False):
+            body=value if binary else json.dumps(value).encode()
+            self.send_response(status);self.send_header("Content-Length",str(len(body)));self.end_headers();self.wfile.write(body)
+        def do_GET(self):
+            path=urlsplit(self.path).path
+            if path=='/v1/projects':return self.reply(200,{'items':state['projects']})
+            if path=='/v1/api-keys':return self.reply(200,{'items':state['keys']})
+            if path.endswith('/nodes'): return self.reply(200,{"items":[node]})
+            if path.endswith('/nodes/'+NODE):return self.reply(200,node)
+            if path.endswith('/files'): return self.reply(200,{"items":[entry[0] for entry in state["files"].values()]})
+            identity=path.split('/')[-2] if path.endswith('/metadata') else path.split('/')[-1]
+            entry=state["files"].get(identity)
+            if not entry:return self.reply(404,{"error":{"code":"not_found"}})
+            if path.endswith('/metadata'):return self.reply(200,entry[0])
+            return self.reply(200,b"corrupted" if state["corrupt"] else entry[1],True)
+        def do_POST(self):
+            if self.path=='/v1/projects':
+                payload=json.loads(self.rfile.read(int(self.headers['Content-Length'])))
+                project={**payload,'id':str(uuid.uuid4()),'encryption_state':'standard'}
+                state['projects'].append(project);return self.reply(200,project)
+            if self.path=='/v1/api-keys':
+                payload=json.loads(self.rfile.read(int(self.headers['Content-Length'])))
+                key={**payload,'id':str(uuid.uuid4())};state['keys'].append(key)
+                return self.reply(200,{**key,'api_key':'npk_'+uuid.uuid4().hex})
+            query=parse_qs(urlsplit(self.path).query)
+            body=self.rfile.read(int(self.headers['Content-Length']))
+            identity=query['id'][0]
+            info={"id":identity,"name":query['name'][0],"node_id":query.get('node_id',[None])[0],"size_bytes":len(body),"sha256":hashlib.sha256(body).hexdigest()}
+            state['files'][identity]=(info,body)
+            return self.reply(200,info)
+        def do_DELETE(self):
+            identity=urlsplit(self.path).path.split('/')[-1];state['files'].pop(identity,None)
+            return self.reply(200,{"archived":True})
+    server=ThreadingHTTPServer(('127.0.0.1',0),Handler)
+    thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
+    store=CredentialStore(tmp_path/'.napseer')
+    store.login({"base_url":f"http://127.0.0.1:{server.server_port}","token":"synthetic-access","account_mode":"operator_account"})
+    atomic_json(store.selection,{"project_id":PROJECT})
+    try:yield state
+    finally:server.shutdown();server.server_close();thread.join(timeout=2)
+
+
+def test_agent_upload_retries_download_and_export_preserve_binary_and_references(tmp_path,storage):
+    body=bytes([0,255,128,13,10,65]);source=tmp_path/'proof.bin';source.write_bytes(body)
+    args={"local_path":str(source),"node_id":NODE}
+    first=outcome(worker(tmp_path,f"m.call_tool_impl('nap_file_upload',{args!r})"))
+    assert first['ok'],first.get('code')
+    identity=first['result']['id']
+    again=outcome(worker(tmp_path,f"m.call_tool_impl('nap_file_upload',{args!r})"))
+    assert again['result']['id']==identity
+    assert len(storage['files'])==1
+    destination=tmp_path/'download.bin'
+    result=outcome(worker(tmp_path,f"m.call_tool_impl('nap_file_download',{{'id':{identity!r},'local_path':{str(destination)!r}}})"))
+    assert result['ok'];assert destination.read_bytes()==body
+    rejected=outcome(worker(tmp_path,f"m.call_tool_impl('nap_file_download',{{'id':{identity!r},'local_path':{str(source)!r}}})"))
+    assert rejected['code']=='destination_unavailable';assert source.read_bytes()==body
+    target=tmp_path/'export.zip'
+    env=dict(os.environ,NAPSEER_PROJECT_ROOT=str(tmp_path),NAPSEER_TELEMETRY='0')
+    for name in ['NAPSEER_AUTH_FILE','NAPSEER_BASE_URL','NAPSEER_PROJECT_ID']:env.pop(name,None)
+    exported=subprocess.run([sys.executable,str(SCRIPT),'export','--path',str(target)],cwd=tmp_path,env=env,capture_output=True,text=True,timeout=15)
+    assert exported.returncode==0
+    assert json.loads(exported.stdout)['status']=='exported'
+    with zipfile.ZipFile(target) as archive:
+        manifest=json.loads(archive.read('manifest.json'))
+        assert manifest['files'][0]['node_id']==NODE
+        assert archive.read('files/'+identity)==body
+        assert json.loads(archive.read('nodes/'+NODE+'.json'))['content_text']=='Recovery procedure'
+        assert not any('auth' in name for name in archive.namelist())
+    storage['corrupt']=True
+    rejected=outcome(worker(tmp_path,f"m.export_project({{'local_path':{str(tmp_path/'broken.zip')!r}}})"))
+    assert rejected['code']=='file_integrity_failed'
+    assert not (tmp_path/'broken.zip').exists()
+    assert not list(tmp_path.glob('.napseer-export-*'))
+    assert outcome(worker(tmp_path,f"m.call_tool_impl('nap_file_archive',{{'id':{identity!r}}})"))['ok']
+    assert not storage['files']
+
+
+def test_cli_key_issuance_writes_private_credential_without_printing_it(tmp_path,storage):
+    env=dict(os.environ,NAPSEER_PROJECT_ROOT=str(tmp_path),NAPSEER_TELEMETRY='0')
+    for name in ['NAPSEER_AUTH_FILE','NAPSEER_BASE_URL','NAPSEER_PROJECT_ID']:env.pop(name,None)
+    created=subprocess.run([sys.executable,str(SCRIPT),'auth','keys','create','--project-id',PROJECT,'--name','Test delegate'],cwd=tmp_path,env=env,capture_output=True,text=True,timeout=15)
+    assert created.returncode==0
+    assert 'npk_' not in created.stdout
+    result=json.loads(created.stdout);path=Path(result['credential_file'])
+    assert path.stat().st_mode & 0o777 == 0o600
+    assert json.loads(path.read_text())['token'].startswith('npk_')
+    assert not (tmp_path/'.napseer/auth.json').exists()
+
+
+def test_two_agents_bootstrap_one_project_with_general_login(tmp_path,storage):
+    store=CredentialStore(tmp_path/'.napseer');store.selection.unlink()
+    first=worker(tmp_path,"m.resolve_project_id({})")
+    second=worker(tmp_path,"m.resolve_project_id({})")
+    one,two=outcome(first),outcome(second)
+    assert one['ok'] and two['ok']
+    assert one['result']==two['result']
+    assert len(storage['projects'])==1
+    assert json.loads(store.user.read_text())['account_mode']=='operator_account'
+    assert 'project_id' not in json.loads(store.user.read_text())
+    assert not store.legacy.exists()
+    assert (tmp_path/'.napseer/project.json').exists()

tests/test_anonymous_auth_recovery.py

@@ -29,37 +29,18 @@ def anonymous_auth(mod):
     }
 
 
-def test_initial_enrollment_persists_refresh_credential():
+def test_unconfigured_agent_requires_user_login_without_enrollment():
     mod = load_module()
     mod.TOKEN = None
-    mod.AUTH = anonymous_auth(mod)
-    mod.BASE_URL = mod.AUTH["base_url"]
-    mod.DEFAULT_PROJECT_ID = None
-    saved = []
     requests = []
-    mod.save_auth = lambda payload: saved.append(dict(payload))
+    mod.request_json = lambda *args, **kwargs: requests.append(args)
-
+    try:
-    def request(method, path, payload=None, token_required=False):
-        requests.append((method, path, payload, token_required))
-        return {
-            "token": {
-                "access_token": "access",
-                "refresh_token": "refresh",
-                "expires_at": "2099-01-01T00:00:00Z",
-                "refresh_expires_at": "",
-            },
-            "worker": {"id": "worker", "agent_id": "agent-id"},
-        }
-
-    mod.request_json = request
         mod.ensure_enrolled({"slug": "example"})
-
+    except mod.SafeToolError as exc:
-    assert saved[0]["refresh_token"] == "refresh"
+        assert exc.code == "auth_required"
-    assert "refresh_expires_at" in saved[0]
+    else:
-    assert [(method, path) for method, path, _, _ in requests] == [
+        raise AssertionError("unconfigured agents must not enroll")
-        ("POST", "/v1/enrollment/token")
+    assert not requests
-    ]
-    assert requests[0][3] is False
 
 
 def test_failed_anonymous_refresh_uses_ssh_recovery():
@@ -107,65 +88,28 @@ def test_failed_anonymous_refresh_uses_ssh_recovery():
     assert saved[0]["refresh_token"] == "recovered-refresh"
 
 
-def test_failed_mixed_state_refresh_uses_ssh_recovery():
+def test_oauth_failure_never_falls_back_to_anonymous_identity():
     mod = load_module()
-    mod.AUTH = {
+    mod.AUTH = {**anonymous_auth(mod), "account_mode": "operator_project"}
-        **anonymous_auth(mod),
+    mod.TOKEN = "expired-access"
-        "account_mode": "operator_project",
+    mod.REFRESH_TOKEN = "revoked-refresh"
-        "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"
+    calls = []
-    mod.sign_with_auth_key = lambda *args: "test-signature"
+    def rejected(*args, **kwargs):
-    saved = []
+        raise mod.SafeToolError("http_401", "Login rejected", status=401)
-    mod.save_auth = lambda payload: saved.append(dict(payload))
+    mod.request_form_json = rejected
-
+    mod.request_json = lambda *args, **kwargs: calls.append(args)
-    def request_form(method, path, payload=None, token_required=False):
+    try:
-        assert (method, path, token_required) == ("POST", "/v1/oauth/token", False)
+        mod.renew_auth()
-        raise RuntimeError("HTTP 401")
+    except mod.SafeToolError as exc:
-
+        assert exc.code == "auth_required"
-    def request(method, path, payload=None, token_required=False):
+    else:
-        if path == "/v1/enrollment/refresh":
+        raise AssertionError("OAuth revocation must require user login")
-            raise RuntimeError("HTTP 401")
+    assert not calls
-        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_unconfigured_agent_requires_user_login_without_enrollment()
     test_failed_anonymous_refresh_uses_ssh_recovery()
-    test_failed_mixed_state_refresh_uses_ssh_recovery()
+    test_oauth_failure_never_falls_back_to_anonymous_identity()
     print("ok: anonymous auth recovery tests passed")

tests/test_bootstrap_entrypoint.py

@@ -0,0 +1,79 @@
+"""Exercise the published `curl /install | python3 -` entry point over HTTP."""
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+
+def test_stdin_bootstrap_installs_verified_runtime_and_preserves_credentials(tmp_path):
+    source = Path(__file__).resolve().parents[1] / "resources/scripts"
+    spec = importlib.util.spec_from_file_location("bootstrap_contract", source / "nap_install.py")
+    installer = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(installer)
+    revision = "f" * 40
+    items, payloads = [], {}
+    for name in installer.SCRIPT_NAMES:
+        content = (source / installer.INSTALL_PATHS[name]).read_text()
+        item = {"name": name, "version": installer.CLI_RELEASE_VERSION,
+                "bytes": len(content.encode()), "sha256": hashlib.sha256(content.encode()).hexdigest(),
+                "source_repo": "https://github.com/napseer/nap-cli", "source_revision": revision,
+                "source_revision_status": "resolved", "contract_version": installer.CLI_DISTRIBUTION_CONTRACT_VERSION,
+                "minimum_contract_version": installer.CLI_MINIMUM_CONTRACT_VERSION,
+                "install_path": installer.INSTALL_PATHS[name], "mode": "0755"}
+        items.append(item)
+        payloads["/v1/scripts/" + name] = {**item, "content": content}
+    payloads["/v1/scripts"] = {"schema_version": installer.CLI_BUNDLE_SCHEMA_VERSION,
+        "bundle_id": "synthetic-bootstrap", "release_version": installer.CLI_RELEASE_VERSION,
+        "contract": {"current": installer.CLI_DISTRIBUTION_CONTRACT_VERSION, "minimum_supported": installer.CLI_MINIMUM_CONTRACT_VERSION},
+        "source": {"repo": "https://github.com/napseer/nap-cli", "revision": revision, "revision_status": "resolved"}, "items": items}
+
+    class Provider(BaseHTTPRequestHandler):
+        def log_message(self, *_args):
+            pass
+
+        def do_GET(self):
+            assert self.path in payloads, "bootstrap must only download its declared bundle"
+            raw = json.dumps(payloads[self.path]).encode()
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(raw)))
+            self.end_headers()
+            self.wfile.write(raw)
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Provider)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    user_data = tmp_path / "user-data"
+    user_data.mkdir(exist_ok=True)
+    auth = user_data / "auth.json"
+    auth.write_text('{"refresh_token":"synthetic-preserved"}')
+    env = {"PATH": os.environ["PATH"], "NAPSEER_HOME": str(tmp_path / "runtime"),
+        "NAPSEER_BIN_DIR": str(tmp_path / "bin"), "NAPSEER_USER_DATA_DIR": str(user_data),
+        "NAPSEER_PROJECT_ROOT": str(tmp_path), "NAPSEER_BASE_URL": f"http://127.0.0.1:{server.server_port}"}
+    try:
+        result = subprocess.run([sys.executable, "-"], input=(source / "nap_install.py").read_text(),
+            cwd=tmp_path, env=env, capture_output=True, text=True, timeout=30)
+        assert result.returncode == 0, "the advertised stdin bootstrap must install without an explicit subcommand"
+        assert json.loads(result.stdout)["status"] == "installed"
+        for name in installer.SCRIPT_NAMES:
+            installed = tmp_path / "runtime/current" / installer.INSTALL_PATHS[name]
+            assert installed.read_bytes() == (source / installer.INSTALL_PATHS[name]).read_bytes()
+        assert auth.read_text() == '{"refresh_token":"synthetic-preserved"}'
+        # Import the actual activated modules, including the newly added dependency.
+        env["PYTHONPATH"] = str(tmp_path / "runtime/current")
+        imported = subprocess.run([sys.executable, "-c", "import napseer_credentials, napseer_mcp_server"],
+            cwd=tmp_path, env=env, capture_output=True, timeout=10)
+        assert imported.returncode == 0, "activated runtime must load all dependencies"
+        result = subprocess.run([sys.executable, str(tmp_path / "bin/nap"), "update"],
+            cwd=tmp_path, env=env, capture_output=True, text=True, timeout=30)
+        assert result.returncode == 0
+        assert auth.read_text() == '{"refresh_token":"synthetic-preserved"}'
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)

tests/test_cli_command_system.py

@@ -3,6 +3,7 @@ import hashlib
 import json
 import pathlib
 import time
+import sys
 import pytest
 
 
@@ -15,6 +16,8 @@ SCRIPT_PATH = (
 
 
 def load_installer():
+    if str(SCRIPT_PATH.parent) not in sys.path:
+        sys.path.insert(0, str(SCRIPT_PATH.parent))
     spec = importlib.util.spec_from_file_location("nap_install_command_system_test", SCRIPT_PATH)
     module = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(module)
@@ -33,7 +36,7 @@ def test_root_help_exposes_only_canonical_commands(capsys):
     output = capsys.readouterr().out
     visible = [name for name, metadata in module.COMMAND_METADATA.items() if metadata["visible"]]
     assert visible == [name for name, _summary in module.CANONICAL_COMMANDS]
-    assert len(visible) == 10
+    assert len(visible) == 13
     for name in visible:
         assert f"  {name}" in output
     for hidden in ("chat", "plan", "lineage", "agent", "feedback", "where", "install"):
@@ -101,6 +104,24 @@ def test_init_and_auth_repair_normalize_to_existing_handlers(monkeypatch):
     ]
 
 
+def test_reindex_routes_to_worker_and_has_side_effect_free_help(monkeypatch, capsys):
+    module = load_installer()
+    calls = []
+    monkeypatch.setattr(
+        module,
+        "run_wrapper",
+        lambda command, args: calls.append((command, args)) or 0,
+    )
+
+    assert module.main(["nap", "reindex", "--limit", "50"]) == 0
+    assert calls == [("reindex", ["--limit", "50"])]
+
+    calls.clear()
+    assert module.main(["nap", "reindex", "--help"]) == 0
+    assert calls == []
+    assert "Usage: nap reindex [--limit N]" in capsys.readouterr().out
+
+
 def test_update_does_not_overwrite_repository_source(monkeypatch, capsys):
     module = load_installer()
     calls = []
@@ -418,7 +439,7 @@ def test_bundle_install_stages_validates_and_activates_atomically(tmp_path, monk
 
     result = module.install_assets()
 
-    assert result["cli_distribution"]["release_version"] == "0.2.2"
+    assert result["cli_distribution"]["release_version"] == module.CLI_RELEASE_VERSION
     assert (install_dir / "current").is_symlink()
     assert (install_dir / "current" / "bundle-manifest.json").is_file()
     assert (install_dir / "current" / "napseer_mcp_server.py").is_file()
@@ -475,10 +496,10 @@ def test_manifest_accepts_a_future_compatible_release():
     module = load_installer()
     manifest, _payloads = fake_bundle(module)
     manifest["release_version"] = "0.2.2"
-    manifest["contract"]["current"] = "2026-08-02"
+    manifest["contract"]["current"] = "2026-09-11"
     for item in manifest["items"]:
         item["version"] = "0.2.2"
-        item["contract_version"] = "2026-08-02"
+        item["contract_version"] = "2026-09-11"
 
     items = module.validated_manifest_items(manifest)
 
@@ -488,11 +509,11 @@ def test_manifest_accepts_a_future_compatible_release():
 def test_manifest_rejects_a_minimum_contract_newer_than_installer():
     module = load_installer()
     manifest, _payloads = fake_bundle(module)
-    manifest["contract"]["current"] = "2026-08-02"
+    manifest["contract"]["current"] = "2026-09-11"
-    manifest["contract"]["minimum_supported"] = "2026-08-02"
+    manifest["contract"]["minimum_supported"] = "2026-09-11"
     for item in manifest["items"]:
-        item["contract_version"] = "2026-08-02"
+        item["contract_version"] = "2026-09-11"
-        item["minimum_contract_version"] = "2026-08-02"
+        item["minimum_contract_version"] = "2026-09-11"
 
     try:
         module.validated_manifest_items(manifest)
@@ -502,6 +523,31 @@ def test_manifest_rejects_a_minimum_contract_newer_than_installer():
         raise AssertionError("incompatible minimum contract must be rejected")
 
 
+def test_legacy_update_preserves_active_runtime_and_bootstrap_recovers(tmp_path, monkeypatch):
+    module = load_installer()
+    monkeypatch.setattr(module, "INSTALL_DIR", tmp_path / "install")
+    monkeypatch.setattr(module, "BIN_DIR", tmp_path / "bin")
+    manifest, payloads = fake_bundle(module)
+    previous = module.INSTALL_DIR / "releases" / "previous"
+    previous.mkdir(parents=True)
+    (previous / "nap_install.py").write_text("# existing release\n")
+    (module.INSTALL_DIR / "current").symlink_to(previous, target_is_directory=True)
+    credentials = module.INSTALL_DIR / "auth.json"
+    credentials.write_text('{"refresh_token":"synthetic-preserved"}')
+    monkeypatch.setattr(module, "bundle_manifest", lambda: manifest)
+    monkeypatch.setattr(module, "fetch_script", lambda name: payloads[name])
+    current_contract = module.CLI_DISTRIBUTION_CONTRACT_VERSION
+    monkeypatch.setattr(module, "CLI_DISTRIBUTION_CONTRACT_VERSION", "2026-08-01")
+    with pytest.raises(RuntimeError, match="newer bootstrap installer"):
+        module.main(["nap", "update"])
+    assert (module.INSTALL_DIR / "current").resolve() == previous
+    assert list((module.INSTALL_DIR / "releases").iterdir()) == [previous]
+    monkeypatch.setattr(module, "CLI_DISTRIBUTION_CONTRACT_VERSION", current_contract)
+    assert module.main(["nap_install.py", "mcp", "install"]) == 0
+    assert (module.INSTALL_DIR / "current" / "napseer_credentials.py").is_file()
+    assert credentials.read_text() == '{"refresh_token":"synthetic-preserved"}'
+
+
 def test_cli_log_compaction_preserves_live_writer_inode(tmp_path):
     module = load_installer()
     log_path = tmp_path / "gateway.log"

tests/test_cli_dispatch_consolidation.py

@@ -36,6 +36,22 @@ def expect_unknown(mod, argv, expected_fragment):
         raise AssertionError(f"expected RuntimeError for {argv}")
 
 
+def test_reindex_cli_rebuilds_with_bounded_page_size():
+    mod = load_module()
+    calls = []
+    mod.reindex_project = lambda args: calls.append(dict(args)) or {
+        "indexed": 7,
+        "graph_complete": True,
+    }
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        mod.cli_main(["napseer_mcp_server.py", "reindex", "--limit", "50"])
+
+    assert calls == [{"limit": 50}]
+    assert '"indexed": 7' in output.getvalue()
+
+
 def test_auth_login_rejects_history_expansion_tail_before_browser():
     mod = load_module()
     mod.operator_account_login = lambda _args: (_ for _ in ()).throw(

tests/test_gateway_relay_auth.py

@@ -15,6 +15,8 @@ def load_module():
 
 def test_renew_auth_serializes_rotating_refresh_tokens():
     mod = load_module()
+    mod.TOKEN = "synthetic-configured-access"
+    mod.REFRESH_TOKEN = "synthetic-configured-refresh"
     active = 0
     maximum = 0
     guard = threading.Lock()

tests/test_live_auth_state_reload.py

@@ -15,6 +15,8 @@ def load_module():
     module = importlib.util.module_from_spec(spec)
     assert spec and spec.loader
     spec.loader.exec_module(module)
+    module.load_project_locator = lambda: None
+    module.require_project_locator_service = lambda _locator: None
     return module
 
 
@@ -53,7 +55,7 @@ def run():
     mod.DEFAULT_PROJECT_ID = None
     assert mod.resolve_project_id({}) == "project-1"
 
-    current = {}
+    current = {"token": "configured-account", "account_mode": "operator_account"}
     mod.DEFAULT_PROJECT_ID = "stale-project"
     bootstrap_calls = []
 
@@ -73,7 +75,7 @@ def run():
         }
     ]
 
-    current = {}
+    current = {"token": "configured-account", "account_mode": "operator_account"}
     mod.DEFAULT_PROJECT_ID = None
 
     def limited_project(_args):

tests/test_mcp_batch.py

@@ -0,0 +1,303 @@
+import importlib.util
+import pathlib
+import sys
+import threading
+import time
+
+import pytest
+
+
+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_batch_test", script_path)
+    module = importlib.util.module_from_spec(spec)
+    assert spec and spec.loader
+    spec.loader.exec_module(module)
+    module.DEFAULT_PROJECT_ID = "project-1"
+    module.CONFIGURED_TOOL_PROFILES = {"all"}
+    return module
+
+
+@pytest.fixture
+def mod(monkeypatch):
+    monkeypatch.delenv("NAPSEER_TOOL_PROFILES", raising=False)
+    return load_module()
+
+
+def action(action_id, tool="nap_node_get", *, depends_on=None, arguments=None):
+    return {
+        "id": action_id,
+        "tool": tool,
+        "reason": f"Run {action_id} for batch verification",
+        "arguments": arguments or {},
+        "depends_on": depends_on or [],
+    }
+
+
+def test_batch_is_a_conservatively_annotated_core_tool(mod):
+    tool = next(item for item in mod.tools() if item["name"] == "nap_batch")
+
+    assert tool["annotations"] == {
+        "title": "nap_batch",
+        "readOnlyHint": False,
+        "destructiveHint": True,
+        "idempotentHint": False,
+        "openWorldHint": False,
+    }
+    assert tool["napseer"]["lock_policy"] == "per-action-mutation-barrier"
+    assert tool["napseer"]["idempotency"] == "mixed-no-replay"
+    schema = tool["inputSchema"]
+    assert schema["properties"]["actions"]["maxItems"] == 32
+    assert schema["properties"]["max_parallel"]["maximum"] == 8
+    assert set(schema["properties"]["actions"]["items"]["required"]) == {
+        "id", "tool", "reason", "arguments",
+    }
+
+
+def test_independent_reads_overlap_and_results_preserve_input_order(mod, monkeypatch):
+    state_lock = threading.Lock()
+    gate = threading.Barrier(2)
+    active = 0
+    max_active = 0
+
+    def fake_call(name, arguments):
+        nonlocal active, max_active
+        with state_lock:
+            active += 1
+            max_active = max(max_active, active)
+        gate.wait(timeout=1)
+        if arguments["value"] == "first":
+            time.sleep(0.04)
+        with state_lock:
+            active -= 1
+        return {"tool": name, "value": arguments["value"]}
+
+    monkeypatch.setattr(mod, "call_tool_impl", fake_call)
+    result = mod.execute_batch({
+        "max_parallel": 2,
+        "actions": [
+            action("first", arguments={"value": "first"}),
+            action("second", tool="nap_context", arguments={"value": "second"}),
+        ],
+    })
+
+    assert result["ok"] is True
+    assert result["failure_policy"] == "continue_on_error"
+    assert result["parallel_read_groups"] == 1
+    assert max_active == 2
+    assert [item["id"] for item in result["results"]] == ["first", "second"]
+    assert [item["result"]["value"] for item in result["results"]] == ["first", "second"]
+
+
+def test_mutation_is_an_exclusive_array_order_barrier(mod, monkeypatch):
+    state_lock = threading.Lock()
+    active_reads = 0
+    active_writes = 0
+    max_reads = 0
+    violations = []
+    completed = []
+
+    def fake_call(name, arguments):
+        nonlocal active_reads, active_writes, max_reads
+        is_read = mod.tool_side_effect(name) == "none"
+        with state_lock:
+            if is_read:
+                if active_writes:
+                    violations.append("read overlapped write")
+                active_reads += 1
+                max_reads = max(max_reads, active_reads)
+            else:
+                if active_reads or active_writes:
+                    violations.append("write was not exclusive")
+                active_writes += 1
+        time.sleep(0.04)
+        with state_lock:
+            if is_read:
+                active_reads -= 1
+            else:
+                active_writes -= 1
+            completed.append(arguments["value"])
+        return {"value": arguments["value"]}
+
+    monkeypatch.setattr(mod, "call_tool_impl", fake_call)
+    result = mod.execute_batch({
+        "actions": [
+            action("read-a", arguments={"value": "read-a"}),
+            action("read-b", tool="nap_context", arguments={"value": "read-b"}),
+            action("write", tool="nap_create_node", arguments={"value": "write"}),
+            action("read-c", arguments={"value": "read-c"}),
+            action("read-d", tool="nap_context", arguments={"value": "read-d"}),
+        ],
+    })
+
+    assert result["ok"] is True
+    assert result["failure_policy"] == "stop_on_error"
+    assert max_reads == 2
+    assert violations == []
+    assert completed.index("write") > max(completed.index("read-a"), completed.index("read-b"))
+    assert completed.index("write") < min(completed.index("read-c"), completed.index("read-d"))
+
+
+@pytest.mark.parametrize(
+    ("actions", "error_code"),
+    [
+        (
+            [
+                action("a", depends_on=["b"]),
+                action("b", tool="nap_context", depends_on=["a"]),
+            ],
+            "batch_dependency_cycle",
+        ),
+        ([action("a", depends_on=["missing"])], "unknown_batch_dependency"),
+        ([action("a", tool="nap_gateway_status")], "batch_tool_not_allowed"),
+        ([action("a", tool="nap_batch")], "batch_tool_not_allowed"),
+    ],
+)
+def test_invalid_graphs_and_capabilities_fail_before_execution(
+    mod, monkeypatch, actions, error_code
+):
+    calls = []
+    monkeypatch.setattr(mod, "call_tool_impl", lambda *args: calls.append(args))
+
+    with pytest.raises(mod.SafeToolError) as error:
+        mod.execute_batch({"actions": actions})
+
+    assert error.value.code == error_code
+    assert calls == []
+
+
+def test_dependency_failure_skips_dependents_but_runs_independent_reads(
+    mod, monkeypatch
+):
+    calls = []
+
+    def fake_call(_name, arguments):
+        calls.append(arguments["value"])
+        if arguments["value"] == "fails":
+            raise mod.SafeToolError("fixture_failure", "fixture failed safely")
+        return {"value": arguments["value"]}
+
+    monkeypatch.setattr(mod, "call_tool_impl", fake_call)
+    result = mod.execute_batch({
+        "failure_policy": "continue_on_error",
+        "actions": [
+            action("fails", arguments={"value": "fails"}),
+            action("dependent", depends_on=["fails"], arguments={"value": "dependent"}),
+            action("independent", tool="nap_context", arguments={"value": "independent"}),
+        ],
+    })
+
+    assert result["ok"] is False
+    assert result["stopped_at"] is None
+    assert [item["status"] for item in result["results"]] == [
+        "failed", "skipped", "succeeded",
+    ]
+    assert set(calls) == {"fails", "independent"}
+
+
+def test_stop_on_error_does_not_schedule_a_later_read_wave(mod, monkeypatch):
+    calls = []
+
+    def fake_call(_name, arguments):
+        calls.append(arguments["value"])
+        raise mod.SafeToolError("fixture_failure", "fixture failed safely")
+
+    monkeypatch.setattr(mod, "call_tool_impl", fake_call)
+    result = mod.execute_batch({
+        "failure_policy": "stop_on_error",
+        "max_parallel": 1,
+        "actions": [
+            action("first", arguments={"value": "first"}),
+            action("second", tool="nap_context", arguments={"value": "second"}),
+        ],
+    })
+
+    assert calls == ["first"]
+    assert result["stopped_at"] == "first"
+    assert [item["status"] for item in result["results"]] == ["failed", "skipped"]
+
+
+def test_dry_run_validates_without_invoking_actions(mod, monkeypatch):
+    calls = []
+    monkeypatch.setattr(mod, "call_tool_impl", lambda *args: calls.append(args))
+
+    result = mod.execute_batch({
+        "dry_run": True,
+        "actions": [
+            action("read", arguments={"node_id": "node-1"}),
+            action("write", tool="nap_create_node", arguments={"name": "note"}),
+        ],
+    })
+
+    assert calls == []
+    assert result["failure_policy"] == "dry_run"
+    assert result["contains_mutations"] is True
+    assert [item["status"] for item in result["results"]] == ["planned", "planned"]
+
+
+def test_cancellation_prevents_unscheduled_actions_from_starting(mod, monkeypatch):
+    cancelled = threading.Event()
+    calls = []
+
+    def fake_call(_name, arguments):
+        calls.append(arguments["value"])
+        cancelled.set()
+        return {"ok": True}
+
+    monkeypatch.setattr(mod, "call_tool_impl", fake_call)
+    mod.MCP_REQUEST_LOCAL.cancelled = cancelled
+    try:
+        with pytest.raises(mod.SafeToolError) as error:
+            mod.execute_batch({
+                "max_parallel": 1,
+                "actions": [
+                    action("first", arguments={"value": "first"}),
+                    action("second", tool="nap_context", arguments={"value": "second"}),
+                ],
+            })
+    finally:
+        del mod.MCP_REQUEST_LOCAL.cancelled
+
+    assert error.value.code == "request_cancelled"
+    assert calls == ["first"]
+
+
+def test_protocol_lists_and_dry_runs_batch(mod):
+    listed = mod.handle({"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
+    names = [item["name"] for item in listed["result"]["tools"]]
+    assert "nap_batch" in names
+
+    called = mod.handle({
+        "jsonrpc": "2.0",
+        "id": 2,
+        "method": "tools/call",
+        "params": {
+            "name": "nap_batch",
+            "arguments": {
+                "dry_run": True,
+                "actions": [action("read", arguments={"node_id": "node-1"})],
+            },
+        },
+    })
+    payload = called["result"]["structuredContent"]
+    assert called["result"]["isError"] is False
+    assert payload["ok"] is True
+    assert payload["results"][0]["status"] == "planned"
+
+
+def test_profile_gated_tool_is_rejected_before_execution(mod, monkeypatch):
+    mod.CONFIGURED_TOOL_PROFILES = set()
+    calls = []
+    monkeypatch.setattr(mod, "call_tool_impl", lambda *args: calls.append(args))
+
+    with pytest.raises(mod.SafeToolError) as error:
+        mod.execute_batch({"actions": [action("plan", tool="nap_plan_list_active")]})
+
+    assert error.value.code == "tool_profile_required"
+    assert calls == []

tests/test_mcp_memory_ergonomics_helpers.py

@@ -147,6 +147,34 @@ def test_node_by_path_returns_canonical_identity_uri(mod):
     assert "Path lookup is index/route resolution only" in result["warnings"][0]
 
 
+def test_node_by_path_resolves_project_anchor_shortcut(mod):
+    def fake_request(method, path, payload=None, **kwargs):
+        assert method == "GET"
+        assert path.endswith("/nodes/by-path?path=%40project")
+        return {
+            "identity": {
+                "node_id": "anchor-1",
+                "canonical_uri": "nap://napseer/project-1/anchor-1",
+                "project_id": "project-1",
+                "lookup_kind": "shortcut",
+                "lookup_value": "@project",
+                "legacy_full_path": "/routes/project/core",
+            },
+            "route": {
+                "id": "anchor-1",
+                "project_id": "project-1",
+                "full_path": "/routes/project/core",
+            },
+        }
+
+    mod.request_json = fake_request
+    result = mod.node_by_path({"shortcut": "@Project"})
+
+    assert result["identity"]["lookup_kind"] == "shortcut"
+    assert result["identity"]["lookup_value"] == "@project"
+    assert result["route"]["full_path"] == "/routes/project/core"
+
+
 def test_path_resolver_then_read_uses_node_id_route(mod):
     calls = []
 
@@ -255,7 +283,7 @@ def test_node_get_rejects_path_reads(mod):
 
 def test_tools_include_node_by_path_descriptor(mod):
     tool_by_name = {item["name"]: item for item in mod.tools()}
-    assert len(tool_by_name) == 51
+    assert len(tool_by_name) == 58
     assert len(tool_by_name) == len(mod.tools())
 
     get_tool = next(item for item in mod.tools() if item["name"] == "nap_node_get")
@@ -263,9 +291,19 @@ def test_tools_include_node_by_path_descriptor(mod):
     assert "node_id" in get_tool["inputSchema"]["properties"]
     assert "uri" in get_tool["inputSchema"]["properties"]
     assert "path" not in get_tool["inputSchema"]["properties"]
+    assert "node_id returned by nap_discover" in get_tool["description"]
+
+    discover_tool = next(item for item in mod.tools() if item["name"] == "nap_discover")
+    assert "node_id on every row" in discover_tool["description"]
+    assert "directly to nap_node_get" in discover_tool["description"]
 
     tool = next(item for item in mod.tools() if item["name"] == "nap_node_by_path")
-    assert tool["inputSchema"]["required"] == ["path"]
+    assert tool["inputSchema"].get("required") is None
+    assert tool["inputSchema"]["oneOf"] == [
+        {"required": ["path"]},
+        {"required": ["shortcut"]},
+    ]
+    assert "shortcut" in tool["inputSchema"]["properties"]
     assert "preview_chars" not in tool["inputSchema"]["properties"]
     assert "does not read full node content" in tool["description"]
 
@@ -425,9 +463,9 @@ def test_tee_and_bulk_create_by_path_update_by_identity(mod):
         "content_text": payload.get("content_text", existing["content_text"]),
     }
     mod.index_node = lambda node: None
-    mod.acquire_project_lock = lambda args: {"id": "lock-1", "lease_token": "lease-1"}
+    mod.acquire_project_lock = lambda args, **kwargs: {"id": "lock-1", "lease_token": "lease-1"}
     mod.lock_headers = lambda lock: {"X-Lock": lock["id"]}
-    mod.release_project_lock = lambda args: None
+    mod.release_project_lock = lambda args, **kwargs: None
     mod.request_json = lambda method, path, payload=None, **kwargs: writes.append((method, path, payload, ())) or {
         **existing,
         "content_text": payload.get("content_text", existing["content_text"]) if payload else existing["content_text"],
@@ -591,6 +629,39 @@ def test_context_and_related_use_identity_inputs(mod):
         mod.find_related_nodes({"path": "/notes/root"})
 
 
+def test_hybrid_context_resolves_path_inside_complete_local_snapshot(mod):
+    with tempfile.TemporaryDirectory() as tmp:
+        state_dir = pathlib.Path(tmp)
+        mod.AUTH_DIR = state_dir
+        mod.INDEX_PATH = state_dir / "index.sqlite"
+        mod.INDEX_LOCK_PATH = state_dir / "index.lock"
+        indexed = node("/notes/local-root", "active", node_type="note")
+        indexed["id"] = "local-root-123"
+        mod.index_node(indexed)
+        mod.set_local_graph_index_complete("project-1", True)
+
+        def unexpected_request(*_args, **_kwargs):
+            raise AssertionError("complete local context must not make an HTTP request")
+
+        mod.request_json = unexpected_request
+        result = mod.get_memory_context({
+            "path": "/notes/local-root",
+            "depth": 0,
+            "mode": "hybrid",
+        })
+
+        assert result["source"] == "local_index"
+        assert result["seed_node_ids"] == ["local-root-123"]
+        assert result["items"][0]["full_path"] == "/notes/local-root"
+        with mod.index_connect() as conn:
+            plan = conn.execute(
+                "EXPLAIN QUERY PLAN SELECT node_id FROM local_index_nodes "
+                "WHERE project_id = ? AND full_path = ? AND archived = 0",
+                ("project-1", "/notes/local-root"),
+            ).fetchall()
+        assert any("local_index_nodes_project_path" in row["detail"] for row in plan)
+
+
 def test_local_helper_ui_forms_submit_identity_for_existing_node_actions():
     source = (pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py").read_text()
     assert "name='node_id' value='{html.escape(item.get('id') or '')}'" in source
@@ -643,6 +714,7 @@ def test_recent_memory_defaults_to_paths(mod):
     assert result["ok"] is True
     assert result["view"] == "paths"
     assert result["items"] == [{
+        "node_id": "id-recent-title",
         "full_path": "/notes/recent-title",
         "type": "note",
         "status": "active",
@@ -671,6 +743,7 @@ def test_recent_memory_accepts_date_aliases_and_returns_dates(mod):
     mod.request_json = fake_request
     result = mod.recent_memory({"since": "2026-05-05", "until": "2026-05-06"})
     assert result["items"] == [{
+        "node_id": "id-recent-title",
         "full_path": "/notes/recent-title",
         "type": "note",
         "status": "active",
@@ -699,6 +772,7 @@ def test_find_defaults_to_paths_and_requests_compact_rest_view(mod):
     assert result["ok"] is True
     assert result["view"] == "paths"
     assert result["items"] == [{
+        "node_id": "id-found",
         "full_path": "/plans/found",
         "type": "plan",
         "status": "planned",
@@ -721,6 +795,8 @@ def test_discover_uses_search_when_keywords_present(mod):
     assert calls and calls[-1]["q"] == "gateway relay"
     assert calls[-1]["view"] == "summary"
     assert "date" in result["awareness"]["date_fields_in_rows"]
+    assert "item.node_id" in result["awareness"]["next_step"]
+    assert "outside discovery" in result["awareness"]["next_step"]
 
 
 def test_classify_query_points_exact_paths_to_context(mod):
@@ -748,7 +824,11 @@ def test_title_view_requests_summary_rest_view_without_content(mod):
     result = mod.list_project_nodes({"view": "titles"})
     assert result["ok"] is True
     assert result["view"] == "titles"
-    assert result["items"] == [{"title": "Readable Title", "full_path": "/notes/titled"}]
+    assert result["items"] == [{
+        "node_id": "id-titled",
+        "title": "Readable Title",
+        "full_path": "/notes/titled",
+    }]
     assert "content_text" not in result["items"][0]
 
 
@@ -781,7 +861,8 @@ def test_index_status_reports_sync_high_water_mark(mod):
         status = mod.index_status({})
         assert status["count"] == 1
         assert status["latest_node_updated_at"] == "2026-05-06T12:00:00Z"
-        assert status["sync_recommended"] is False
+        assert status["graph_complete"] is False
+        assert status["sync_recommended"] is True
 
 
 def test_index_sync_fetches_incremental_changes_and_removes_archived_nodes(mod):
@@ -794,6 +875,7 @@ def test_index_sync_fetches_incremental_changes_and_removes_archived_nodes(mod):
         existing["id"] = "existing-node"
         existing["updated_at"] = "2026-05-06T12:00:00Z"
         mod.index_node(existing)
+        mod.set_local_graph_index_complete("project-1", True)
 
         fresh = node("/notes/fresh", "active", content="fresh sync needle")
         fresh["id"] = "fresh-node"
@@ -817,6 +899,8 @@ def test_index_sync_fetches_incremental_changes_and_removes_archived_nodes(mod):
         assert result["seen"] == 2
         assert result["indexed"] == 1
         assert result["removed"] == 1
+        assert result["mode"] == "incremental"
+        assert result["index"]["graph_complete"] is True
         assert calls
         local = mod.search_local_index({"q": "fresh needle"})
         assert [item["id"] for item in local["items"]] == ["fresh-node"]

tests/test_mcp_memory_node_encryption.py

@@ -194,8 +194,13 @@ def run():
             "precondition": {"revision": "r1", "read_fingerprint": "fp1"},
             "content_op": {"op": "replace_all", "content_text": "plaintext"},
         })
-    except RuntimeError as exc:
+    except mod.SafeToolError as exc:
-        assert "cannot safely patch encrypted content or metadata" in str(exc)
+        assert exc.code == "encrypted_patch_requires_full_replacement"
+        assert "cannot safely patch client-E2E encrypted content or metadata" in str(exc)
+        assert "nap_bulk" in str(exc)
+        assert "fresh read" in str(exc)
+        assert "nap_tee" not in str(exc)
+        assert "nap_patch" not in str(exc)
     else:
         raise AssertionError("nap_node_patch content updates must fail closed for encrypted projects")
 
@@ -296,7 +301,12 @@ def run():
     active = mod.list_project_nodes({"folder_path": "/plans", "active_only": True})
     assert [item["full_path"] for item in active["items"]] == ["/plans/active"]
     status_filtered = mod.list_project_nodes({"folder_path": "/plans", "status": ["completed"], "view": "paths"})
-    assert status_filtered["items"] == [{"full_path": "/plans/done", "type": "note", "status": "completed"}]
+    assert status_filtered["items"] == [{
+        "node_id": "done",
+        "full_path": "/plans/done",
+        "type": "note",
+        "status": "completed",
+    }]
 
     def search_response(method, path, payload=None, **kwargs):
         assert "q=" in path
@@ -383,9 +393,9 @@ def run():
     assert patch_body["encrypted_content_envelope"]["aad_subject"] == "node_id:node-1"
 
     bulk_writes = []
-    mod.acquire_project_lock = lambda args: {"id": "lock-1", "lease_token": "lease-1"}
+    mod.acquire_project_lock = lambda args, **kwargs: {"id": "lock-1", "lease_token": "lease-1"}
     mod.lock_headers = lambda lock: {"X-Lock": lock["id"]}
-    mod.release_project_lock = lambda args: None
+    mod.release_project_lock = lambda args, **kwargs: None
     mod.try_get_node_by_path = lambda path, allow_agent=False: None
     mod.request_json = lambda method, path, payload=None, **kwargs: bulk_writes.append((method, path, payload, kwargs)) or {
         "id": "node-bulk",

tests/test_mcp_memory_surface_efficiency.py

@@ -2,6 +2,7 @@
 """Contract tests for the reduced, bounded agent-memory surface."""
 
 import importlib.util
+import json
 import pathlib
 import sys
 import tempfile
@@ -24,15 +25,19 @@ def load_module():
 def mod(monkeypatch):
     monkeypatch.delenv("NAPSEER_TOOL_PROFILES", raising=False)
     module = load_module()
+    module._auth_state = tempfile.TemporaryDirectory()
+    module.AUTH_PATH = pathlib.Path(module._auth_state.name) / "auth.json"
+    module.AUTH = {"account_id": "acct-1", "token": "token"}
     module.CONFIGURED_TOOL_PROFILES = None
     return module
 
 
-def test_default_surface_is_exactly_the_core_fourteen(mod):
+def test_default_surface_includes_knowledge_files_and_coordination(mod):
     assert [tool["name"] for tool in mod.tools()] == [
         "nap_apropos", "nap_man", "nap_doctor", "nap_whoami",
+        "nap_operation_list", "nap_operation", "nap_file_list", "nap_file_upload", "nap_file_download", "nap_file_archive",
         "nap_discover", "nap_context", "nap_node_by_path", "nap_node_get",
-        "nap_create_node", "nap_node_patch", "nap_bulk", "nap_ln",
+        "nap_create_node", "nap_node_patch", "nap_bulk", "nap_batch", "nap_ln",
         "nap_mv", "nap_rm",
     ]
 
@@ -64,6 +69,67 @@ def test_pruned_and_profile_gated_calls_are_actionable(mod):
     assert "workflow" in gated.value.safe_message
 
 
+def test_removed_write_tools_and_unknown_manual_lookups_are_actionable(mod):
+    tee = mod.man_tool({"tool": "nap_tee"})
+    assert tee["removed"] is True
+    assert "nap_create_node" in tee["replacement"]
+    assert "nap_bulk" in tee["replacement"]
+
+    patch = mod.man_tool({"tool": "nap_patch"})
+    assert patch["removed"] is True
+    assert "nap_node_patch" in patch["replacement"]
+    assert "nap_bulk" in patch["replacement"]
+
+    for name in ("nap_tee", "nap_patch"):
+        with pytest.raises(mod.SafeToolError) as removed:
+            mod.call_tool_impl(name, {})
+        assert removed.value.code == "tool_removed"
+
+    with pytest.raises(mod.SafeToolError) as unknown:
+        mod.man_tool({"tool": "not_a_real_tool"})
+    assert unknown.value.code == "unknown_tool"
+    assert "nap_apropos" in unknown.value.safe_message
+
+
+def test_node_patch_rejects_malformed_arguments_before_project_or_node_reads(mod):
+    mod.resolve_project_id = lambda args: (_ for _ in ()).throw(
+        AssertionError("invalid arguments must fail before project resolution")
+    )
+    mod.get_node_by_id = lambda args: (_ for _ in ()).throw(
+        AssertionError("invalid arguments must fail before node reads")
+    )
+    base = {
+        "node_id": "node-1",
+        "precondition": {"revision": "r1", "read_fingerprint": "fp1"},
+    }
+
+    with pytest.raises(mod.SafeToolError) as legacy_metadata:
+        mod.guarded_node_patch({**base, "metadata_patch": {"set": {"status": "done"}}})
+    assert legacy_metadata.value.code == "invalid_arguments"
+    assert "metadata_patch" in legacy_metadata.value.safe_message
+
+    with pytest.raises(mod.SafeToolError) as legacy_content:
+        mod.guarded_node_patch({
+            **base,
+            "content_op": {"kind": "replace_all", "content_text": "updated"},
+        })
+    assert legacy_content.value.code == "invalid_arguments"
+    assert "use 'op'" in legacy_content.value.safe_message
+
+
+def test_node_patch_argument_validation_accepts_canonical_shapes(mod):
+    mod.validate_node_patch_arguments({
+        "node_id": "node-1",
+        "precondition": {"revision": "r1", "read_fingerprint": "fp1"},
+        "content_op": {"op": "replace_all", "content_text": "updated"},
+        "merge_metadata": {"status": "done"},
+        "tag_ops": {"add": ["verified"]},
+        "link_ops": [
+            {"op": "add", "path": "/plans/example", "relation": "implements"}
+        ],
+    })
+
+
 def test_context_and_node_read_schemas_are_bounded(mod):
     by_name = {tool["name"]: tool for tool in mod.all_tools()}
     context = by_name["nap_context"]["inputSchema"]["properties"]
@@ -73,6 +139,10 @@ def test_context_and_node_read_schemas_are_bounded(mod):
     assert "relations" in context
     assert "max_edges" in context
     assert "max_per_level" in context
+    assert context["node_ids"]["maxItems"] == 10
+    assert context["paths"]["maxItems"] == 10
+    assert context["shortcuts"]["maxItems"] == 10
+    assert context["uris"]["maxItems"] == 10
 
     node_get = by_name["nap_node_get"]["inputSchema"]["properties"]
     assert node_get["view"]["enum"] == ["content", "outline", "edit"]
@@ -83,10 +153,47 @@ def test_tool_text_does_not_duplicate_large_structured_payload(mod):
     payload = {"ok": True, "items": [{"content_text": "x" * 20_000}]}
     result = mod.tool_result(payload)
     assert result["structuredContent"] == payload
-    assert len(result["content"][0]["text"]) < 200
+    assert len(result["content"][0]["text"]) < 2_000
     assert "20000" not in result["content"][0]["text"]
 
 
+def test_tool_text_keeps_small_structured_results_useful_for_text_only_clients(mod):
+    payload = {
+        "status": "current",
+        "full_path": "/docs/useful",
+        "view": "content",
+        "content_text": "A small useful body.",
+    }
+
+    result = mod.tool_result(payload)
+
+    assert json.loads(result["content"][0]["text"]) == payload
+
+
+def test_large_graph_text_fallback_contains_bounded_node_and_edge_previews(mod):
+    payload = {
+        "ok": True,
+        "nodes": [
+            {"id": f"node-{index}", "full_path": f"/docs/{index}", "title": f"Node {index}"}
+            for index in range(100)
+        ],
+        "edges": [
+            {"source": f"node-{index}", "target": f"node-{index + 1}", "relation": "references"}
+            for index in range(99)
+        ],
+        "warnings": ["Use first-class links."],
+    }
+
+    fallback = json.loads(mod.tool_result(payload)["content"][0]["text"])
+
+    assert fallback["nodes_count"] == 100
+    assert fallback["edges_count"] == 99
+    assert len(fallback["nodes_preview"]) == 5
+    assert len(fallback["edges_preview"]) == 5
+    assert fallback["warnings"] == ["Use first-class links."]
+    assert fallback["structured_result"] is True
+
+
 def test_local_context_returns_title_radius_without_node_bodies(mod):
     state = tempfile.TemporaryDirectory()
     mod._surface_state = state
@@ -123,6 +230,315 @@ def test_local_context_returns_title_radius_without_node_bodies(mod):
     assert result["source"] == "local_index"
 
 
+def test_local_context_resolves_anchor_shortcut_without_extra_tool(mod):
+    state = tempfile.TemporaryDirectory()
+    mod._surface_shortcut_state = state
+    mod.AUTH_DIR = pathlib.Path(state.name)
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
+    mod.index_node({
+        "id": "anchor", "project_id": "project-1",
+        "full_path": "/routes/project/core", "folder_path": "/routes/project",
+        "name": "core", "type": "anchor_route", "aliases": ["shortcut.project"],
+        "metadata": {"title": "Project"}, "tags": [],
+        "links": [{"node_id": "definition", "relation": "defines"}],
+        "content_text": "route body", "updated_at": "2026-08-16T12:00:00Z",
+    })
+    mod.index_node({
+        "id": "definition", "project_id": "project-1",
+        "full_path": "/documentation/product/what-we-are-building",
+        "folder_path": "/documentation/product", "name": "what-we-are-building",
+        "type": "definition", "aliases": [], "metadata": {"title": "Product"},
+        "tags": [], "links": [], "content_text": "definition body",
+        "updated_at": "2026-08-16T12:01:00Z",
+    })
+
+    result = mod.get_memory_context({
+        "shortcut": "@Project", "mode": "local", "depth": 1,
+        "direction": "outgoing",
+    })
+
+    assert result["seed_node_ids"] == ["anchor"]
+    assert [item["id"] for item in result["nodes"]] == ["anchor", "definition"]
+    assert all("content_text" not in item for item in result["nodes"])
+
+
+def test_duplicate_local_anchor_shortcuts_fail_closed(mod):
+    state = tempfile.TemporaryDirectory()
+    mod._surface_duplicate_shortcut_state = state
+    mod.AUTH_DIR = pathlib.Path(state.name)
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
+    for index in range(2):
+        mod.index_node({
+            "id": f"anchor-{index}", "project_id": "project-1",
+            "full_path": f"/routes/project/{index}", "folder_path": "/routes/project",
+            "name": str(index), "type": "anchor_route", "aliases": ["shortcut.project"],
+            "metadata": {}, "tags": [], "links": [], "content_text": "",
+            "updated_at": f"2026-08-16T12:0{index}:00Z",
+        })
+
+    with pytest.raises(mod.SafeToolError) as ambiguous:
+        mod.get_memory_context({"shortcut": "@project", "mode": "local"})
+
+    assert ambiguous.value.code == "anchor_shortcut_ambiguous"
+
+
+def test_anchor_integrity_reports_budgets_duplicates_and_broken_links(mod):
+    state = tempfile.TemporaryDirectory()
+    mod._surface_anchor_health_state = state
+    mod.AUTH_DIR = pathlib.Path(state.name)
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
+    for index in range(2):
+        mod.index_node({
+            "id": f"anchor-{index}", "project_id": "project-1",
+            "full_path": f"/routes/project/{index}", "folder_path": "/routes/project",
+            "name": str(index), "type": "anchor_route", "aliases": ["shortcut.project"],
+            "metadata": {}, "tags": [],
+            "links": ([{"path": "/missing", "relation": "routes-to"}] if index == 0 else []),
+            "content_text": "", "updated_at": f"2026-08-16T12:0{index}:00Z",
+        })
+    mod.set_local_graph_index_complete("project-1", True)
+
+    report = mod.local_anchor_integrity()
+
+    assert report["ok"] is False
+    assert report["graph_complete"] is True
+    assert report["duplicate_shortcuts"][0]["shortcut"] == "@project"
+    assert report["broken_anchor_links"] == [{
+        "source": "/routes/project/0",
+        "target": "/missing",
+        "relation": "routes-to",
+    }]
+
+
+def test_local_context_reports_partial_and_total_missing_seeds_actionably(mod):
+    state = tempfile.TemporaryDirectory()
+    mod._surface_missing_seed_state = state
+    mod.AUTH_DIR = pathlib.Path(state.name)
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
+    mod.index_node({
+        "id": "root", "project_id": "project-1", "full_path": "/docs/root",
+        "folder_path": "/docs", "name": "root", "type": "note",
+        "metadata": {"title": "Root"}, "tags": [], "links": [],
+        "content_text": "body", "updated_at": "2026-08-16T12:00:00Z",
+    })
+
+    partial = mod.get_memory_context({
+        "node_ids": ["root", "missing"], "mode": "local", "depth": 0,
+    })
+
+    assert partial["requested_seed_node_ids"] == ["root", "missing"]
+    assert partial["seed_node_ids"] == ["root"]
+    assert partial["missing_seed_node_ids"] == ["missing"]
+    assert any("active project" in warning for warning in partial["warnings"])
+
+    with pytest.raises(mod.SafeToolError) as missing:
+        mod.get_memory_context({"node_id": "missing", "mode": "local"})
+    assert missing.value.code == "context_seed_not_found"
+    assert "nap_whoami" in missing.value.safe_message
+    assert "nap_discover" in missing.value.safe_message
+
+
+def test_context_translates_server_not_found_into_seed_diagnostic(mod):
+    def missing_request(*_args, **_kwargs):
+        raise mod.SafeToolError(
+            "http_404",
+            "Napseer GET request failed with HTTP 404.",
+            status=404,
+            service_code="not_found",
+        )
+
+    mod.request_json = missing_request
+
+    with pytest.raises(mod.SafeToolError) as missing:
+        mod.get_memory_context({"node_id": "missing", "mode": "server"})
+
+    assert missing.value.code == "context_seed_not_found"
+    assert missing.value.status == 404
+    assert mod.safe_exception_payload(missing.value)["service_code"] == "not_found"
+
+
+def test_context_rejects_cross_project_uri_before_graph_reads(mod):
+    with pytest.raises(mod.SafeToolError) as mismatch:
+        mod.get_memory_context({
+            "uri": "nap://other-project/project-2/node-1",
+            "mode": "server",
+        })
+
+    assert mismatch.value.code == "context_project_mismatch"
+    assert "nap_whoami" in mismatch.value.safe_message
+
+
+def test_local_context_projects_requested_view_and_bounds_frontier(mod):
+    state = tempfile.TemporaryDirectory()
+    mod._surface_view_state = state
+    mod.AUTH_DIR = pathlib.Path(state.name)
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
+    children = []
+    for index in range(5):
+        children.append({
+            "id": f"child-{index}", "project_id": "project-1",
+            "full_path": f"/docs/child-{index}", "folder_path": "/docs",
+            "name": f"child-{index}", "type": "note", "metadata": {},
+            "tags": [], "links": [], "content_text": "",
+            "updated_at": f"2026-08-16T12:0{index + 1}:00Z",
+        })
+    mod.index_node({
+        "id": "root", "project_id": "project-1", "full_path": "/docs/root",
+        "folder_path": "/docs", "name": "root", "type": "note",
+        "metadata": {"title": "Root"}, "tags": [],
+        "links": [
+            {"node_id": child["id"], "relation": "references"}
+            for child in children
+        ],
+        "content_text": "", "updated_at": "2026-08-16T12:00:00Z",
+    })
+    for child in children:
+        mod.index_node(child)
+
+    result = mod.get_memory_context({
+        "node_id": "root", "mode": "local", "depth": 1,
+        "max_nodes": 2, "max_per_level": 1, "view": "paths",
+    })
+
+    assert result["view"] == "paths"
+    assert "title" not in result["nodes"][0]
+    assert set(result["nodes"][0]) == {
+        "id", "full_path", "type", "status", "updated_at", "depth"
+    }
+    assert len(result["frontier"]) <= 1
+    assert "max_frontier" in result["truncation_reasons"]
+    assert result["node_count"] == 2
+    assert result["edge_count"] == 1
+
+
+def test_context_explains_empty_first_class_graph(mod):
+    state = tempfile.TemporaryDirectory()
+    mod._surface_empty_graph_state = state
+    mod.AUTH_DIR = pathlib.Path(state.name)
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
+    mod.index_node({
+        "id": "root", "project_id": "project-1", "full_path": "/docs/root",
+        "folder_path": "/docs", "name": "root", "type": "note",
+        "metadata": {"title": "Root"}, "tags": [], "links": [],
+        "content_text": "mentions /docs/other only in prose",
+        "updated_at": "2026-08-16T12:00:00Z",
+    })
+
+    result = mod.get_memory_context({"node_id": "root", "mode": "local", "depth": 1})
+
+    assert any("Prose mentions are not traversed" in warning for warning in result["warnings"])
+    assert "nap_ln" in result["next_action"]
+
+
+def test_hybrid_context_falls_back_until_local_graph_coverage_is_complete(mod):
+    state = tempfile.TemporaryDirectory()
+    mod._surface_fallback_state = state
+    mod.AUTH_DIR = pathlib.Path(state.name)
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
+    mod.index_node({
+        "id": "root", "project_id": "project-1", "full_path": "/docs/root",
+        "folder_path": "/docs", "name": "root", "type": "note",
+        "metadata": {"title": "Root"}, "tags": [], "links": [],
+        "content_text": "body", "updated_at": "2026-08-16T12:00:00Z",
+    })
+    calls = []
+
+    def server_context(_args, seed_ids):
+        calls.append(seed_ids)
+        return {"ok": True, "nodes": [{"id": "root"}], "edges": [], "source": "server"}
+
+    mod.server_memory_context = server_context
+
+    result = mod.get_memory_context({"node_id": "root", "mode": "hybrid"})
+
+    assert result["source"] == "server"
+    assert result["local_fallback"]
+    assert calls == [["root"]]
+
+
+def test_local_context_does_not_emit_edges_beyond_requested_depth(mod):
+    state = tempfile.TemporaryDirectory()
+    mod._surface_depth_state = state
+    mod.AUTH_DIR = pathlib.Path(state.name)
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
+    nodes = [
+        {
+            "id": "root", "project_id": "project-1", "full_path": "/docs/root",
+            "folder_path": "/docs", "name": "root", "type": "note", "metadata": {},
+            "tags": [], "links": [
+                {"node_id": "left", "relation": "references"},
+                {"node_id": "right", "relation": "references"},
+            ], "content_text": "", "updated_at": "2026-08-16T12:00:00Z",
+        },
+        {
+            "id": "left", "project_id": "project-1", "full_path": "/docs/left",
+            "folder_path": "/docs", "name": "left", "type": "note", "metadata": {},
+            "tags": [], "links": [{"node_id": "right", "relation": "explains"}],
+            "content_text": "", "updated_at": "2026-08-16T12:01:00Z",
+        },
+        {
+            "id": "right", "project_id": "project-1", "full_path": "/docs/right",
+            "folder_path": "/docs", "name": "right", "type": "note", "metadata": {},
+            "tags": [], "links": [], "content_text": "",
+            "updated_at": "2026-08-16T12:02:00Z",
+        },
+    ]
+    for item in nodes:
+        mod.index_node(item)
+
+    result = mod.get_memory_context({
+        "node_id": "root", "mode": "local", "depth": 1, "direction": "both",
+    })
+
+    assert {item["id"] for item in result["nodes"]} == {"root", "left", "right"}
+    assert result["edges"] == [
+        {"source": "root", "target": "left", "relation": "references"},
+        {"source": "root", "target": "right", "relation": "references"},
+    ]
+
+
+def test_index_sync_upgrades_pre_graph_index_with_full_reindex(mod):
+    state = tempfile.TemporaryDirectory()
+    mod._surface_upgrade_state = state
+    mod.AUTH_DIR = pathlib.Path(state.name)
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
+    mod.index_node({
+        "id": "old", "project_id": "project-1", "full_path": "/docs/old",
+        "folder_path": "/docs", "name": "old", "type": "note",
+        "metadata": {}, "tags": [], "links": [], "content_text": "old",
+        "updated_at": "2026-08-16T12:00:00Z",
+    })
+    fresh = {
+        "id": "fresh", "project_id": "project-1", "full_path": "/docs/fresh",
+        "folder_path": "/docs", "name": "fresh", "type": "note",
+        "metadata": {}, "tags": [],
+        "links": [{"node_id": "target", "relation": "references"}],
+        "content_text": "fresh", "updated_at": "2026-08-16T12:01:00Z",
+    }
+    mod.request_json = lambda *_args, **_kwargs: {"items": [fresh], "next_cursor": None}
+
+    result = mod.sync_local_index({})
+
+    assert result["mode"] == "full_reindex"
+    assert result["reason"] == "local_graph_index_incomplete"
+    assert result["graph_complete"] is True
+    assert result["index"]["graph_complete"] is True
+    with mod.index_connect() as conn:
+        assert conn.execute(
+            "SELECT COUNT(*) FROM local_index_edges WHERE project_id = ?",
+            ("project-1",),
+        ).fetchone()[0] == 1
+
+
 def test_discovery_sort_and_returned_item_facets(mod):
     mod.list_project_nodes = lambda args: {
         "items": [

tests/test_mcp_search_latency.py

@@ -1,7 +1,11 @@
 import importlib.util
 import pathlib
 import sys
+import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
 
 
 def load_module():
@@ -30,7 +34,14 @@ def test_natural_language_discovery_uses_one_backend_search_request(tmp_path):
 
     def request(method, path, payload=None, **_kwargs):
         calls.append((method, path, payload))
-        return {"items": [], "next_cursor": None, "query_analysis": {"mode": "terms"}}
+        return {
+            "items": [],
+            "next_cursor": None,
+            "query_analysis": {"mode": "terms"},
+            "search_strategy": "identity_fts",
+            "search_coverage": "identity_only",
+            "search_projection": "postgres_identity_v1",
+        }
 
     mod.request_json = request
 
@@ -46,6 +57,9 @@ def test_natural_language_discovery_uses_one_backend_search_request(tmp_path):
     assert len(calls) == 1
     assert calls[0][0] == "GET"
     assert "/nodes?" in calls[0][1]
+    assert result["search_strategy"] == "identity_fts"
+    assert result["search_coverage"] == "identity_only"
+    assert result["search_projection"] == "postgres_identity_v1"
 
 
 def test_sparse_index_search_latency_does_not_scale_with_query_variants(tmp_path):
@@ -74,3 +88,433 @@ def test_sparse_index_search_latency_does_not_scale_with_query_variants(tmp_path
     elapsed = time.monotonic() - started
     assert len(calls) == 1
     assert elapsed < 0.15
+
+
+def test_complete_local_search_skips_remote_encryption_preflight_and_search(tmp_path):
+    mod = load_module()
+    mod.INDEX_PATH = tmp_path / "index.sqlite"
+    mod.INDEX_PATH.touch()
+    mod.resolve_project_id = lambda _args: "project-1"
+    mod.project_encryption_state = lambda _project_id: "unknown"
+    items = [
+        {
+            "id": f"node-{index}",
+            "project_id": "project-1",
+            "full_path": f"/notes/result-{index}",
+            "name": f"result-{index}",
+            "type": "note",
+            "tags": [],
+            "updated_at": "2026-08-22T00:00:00Z",
+        }
+        for index in range(5)
+    ]
+    mod.local_index_diagnostics = lambda _project_id: {
+        "exists": True,
+        "graph_complete": True,
+    }
+    mod.search_local_index = lambda _args: {
+        "items": items,
+        "query": {"terms": ["result"]},
+        "search_strategies": [],
+        "index": {"exists": True, "graph_complete": True, "used": True},
+    }
+
+    def unexpected_request(*_args, **_kwargs):
+        raise AssertionError("complete local search must not make an HTTP request")
+
+    mod.request_json = unexpected_request
+    result = mod.search_memory({"q": "result", "limit": 5, "view": "summary"})
+
+    assert result["sources"] == ["local"]
+    assert result["search_strategy"] == "local_fts"
+    assert result["search_coverage"] == "identity_and_local_body"
+    assert result["search_projection"] == "sqlite_fts5_current"
+    assert len(result["items"]) == 5
+    assert all(item["node_id"].startswith("node-") for item in result["items"])
+
+
+def indexed_node(path, *, node_type="note", tags=None, status="active", content=""):
+    name = path.rsplit("/", 1)[-1]
+    return {
+        "id": f"node-{path.strip('/').replace('/', '-')}",
+        "project_id": "project-1",
+        "full_path": path,
+        "folder_path": path.rsplit("/", 1)[0] or None,
+        "name": name,
+        "title": name,
+        "type": node_type,
+        "tags": tags or [],
+        "aliases": [],
+        "links": [],
+        "metadata": {"status": status},
+        "content_text": content,
+        "encryption_state": "legacy_plaintext",
+        "updated_at": "2026-08-22T00:00:00Z",
+    }
+
+
+def configure_local_index(mod, tmp_path):
+    mod.AUTH_DIR = tmp_path
+    mod.INDEX_PATH = tmp_path / "index.sqlite"
+    mod.INDEX_LOCK_PATH = tmp_path / "index.lock"
+    mod.resolve_project_id = lambda _args: "project-1"
+
+
+def test_single_pass_ranking_prefers_exact_canonical_identity(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    mod.index_node(
+        indexed_node(
+            "/rules/memory-source-policy",
+            node_type="rule",
+            tags=["memory", "policy"],
+            content="Canonical memory source rule.",
+        )
+    )
+    for index in range(12):
+        mod.index_node(
+            indexed_node(
+                f"/work/current/incidental-{index}",
+                content="memory source policy " * (index + 2),
+            )
+        )
+
+    result = mod.search_local_index({"q": "memory source policy", "limit": 5})
+
+    assert result["items"][0]["full_path"] == "/rules/memory-source-policy"
+    assert "exact_identity" in result["items"][0]["match_reason"]
+    assert "full_path" in result["items"][0]["matched_fields"]
+    assert result["search_strategies"] == [
+        {
+            "strategy": "single_pass_weighted_candidates",
+            "matched": 13,
+            "returned": 5,
+            "query_count": 1,
+            "candidate_limit": 30,
+        }
+    ]
+
+
+def test_complete_local_compact_browse_uses_keyset_cursor_without_http(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    for index in range(7):
+        item = indexed_node(
+            f"/plans/item-{index}",
+            node_type="plan",
+            tags=["performance"],
+            status="active",
+        )
+        item["updated_at"] = f"2026-08-2{index}T00:00:00Z"
+        mod.index_node(item)
+    mod.set_local_graph_index_complete("project-1", True)
+
+    def unexpected_request(*_args, **_kwargs):
+        raise AssertionError("complete local compact browse must not make an HTTP request")
+
+    mod.request_json = unexpected_request
+    first = mod.discover_memory(
+        {
+            "folder_path": "/plans",
+            "tag": "performance",
+            "limit": 3,
+            "view": "paths",
+        }
+    )
+    second = mod.discover_memory(
+        {
+            "folder_path": "/plans",
+            "tag": "performance",
+            "limit": 3,
+            "view": "paths",
+            "cursor": first["next_cursor"],
+        }
+    )
+
+    assert first["source"] == "local_index"
+    assert first["search_strategy"] == "local_snapshot_browse"
+    assert first["search_projection"] == "sqlite_compact_identity_v1"
+    assert first["freshness"]["graph_complete"] is True
+    assert first["next_cursor"].startswith(mod.LOCAL_BROWSE_CURSOR_PREFIX)
+    assert [item["updated_at"] for item in first["items"]] == sorted(
+        (item["updated_at"] for item in first["items"]), reverse=True
+    )
+    assert not ({item["node_id"] for item in first["items"]} & {item["node_id"] for item in second["items"]})
+    assert all(item["node_id"] for item in first["items"] + second["items"])
+
+
+def test_local_compact_browse_cursor_is_bound_to_filters(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    for index in range(3):
+        mod.index_node(indexed_node(f"/plans/item-{index}", node_type="plan"))
+    mod.set_local_graph_index_complete("project-1", True)
+
+    first = mod.discover_memory({"folder_path": "/plans", "limit": 1, "view": "summary"})
+    with pytest.raises(RuntimeError, match="cursor filters do not match"):
+        mod.discover_memory(
+            {
+                "folder_path": "/decisions",
+                "limit": 1,
+                "view": "summary",
+                "cursor": first["next_cursor"],
+            }
+        )
+
+
+def test_local_compact_browse_title_order_is_global_across_pages(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    for title in ["Zulu", "alpha", "Mike", "bravo"]:
+        item = indexed_node(f"/notes/{title}")
+        item["title"] = title
+        mod.index_node(item)
+    mod.set_local_graph_index_complete("project-1", True)
+
+    first = mod.discover_memory({"limit": 2, "view": "titles", "sort": "title_asc"})
+    second = mod.discover_memory(
+        {"limit": 2, "view": "titles", "sort": "title_asc", "cursor": first["next_cursor"]}
+    )
+
+    assert [item["title"] for item in first["items"] + second["items"]] == [
+        "alpha",
+        "bravo",
+        "Mike",
+        "Zulu",
+    ]
+
+
+def test_compact_browse_falls_back_once_when_local_contract_is_incomplete(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    mod.index_node(indexed_node("/plans/local", node_type="plan"))
+    calls = []
+
+    def request(method, path, payload=None, **_kwargs):
+        calls.append((method, path, payload))
+        return {"items": [], "next_cursor": None}
+
+    mod.request_json = request
+    result = mod.discover_memory({"folder_path": "/plans", "limit": 5, "view": "paths"})
+
+    assert result["ok"] is True
+    assert len(calls) == 1
+    assert calls[0][0] == "GET"
+    assert "/nodes?" in calls[0][1]
+    with pytest.raises(RuntimeError, match="complete active-node index"):
+        mod.discover_memory(
+            {"folder_path": "/plans", "limit": 5, "view": "paths", "mode": "local"}
+        )
+
+
+def test_local_search_reads_overlap_without_writer_file_lock(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    for index in range(4):
+        mod.index_node(indexed_node(f"/notes/result-{index}", content="shared needle"))
+    mod.set_local_graph_index_complete("project-1", True)
+
+    def unexpected_writer_lock(*_args, **_kwargs):
+        raise AssertionError("read-only local search must not acquire the writer file lock")
+
+    mod.index_file_lock = unexpected_writer_lock
+    original_rank = mod.rank_search_items
+    rendezvous = threading.Barrier(2)
+
+    def overlapping_rank(items, analysis, limit=None):
+        rendezvous.wait(timeout=1)
+        return original_rank(items, analysis, limit=limit)
+
+    mod.rank_search_items = overlapping_rank
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        futures = [
+            executor.submit(mod.search_local_index, {"q": "shared needle", "limit": 2})
+            for _ in range(2)
+        ]
+        results = [future.result(timeout=2) for future in futures]
+
+    assert all(len(result["items"]) == 2 for result in results)
+
+
+def test_index_schema_initializes_once_and_revalidates_replacement(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    original_initialize = mod.initialize_index_schema
+    calls = []
+
+    def counted_initialize(conn):
+        calls.append(str(mod.INDEX_PATH))
+        return original_initialize(conn)
+
+    mod.initialize_index_schema = counted_initialize
+    with mod.index_connect():
+        pass
+    with mod.index_connect():
+        pass
+    assert len(calls) == 1
+
+    replacement = tmp_path / "replacement.sqlite"
+    replacement.touch()
+    replacement.replace(mod.INDEX_PATH)
+    with mod.index_connect():
+        pass
+    assert len(calls) == 2
+
+
+def test_discover_contract_advertises_local_routing_and_pagination():
+    mod = load_module()
+    tool = next(item for item in mod.tools() if item["name"] == "nap_discover")
+    properties = tool["inputSchema"]["properties"]
+
+    assert "cursor" in properties
+    assert properties["mode"]["enum"] == ["hybrid", "server", "local"]
+
+
+def test_query_planner_normalizes_plurals_and_reports_intent():
+    mod = load_module()
+
+    analysis = mod.analyze_search_query(
+        "where did we decide how search queries and policies should work"
+    )
+
+    assert "query" in analysis["terms"]
+    assert "querie" not in analysis["terms"]
+    assert analysis["inferred_intent"] == "decision"
+    assert len(mod.fts_query_variants(analysis["raw"])[1]) == 1
+
+
+def test_local_search_pushes_structured_filters_into_candidate_query(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    nodes = [
+        indexed_node(
+            "/decisions/search-governance",
+            node_type="decision",
+            tags=["search"],
+            content="search governance",
+        ),
+        indexed_node(
+            "/decisions/old-search-governance",
+            node_type="decision",
+            tags=["search"],
+            status="superseded",
+            content="search governance",
+        ),
+        indexed_node(
+            "/plans/search-governance",
+            node_type="plan",
+            tags=["search"],
+            content="search governance",
+        ),
+        indexed_node(
+            "/decisions/other-search",
+            node_type="decision",
+            tags=["other"],
+            content="search governance",
+        ),
+    ]
+    for node in nodes:
+        mod.index_node(node)
+
+    result = mod.search_local_index(
+        {
+            "q": "search governance",
+            "folder_path": "/decisions",
+            "tag": "search",
+            "type": "decision",
+            "status": "active",
+            "limit": 10,
+        }
+    )
+
+    assert [item["full_path"] for item in result["items"]] == [
+        "/decisions/search-governance"
+    ]
+    assert result["index"]["query_count"] == 1
+    assert set(result["index"]["filters_applied"]) >= {
+        "folder_path",
+        "tag",
+        "type",
+        "status",
+    }
+
+
+def test_explicit_intent_boosts_without_filtering(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    mod.index_node(
+        indexed_node(
+            "/decisions/gateway-relay-authentication",
+            node_type="decision",
+            content="gateway relay authentication",
+        )
+    )
+    mod.index_node(
+        indexed_node(
+            "/work/current/gateway-relay-authentication",
+            content="gateway relay authentication",
+        )
+    )
+
+    result = mod.search_local_index(
+        {"q": "gateway relay authentication", "intent": "decision", "limit": 10}
+    )
+
+    assert result["items"][0]["full_path"].startswith("/decisions/")
+    assert len(result["items"]) == 2
+
+
+def test_complete_filtered_local_search_stays_off_network(tmp_path):
+    mod = load_module()
+    configure_local_index(mod, tmp_path)
+    mod.index_node(
+        indexed_node(
+            "/documentation/facturacion-electronica",
+            tags=["billing"],
+            content="facturación electrónica",
+        )
+    )
+    mod.index_node(
+        indexed_node(
+            "/work/current/facturacion-migration",
+            tags=["migration"],
+            content="facturación electrónica",
+        )
+    )
+    mod.set_local_graph_index_complete("project-1", True)
+
+    def unexpected_request(*_args, **_kwargs):
+        raise AssertionError("a complete filtered local result must not use the network")
+
+    mod.request_json = unexpected_request
+    result = mod.search_memory(
+        {
+            "q": "facturación",
+            "tag": "billing",
+            "limit": 1,
+            "view": "summary",
+        }
+    )
+
+    assert result["sources"] == ["local"]
+    assert [item["full_path"] for item in result["items"]] == [
+        "/documentation/facturacion-electronica"
+    ]
+
+
+def test_discovery_schema_documents_dynamic_sort_and_intent():
+    mod = load_module()
+    tool = next(item for item in mod.tools() if item["name"] == "nap_discover")
+    properties = tool["inputSchema"]["properties"]
+
+    assert properties["intent"]["enum"] == [
+        "any",
+        "decision",
+        "rule",
+        "plan",
+        "task",
+        "implementation",
+        "documentation",
+    ]
+    assert "default" not in properties["sort"]
+    assert "relevance" in properties["sort"]["description"]

tests/test_mcp_supervisor.py

@@ -70,10 +70,26 @@ for line in sys.stdin:
 
 
 class SupervisorClient:
-    def __init__(self, worker_path: pathlib.Path, project_root: pathlib.Path | None = None):
+    def __init__(
+        self,
+        worker_path: pathlib.Path,
+        project_root: pathlib.Path | None = None,
+        watch_paths: tuple[pathlib.Path, ...] = (),
+        response_timeout: int = 5,
+        cancel_grace: int = 5,
+        max_in_flight: int = 32,
+        write_timeout: int = 5,
+    ):
         environment = os.environ.copy()
         environment["NAPSEER_MCP_WORKER_PATH"] = str(worker_path)
-        environment["NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS"] = "5"
+        if watch_paths:
+            environment["NAPSEER_MCP_WORKER_WATCH_PATHS"] = os.pathsep.join(
+                str(path) for path in watch_paths
+            )
+        environment["NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS"] = str(response_timeout)
+        environment["NAPSEER_MCP_CANCEL_GRACE_SECONDS"] = str(cancel_grace)
+        environment["NAPSEER_MCP_MAX_IN_FLIGHT_REQUESTS"] = str(max_in_flight)
+        environment["NAPSEER_MCP_WRITE_TIMEOUT_SECONDS"] = str(write_timeout)
         environment["NAPSEER_TELEMETRY"] = "0"
         if project_root is not None:
             environment["NAPSEER_PROJECT_ROOT"] = str(project_root)
@@ -83,6 +99,7 @@ class SupervisorClient:
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             env=environment,
+            bufsize=0,
         )
 
     def request(self, request_id: int, method: str, params=None):
@@ -272,6 +289,33 @@ def test_reloads_replaced_worker_between_requests(tmp_path):
         client.close()
 
 
+def test_reloads_stable_launcher_when_watched_runtime_is_replaced(tmp_path):
+    runtime_path = tmp_path / "runtime.py"
+    launcher_path = tmp_path / "launcher.py"
+    write_worker(runtime_path, 1)
+    launcher_path.write_text(
+        "import runpy\n"
+        f"runpy.run_path({str(runtime_path)!r}, run_name='__main__')\n",
+        encoding="utf-8",
+    )
+    client = SupervisorClient(launcher_path, watch_paths=(runtime_path,))
+    try:
+        first = client.request(1, "initialize")
+        launcher_identity = launcher_path.stat().st_mtime_ns
+
+        time.sleep(0.01)
+        write_worker(runtime_path, 2)
+        os.utime(runtime_path, None)
+
+        second = client.request(2, "tools/list")
+
+        assert first["result"]["version"] == 1
+        assert second["result"]["version"] == 2
+        assert launcher_path.stat().st_mtime_ns == launcher_identity
+    finally:
+        client.close()
+
+
 def test_worker_crash_returns_error_without_closing_transport(tmp_path):
     worker_path = tmp_path / "worker.py"
     write_worker(worker_path, 1)
@@ -285,3 +329,201 @@ def test_worker_crash_returns_error_without_closing_transport(tmp_path):
         assert client.process.poll() is None
     finally:
         client.close()
+
+
+def write_lifecycle_worker(path, version=1):
+    # The actual worker dispatcher owns cancellation/completion; the handler
+    # supplies deterministic slow and uncooperative work without credentials.
+    path.write_text(f'''import json, os, pathlib, runpy, sys, time
+sys.path.insert(0, {str(REAL_WORKER.parent)!r})
+mod = runpy.run_path({str(REAL_WORKER)!r})
+root = pathlib.Path({str(path.parent)!r})
+def handle(message):
+    name = (message.get("params") or {{}}).get("name")
+    if name in ("slow", "stuck"):
+        (root / (str(message["id"]) + ".started")).write_text("started")
+        try:
+            deadline = time.monotonic() + 10
+            while time.monotonic() < deadline:
+                if name == "slow":
+                    mod["raise_if_mcp_request_cancelled"]()
+                time.sleep(0.01)
+        finally:
+            time.sleep(0.3)
+            (root / (str(message["id"]) + ".cleaned")).write_text("cleaned")
+    return mod["rpc_result"](message.get("id"), {{"pid": os.getpid(), "version": {version}}})
+runtime = mod["McpRequestRuntime"](handler=handle)
+try:
+    for line in sys.stdin:
+        runtime.dispatch(json.loads(line))
+finally:
+    runtime.close()
+''')
+
+
+def wait_for_path(path, timeout=3):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if path.exists():
+            return
+        time.sleep(0.01)
+    raise AssertionError("worker did not reach the expected lifecycle boundary")
+
+
+def test_cancelled_work_counts_until_cleanup_before_source_replacement(tmp_path):
+    worker_path = tmp_path / "worker.py"
+    write_lifecycle_worker(worker_path)
+    client = SupervisorClient(worker_path, project_root=tmp_path)
+    try:
+        client.send(1, "tools/call", {"name": "slow"})
+        wait_for_path(tmp_path / "1.started")
+        client.notify("notifications/cancelled", {"requestId": 1})
+        write_lifecycle_worker(worker_path, version=2)
+        during_cleanup = client.request(2, "tools/list")
+        assert during_cleanup["result"]["version"] == 1
+        wait_for_path(tmp_path / "1.cleaned")
+        time.sleep(0.05)
+        after_cleanup = client.request(3, "tools/list")
+        assert after_cleanup["result"]["version"] == 2
+        assert after_cleanup["result"]["pid"] != during_cleanup["result"]["pid"]
+    finally:
+        client.close()
+
+
+def test_timeout_cancels_work_and_keeps_transport_usable(tmp_path):
+    worker_path = tmp_path / "worker.py"
+    write_lifecycle_worker(worker_path)
+    client = SupervisorClient(worker_path, project_root=tmp_path, response_timeout=1)
+    try:
+        before = client.request(1, "tools/list")
+        failed = client.request(2, "tools/call", {"name": "slow"})
+        assert failed["error"]["code"] == -32098
+        wait_for_path(tmp_path / "2.cleaned")
+        recovered = client.request(3, "tools/list")
+        assert recovered["result"]["pid"] == before["result"]["pid"]
+    finally:
+        client.close()
+
+
+def test_cancelled_work_cannot_bypass_admission_limit(tmp_path):
+    worker_path = tmp_path / "worker.py"
+    write_lifecycle_worker(worker_path)
+    client = SupervisorClient(worker_path, project_root=tmp_path, max_in_flight=2)
+    try:
+        client.send(1, "tools/call", {"name": "stuck"})
+        wait_for_path(tmp_path / "1.started")
+        client.notify("notifications/cancelled", {"requestId": 1})
+        client.send(2, "tools/call", {"name": "stuck"})
+        wait_for_path(tmp_path / "2.started")
+        client.send(3, "tools/call", {"name": "stuck"})
+        rejected = client.receive(timeout=1)
+        assert rejected["id"] == 3
+        assert rejected["error"]["code"] == -32097
+        assert not (tmp_path / "3.started").exists()
+    finally:
+        client.close()
+
+
+def test_uncooperative_timeout_restarts_worker_with_no_replay(tmp_path):
+    worker_path = tmp_path / "worker.py"
+    write_lifecycle_worker(worker_path)
+    client = SupervisorClient(worker_path, project_root=tmp_path,
+                              response_timeout=1, cancel_grace=1)
+    try:
+        before = client.request(1, "tools/list")
+        failed = client.request(2, "tools/call", {"name": "stuck"})
+        assert failed["error"]["code"] == -32098
+        time.sleep(1.3)
+        recovered = client.request(3, "tools/list")
+        assert recovered["result"]["pid"] != before["result"]["pid"]
+        assert not (tmp_path / "2.cleaned").exists()
+        assert client.process.poll() is None
+    finally:
+        client.close()
+
+
+def test_reused_id_is_rejected_until_cancelled_request_finishes(tmp_path):
+    worker_path = tmp_path / "worker.py"
+    write_lifecycle_worker(worker_path)
+    client = SupervisorClient(worker_path, project_root=tmp_path)
+    try:
+        client.send(1, "tools/call", {"name": "slow"})
+        wait_for_path(tmp_path / "1.started")
+        client.notify("notifications/cancelled", {"requestId": 1})
+        rejected = client.request(1, "tools/list")
+        assert rejected["error"]["code"] == -32097
+        wait_for_path(tmp_path / "1.cleaned")
+        time.sleep(0.05)
+        assert client.request(1, "tools/list")["result"]["version"] == 1
+    finally:
+        client.close()
+
+
+def test_blocked_worker_stdin_has_a_deadline_and_recovers(tmp_path):
+    worker_path = tmp_path / "worker.py"
+    worker_path.write_text("import time\ntime.sleep(60)\n")
+    client = SupervisorClient(worker_path, project_root=tmp_path, write_timeout=1)
+    try:
+        client.send(1, "tools/call", {"name": "large", "arguments": {"data": "x" * 262144}})
+        failed = client.receive(timeout=3)
+        assert failed["error"]["code"] == -32098
+        write_worker(worker_path, 2)
+        assert client.request(2, "tools/list")["result"]["version"] == 2
+    finally:
+        client.close()
+
+
+def test_eof_stops_worker_with_unfinished_requests(tmp_path):
+    worker_path = tmp_path / "worker.py"
+    write_lifecycle_worker(worker_path)
+    client = SupervisorClient(worker_path, project_root=tmp_path)
+    try:
+        worker_pid = client.request(1, "tools/list")["result"]["pid"]
+        client.send(2, "tools/call", {"name": "stuck"})
+        wait_for_path(tmp_path / "2.started")
+        client.close()
+        assert client.process.returncode == 0
+        try:
+            os.kill(worker_pid, 0)
+        except ProcessLookupError:
+            pass
+        else:
+            raise AssertionError("worker survived supervisor EOF")
+    finally:
+        client.close()
+
+
+def test_initialize_timeout_retires_uncancellable_worker(tmp_path):
+    worker_path = tmp_path / "worker.py"
+    worker_path.write_text('''import json, os, sys, time
+for line in sys.stdin:
+    msg = json.loads(line)
+    if msg.get("method") == "initialize":
+        time.sleep(10)
+    print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"pid": os.getpid()}}), flush=True)
+''')
+    client = SupervisorClient(worker_path, project_root=tmp_path, response_timeout=1)
+    try:
+        before = client.request(1, "tools/list")
+        failed = client.request(2, "initialize")
+        assert failed["error"]["code"] == -32098
+        recovered = client.request(3, "tools/list")
+        assert recovered["result"]["pid"] != before["result"]["pid"]
+    finally:
+        client.close()
+
+
+def test_real_worker_contract_exposes_recovery_without_private_notifications(tmp_path, monkeypatch):
+    monkeypatch.setenv("NAPSEER_TOOL_PROFILES", "maintenance")
+    client = SupervisorClient(REAL_WORKER, project_root=tmp_path)
+    try:
+        assert "result" in client.request(1, "initialize")
+        response = client.request(2, "tools/call", {"name": "nap_contract", "arguments": {}})
+        contract = response["result"]["structuredContent"]["request_lifecycle"]
+        assert contract["max_unfinished_requests"] == 32
+        assert contract["supervised"] is True
+        assert contract["rejected_error_code"] == -32097
+        assert contract["uncertain_error_code"] == -32098
+        assert not (tmp_path / ".napseer").exists()
+    finally:
+        client.close()

tests/test_mcp_worker_lifecycle.py

@@ -4,6 +4,8 @@ import sys
 import threading
 import time
 
+import pytest
+
 
 def load_module():
     script_path = (
@@ -164,6 +166,7 @@ def test_mutating_tools_are_serialized_while_reads_can_overlap(monkeypatch):
 
 def test_concurrent_first_reads_share_one_project_bootstrap(tmp_path, monkeypatch):
     mod = load_module()
+    mod.TOKEN = "synthetic-configured-access"
     mod.AUTH_PATH = tmp_path / "missing-auth.json"
     mod.DEFAULT_PROJECT_ID = None
     monkeypatch.setattr(mod, "refresh_public_auth_state", lambda: None)
@@ -242,3 +245,160 @@ def test_auth_renewal_reuses_token_refreshed_by_another_request(monkeypatch):
         "refresh_expires_at": "later",
         "project_id": "project-1",
     }
+
+
+def test_tool_call_notification_cannot_execute_a_mutation(monkeypatch):
+    mod = load_module()
+    calls = []
+    monkeypatch.setattr(mod, "call_tool", lambda *args: calls.append(args))
+    assert mod.handle({"jsonrpc": "2.0", "method": "tools/call", "params": {
+        "name": "nap_create_node", "arguments": {"name": "unacknowledged"},
+    }}) is None
+    assert calls == []
+
+
+def test_worker_admission_is_bounded_and_recovers():
+    mod = load_module()
+    release = threading.Event()
+    emitted = []
+    executed = []
+
+    def handler(message):
+        executed.append(message["id"])
+        release.wait(2)
+        return mod.rpc_result(message["id"], {"ok": True})
+
+    runtime = mod.McpRequestRuntime(handler=handler, sender=emitted.append,
+                                    max_workers=2, max_in_flight=2)
+    try:
+        for request_id in (1, 2, 3):
+            runtime.dispatch({"jsonrpc": "2.0", "id": request_id, "method": "tools/call"})
+        wait_until(lambda: any(item["id"] == 3 for item in emitted))
+        rejection = next(item for item in emitted if item["id"] == 3)
+        assert rejection["error"]["code"] == -32097
+        assert 3 not in executed
+        release.set()
+        wait_until(lambda: len(emitted) == 3)
+        runtime.dispatch({"jsonrpc": "2.0", "id": 4, "method": "tools/call"})
+        wait_until(lambda: any(item["id"] == 4 for item in emitted))
+    finally:
+        release.set()
+        runtime.close()
+
+
+def test_cancelled_mutation_waiting_for_lock_never_starts(monkeypatch):
+    mod = load_module()
+    cancelled = threading.Event()
+    waiting = threading.Event()
+    calls = []
+    errors = []
+    monkeypatch.setattr(mod, "call_tool_impl", lambda *args: calls.append(args))
+    monkeypatch.setattr(mod, "send_telemetry_event_async", lambda *a, **kw: None)
+
+    def run():
+        mod.MCP_REQUEST_LOCAL.cancelled = cancelled
+        waiting.set()
+        try:
+            mod.call_tool("nap_create_node", {})
+        except mod.SafeToolError as exc:
+            errors.append(exc)
+
+    with mod.MCP_MUTATION_LOCK:
+        thread = threading.Thread(target=run)
+        thread.start()
+        assert waiting.wait(1)
+        cancelled.set()
+    thread.join(2)
+    assert not thread.is_alive()
+    assert calls == []
+    assert len(errors) == 1
+
+
+def test_cancelled_queue_entries_remain_bounded_until_consumed():
+    mod = load_module()
+    release = threading.Event()
+    executed = []
+    emitted = []
+
+    def handler(message):
+        executed.append(message["id"])
+        release.wait(2)
+        return mod.rpc_result(message["id"], {"ok": True})
+
+    runtime = mod.McpRequestRuntime(handler=handler, sender=emitted.append,
+                                    max_workers=2, max_in_flight=3)
+    try:
+        for request_id in (1, 2, 3):
+            runtime.dispatch({"id": request_id, "method": "tools/call"})
+        wait_until(lambda: len(executed) == 2)
+        runtime.cancel(3)
+        runtime.dispatch({"id": 4, "method": "tools/call"})
+        wait_until(lambda: any(item["id"] == 4 for item in emitted))
+        assert emitted[0]["error"]["code"] == -32097
+        release.set()
+        wait_until(lambda: len(emitted) == 3)
+        assert executed == [1, 2]
+    finally:
+        release.set()
+        runtime.close()
+
+
+@pytest.mark.parametrize("cancel_at", ["acquire", "operation"])
+def test_cancelled_write_releases_its_original_project_lock(tmp_path, monkeypatch, cancel_at):
+    import http.server
+    import json
+
+    monkeypatch.setenv("NAPSEER_PROJECT_ROOT", str(tmp_path))
+    mod = load_module()
+    cancelled = threading.Event()
+    requests = []
+    mutation_ran = []
+
+    class Handler(http.server.BaseHTTPRequestHandler):
+        def log_message(self, *args):
+            pass
+
+        def do_POST(self):
+            self.rfile.read(int(self.headers.get("Content-Length", 0)))
+            requests.append(("POST", self.path))
+            if cancel_at == "acquire":
+                cancelled.set()
+            body = json.dumps({"id": "lock-test", "lease_token": "synthetic-lease"}).encode()
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def do_DELETE(self):
+            requests.append(("DELETE", self.path))
+            self.send_response(204)
+            self.end_headers()
+
+    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    mod.BASE_URL = f"http://127.0.0.1:{server.server_port}"
+    mod.TOKEN = "synthetic-access"
+    mod.DEFAULT_PROJECT_ID = "original-project"
+    monkeypatch.setattr(mod, "refresh_public_auth_state", lambda: None)
+    mod.MCP_REQUEST_LOCAL.cancelled = cancelled
+
+    def operation(_headers):
+        mutation_ran.append(True)
+        mod.DEFAULT_PROJECT_ID = "different-project"
+        cancelled.set()
+        mod.raise_if_mcp_request_cancelled()
+
+    try:
+        with pytest.raises(mod.SafeToolError):
+            mod.with_project_lock({}, operation)
+        assert requests == [
+            ("POST", "/v1/projects/original-project/locks"),
+            ("DELETE", "/v1/projects/original-project/locks/lock-test"),
+        ]
+        assert mutation_ran == ([] if cancel_at == "acquire" else [True])
+    finally:
+        del mod.MCP_REQUEST_LOCAL.cancelled
+        server.shutdown()
+        server.server_close()
+        thread.join(2)

tests/test_project_attach_oauth.py

@@ -2,6 +2,7 @@
 """Smoke-test existing-project OAuth attachment for folder bootstrap."""
 
 import importlib.util
+import json
 import pathlib
 import sys
 import tempfile
@@ -17,6 +18,10 @@ def load_module():
     module = importlib.util.module_from_spec(spec)
     assert spec and spec.loader
     spec.loader.exec_module(module)
+    module.load_project_locator = lambda: None
+    module.require_project_locator_service = lambda _locator: None
+    module.write_project_locator = lambda *_args, **_kwargs: {"declared": False}
+    module.project_locator_status = lambda _locator=None: {"declared": False}
     return module
 
 
@@ -55,12 +60,12 @@ def run():
 
     result = mod.operator_project_attach({"open_browser": False})
 
+    saves.insert(0, json.loads(pathlib.Path(result["auth_path"]).read_text()))
     assert saves[0]["token"] == "oauth-token"
     assert saves[0]["refresh_token"] == "oauth-refresh-token"
     assert saves[0]["project_id"] == PROJECT_ID
     assert saves[0]["account_mode"] == "operator_project"
-    assert replacements[0] == mod.WORKER_BINDING_AUTH_KEYS
+    assert mod.WORKER_BINDING_AUTH_KEYS.isdisjoint(saves[0])
-    assert "project_id" not in replacements[0]
     assert "napseer.projects.read" in saves[0]["oauth_scope"]
     assert saves[1]["project_slug"] == "existing-project"
     assert result["project_id"] == PROJECT_ID
@@ -77,6 +82,7 @@ def run():
         "scope": args["scope"],
     }
     result = mod.operator_account_login({"open_browser": False})
+    saves.insert(0, json.loads(pathlib.Path(result["auth_path"]).read_text()))
     assert saves[0]["refresh_token"] == "account-refresh-token"
     assert saves[0]["account_mode"] == "operator_account"
     assert "project_id" not in saves[0]

tests/test_project_create_oauth.py

@@ -59,6 +59,7 @@ def test_account_login_requests_create_and_runtime_scopes():
 
     result = mod.operator_account_login({"open_browser": False})
 
+    saves.append(json.loads(pathlib.Path(result["auth_path"]).read_text()))
     requested = set(captured[0]["scope"].split())
     assert "napseer.projects.write" in requested
     assert set(mod.LOCAL_PROJECT_OAUTH_SCOPE.split()).issubset(requested)
@@ -84,23 +85,27 @@ def test_legacy_account_scope_fails_before_http_with_recovery():
     assert not requests
 
 
-def test_successful_create_transitions_local_state_to_project_mode():
+def test_successful_create_preserves_general_account_mode():
     mod = load_module()
     configure_account_state(mod, mod.OPERATOR_ACCOUNT_OAUTH_SCOPE)
     saves = []
+    locators = []
     mod.request_json = lambda method, path, payload=None: {
         "id": PROJECT_ID,
         "slug": payload["slug"],
         "name": payload["name"],
     }
     mod.save_auth = lambda updates: saves.append(dict(updates))
+    mod.write_project_locator = lambda project: locators.append(dict(project))
 
     project, status, _message = mod.create_project_with_state(project_args())
 
     assert status == "created"
     assert project["id"] == PROJECT_ID
     assert saves[0]["project_id"] == PROJECT_ID
-    assert saves[0]["account_mode"] == "operator_project"
+    assert "account_mode" not in saves[0]
+    assert mod.AUTH["account_mode"] == "operator_account"
+    assert locators == [project]
 
 
 def test_server_scope_denial_is_actionable():
@@ -159,7 +164,7 @@ def test_anonymous_limits_are_actionable_without_exposing_response_details():
 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_successful_create_preserves_general_account_mode()
     test_server_scope_denial_is_actionable()
     test_anonymous_limits_are_actionable_without_exposing_response_details()
     print("ok: project create OAuth lifecycle passed")

tests/test_project_init_wizard.py

@@ -16,6 +16,10 @@ def load_module():
     module = importlib.util.module_from_spec(spec)
     assert spec and spec.loader
     spec.loader.exec_module(module)
+    module.load_project_locator = lambda: None
+    module.require_project_locator_service = lambda _locator: None
+    module.write_project_locator = lambda *_args, **_kwargs: {"declared": False}
+    module.project_locator_status = lambda _locator=None: {"declared": False}
     return module
 
 

tests/test_project_locator.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Contract tests for the committed, non-secret Napseer project locator."""
+
+import importlib.util
+import json
+import pathlib
+import subprocess
+import sys
+import tempfile
+
+
+PROJECT_ID = "33333333-3333-3333-3333-333333333333"
+OTHER_PROJECT_ID = "44444444-4444-4444-4444-444444444444"
+
+
+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_project_locator_test", script_path)
+    module = importlib.util.module_from_spec(spec)
+    assert spec and spec.loader
+    spec.loader.exec_module(module)
+    return module
+
+
+def configure_state(mod, root):
+    state_dir = pathlib.Path(root) / ".napseer"
+    mod.AUTH_DIR = state_dir
+    mod.AUTH_PATH = state_dir / "auth.json"
+    mod.BASE_URL = "https://api.example.test"
+    mod.TOKEN = None
+    mod.DEFAULT_PROJECT_ID = None
+    mod.AUTH = {}
+    return state_dir
+
+
+def test_locator_is_deterministic_committable_and_secret_free():
+    mod = load_module()
+    with tempfile.TemporaryDirectory() as directory:
+        root = pathlib.Path(directory)
+        subprocess.run(["git", "init", "-q"], cwd=root, check=True)
+        state_dir = configure_state(mod, root)
+        mod.DEFAULT_PROJECT_ID = PROJECT_ID
+
+        result = mod.write_project_locator({"id": PROJECT_ID, "slug": "portable-project"})
+        locator = json.loads((state_dir / "project.json").read_text(encoding="utf-8"))
+
+        assert locator == {
+            "schema": "napseer.project.v1",
+            "base_url": "https://api.example.test",
+            "project_id": PROJECT_ID,
+            "project_slug": "portable-project",
+        }
+        assert result["declared"] is True
+        assert result["matches_local_auth"] is True
+        forbidden = {
+            "token", "access_token", "refresh_token", "account_id", "agent_id",
+            "worker_id", "claim_token", "claim_url", "passphrase", "encryption_state",
+        }
+        assert forbidden.isdisjoint(locator)
+
+        (state_dir / "auth.json").write_text('{"token":"private"}\n', encoding="utf-8")
+        (state_dir / "gateway-auth.json").write_text('{"token":"private"}\n', encoding="utf-8")
+        assert subprocess.run(
+            ["git", "check-ignore", "--quiet", ".napseer/auth.json"], cwd=root
+        ).returncode == 0
+        assert subprocess.run(
+            ["git", "check-ignore", "--quiet", ".napseer/gateway-auth.json"], cwd=root
+        ).returncode == 0
+        assert subprocess.run(
+            ["git", "check-ignore", "--quiet", ".napseer/project.json"], cwd=root
+        ).returncode == 1
+        assert subprocess.run(
+            ["git", "check-ignore", "--quiet", ".napseer/.gitignore"], cwd=root
+        ).returncode == 1
+
+
+def test_locator_rejects_extra_secret_fields_and_mismatched_auth():
+    mod = load_module()
+    with tempfile.TemporaryDirectory() as directory:
+        state_dir = configure_state(mod, directory)
+        state_dir.mkdir(parents=True)
+        (state_dir / "project.json").write_text(
+            json.dumps({
+                "schema": "napseer.project.v1",
+                "base_url": "https://api.example.test",
+                "project_id": PROJECT_ID,
+                "project_slug": "portable-project",
+                "token": "must-not-be-accepted",
+            }),
+            encoding="utf-8",
+        )
+        try:
+            mod.load_project_locator()
+        except mod.SafeToolError as exc:
+            assert exc.code == "project_locator_invalid"
+        else:
+            raise AssertionError("locator with an extra secret field must be rejected")
+
+        (state_dir / "project.json").unlink()
+        mod.write_project_locator({"id": PROJECT_ID, "slug": "portable-project"})
+        mod.DEFAULT_PROJECT_ID = OTHER_PROJECT_ID
+        mod.refresh_public_auth_state = lambda: None
+        try:
+            mod.resolve_project_id({})
+        except mod.SafeToolError as exc:
+            assert exc.code == "project_locator_mismatch"
+        else:
+            raise AssertionError("local auth/locator mismatch must fail closed")
+
+
+def test_fresh_clone_requires_attach_and_never_bootstraps_anonymous_project():
+    mod = load_module()
+    with tempfile.TemporaryDirectory() as directory:
+        configure_state(mod, directory)
+        mod.write_project_locator({"id": PROJECT_ID, "slug": "portable-project"})
+        enrollment_calls = []
+        bootstrap_calls = []
+        mod.ensure_enrolled = lambda args: enrollment_calls.append(dict(args))
+        mod.bootstrap_project = lambda args: bootstrap_calls.append(dict(args))
+
+        result = mod.cli_project_init([])
+        assert result["status"] == "project_access_required"
+        assert result["project_id"] == PROJECT_ID
+        assert result["next"]["attach_project"] == "nap project attach"
+        assert enrollment_calls == []
+        assert bootstrap_calls == []
+
+        mod.refresh_public_auth_state = lambda: None
+        try:
+            mod.resolve_project_id({})
+        except mod.SafeToolError as exc:
+            assert exc.code == "project_access_required"
+            assert "will not create a duplicate anonymous project" in exc.safe_message
+        else:
+            raise AssertionError("first MCP project resolution must require attach")
+
+
+def test_attach_forwards_and_verifies_repository_project_hint():
+    mod = load_module()
+    with tempfile.TemporaryDirectory() as directory:
+        configure_state(mod, directory)
+        mod.write_project_locator({"id": PROJECT_ID, "slug": "portable-project"})
+        oauth_calls = []
+        saves = []
+
+        def oauth(args):
+            oauth_calls.append(dict(args))
+            return {
+                "api_base_url": "https://api.example.test",
+                "access_token": "local-test-token",
+                "refresh_token": "local-test-refresh",
+                "expires_in": 3600,
+                "scope": args["scope"],
+                "project_id": PROJECT_ID,
+            }
+
+        mod.oauth_loopback_authorize = oauth
+        mod.replace_public_auth_state = lambda updates, clear_keys=(): saves.append(dict(updates))
+        mod.save_auth_file_credentials = lambda updates: saves.append(dict(updates))
+        mod.request_json = lambda method, path: {
+            "id": PROJECT_ID,
+            "slug": "portable-project",
+            "name": "Portable Project",
+            "encryption_state": "standard",
+        }
+
+        result = mod.operator_project_attach({"open_browser": False})
+        assert oauth_calls[0]["project_id_hint"] == PROJECT_ID
+        assert result["project_id"] == PROJECT_ID
+        assert result["project_locator"]["project_id"] == PROJECT_ID
+        credentials = json.loads(pathlib.Path(result["auth_path"]).read_text())
+        assert credentials["project_id"] == PROJECT_ID
+
+
+if __name__ == "__main__":
+    test_locator_is_deterministic_committable_and_secret_free()
+    test_locator_rejects_extra_secret_fields_and_mismatched_auth()
+    test_fresh_clone_requires_attach_and_never_bootstraps_anonymous_project()
+    test_attach_forwards_and_verifies_repository_project_hint()
+    print("ok: committed project locator contract passed")

tests/test_standalone_smokes.py

@@ -23,6 +23,7 @@ SMOKES = [
     "test_project_attach_oauth.py",
     "test_project_encryption_enable_payload.py",
     "test_project_init_wizard.py",
+    "test_project_locator.py",
     "test_two_passphrase_boundary.py",
 ]
 

tests/test_user_credentials.py

@@ -0,0 +1,284 @@
+"""Real workers and synthetic HTTP provider: user sessions across repositories."""
+import importlib.util
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs
+
+import pytest
+
+SCRIPT = Path(__file__).resolve().parents[1] / "resources/scripts/napseer_mcp_server.py"
+sys.path.insert(0, str(SCRIPT.parent))
+from napseer_credentials import CredentialStore, atomic_json, read_json
+
+
+@pytest.fixture
+def provider():
+    state = {"refreshes": 0, "requests": 0, "failure": None, "revoked": False}
+    mutex = threading.Lock()
+    state["refresh_started"] = threading.Event()
+    state["refresh_release"] = threading.Event()
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, *_args):
+            pass
+
+        def reply(self, status, value):
+            data = json.dumps(value).encode()
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(data)))
+            self.end_headers()
+            self.wfile.write(data)
+
+        def do_GET(self):
+            with mutex:
+                state["requests"] += 1
+            if self.headers.get("Authorization") == "Bearer fresh-access" and not state["revoked"]:
+                self.reply(200, {"ok": True})
+            else:
+                self.reply(401, {"error": {"code": "unauthorized"}})
+
+        def do_POST(self):
+            form = parse_qs(self.rfile.read(int(self.headers["Content-Length"])).decode())
+            if self.path == "/v1/oauth/revoke":
+                state["revoked"] = True
+                return self.reply(200, {})
+            assert self.path == "/v1/oauth/token", "no enrollment fallback is allowed"
+            if state["failure"] == "transport":
+                return self.reply(503, {"error": {"code": "unavailable"}})
+            if state["failure"] == "malformed":
+                return self.reply(200, {"access_token": "incomplete"})
+            if form.get("refresh_token") != ["old-refresh"] or state["refreshes"]:
+                return self.reply(401, {"error": {"code": "invalid_grant"}})
+            if state.get("hold_refresh"):
+                state["refresh_started"].set()
+                assert state["refresh_release"].wait(5)
+            with mutex:
+                state["refreshes"] += 1
+            self.reply(200, {"access_token": "fresh-access", "refresh_token": "fresh-refresh",
+                "expires_in": 3600, "refresh_expires_at": "2099-01-01T00:00:00Z"})
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{server.server_port}", state
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+
+
+def seed(root, url):
+    root.mkdir(exist_ok=True)
+    store = CredentialStore(root / ".napseer")
+    store.login({"base_url": url, "token": "old-access", "refresh_token": "old-refresh",
+                 "account_mode": "operator_account", "oauth_client_id": "nap-cli"})
+    return store
+
+
+def worker(root, expression):
+    code = "import importlib.util, json, sys\n" + (
+        "spec = importlib.util.spec_from_file_location('worker', sys.argv[1])\n"
+        "m = importlib.util.module_from_spec(spec)\nspec.loader.exec_module(m)\n"
+        "try:\n    result = " + expression + "\n    print(json.dumps({'ok': True, 'result': result}))\n"
+        "except Exception as exc:\n    print(json.dumps({'ok': False, 'code': getattr(exc, 'code', type(exc).__name__)}))\n"
+    )
+    env = dict(os.environ, NAPSEER_PROJECT_ROOT=str(root), NAPSEER_TELEMETRY="0")
+    for key in ("NAPSEER_AUTH_FILE", "NAPSEER_BASE_URL", "NAPSEER_PROJECT_ID"):
+        env.pop(key, None)
+    return subprocess.Popen([sys.executable, "-c", code, str(SCRIPT)], cwd=root, env=env,
+                            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+
+
+def outcome(process):
+    out, _err = process.communicate(timeout=15)
+    assert process.returncode == 0, "worker must exit without exposing credential state"
+    return json.loads(out)
+
+
+def test_shared_login_refreshes_once_across_real_processes_and_survives_restart(tmp_path, provider):
+    url, state = provider
+    first, second = tmp_path / "first", tmp_path / "second"
+    store = seed(first, url)
+    second.mkdir()
+    processes = [worker(root, "m.request_json('GET', '/v1/projects')") for root in [first, second]]
+    assert all(outcome(p)["ok"] for p in processes)
+    assert state["refreshes"] == 1
+    assert outcome(worker(second, "m.request_json('GET', '/v1/projects')"))["ok"]
+    assert state["refreshes"] == 1
+    assert store.user.stat().st_mode & 0o777 == 0o600
+    assert not (first / ".napseer/auth.json").exists()
+    assert not (second / ".napseer/auth.json").exists()
+
+
+@pytest.mark.parametrize("failure", ["transport", "malformed"])
+def test_refresh_failure_preserves_credentials_and_recovers(tmp_path, provider, failure):
+    url, state = provider
+    store = seed(tmp_path, url)
+    before = store.user.read_bytes()
+    state["failure"] = failure
+    assert not outcome(worker(tmp_path, "m.request_json('GET', '/v1/projects')"))["ok"]
+    assert store.user.read_bytes() == before
+    state["failure"] = None
+    assert outcome(worker(tmp_path, "m.request_json('GET', '/v1/projects')"))["ok"]
+
+
+def test_project_override_does_not_fall_back_to_user_authority(tmp_path, provider):
+    url, state = provider
+    store = seed(tmp_path, url)
+    store.login({"base_url": url, "token": "npk_expired", "account_mode": "api_key"}, project=True)
+    result = outcome(worker(tmp_path, "m.request_json('GET', '/v1/projects')"))
+    assert not result["ok"]
+    assert state["refreshes"] == 0
+    assert read_json(store.user)["token"] == "old-access"
+    store.logout(project=True)
+    assert not outcome(worker(tmp_path, "m.request_json('GET', '/v1/projects')"))["ok"]
+    assert state["refreshes"] == 0
+
+
+def test_unconfigured_agent_never_enrolls_or_creates_project(tmp_path):
+    result = outcome(worker(tmp_path, "m.resolve_project_id({})"))
+    assert result == {"ok": False, "code": "auth_required"}
+    assert not (tmp_path / ".napseer").exists()
+
+
+def test_logout_revokes_and_next_worker_observes_tombstone(tmp_path, provider):
+    url, state = provider
+    store = seed(tmp_path, url)
+    assert outcome(worker(tmp_path, "m.operator_logout({})"))["ok"]
+    assert state["revoked"]
+    assert read_json(store.user)["logged_out"]
+    assert not outcome(worker(tmp_path, "m.request_json('GET', '/v1/projects')"))["ok"]
+    assert state["requests"] == 0
+
+
+def test_global_credentials_keep_repository_selections_independent(tmp_path, provider):
+    url, _ = provider
+    first = tmp_path / "first"
+    store = seed(first, url)
+    other = CredentialStore(tmp_path / "second/.napseer")
+    store.write({**store.read(), "project_id": "first"})
+    other.write({**other.read(), "project_id": "second"})
+    assert "project_id" not in read_json(store.user)
+    assert store.read()["project_id"] == "first"
+    assert other.read()["project_id"] == "second"
+
+
+def test_logout_waits_for_inflight_rotation_and_cannot_be_undone(tmp_path, provider):
+    url,state=provider
+    store=seed(tmp_path,url)
+    state["hold_refresh"]=True
+    refresh=worker(tmp_path,"m.request_json('GET','/v1/projects')")
+    assert state["refresh_started"].wait(5)
+    logout=worker(tmp_path,"m.operator_logout({})")
+    state["refresh_release"].set()
+    outcome(refresh)
+    assert outcome(logout)["ok"]
+    assert state["revoked"]
+    assert read_json(store.user).get("logged_out")
+    assert not outcome(worker(tmp_path,"m.request_json('GET','/v1/projects')"))["ok"]
+
+
+def test_refresh_pins_credential_source_when_override_appears(tmp_path, provider):
+    url,state=provider
+    store=seed(tmp_path,url)
+    state["hold_refresh"]=True
+    refresh=worker(tmp_path,"m.request_json('GET','/v1/projects')")
+    assert state["refresh_started"].wait(5)
+    override={"base_url":url,"token":"npk_project_test","account_mode":"api_key"}
+    store.login(override,project=True)
+    state["refresh_release"].set()
+    assert not outcome(refresh)["ok"], "retry observes the explicit override"
+    assert read_json(store.project)==override
+    assert read_json(store.user)["token"]=="fresh-access"
+
+
+def test_legacy_migration_preserves_authority_and_general_selection_is_explicit(tmp_path,provider):
+    url,_=provider
+    store=seed(tmp_path,url)
+    legacy={"base_url":url,"token":"legacy-project","account_mode":"operator_project","project_id":"test-project"}
+    atomic_json(store.legacy,legacy)
+    assert store.selected()[1]=="legacy_project"
+    store.migrate()
+    assert not store.legacy.exists()
+    assert store.read()["token"]=="legacy-project"
+    store.use_user()
+    assert store.read()["token"]=="old-access"
+    assert read_json(store.project)==legacy
+
+
+def test_failed_login_initialization_releases_actual_loopback_socket(tmp_path,monkeypatch):
+    import socket
+    spec=importlib.util.spec_from_file_location("cleanup_worker",SCRIPT)
+    module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
+    reserved=socket.socket();reserved.bind(("127.0.0.1",0));port=reserved.getsockname()[1];reserved.close()
+    monkeypatch.setattr(module,"request_json",lambda *a,**k: (_ for _ in ()).throw(RuntimeError("synthetic unavailable")))
+    with pytest.raises(RuntimeError):
+        module.oauth_loopback_authorize({"scope":"test","flow":"operator_login","port":port,"open_browser":False})
+    probe=socket.socket()
+    try: probe.bind(("127.0.0.1",port))
+    finally: probe.close()
+
+
+def test_native_login_uses_requested_service_pkce_callback_and_shared_store(tmp_path,monkeypatch):
+    import base64
+    import hashlib
+    import urllib.request
+    from urllib.parse import urlsplit, urlencode
+    state={}
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self,*_args): pass
+        def do_POST(self):
+            raw=self.rfile.read(int(self.headers['Content-Length']))
+            if self.path.endswith('/authorizations'):
+                state['authorization']=json.loads(raw)
+                payload={"authorization_url":"http://synthetic-browser.invalid/consent"}
+            else:
+                form=parse_qs(raw.decode());state['exchanged']=True
+                expected=base64.urlsafe_b64encode(hashlib.sha256(form['code_verifier'][0].encode()).digest()).decode().rstrip('=')
+                assert expected==state['authorization']['code_challenge']
+                assert form['code']==['synthetic-one-time-code']
+                payload={"access_token":"synthetic-native-access","refresh_token":"synthetic-native-refresh","expires_in":3600}
+            body=json.dumps(payload).encode();self.send_response(200);self.end_headers();self.wfile.write(body)
+    server=ThreadingHTTPServer(('127.0.0.1',0),Handler)
+    thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
+    spec=importlib.util.spec_from_file_location('native_login_worker',SCRIPT)
+    module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
+    monkeypatch.setattr(module,'AUTH_DIR',tmp_path/'.napseer')
+    monkeypatch.setattr(module,'AUTH_PATH',tmp_path/'.napseer/auth.json')
+    monkeypatch.setattr(module,'BASE_URL','http://127.0.0.1:1')
+    store = CredentialStore(tmp_path/'.napseer')
+    store.user.parent.mkdir(parents=True,exist_ok=True)
+    store.user.write_text('{broken synthetic state')
+    def consent(_url):
+        grant=state['authorization']
+        callback=grant['redirect_uri']+'?'+urlencode({'code':'synthetic-one-time-code','state':grant['state']})
+        with urllib.request.urlopen(callback,timeout=3) as response: assert response.status==200
+        return True
+    monkeypatch.setattr(module.webbrowser,'open',consent)
+    try:
+        result=module.operator_account_login({'base_url':f'http://127.0.0.1:{server.server_port}','timeout_seconds':5})
+        assert result['status']=='authenticated';assert state['exchanged']
+        store=CredentialStore(tmp_path/'.napseer')
+        assert store.user.exists();assert not store.legacy.exists()
+        assert read_json(store.user)['base_url']==f'http://127.0.0.1:{server.server_port}'
+    finally:server.shutdown();server.server_close();thread.join(timeout=2)
+
+
+def test_corrupt_credentials_allow_help_but_never_authenticated_fallback(tmp_path):
+    store=CredentialStore(tmp_path/'.napseer')
+    store.user.parent.mkdir(parents=True,exist_ok=True)
+    store.user.write_text('{broken synthetic state')
+    result=outcome(worker(tmp_path,"m.request_json('GET','/v1/projects')"))
+    assert result['code']=='CredentialError'
+    assert store.user.read_text()=='{broken synthetic state'
+    env=dict(os.environ,NAPSEER_PROJECT_ROOT=str(tmp_path))
+    help_result=subprocess.run([sys.executable,str(SCRIPT),'auth','--help'],cwd=tmp_path,env=env,capture_output=True,text=True,timeout=10)
+    assert help_result.returncode==0
+    assert 'nap auth login' in help_result.stdout