"""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)