Provision the E2E memory secret into the local vault

`nap gateway vault setup` unwraps the project memory secret once (prompt
on a terminal, or the existing argument/env paths) and stores the
secret, never the passphrase, in memory-secrets.vault.json under the
existing 0600 local vault key. The CLI and MCP workers load it
automatically and MCP never prompts, per
/decisions/local-gateway-no-interactive-unlock and
/decisions/2026-09-29-memory-secret-local-vault. Older versions stay
readable; a rotated active version reports memory_secret_rotated with
the setup next action. `vault forget` revokes local access, including
for running workers. Env passphrases still work, are persisted into the
vault, and are reported as deprecated. Search warns
e2e_index_not_provisioned instead of e2e_index_locked. Local key
creation is 0600 from the start and never replaces a published key.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

resources/scripts/nap_install.py

@@ -1042,8 +1042,12 @@ def print_gateway_help(topic=None):
   nap gateway schedule delete [ID | --schedule ID]""")
         return 0
     if topic == "vault":
-        print("Usage: nap gateway vault [status|process|rotate-secret] [--kind memory] [--all] [--project-id ID] [--vault-passphrase TEXT]")
+        print("Usage: nap gateway vault [status|setup|forget|process|rotate-secret] [--kind memory] [--all] [--project-id ID] [--vault-passphrase TEXT]")
-        print("  status        Show local gateway state and pending setup requests.")
+        print("  status        Show local gateway state, memory secret provisioning, and pending setup requests.")
+        print("  setup         Store the project memory secret in local encrypted storage for this worktree.")
+        print("                Prompts once for the project vault passphrase; the passphrase is never stored.")
+        print("                The CLI and MCP workers then load the secret automatically.")
+        print("  forget        Remove the locally stored memory secret for this worktree.")
         print("  process       Upload opaque client-wrapped key bundle records for backend-owned HashiCorp storage.")
         print("  rotate-secret Rotate the memory project secret from the gateway worker path.")
         return 0
@@ -1054,6 +1058,7 @@ def print_gateway_help(topic=None):
     print("  nap gateway repair                    Create dedicated gateway worker credentials.")
     print("  nap gateway terminal help             Show terminal lifecycle commands.")
     print("  nap gateway schedule help             Show schedule lifecycle commands.")
+    print("  nap gateway vault setup               Store the E2E memory secret locally (prompts once).")
     print("  nap gateway vault help                Show vault commands.")
     return 0
 
@@ -1308,6 +1313,10 @@ def mcp_runtime_probe(timeout_seconds=6):
                 "verified": False,
             },
             "remote_project_access": "not_tested",
+            "memory_secret": (local_payload or {}).get("memory_secret"),
+            "deprecations": [
+                item.get("code") for item in (local_payload or {}).get("deprecations") or [] if isinstance(item, dict)
+            ],
         }
     except Exception as exc:
         return {
@@ -1417,6 +1426,18 @@ def doctor_status():
                 else "inspect the failed_stage field and local project binding"
             ),
         })
+    if probe.get("memory_secret") == "rotated":
+        setup.append({
+            "code": "memory_secret_rotated",
+            "message": "the project memory secret was rotated after it was stored locally",
+            "next": "nap gateway vault setup",
+        })
+    if "vault_passphrase_env_deprecated" in (probe.get("deprecations") or []):
+        setup.append({
+            "code": "vault_passphrase_env_deprecated",
+            "message": "NAPSEER_VAULT_PASSPHRASE / NAPSEER_MASTER_PASSPHRASE are deprecated for agents; store the memory secret locally instead",
+            "next": "nap gateway vault setup",
+        })
     status = "repair_required" if issues else ("setup_available" if setup else "ok")
     return {
         "status": status,

tests/test_cli_dispatch_consolidation.py

@@ -188,6 +188,13 @@ def test_gateway_vault_aliases_rejected():
         expect_unknown(mod, args, "gateway vault action: list")
 
 
+def test_gateway_vault_unlock_verbs_point_to_setup_and_forget():
+    """The memory secret is provisioned once, not unlocked per session."""
+    mod = load_module()
+    for action in ("unlock", "lock", "provision", "remove", "reset"):
+        expect_unknown(mod, ["napseer_mcp_server.py", "gateway", "vault", action], "nap gateway vault setup")
+
+
 def test_gateway_rotate_passphrase_aliases_rejected():
     mod = load_module()
     for args in [

tests/test_mcp_e2e_index_shard.py

@@ -229,20 +229,23 @@ def test_e2e_body_term_is_findable_with_snippet_while_unlocked(env):
     assert not discovered.get("warnings")
 
 
-def test_locked_secret_yields_no_e2e_hits_and_an_explicit_warning(env):
+def test_unprovisioned_secret_yields_no_e2e_hits_and_an_explicit_warning(env):
     mod, provider, _ = env
     seed(mod, provider)
     lock(mod)
 
     local = local_search(mod, BODY_TERM)
     assert hit_ids(local) == []
-    assert codes(local) == ["e2e_index_locked"]
+    assert codes(local) == ["e2e_index_not_provisioned"]
+    warning = local["warnings"][0]
+    assert warning["memory_secret"] == "not_provisioned"
+    assert warning["next_action"] == "nap gateway vault setup"
     response = mod.search_memory({"q": BODY_TERM, "mode": "local", "view": "summary"})
     assert response["items"] == []
-    assert "e2e_index_locked" in codes(response)
+    assert "e2e_index_not_provisioned" in codes(response)
     # Identity fields of the E2E node remain searchable from the main index.
     assert hit_ids(local_search(mod, "secretnote")) == ["secretnote"]
-    # A locked reindex never prompts and writes no shard.
+    # An unprovisioned reindex never prompts and writes no shard.
     shard_file(mod).unlink()
     assert "e2e_index" in mod.reindex_project({})
     assert not shard_file(mod).exists()

tests/test_mcp_memory_node_encryption.py

@@ -116,6 +116,8 @@ def run():
     mod = load_module()
     test_state = tempfile.TemporaryDirectory()
     mod.AUTH_PATH = pathlib.Path(test_state.name) / "auth.json"
+    mod.MEMORY_SECRET_VAULT_PATH = pathlib.Path(test_state.name) / "memory-secrets.vault.json"
+    mod.LOCAL_VAULT_KEY_PATH = pathlib.Path(test_state.name) / "vault.local-key"
     mod.AUTH = {"account_id": ACCOUNT_ID, "token": "token", "token_expires_at": "later"}
     mod.PROJECT_VAULT_PASSPHRASE = PASSPHRASE
     mod.DEFAULT_PROJECT_ID = "project-1"
@@ -485,9 +487,11 @@ def run():
         mod.MEMORY_SECRET_CACHE.clear()
         mod.ACTIVE_MEMORY_SECRET_VERSIONS.clear()
         mod.PROJECT_VAULT_PASSPHRASE = None
+        # Earlier passphrase unwraps persisted the secret locally; remove it.
+        mod.memory_secret_vault_forget("project-1")
         locked = mod.search_local_index({"q": "needle", "limit": 5})
         assert locked["items"] == []
-        assert [warning["code"] for warning in locked["warnings"]] == ["e2e_index_locked"]
+        assert [warning["code"] for warning in locked["warnings"]] == ["e2e_index_not_provisioned"]
 
     print("ok: local MCP memory node encryption smoke passed")
 

tests/test_memory_secret_local_vault.py

@@ -0,0 +1,674 @@
+"""Project memory secret provisioned once into the local vault, loaded automatically.
+
+Runs the real wrapper (in-process and as fresh CLI/MCP processes) against a
+disposable HTTP provider that serves a memory secret wrapped with a fixture
+passphrase. The fixture secret and passphrase are random per test and every
+tool output, error, log line and state file is scanned for them (raw, base64,
+hex) without ever printing them.
+"""
+
+import base64
+import importlib.util
+import json
+import os
+import pathlib
+import pty
+import queue
+import re
+import select
+import stat
+import subprocess
+import sys
+import termios
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs, urlsplit
+
+import pytest
+
+SCRIPT = pathlib.Path(__file__).resolve().parents[1] / "resources/scripts/napseer_mcp_server.py"
+PROJECT = "project-vault-1"
+ACCOUNT = "22222222-2222-2222-2222-222222222222"
+BODY_TERM = "zephyrquartz"
+BODY = f"calibrate the {BODY_TERM} manifold before launch"
+SETUP_COMMAND = "nap gateway vault setup"
+
+
+# -- fixture material ---------------------------------------------------------
+
+class Material:
+    def __init__(self):
+        self.passphrase = "fixture-" + os.urandom(12).hex()
+        self.secrets = {}
+
+    def secret(self, version):
+        return self.secrets.setdefault(version, os.urandom(32))
+
+    def forms(self):
+        forms = [self.passphrase.encode("utf-8")]
+        for value in self.secrets.values():
+            encoded = base64.b64encode(value)
+            forms += [value, encoded, encoded.rstrip(b"="), base64.urlsafe_b64encode(value).rstrip(b"="),
+                      value.hex().encode("ascii")]
+        return forms
+
+
+def assert_no_leak(material, label, blob):
+    data = blob if isinstance(blob, bytes) else json.dumps(blob, default=str).encode("utf-8") \
+        if not isinstance(blob, str) else blob.encode("utf-8")
+    for form in material.forms():
+        if form in data:
+            pytest.fail(f"fixture secret material leaked into {label}")  # never print the value
+
+
+def assert_state_dir_clean(material, *roots):
+    for root in roots:
+        for path in pathlib.Path(root).rglob("*"):
+            if path.is_file():
+                assert_no_leak(material, f"state file {path.name}", path.read_bytes())
+
+
+# -- disposable provider ------------------------------------------------------
+
+def secret_record(material, mod, version, wrapping_epoch=1):
+    previous = mod.PROJECT_VAULT_PASSPHRASE
+    mod.PROJECT_VAULT_PASSPHRASE = material.passphrase
+    try:
+        bundle = mod.generate_project_data_key_bundle(PROJECT, ACCOUNT, data_key_epoch=version)
+        bundle["keys"]["memory"]["versions"][str(version)]["key_b64"] = base64.b64encode(
+            material.secret(version)).decode("ascii")
+        wrapped = mod.wrapped_project_key_bundle_record(
+            PROJECT, ACCOUNT, "memory", bundle,
+            wrapping_epoch=wrapping_epoch, bundle_version=version, data_key_epoch=version)
+    finally:
+        mod.PROJECT_VAULT_PASSPHRASE = previous
+    return {
+        "secret_kind": "memory", "version": version, "project_secret_version": version,
+        "wrapping_epoch": wrapping_epoch, "bundle_version": version, "data_key_epoch": version,
+        "wrapped_key_bundle": wrapped, "created_at": "2026-09-01T00:00:00Z", "last_rotated_at": "2026-09-01T00:00:00Z",
+    }
+
+
+def e2e_node(mod, material, name, version, content, minute=1):
+    secret = {
+        "project_id": PROJECT, "secret_kind": "memory", "version": version, "wrapping_epoch": 1,
+        "bundle_version": version, "data_key_epoch": version, "secret_bytes": material.secret(version),
+    }
+    stamp = f"2026-09-01T12:{minute:02d}:00Z"
+    return {
+        "id": name, "project_id": PROJECT, "full_path": f"/notes/{name}", "folder_path": "/notes",
+        "name": name, "type": "note", "metadata": {}, "tags": [], "aliases": [], "links": [],
+        "content_text": "", "updated_at": stamp, "archived_at": None, "encryption_state": "encrypted",
+        "project_secret_version": version,
+        "encrypted_content_envelope": mod.encrypt_memory_bytes(
+            PROJECT, secret, "node_content", content.encode("utf-8"), {"full_path": f"/notes/{name}"},
+            "text/plain; charset=utf-8"),
+        "encrypted_metadata_envelope": None,
+    }
+
+
+class Provider:
+    def __init__(self):
+        self.nodes = {}
+        self.events = []
+        self.requests = []
+        self.active = None
+        self.versions = {}
+        self.vault_down = False
+
+    def publish(self, record):
+        self.active = record
+        self.versions[record["version"]] = record
+
+    def put(self, node):
+        self.nodes[node["id"]] = node
+        self.events.append({"type": "node_created", "node_id": node["id"], "full_path": node["full_path"],
+                            "new_revision": node["updated_at"]})
+
+
+def start_provider(provider):
+    vault_re = re.compile(rf"/v1/projects/{PROJECT}/vault/secrets/memory/(active|versions/(\d+))")
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, *_args):
+            pass
+
+        def reply(self, status, payload):
+            body = json.dumps(payload).encode()
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def do_POST(self):
+            provider.requests.append(("POST", urlsplit(self.path).path))
+            self.reply(404, {"error": {"code": "not_found", "message": "not served"}})
+
+        def do_GET(self):
+            url = urlsplit(self.path)
+            query = parse_qs(url.query)
+            provider.requests.append(("GET", url.path))
+            match = vault_re.fullmatch(url.path)
+            if match:
+                if provider.vault_down:
+                    return self.reply(503, {"error": {"code": "unavailable", "message": "vault down"}})
+                record = provider.active if match.group(1) == "active" else provider.versions.get(int(match.group(2)))
+                if record is None:
+                    return self.reply(404, {"error": {"code": "not_found", "message": "no such secret"}})
+                return self.reply(200, record)
+            if url.path.endswith("/encryption/status"):
+                return self.reply(200, {"state": "encrypted"})
+            if url.path.endswith("/changes"):
+                after = query.get("after", ["head"])[0]
+                start = len(provider.events) if after == "head" else int(after[1:])
+                page = provider.events[start:]
+                items = [{**event, "cursor": f"c{start + i + 1}", "project_id": PROJECT, "changed_fields": [],
+                          "created_at": "2026-09-01T12:00:00Z"} for i, event in enumerate(page)]
+                return self.reply(200, {"items": items, "next_cursor": f"c{start + len(page)}", "has_more": False})
+            if url.path == f"/v1/projects/{PROJECT}/nodes":
+                rows = sorted(provider.nodes.values(), key=lambda n: (n["updated_at"], n["id"]), reverse=True)
+                return self.reply(200, {"items": rows, "next_cursor": None})
+            node_id = url.path.rsplit("/", 1)[-1]
+            if "/nodes/" in url.path and node_id in provider.nodes:
+                return self.reply(200, provider.nodes[node_id])
+            return self.reply(404, {"error": {"code": "not_found", "message": "not served"}})
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    return server, thread
+
+
+def load_module(name, state_dir, base_url):
+    spec = importlib.util.spec_from_file_location(name, SCRIPT)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    mod.AUTH_DIR = state_dir
+    mod.INDEX_PATH = state_dir / "index.sqlite"
+    mod.INDEX_LOCK_PATH = state_dir / "index.lock"
+    mod.VAULT_PATH = state_dir / "vault.json"
+    mod.LOCAL_VAULT_KEY_PATH = state_dir / "vault.local-key"
+    mod.MEMORY_SECRET_VAULT_PATH = state_dir / "memory-secrets.vault.json"
+    mod.BASE_URL = base_url
+    mod.TOKEN = "disposable-test-token"
+    mod.AUTH = {**mod.AUTH, "account_id": ACCOUNT, "project_id": PROJECT}
+    mod.DEFAULT_PROJECT_ID = PROJECT
+    mod.CONFIGURED_TOOL_PROFILES = {"all"}
+    mod.refresh_public_auth_state = lambda: None
+    mod.resolve_project_id = lambda args: args.get("project_id", PROJECT)
+    mod.project_encryption_state = lambda _project_id: "encrypted"
+    return mod
+
+
+@pytest.fixture
+def env(tmp_path, monkeypatch):
+    for name in ("NAPSEER_VAULT_PASSPHRASE", "NAPSEER_MASTER_PASSPHRASE", "NAPSEER_LOCAL_VAULT_KEY"):
+        monkeypatch.delenv(name, raising=False)
+    monkeypatch.setenv("NAPSEER_TOOL_PROFILES", "all")
+    monkeypatch.setenv("NAPSEER_TELEMETRY", "0")
+    sqlite_tmp = tmp_path / "sqlite-tmp"
+    sqlite_tmp.mkdir()
+    monkeypatch.setenv("SQLITE_TMPDIR", str(sqlite_tmp))
+    monkeypatch.syspath_prepend(str(SCRIPT.parent))
+    provider = Provider()
+    server, thread = start_provider(provider)
+    base_url = f"http://127.0.0.1:{server.server_port}"
+    state_dir = tmp_path / "state"
+    mod = load_module("memory_vault_worker_a", state_dir, base_url)
+    material = Material()
+    provider.publish(secret_record(material, mod, 1))
+    provider.put(e2e_node(mod, material, "secretnote", 1, BODY))
+    try:
+        yield mod, provider, material, state_dir, base_url
+    finally:
+        mod.e2e_shards_close(flush=False)
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+
+
+def fresh_process_state(mod):
+    """What a new worker starts with: no cached secret, no passphrase."""
+    mod.e2e_shards_close(flush=True)
+    mod.MEMORY_SECRET_CACHE.clear()
+    mod.ACTIVE_MEMORY_SECRET_VERSIONS.clear()
+    mod.MEMORY_SECRET_ROTATION_OBSERVED.clear()
+    mod.PROJECT_VAULT_PASSPHRASE = None
+
+
+def error_of(callable_, *args):
+    with pytest.raises(Exception) as info:
+        callable_(*args)
+    return info.value
+
+
+# -- in-process behavior ------------------------------------------------------
+
+def test_setup_stores_secret_not_passphrase_and_loads_without_it(env, capsys):
+    mod, provider, material, state_dir, _ = env
+    before = error_of(mod.active_memory_secret, PROJECT)
+    assert getattr(before, "code", None) == "memory_secret_not_provisioned"
+    assert mod.memory_secret_status(PROJECT)["memory_secret"] == "not_provisioned"
+
+    result = mod.memory_secret_setup({"vault_passphrase": material.passphrase})
+    assert result["status"] == "provisioned" and result["memory_secret"] == "provisioned"
+    assert result["passphrase_stored"] is False and result["passphrase_source"] == "argument"
+    assert result["active_identity"] == {"version": 1, "wrapping_epoch": 1, "bundle_version": 1, "data_key_epoch": 1}
+    assert mod.PROJECT_VAULT_PASSPHRASE is None  # not retained after setup
+
+    vault = mod.MEMORY_SECRET_VAULT_PATH
+    assert stat.S_IMODE(vault.stat().st_mode) == 0o600
+    assert stat.S_IMODE(mod.LOCAL_VAULT_KEY_PATH.stat().st_mode) == 0o600
+    assert json.loads(vault.read_text())["schema"] == "napseer.memory-secret-vault.v1"
+
+    fresh_process_state(mod)
+    requests_before = len(provider.requests)
+    secret = mod.active_memory_secret(PROJECT)
+    assert secret["secret_bytes"] == material.secret(1)
+    # Identity comes from the server; no unwrap or passphrase was involved.
+    assert provider.requests[requests_before:] == [("GET", f"/v1/projects/{PROJECT}/vault/secrets/memory/active")]
+    whoami = mod.gateway_status()
+    assert whoami["memory_secret"] == "provisioned" and "deprecations" not in whoami
+    assert whoami["memory_secret_versions"] == [1]
+
+    captured = capsys.readouterr()
+    for label, blob in (("setup result", result), ("whoami", whoami), ("stdout", captured.out),
+                        ("stderr", captured.err)):
+        assert_no_leak(material, label, blob)
+    assert_state_dir_clean(material, state_dir)
+
+
+def test_e2e_search_uses_provisioned_secret_and_reports_not_provisioned(env):
+    mod, provider, material, state_dir, _ = env
+    mod.reindex_project({})
+    unprovisioned = mod.search_local_index({"q": BODY_TERM, "limit": 5})
+    assert unprovisioned["items"] == []
+    warning = unprovisioned["warnings"][0]
+    assert warning["code"] == "e2e_index_not_provisioned"
+    assert warning["memory_secret"] == "not_provisioned" and warning["next_action"] == SETUP_COMMAND
+
+    mod.memory_secret_setup({"vault_passphrase": material.passphrase})
+    fresh_process_state(mod)
+    mod.sync_local_index({})
+    found = mod.search_local_index({"q": BODY_TERM, "limit": 5})
+    assert [item["id"] for item in found["items"]] == ["secretnote"]
+    assert found["items"][0]["search_source"] == "local_e2e" and not found.get("warnings")
+    mod.e2e_shards_close(flush=True)
+    assert_state_dir_clean(material, state_dir)
+
+
+def test_rotation_is_reported_older_versions_still_read_and_passphrase_reprovisions(env):
+    mod, provider, material, state_dir, _ = env
+    mod.memory_secret_setup({"vault_passphrase": material.passphrase})
+    provider.publish(secret_record(material, mod, 2))
+    provider.put(e2e_node(mod, material, "newnote", 2, "second body", minute=2))
+
+    fresh_process_state(mod)
+    rotated = error_of(mod.active_memory_secret, PROJECT)
+    assert rotated.code == "memory_secret_rotated"
+    assert rotated.details == {"memory_secret": "rotated", "next_action": SETUP_COMMAND, "active_version": 2}
+    assert mod.gateway_status()["memory_secret"] == "rotated"
+    assert mod.gateway_status()["memory_secret_next_action"] == SETUP_COMMAND
+    # The stored entry still serves nodes written under the older version...
+    assert mod.memory_secret_for_read(PROJECT, 1)["secret_bytes"] == material.secret(1)
+    assert mod.decrypt_node_for_return(provider.nodes["secretnote"], project_id=PROJECT)["content_text"] == BODY
+    # ...but never the rotated active version.
+    assert error_of(mod.memory_secret_for_read, PROJECT, 2).code == "memory_secret_rotated"
+    mod.reindex_project({})
+    search = mod.search_local_index({"q": BODY_TERM, "limit": 5})
+    assert search["items"] == []
+    assert [(w["code"], w.get("memory_secret")) for w in search["warnings"]] == [("e2e_index_not_provisioned", "rotated")]
+
+    # A doctor in a fresh process detects the rotation from the server identity.
+    fresh_process_state(mod)
+    mod.local_anchor_integrity = lambda: {}
+    doctor = mod.doctor({})
+    assert doctor["memory_secret"]["memory_secret"] == "rotated"
+    assert any(SETUP_COMMAND in warning for warning in doctor["warnings"])
+
+    # A passphrase supplied through the tool-argument path re-provisions automatically.
+    mod.set_project_vault_passphrase(material.passphrase)
+    assert mod.active_memory_secret(PROJECT)["version"] == 2
+    fresh_process_state(mod)
+    assert mod.active_memory_secret(PROJECT)["secret_bytes"] == material.secret(2)
+    assert mod.gateway_status()["memory_secret"] == "provisioned"
+    assert mod.gateway_status()["memory_secret_versions"] == [2, 1]
+    for blob in (rotated.details, str(rotated), doctor):
+        assert_no_leak(material, "rotation output", blob)
+    mod.e2e_shards_close(flush=True)
+    assert_state_dir_clean(material, state_dir)
+
+
+def test_setup_after_rotation_stores_active_and_older_versions(env):
+    mod, provider, material, _state_dir, _ = env
+    provider.publish(secret_record(material, mod, 2))
+    provider.publish(secret_record(material, mod, 3))
+    del provider.versions[1]  # an unavailable older version is reported, not fatal
+    result = mod.memory_secret_setup({"vault_passphrase": material.passphrase})
+    assert result["active_identity"]["version"] == 3
+    assert result["older_versions_stored"] == [2] and result["older_versions_unavailable"] == [1]
+    fresh_process_state(mod)
+    assert mod.memory_secret_for_read(PROJECT, 2)["secret_bytes"] == material.secret(2)
+
+
+def test_forget_revokes_access_in_this_and_other_running_processes(env):
+    mod, provider, material, state_dir, base_url = env
+    mod.memory_secret_setup({"vault_passphrase": material.passphrase})
+    mod.reindex_project({})
+    assert [item["id"] for item in mod.search_local_index({"q": BODY_TERM})["items"]] == ["secretnote"]
+    mod.e2e_shards_close(flush=True)
+    assert mod.e2e_shard_path(PROJECT).exists()
+
+    # Another running worker has the secret cached in memory.
+    other = load_module("memory_vault_worker_b", state_dir, base_url)
+    other.MEMORY_SECRET_ROTATION_OBSERVED.clear()
+    assert other.active_memory_secret(PROJECT)["secret_bytes"] == material.secret(1)
+
+    removed = mod.memory_secret_forget({})
+    assert removed["status"] == "removed" and removed["removed_entries"] == 1
+    assert removed["memory_secret"] == "not_provisioned"
+    assert not mod.MEMORY_SECRET_VAULT_PATH.exists()
+    assert not mod.e2e_shard_path(PROJECT).exists()
+
+    assert error_of(mod.active_memory_secret, PROJECT).code == "memory_secret_not_provisioned"
+    # The other worker drops its cached copy on the next memory access.
+    assert error_of(other.active_memory_secret, PROJECT).code == "memory_secret_not_provisioned"
+    assert other.e2e_cached_memory_secret(PROJECT) is None
+    search = mod.search_local_index({"q": BODY_TERM})
+    assert search["items"] == [] and search["warnings"][0]["code"] == "e2e_index_not_provisioned"
+    assert mod.memory_secret_forget({})["status"] == "not_provisioned"
+    assert_no_leak(material, "forget output", removed)
+
+
+@pytest.mark.parametrize("damage", ["tampered", "wrong_local_key", "missing_local_key", "garbage"])
+def test_damaged_vault_fails_closed_without_leaking(env, capsys, damage):
+    mod, provider, material, state_dir, _ = env
+    mod.memory_secret_setup({"vault_passphrase": material.passphrase})
+    vault = mod.MEMORY_SECRET_VAULT_PATH
+    if damage == "tampered":
+        envelope = json.loads(vault.read_text())
+        ciphertext = bytearray(base64.b64decode(envelope["cipher"]["ciphertext"]))
+        ciphertext[5] ^= 0x01
+        envelope["cipher"]["ciphertext"] = base64.b64encode(bytes(ciphertext)).decode()
+        vault.write_text(json.dumps(envelope))
+    elif damage == "wrong_local_key":
+        mod.LOCAL_VAULT_KEY_PATH.write_text(base64.b64encode(os.urandom(32)).decode() + "\n")
+    elif damage == "missing_local_key":
+        mod.LOCAL_VAULT_KEY_PATH.unlink()
+    else:
+        vault.write_bytes(b"{not json")
+
+    fresh_process_state(mod)
+    failure = error_of(mod.active_memory_secret, PROJECT)
+    assert failure.code == "memory_secret_not_provisioned"
+    status = mod.gateway_status()
+    assert status["memory_secret"] == "not_provisioned"
+    assert status["memory_secret_detail"] == "local_vault_unreadable"
+    mod.reindex_project({})
+    search = mod.search_local_index({"q": BODY_TERM})
+    assert search["items"] == [] and search["warnings"][0]["code"] == "e2e_index_not_provisioned"
+    payload = mod.safe_exception_payload(failure)
+    captured = capsys.readouterr()
+    for label, blob in (("error", payload), ("status", status), ("search", search),
+                        ("stdout", captured.out), ("stderr", captured.err)):
+        assert_no_leak(material, label, blob)
+
+    # An explicit setup replaces the unreadable vault.
+    mod.memory_secret_setup({"vault_passphrase": material.passphrase})
+    fresh_process_state(mod)
+    assert mod.active_memory_secret(PROJECT)["secret_bytes"] == material.secret(1)
+
+
+def test_wrong_passphrase_stores_nothing(env):
+    mod, _provider, material, state_dir, _ = env
+    failure = error_of(mod.memory_secret_setup, {"vault_passphrase": "not-" + material.passphrase})
+    assert failure.code == "memory_secret_unwrap_failed"
+    assert not mod.MEMORY_SECRET_VAULT_PATH.exists()
+    assert_no_leak(material, "unwrap failure", mod.safe_exception_payload(failure))
+
+
+def test_env_passphrase_still_works_reports_deprecation_and_is_persisted(env, monkeypatch, capsys):
+    mod, provider, material, state_dir, _ = env
+    monkeypatch.setenv("NAPSEER_VAULT_PASSPHRASE", material.passphrase)
+    assert mod.active_memory_secret(PROJECT)["secret_bytes"] == material.secret(1)
+    status = mod.gateway_status()
+    assert status["memory_secret"] == "provisioned"
+    assert [item["code"] for item in status["deprecations"]] == ["vault_passphrase_env_deprecated"]
+    assert status["deprecations"][0]["variables"] == ["NAPSEER_VAULT_PASSPHRASE"]
+    assert status["deprecations"][0]["next_action"] == SETUP_COMMAND
+    mod.local_anchor_integrity = lambda: {}
+    doctor = mod.doctor({})
+    assert any("deprecated" in warning for warning in doctor["warnings"])
+    log = capsys.readouterr().err
+    assert '"event":"vault_passphrase_env_deprecated"' in log
+
+    # Later processes no longer need the variable.
+    monkeypatch.delenv("NAPSEER_VAULT_PASSPHRASE")
+    fresh_process_state(mod)
+    assert mod.active_memory_secret(PROJECT)["secret_bytes"] == material.secret(1)
+    assert "deprecations" not in mod.gateway_status()
+    for label, blob in (("status", status), ("doctor", doctor), ("log", log)):
+        assert_no_leak(material, label, blob)
+    assert_state_dir_clean(material, state_dir)
+
+
+def test_mcp_mode_never_prompts_even_on_a_tty(env, monkeypatch):
+    mod, _provider, _material, _state_dir, _ = env
+
+    class Tty:
+        def isatty(self):
+            return True
+
+    def forbidden_prompt(*_args, **_kwargs):
+        raise AssertionError("MCP must never prompt")
+
+    monkeypatch.setattr(mod.sys, "stdin", Tty())
+    monkeypatch.setattr(mod.getpass, "getpass", forbidden_prompt)
+    mod.MCP_SERVING = True
+    assert error_of(mod.active_memory_secret, PROJECT).code == "memory_secret_not_provisioned"
+    assert "NAPSEER_VAULT_PASSPHRASE" in str(error_of(mod.project_bundle_passphrase))
+    assert error_of(mod.memory_secret_setup, {}).code == "passphrase_required"
+    assert error_of(mod.decrypt_node_for_return, _provider.nodes["secretnote"], PROJECT).code == \
+        "memory_secret_not_provisioned"
+
+
+# -- fresh processes ----------------------------------------------------------
+
+def child_env(user_data, base_url, **extra):
+    env = {key: value for key, value in os.environ.items()
+           if not key.startswith("NAPSEER_") or key == "NAPSEER_CRYPTO_BACKEND"}
+    env.update({
+        "NAPSEER_USER_DATA_DIR": str(user_data),
+        "NAPSEER_BASE_URL": base_url,
+        "NAPSEER_TOKEN": "disposable-test-token",
+        "NAPSEER_PROJECT_ID": PROJECT,
+        "NAPSEER_ACCOUNT_ID": ACCOUNT,
+        "NAPSEER_TELEMETRY": "0",
+        "NAPSEER_TOOL_PROFILES": "all",
+        "PYTHONDONTWRITEBYTECODE": "1",
+    })
+    env.update(extra)
+    return env
+
+
+def no_echo_pty():
+    master, slave = pty.openpty()
+    attrs = termios.tcgetattr(slave)
+    attrs[3] &= ~termios.ECHO
+    termios.tcsetattr(slave, termios.TCSANOW, attrs)
+    return master, slave
+
+
+def run_interactive_setup(env, cwd, passphrase):
+    """`gateway vault setup` on a terminal: the passphrase is typed at the prompt."""
+    master, slave = no_echo_pty()
+    process = subprocess.Popen(
+        [sys.executable, str(SCRIPT), "gateway", "vault", "setup"],
+        stdin=slave, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd, env=env, start_new_session=True,
+    )
+    os.close(slave)
+    stderr = b""
+    deadline = time.monotonic() + 30
+    try:
+        while b"passphrase" not in stderr.lower():
+            remaining = deadline - time.monotonic()
+            ready, _, _ = select.select([process.stderr], [], [], max(0.0, remaining))
+            if not ready:
+                raise AssertionError("setup did not prompt for the passphrase")
+            chunk = os.read(process.stderr.fileno(), 4096)
+            if not chunk:
+                break
+            stderr += chunk
+        os.write(master, passphrase.encode("utf-8") + b"\n")
+        stdout, rest = process.communicate(timeout=60)
+    finally:
+        if process.poll() is None:
+            process.kill()
+            process.wait()
+        os.close(master)
+    return process.returncode, stdout, stderr + rest
+
+
+class McpProcess:
+    def __init__(self, env, cwd, tty_stdin=False):
+        self.master = None
+        if tty_stdin:
+            self.master, slave = no_echo_pty()
+            stdin = slave
+        else:
+            stdin = subprocess.PIPE
+        self.process = subprocess.Popen(
+            [sys.executable, "-u", str(SCRIPT)], stdin=stdin, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+            cwd=cwd, env=env, start_new_session=True,
+        )
+        if tty_stdin:
+            os.close(slave)
+        self.lines = queue.Queue()
+        self.stdout = []
+        self.stderr = b""
+        threading.Thread(target=self._read_stdout, daemon=True).start()
+
+    def _read_stdout(self):
+        for line in self.process.stdout:
+            self.stdout.append(line)
+            self.lines.put(line)
+
+    def send(self, message):
+        data = (json.dumps(message) + "\n").encode("utf-8")
+        if self.master is not None:
+            os.write(self.master, data)
+        else:
+            self.process.stdin.write(data)
+            self.process.stdin.flush()
+
+    def request(self, request_id, method, params=None, timeout=60):
+        self.send({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params or {}})
+        deadline = time.monotonic() + timeout
+        while True:
+            try:
+                message = json.loads(self.lines.get(timeout=max(0.01, deadline - time.monotonic())))
+            except queue.Empty:
+                raise AssertionError(f"no MCP response to {method}") from None
+            if message.get("id") == request_id:
+                return message
+
+    def call(self, request_id, name, arguments=None):
+        return self.request(request_id, "tools/call", {"name": name, "arguments": arguments or {}})["result"]
+
+    def start(self):
+        self.request(1, "initialize", {"protocolVersion": "2025-11-25", "capabilities": {},
+                                       "clientInfo": {"name": "memory-vault-test", "version": "1"}})
+        self.send({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}})
+
+    def close(self):
+        if self.master is not None:
+            os.close(self.master)
+        elif self.process.stdin:
+            self.process.stdin.close()
+        try:
+            self.process.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            self.process.kill()
+            self.process.wait()
+        self.stderr = self.process.stderr.read()
+        return b"".join(self.stdout), self.stderr
+
+
+@pytest.fixture
+def processes(env, tmp_path):
+    mod, provider, material, _state_dir, base_url = env
+    user_data = tmp_path / "user-data"
+    workdir = tmp_path / "work"
+    workdir.mkdir()
+    return mod, provider, material, user_data, workdir, child_env(user_data, base_url)
+
+
+def test_interactive_setup_then_fresh_mcp_process_searches_e2e_automatically(processes):
+    mod, provider, material, user_data, workdir, environment = processes
+    code, stdout, stderr = run_interactive_setup(environment, workdir, material.passphrase)
+    assert code == 0, stderr.decode(errors="replace")[-2000:]
+    result = json.loads(stdout)
+    assert result["status"] == "provisioned" and result["passphrase_source"] == "prompt"
+    assert result["passphrase_stored"] is False
+    assert_no_leak(material, "setup stdout", stdout)
+    assert_no_leak(material, "setup stderr", stderr)
+
+    # A new CLI process builds the encrypted E2E index without any passphrase.
+    reindexed = subprocess.run([sys.executable, str(SCRIPT), "reindex"], cwd=workdir, env=environment,
+                               capture_output=True, timeout=120, stdin=subprocess.DEVNULL)
+    assert reindexed.returncode == 0, reindexed.stderr.decode(errors="replace")[-2000:]
+    assert json.loads(reindexed.stdout)["e2e_index"]["status"] == "applied"
+    assert_no_leak(material, "reindex output", reindexed.stdout + reindexed.stderr)
+
+    # A new MCP worker: no passphrase anywhere in its environment.
+    client = McpProcess(environment, workdir)
+    try:
+        client.start()
+        whoami = client.call(2, "nap_whoami")["structuredContent"]
+        assert whoami["memory_secret"] == "provisioned" and "deprecations" not in whoami
+        found = client.call(3, "nap_discover", {"q": BODY_TERM, "mode": "local"})
+        assert found["isError"] is False, found["structuredContent"]
+        found = found["structuredContent"]
+        assert [item["node_id"] for item in found["items"]] == ["secretnote"]
+        assert found["items"][0]["search_source"] == "local_e2e" and not found.get("warnings")
+        node = client.call(4, "nap_node_get", {"node_id": "secretnote"})["structuredContent"]
+        assert BODY in json.dumps(node)
+    finally:
+        out, err = client.close()
+    assert_no_leak(material, "mcp stdout", out)
+    assert_no_leak(material, "mcp stderr", err)
+    assert_state_dir_clean(material, user_data, workdir)
+    assert not (workdir / ".napseer").exists()
+
+    # Removal through the CLI revokes access for the next process.
+    forgot = subprocess.run([sys.executable, str(SCRIPT), "gateway", "vault", "forget"], cwd=workdir,
+                            env=environment, capture_output=True, timeout=60, stdin=subprocess.DEVNULL)
+    assert forgot.returncode == 0 and json.loads(forgot.stdout)["status"] == "removed"
+    client = McpProcess(environment, workdir)
+    try:
+        client.start()
+        failed = client.call(2, "nap_node_get", {"node_id": "secretnote"})
+        assert failed["isError"] is True
+        assert failed["structuredContent"]["error"]["code"] == "memory_secret_not_provisioned"
+    finally:
+        client.close()
+
+
+def test_mcp_worker_on_a_tty_never_prompts(processes):
+    _mod, _provider, material, user_data, workdir, environment = processes
+    client = McpProcess(environment, workdir, tty_stdin=True)
+    try:
+        client.start()
+        failed = client.call(2, "nap_node_get", {"node_id": "secretnote"})
+        assert failed["isError"] is True
+        error = failed["structuredContent"]["error"]
+        assert error["code"] == "memory_secret_not_provisioned"
+        assert error["details"]["next_action"] == SETUP_COMMAND
+        # stdin was not consumed by a prompt: the worker keeps answering.
+        assert client.call(3, "nap_whoami")["structuredContent"]["memory_secret"] == "not_provisioned"
+    finally:
+        _out, err = client.close()
+    assert b"passphrase:" not in err.lower()
+    assert_state_dir_clean(material, user_data)

tests/test_nap_install_gateway_routes.py

@@ -36,6 +36,11 @@ def run():
 
     assert mod.handle_gateway(["vault", "rotate-secret", "--kind", "memory"]) == 0
     assert calls[-1] == ("gateway", ["vault", "rotate-secret", "--kind", "memory"])
+    # Memory secret provisioning/removal route to the wrapper, which owns the prompt.
+    assert mod.handle_gateway(["vault", "setup"]) == 0
+    assert calls[-1] == ("gateway", ["vault", "setup"])
+    assert mod.handle_gateway(["vault", "forget"]) == 0
+    assert calls[-1] == ("gateway", ["vault", "forget"])
 
     try:
         mod.handle_gateway(["process", "--all"])