mempalace/tests/test_mcp_server.py

"""
test_mcp_server.py — Tests for the MCP server tool handlers and dispatch.

Tests each tool handler directly (unit-level) and the handle_request
dispatch layer (integration-level). Uses isolated palace + KG fixtures
via monkeypatch to avoid touching real data.
"""

from datetime import datetime
import json
import os
import subprocess
import sys
from unittest.mock import MagicMock

import pytest


# ── MCP entry point: PYTHONPATH stripping ────────────────────────────────


_MCP_LEAK_PREFIX = "/__mempalace_mcp_leak_sentinel__"


def test_mcp_main_strips_leaked_pythonpath_from_env():
    """mempalace.mcp_server:main must drop PYTHONPATH from the process env
    so any subprocess this server spawns starts clean. Mirrors the
    sys.path-filter test in test_init.py but for the env half of the
    split fix. See #1423.

    Three assertions cover the full split contract:
    - ENV_MID (after import, before main) is preserved verbatim:
      regression detector for someone moving the env pop back into
      __init__.py.
    - SENTINEL_IN_PATH is False at import time: package-level sys.path
      filter half of the split actually ran.
    - ENV_AFTER (after main) is None: MCP entry-point env strip ran.

    The main loop reads JSON-RPC lines from stdin until EOF; closing
    stdin makes readline() return '' and exits the loop cleanly, which
    lets us observe the post-main env state. Probes go to stderr because
    mcp_server redirects stdout at import time for clean JSON-RPC."""
    expected_env = f"{_MCP_LEAK_PREFIX}/a{os.pathsep}{_MCP_LEAK_PREFIX}/b"
    env = os.environ.copy()
    env["PYTHONPATH"] = expected_env
    code = (
        "import os, sys\n"
        "from mempalace.mcp_server import main\n"
        f"prefix = {_MCP_LEAK_PREFIX!r}\n"
        "sys.stderr.write('ENV_MID: ' + repr(os.environ.get('PYTHONPATH')) + '\\n')\n"
        "sys.stderr.write('SENTINEL_IN_PATH: ' + repr(any(prefix in (p or '') for p in sys.path)) + '\\n')\n"
        "main()\n"
        "sys.stderr.write('ENV_AFTER: ' + repr(os.environ.get('PYTHONPATH')) + '\\n')\n"
    )
    result = subprocess.run(
        [sys.executable, "-c", code],
        env=env,
        input="",  # empty stdin → readline() returns '' → loop breaks
        capture_output=True,
        text=True,
        check=False,
        timeout=30,
    )
    diag = f"rc={result.returncode}; stdout={result.stdout!r}; stderr={result.stderr!r}"
    assert result.returncode == 0, f"subprocess failed: {diag}"
    assert f"ENV_MID: {expected_env!r}" in result.stderr, (
        f"package import unexpectedly stripped env (regression in __init__.py): {diag}"
    )
    assert "SENTINEL_IN_PATH: False" in result.stderr, (
        f"package import did not filter sys.path (regression in __init__.py): {diag}"
    )
    assert "ENV_AFTER: None" in result.stderr, f"MCP server did not strip PYTHONPATH: {diag}"


def _patch_mcp_server(monkeypatch, config, kg):
    """Patch the mcp_server module globals to use test fixtures."""
    from mempalace import mcp_server

    monkeypatch.setattr(mcp_server, "_config", config)
    # Accept varargs because production ``_get_kg`` now takes an optional
    # canonical_path; ``_call_kg`` passes the captured key through.
    monkeypatch.setattr(mcp_server, "_get_kg", lambda *a, **kw: kg)


def _get_collection(palace_path, create=False):
    """Helper to get collection from test palace.

    Returns (client, collection) so callers can clean up the client
    when they are done.
    """
    import chromadb

    client = chromadb.PersistentClient(path=palace_path)
    if create:
        return (
            client,
            client.get_or_create_collection("mempalace_drawers", metadata={"hnsw:space": "cosine"}),
        )
    return client, client.get_collection("mempalace_drawers")


# ── Cold-start diagnostics (#1495) ──────────────────────────────────────


class TestColdStartDiagnostics:
    """``MEMPALACE_LOG_FILE`` + ``MEMPALACE_EAGER_WARMUP`` (#1495).

    Each test runs ``main()`` in a fresh ``subprocess`` because

    * ``_init_logging`` uses ``logging.basicConfig(force=True)`` which
      would otherwise reset pytest's ``caplog`` handlers across cases,
    * ``ChromaBackend._resolve_embedding_function`` is a class-level
      attribute that test monkeypatching mutates globally,
    * The whole point of the new env vars is process-startup behaviour
      and must be exercised under a real ``main()`` boot path.

    Pattern mirrors ``test_mcp_main_strips_leaked_pythonpath_from_env``.
    ``_run_main`` injects ``extra_code`` as a hard-coded ``-c`` source
    fragment from this file only (no untrusted input flows in); the
    subprocess argv form ``[sys.executable, "-c", code]`` avoids shell
    interpretation entirely.
    """

    @staticmethod
    def _run_main(env_overrides: dict, extra_code: str = "", timeout: int = 30):
        env = {
            k: v
            for k, v in os.environ.items()
            if k not in env_overrides or env_overrides[k] is not None
        }
        for k, v in env_overrides.items():
            if v is None:
                env.pop(k, None)
            else:
                env[k] = v
        code = extra_code + "from mempalace.mcp_server import main\nmain()\n"
        return subprocess.run(
            [sys.executable, "-c", code],
            env=env,
            input="",  # empty stdin → readline() returns '' → loop breaks immediately
            capture_output=True,
            text=True,
            check=False,
            timeout=timeout,
        )

    def test_log_file_unset_attaches_only_stream_handler(self, tmp_path):
        marker = tmp_path / "handlers.txt"
        env_overrides = {"MEMPALACE_LOG_FILE": None}
        extra = (
            "import logging, pathlib\n"
            "from mempalace import mcp_server  # noqa: F401 — triggers _init_logging()\n"
            f"pathlib.Path({str(marker)!r}).write_text("
            "','.join(type(h).__name__ for h in logging.getLogger().handlers)"
            ")\n"
            "raise SystemExit(0)\n"
        )
        result = self._run_main(env_overrides, extra_code=extra)
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert marker.read_text().split(",") == ["StreamHandler"], marker.read_text()

    def test_log_file_empty_string_attaches_only_stream_handler(self, tmp_path):
        marker = tmp_path / "handlers.txt"
        env_overrides = {"MEMPALACE_LOG_FILE": "   "}  # whitespace counts as unset after .strip()
        extra = (
            "import logging, pathlib\n"
            "from mempalace import mcp_server  # noqa: F401\n"
            f"pathlib.Path({str(marker)!r}).write_text("
            "','.join(type(h).__name__ for h in logging.getLogger().handlers)"
            ")\n"
            "raise SystemExit(0)\n"
        )
        result = self._run_main(env_overrides, extra_code=extra)
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert marker.read_text().split(",") == ["StreamHandler"], marker.read_text()

    def test_log_file_set_attaches_file_handler_and_persists_startup_line(self, tmp_path):
        log_path = tmp_path / "mcp.log"
        result = self._run_main({"MEMPALACE_LOG_FILE": str(log_path)})
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert log_path.exists(), f"log file missing; stderr={result.stderr!r}"
        body = log_path.read_text(encoding="utf-8")
        assert "MemPalace MCP Server starting" in body, body

    def test_log_file_invalid_path_falls_back_to_stderr_with_warning(self, tmp_path):
        # Unique directory name we can grep for cross-platform without
        # depending on path-separator formatting in the %r warning value.
        missing_dir = "missing_dir_for_1495"
        bad_path = tmp_path / missing_dir / "mcp.log"
        result = self._run_main({"MEMPALACE_LOG_FILE": str(bad_path)})
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        # Invalid path must NOT crash the server, must surface a warning, must
        # NOT create the file (the missing-directory ancestor is the failure).
        # Warning must name MEMPALACE_LOG_FILE so the operator knows the source.
        assert "could not be opened" in result.stderr, result.stderr
        assert "MEMPALACE_LOG_FILE" in result.stderr, result.stderr
        assert missing_dir in result.stderr, result.stderr
        assert not bad_path.exists()

    @staticmethod
    def _make_fake_palace(tmp_path):
        """Create just enough on disk for ``_maybe_eager_warmup_embedder``'s
        fresh-install pre-check to pass (``chroma.sqlite3`` exists).

        Returns the palace dir as a string. The file is empty — production
        code must not read its bytes during pre-check; only its existence
        gates whether warmup proceeds to the chromadb client open.
        """
        palace = tmp_path / "palace"
        palace.mkdir()
        (palace / "chroma.sqlite3").touch()
        return str(palace)

    @staticmethod
    def _spy_get_collection_extra(marker_path, return_expr="None"):
        """Render an ``extra_code`` fragment that monkeypatches ``_get_collection``.

        ``marker_path`` records that the spy fired; ``return_expr`` is a Python
        expression evaluated inside the subprocess for the call's return value
        (e.g. ``"None"`` or ``"_FakeCol()"``).
        """
        return (
            "import pathlib\n"
            "from mempalace import mcp_server\n"
            "def _spy_get_collection(create=False):\n"
            f"    pathlib.Path({str(marker_path)!r}).write_text('called')\n"
            f"    return {return_expr}\n"
            "mcp_server._get_collection = _spy_get_collection\n"
        )

    def test_eager_warmup_off_by_default_does_not_open_collection(self, tmp_path):
        marker = tmp_path / "called.txt"
        palace = self._make_fake_palace(tmp_path)
        result = self._run_main(
            {"MEMPALACE_EAGER_WARMUP": None, "MEMPALACE_PALACE_PATH": palace},
            extra_code=self._spy_get_collection_extra(marker),
        )
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert not marker.exists(), "warmup ran despite env var being unset"

    @pytest.mark.parametrize("value", ["0", "false", "no", "off", "FALSE"])
    def test_eager_warmup_explicit_falsy_skips_collection_open_without_warning(
        self, tmp_path, value
    ):
        marker = tmp_path / "called.txt"
        palace = self._make_fake_palace(tmp_path)
        result = self._run_main(
            {"MEMPALACE_EAGER_WARMUP": value, "MEMPALACE_PALACE_PATH": palace},
            extra_code=self._spy_get_collection_extra(marker),
        )
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert not marker.exists(), f"warmup ran for explicit-falsy value {value!r}"
        assert "not recognized" not in result.stderr, (
            f"explicit-falsy {value!r} should not log a warning; stderr={result.stderr!r}"
        )

    @pytest.mark.parametrize("value", ["tru", "maybe", "ENABLED", "2"])
    def test_eager_warmup_unrecognized_value_warns_and_skips_collection_open(self, tmp_path, value):
        marker = tmp_path / "called.txt"
        palace = self._make_fake_palace(tmp_path)
        result = self._run_main(
            {"MEMPALACE_EAGER_WARMUP": value, "MEMPALACE_PALACE_PATH": palace},
            extra_code=self._spy_get_collection_extra(marker),
        )
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert not marker.exists(), f"warmup ran despite unrecognized value {value!r}"
        assert "not recognized" in result.stderr, result.stderr

    @pytest.mark.parametrize("value", ["1", "true", "YES", "On"])
    def test_eager_warmup_truthy_opens_collection_and_invokes_query(self, tmp_path, value):
        """C1 (#1495): warmup must call ``col.query(...)`` — not just open the collection.

        ChromaDB's ``ONNXMiniLM_L6_V2.__init__`` only imports ``onnxruntime``;
        ``InferenceSession`` and model download happen inside ``__call__``,
        which the chromadb query path drives. Pinning both call sites here
        prevents a regression to a no-op resolver-only warmup (the same
        failure mode silent-failure-hunter flagged in initial review).
        Reporter's #1495 proposal: same path covers HNSW cold-load too.
        """
        open_marker = tmp_path / "open_called.txt"
        query_marker = tmp_path / "query_called.txt"
        palace = self._make_fake_palace(tmp_path)
        extra = (
            "import pathlib\n"
            "from mempalace import mcp_server\n"
            "class _FakeCol:\n"
            "    def query(self, **kwargs):\n"
            f"        pathlib.Path({str(query_marker)!r}).write_text(repr(kwargs))\n"
            "        return {'ids': [[]], 'distances': [[]], 'documents': [[]]}\n"
            "_fake_col = _FakeCol()\n"
            "def _spy_get_collection(create=False):\n"
            f"    pathlib.Path({str(open_marker)!r}).write_text('open')\n"
            "    return _fake_col\n"
            "mcp_server._get_collection = _spy_get_collection\n"
        )
        result = self._run_main(
            {"MEMPALACE_EAGER_WARMUP": value, "MEMPALACE_PALACE_PATH": palace},
            extra_code=extra,
        )
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert open_marker.exists(), (
            f"_get_collection not called for {value!r}; stderr={result.stderr!r}"
        )
        assert query_marker.exists(), (
            f"col.query not invoked for {value!r} — warmup is a no-op "
            f"(would let cold-load hit first MCP call); stderr={result.stderr!r}"
        )
        # query was called with the sentinel probe text and n_results=1.
        kwargs_repr = query_marker.read_text()
        assert "__mempalace_warmup_probe__" in kwargs_repr, kwargs_repr
        assert "n_results" in kwargs_repr and "1" in kwargs_repr, kwargs_repr
        # Success path logs embedder + HNSW readiness + palace + device for ops.
        assert "embedder + HNSW ready" in result.stderr, result.stderr
        assert f"palace={palace}" in result.stderr, result.stderr

    def test_eager_warmup_fresh_install_skips_without_creating_palace(self, tmp_path):
        """Real integration test (no monkeypatch): an empty palace dir with no
        ``chroma.sqlite3`` must trigger the pre-check skip path BEFORE any
        chromadb call materializes the palace scaffold on disk.

        This pins three behaviours simultaneously:

        1. ``returncode == 0`` — fresh install does not crash the server.
        2. ``chroma.sqlite3`` is NOT created — warmup respects the
           "no on-disk state before ``mempalace init``" contract from
           CLAUDE.md ("Incremental only"). A regression that drops the
           pre-check would let chromadb's ``PersistentClient(path=...)``
           materialize the palace dir.
        3. ``"nothing to warm"`` lands in stderr — the documented INFO
           message actually fires (the previous test that asserted this
           via a monkeypatched ``_get_collection`` was tautological because
           the real ``_get_collection`` swallows ``NotFoundError`` into
           ``return None`` and silently materializes the palace).
        4. No chromadb retry tracebacks ("attempt N/2 failed") leak into
           stderr — those are the noise this PR exists to reduce.
        """
        palace = tmp_path / "fresh_palace"
        palace.mkdir()
        # Confirm precondition: no chroma.sqlite3 exists before main().
        db_path = palace / "chroma.sqlite3"
        assert not db_path.exists()
        result = self._run_main(
            {"MEMPALACE_EAGER_WARMUP": "1", "MEMPALACE_PALACE_PATH": str(palace)},
        )
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert "nothing to warm" in result.stderr, result.stderr
        assert "collection open failed" not in result.stderr, result.stderr
        assert "warmup query failed" not in result.stderr, result.stderr
        assert "embedder + HNSW ready" not in result.stderr, result.stderr
        assert "attempt 1/2 failed" not in result.stderr, result.stderr
        assert "attempt 2/2 failed" not in result.stderr, result.stderr
        # Pin the no-side-effect contract: the warmup MUST NOT create the
        # palace scaffold on disk before the user runs ``mempalace init``.
        assert not db_path.exists(), (
            f"warmup materialized chroma.sqlite3 in a fresh palace dir "
            f"(violates 'Incremental only' from CLAUDE.md); stderr={result.stderr!r}"
        )

    def test_eager_warmup_collection_returning_none_surfaces_warning(self, tmp_path):
        """_get_collection retries internally and returns None on persistent
        failure (mcp_server.py:373). Warmup must not log a misleading
        success line in that case."""
        palace = self._make_fake_palace(tmp_path)
        extra = self._spy_get_collection_extra(tmp_path / "called.txt", return_expr="None")
        result = self._run_main(
            {"MEMPALACE_EAGER_WARMUP": "1", "MEMPALACE_PALACE_PATH": palace},
            extra_code=extra,
        )
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert "_get_collection returned None" in result.stderr, result.stderr
        assert "embedder + HNSW ready" not in result.stderr, result.stderr

    def test_eager_warmup_collection_open_failure_logs_and_does_not_block_server(self, tmp_path):
        palace = self._make_fake_palace(tmp_path)
        extra = (
            "from mempalace import mcp_server\n"
            "def _boom(create=False):\n"
            "    raise RuntimeError('synthetic-collection-open-fail-1495')\n"
            "mcp_server._get_collection = _boom\n"
        )
        result = self._run_main(
            {"MEMPALACE_EAGER_WARMUP": "1", "MEMPALACE_PALACE_PATH": palace},
            extra_code=extra,
        )
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert "collection open failed" in result.stderr, result.stderr
        assert "synthetic-collection-open-fail-1495" in result.stderr, result.stderr
        # palace + error class included in the diagnostic
        assert f"palace={palace}" in result.stderr, result.stderr
        assert "error=RuntimeError" in result.stderr, result.stderr

    def test_eager_warmup_query_failure_logs_and_persists_to_log_file(self, tmp_path):
        """Query may raise (broken HNSW, network failure during ONNX download,
        runtime decoder error). Server stays up and the diagnostic lands in
        both stderr AND ``MEMPALACE_LOG_FILE`` — the latter is the whole
        point of #1495 for ops debugging the original -32000."""
        palace = self._make_fake_palace(tmp_path)
        log_path = tmp_path / "mcp.log"
        extra = (
            "from mempalace import mcp_server\n"
            "class _BadCol:\n"
            "    def query(self, **kwargs):\n"
            "        raise RuntimeError('synthetic-query-fail-1495')\n"
            "mcp_server._get_collection = lambda create=False: _BadCol()\n"
        )
        result = self._run_main(
            {
                "MEMPALACE_EAGER_WARMUP": "1",
                "MEMPALACE_PALACE_PATH": palace,
                "MEMPALACE_LOG_FILE": str(log_path),
            },
            extra_code=extra,
        )
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        assert "warmup query failed" in result.stderr, result.stderr
        assert "synthetic-query-fail-1495" in result.stderr, result.stderr
        assert f"palace={palace}" in result.stderr, result.stderr
        assert "error=RuntimeError" in result.stderr, result.stderr
        assert log_path.exists(), f"log file not created; stderr={result.stderr!r}"
        body = log_path.read_text(encoding="utf-8")
        assert "warmup query failed" in body, body
        assert "synthetic-query-fail-1495" in body, body

    def test_log_file_path_with_embedded_newline_does_not_crash(self, tmp_path):
        """``MEMPALACE_LOG_FILE`` containing a newline (rare misconfig from
        a YAML/env file copy-paste) must fall through the (OSError, ValueError)
        catch rather than escape as an unhandled exception at import time."""
        # Embedding \n inside a path component triggers ValueError on POSIX
        # ("embedded null byte" raises on OS-level open) or OSError depending
        # on platform — both should land in the fail-soft branch.
        bad_path = str(tmp_path / "with\nnewline" / "mcp.log")
        result = self._run_main({"MEMPALACE_LOG_FILE": bad_path})
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        # Server proceeds with stderr-only and surfaces the env-var-named
        # warning so ops can correlate the misconfig.
        assert "could not be opened" in result.stderr, result.stderr
        assert "MEMPALACE_LOG_FILE" in result.stderr, result.stderr

    def test_log_file_invalid_path_failure_surfaces_before_first_log_record(self, tmp_path):
        """Behavioural pin: ``delay=True`` MUST NOT be used on the FileHandler.

        With ``delay=True`` an invalid path raises inside ``emit()`` at runtime,
        unhandled, defeating the fail-soft contract documented in ``_init_logging``.
        This test pins the eager-open semantics by checking that the warning lands
        BEFORE the ``MemPalace MCP Server starting...`` banner — proving that
        ``FileHandler.__init__`` raised and was caught at module import."""
        bad_path = tmp_path / "regression_pin_dir" / "mcp.log"
        result = self._run_main({"MEMPALACE_LOG_FILE": str(bad_path)})
        assert result.returncode == 0, f"stderr={result.stderr!r}"
        warning_pos = result.stderr.find("could not be opened")
        banner_pos = result.stderr.find("MemPalace MCP Server starting")
        assert warning_pos != -1, f"warning missing; stderr={result.stderr!r}"
        assert banner_pos != -1, f"banner missing; stderr={result.stderr!r}"
        assert warning_pos < banner_pos, (
            f"warning at {warning_pos} must precede banner at {banner_pos} — "
            f"if banner is first, FileHandler was opened lazily (delay=True regression). "
            f"stderr={result.stderr!r}"
        )


# ── Protocol Layer ──────────────────────────────────────────────────────


class TestHandleRequest:
    def test_initialize(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request({"method": "initialize", "id": 1, "params": {}})
        assert resp["result"]["serverInfo"]["name"] == "mempalace"
        assert resp["id"] == 1

    def test_initialize_negotiates_client_version(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request(
            {
                "method": "initialize",
                "id": 1,
                "params": {"protocolVersion": "2025-11-25"},
            }
        )
        assert resp["result"]["protocolVersion"] == "2025-11-25"

    def test_initialize_negotiates_older_supported_version(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request(
            {
                "method": "initialize",
                "id": 1,
                "params": {"protocolVersion": "2025-03-26"},
            }
        )
        assert resp["result"]["protocolVersion"] == "2025-03-26"

    def test_initialize_unknown_version_falls_back_to_latest(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request(
            {
                "method": "initialize",
                "id": 1,
                "params": {"protocolVersion": "9999-12-31"},
            }
        )
        from mempalace.mcp_server import SUPPORTED_PROTOCOL_VERSIONS

        assert resp["result"]["protocolVersion"] == SUPPORTED_PROTOCOL_VERSIONS[0]

    def test_initialize_missing_version_uses_oldest(self):
        from mempalace.mcp_server import handle_request, SUPPORTED_PROTOCOL_VERSIONS

        resp = handle_request({"method": "initialize", "id": 1, "params": {}})
        assert resp["result"]["protocolVersion"] == SUPPORTED_PROTOCOL_VERSIONS[-1]

    def test_notifications_initialized_returns_none(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request({"method": "notifications/initialized", "id": None, "params": {}})
        assert resp is None

    def test_ping_returns_empty_result(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request({"method": "ping", "id": 11, "params": {}})
        assert resp["id"] == 11
        assert resp["result"] == {}

    def test_tools_list(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request({"method": "tools/list", "id": 2, "params": {}})
        tools = resp["result"]["tools"]
        names = {t["name"] for t in tools}
        assert "mempalace_status" in names
        assert "mempalace_search" in names
        assert "mempalace_add_drawer" in names
        assert "mempalace_kg_add" in names

    def test_null_arguments_does_not_hang(self, monkeypatch, config, palace_path, seeded_kg):
        """Sending arguments: null should return a result, not hang (#394)."""
        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import handle_request

        _client, _col = _get_collection(palace_path, create=True)
        del _client
        resp = handle_request(
            {
                "method": "tools/call",
                "id": 10,
                "params": {"name": "mempalace_status", "arguments": None},
            }
        )
        assert "error" not in resp
        assert resp["result"] is not None

    def test_unknown_tool(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request(
            {
                "method": "tools/call",
                "id": 3,
                "params": {"name": "nonexistent_tool", "arguments": {}},
            }
        )
        assert resp["error"]["code"] == -32601

    def test_tools_call_missing_params(self):
        from mempalace.mcp_server import handle_request

        for bad_params in [None, {}, {"arguments": {}}]:
            resp = handle_request(
                {
                    "method": "tools/call",
                    "id": 15,
                    "params": bad_params,
                }
            )
            assert resp["error"]["code"] == -32602
            assert "Invalid params" in resp["error"]["message"]

    def test_unknown_method(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request({"method": "unknown/method", "id": 4, "params": {}})
        assert resp["error"]["code"] == -32601

    def test_any_notification_returns_none(self):
        """All notifications/* methods should return None (no response)."""
        from mempalace.mcp_server import handle_request

        for method in [
            "notifications/initialized",
            "notifications/cancelled",
            "notifications/progress",
            "notifications/roots/list_changed",
        ]:
            resp = handle_request({"method": method, "params": {}})
            assert resp is None, f"{method} should return None"

    def test_unknown_method_no_id_returns_none(self):
        """Messages without id (notifications) must never get a response."""
        from mempalace.mcp_server import handle_request

        resp = handle_request({"method": "unknown/thing", "params": {}})
        assert resp is None

    def test_malformed_method_none(self):
        """method=None or missing should not crash."""
        from mempalace.mcp_server import handle_request

        # Explicit None
        resp = handle_request({"method": None, "params": {}})
        assert resp is None  # no id → no response

        # Missing method entirely
        resp = handle_request({"params": {}})
        assert resp is None

        # method=None with id → should return error, not crash
        resp = handle_request({"method": None, "id": 99, "params": {}})
        assert resp["error"]["code"] == -32601

    @pytest.mark.parametrize("payload", [None, [], "plain", 42, True])
    def test_handle_request_invalid_payload_returns_jsonrpc_error(self, payload):
        from mempalace.mcp_server import handle_request

        resp = handle_request(payload)
        assert resp == {
            "jsonrpc": "2.0",
            "id": None,
            "error": {"code": -32600, "message": "Invalid Request"},
        }

    def test_tools_call_dispatches(self, monkeypatch, config, palace_path, seeded_kg):
        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import handle_request

        # Create a collection so status works
        _client, _col = _get_collection(palace_path, create=True)
        del _client

        resp = handle_request(
            {
                "method": "tools/call",
                "id": 5,
                "params": {"name": "mempalace_status", "arguments": {}},
            }
        )
        assert "result" in resp
        content = json.loads(resp["result"]["content"][0]["text"])
        assert "total_drawers" in content


# ── Read Tools ──────────────────────────────────────────────────────────


class TestReadTools:
    def test_status_cold_start_no_collection(self, monkeypatch, config, palace_path, kg):
        """Status on a valid palace with no ChromaDB collection yet (#830).

        After `mempalace init`, chroma.sqlite3 exists but the mempalace_drawers
        collection has not been created (no mine or add_drawer yet).  Status
        should return total_drawers: 0, not 'No palace found'.
        """
        import chromadb

        _patch_mcp_server(monkeypatch, config, kg)
        # Create the DB file (init does this) but NOT the collection
        client = chromadb.PersistentClient(path=palace_path)
        del client
        from mempalace.mcp_server import tool_status

        result = tool_status()
        assert "error" not in result, f"cold-start should not error: {result}"
        assert result["total_drawers"] == 0

    def test_status_empty_palace(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_status

        result = tool_status()
        assert result["total_drawers"] == 0
        assert result["wings"] == {}

    def test_status_with_data(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_status

        result = tool_status()
        assert result["total_drawers"] == 4
        assert "project" in result["wings"]
        assert "notes" in result["wings"]

    def test_status_handles_none_metadata_without_partial(
        self, monkeypatch, config, palace_path, kg
    ):
        """tool_status must not crash or go partial when the metadata cache
        returns a ``None`` entry — palaces can contain drawers with no
        metadata (older mining paths, third-party writes). Before the guard,
        ``m.get("wing")`` raised AttributeError mid-tally and the result
        carried ``"error"`` + ``"partial": True`` even though the data was
        perfectly fetchable."""
        from unittest.mock import patch as _patch

        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_status

        # Inject a metadata cache where one entry is None
        with _patch("mempalace.mcp_server._get_collection") as mock_get_col:
            fake_col = type("C", (), {"count": lambda self: 2})()
            mock_get_col.return_value = fake_col
            with _patch(
                "mempalace.mcp_server._get_cached_metadata",
                return_value=[{"wing": "proj", "room": "r"}, None],
            ):
                result = tool_status()

        # The None-metadata drawer falls under 'unknown/unknown' — no crash,
        # no partial flag.
        assert "error" not in result
        assert result.get("partial") is not True
        assert result["total_drawers"] == 2
        assert result["wings"].get("proj") == 1
        assert result["wings"].get("unknown") == 1

    def test_list_wings(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_list_wings

        result = tool_list_wings()
        assert result["wings"]["project"] == 3
        assert result["wings"]["notes"] == 1

    def test_list_rooms_all(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_list_rooms

        result = tool_list_rooms()
        assert "backend" in result["rooms"]
        assert "frontend" in result["rooms"]
        assert "planning" in result["rooms"]

    def test_list_rooms_filtered(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_list_rooms

        result = tool_list_rooms(wing="project")
        assert "backend" in result["rooms"]
        assert "planning" not in result["rooms"]

    def test_get_taxonomy(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_get_taxonomy

        result = tool_get_taxonomy()
        assert result["taxonomy"]["project"]["backend"] == 2
        assert result["taxonomy"]["project"]["frontend"] == 1
        assert result["taxonomy"]["notes"]["planning"] == 1

    def test_no_palace_returns_error(self, monkeypatch, config, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_status

        result = tool_status()
        assert "error" in result


# ── Regression: None-metadata safety (issue #1426) ──────────────────────


class TestNoneMetadataSafety:
    """Regression coverage for issue #1426.

    ChromaDB's ``col.get()`` / ``col.query()`` can return ``None`` for the
    metadata cell of a partially-flushed row or any row written without
    metadata in older formats. Before the ``_safe_meta`` boundary helper,
    indexing the result yielded ``None``, the next ``.get(...)`` raised
    ``AttributeError: 'NoneType' object has no attribute 'get'``, and the
    handler crashed before the ``DELETE FROM embeddings_queue`` cleanup
    step — so the queue grew without bound while writes kept appearing
    successful.

    Each test simulates Chroma returning ``None`` in the metadatas list
    via a stub collection — Chroma's own write path rejects ``None`` at
    insert time, so we can't reproduce the upstream state by writing
    bad data through the real backend. Mocking ``_get_collection`` lets
    us assert the handler tolerates the failure mode that actually shows
    up in the wild.
    """

    def test_safe_meta_helper_coerces_none_to_empty_dict(self):
        from mempalace.mcp_server import _safe_meta

        assert _safe_meta(None) == {}
        assert _safe_meta({}) == {}
        assert _safe_meta({"wing": "x"}) == {"wing": "x"}
        # Defensive against other non-dict types Chroma might return on
        # malformed rows — coerce, don't crash.
        assert _safe_meta("not a dict") == {}
        assert _safe_meta(["wing", "x"]) == {}

    def test_get_drawer_tolerates_none_metadata(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from unittest.mock import MagicMock

        from mempalace import mcp_server

        stub_col = MagicMock()
        stub_col.get.return_value = {
            "ids": ["drawer_none_meta"],
            "documents": ["verbatim body"],
            "metadatas": [None],
        }
        monkeypatch.setattr(mcp_server, "_get_collection", lambda create=False: stub_col)

        result = mcp_server.tool_get_drawer("drawer_none_meta")
        assert "error" not in result
        assert result["drawer_id"] == "drawer_none_meta"
        # Missing metadata reduces to empty defaults — no crash, no leak.
        assert result["wing"] == ""
        assert result["room"] == ""
        assert result["content"] == "verbatim body"

    def test_list_drawers_tolerates_none_metadata(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from unittest.mock import MagicMock

        from mempalace import mcp_server

        stub_col = MagicMock()
        stub_col.get.return_value = {
            "ids": ["drawer_a", "drawer_b"],
            "documents": ["body a", "body b"],
            "metadatas": [None, {"wing": "ok", "room": "fine"}],
        }
        stub_col.count.return_value = 2
        monkeypatch.setattr(mcp_server, "_get_collection", lambda create=False: stub_col)

        result = mcp_server.tool_list_drawers()
        assert result["count"] == 2
        assert result["drawers"][0]["wing"] == ""
        assert result["drawers"][0]["room"] == ""
        assert result["drawers"][1]["wing"] == "ok"
        assert result["drawers"][1]["room"] == "fine"

    def test_update_drawer_tolerates_none_metadata(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from unittest.mock import MagicMock

        from mempalace import mcp_server

        stub_col = MagicMock()
        stub_col.get.return_value = {
            "ids": ["drawer_none_meta"],
            "documents": ["old body"],
            "metadatas": [None],
        }
        monkeypatch.setattr(mcp_server, "_get_collection", lambda create=False: stub_col)

        result = mcp_server.tool_update_drawer("drawer_none_meta", wing="recovered")
        # Should succeed: old_meta is coerced to {}, new wing slots in cleanly.
        assert result.get("success") is True
        # Confirm the update call carried the new wing without inheriting None.
        update_call = stub_col.update.call_args
        assert update_call is not None
        new_meta = update_call.kwargs["metadatas"][0]
        assert new_meta["wing"] == "recovered"

    def test_delete_drawer_audit_log_tolerates_none_metadata(
        self, monkeypatch, config, palace_path, kg
    ):
        _patch_mcp_server(monkeypatch, config, kg)
        from unittest.mock import MagicMock

        from mempalace import mcp_server

        stub_col = MagicMock()
        stub_col.get.return_value = {
            "ids": ["drawer_none_meta"],
            "documents": ["doomed body"],
            "metadatas": [None],
        }
        monkeypatch.setattr(mcp_server, "_get_collection", lambda create=False: stub_col)

        # Should reach the delete call without AttributeError on the audit-log path.
        result = mcp_server.tool_delete_drawer("drawer_none_meta")
        assert result["success"] is True
        stub_col.delete.assert_called_once_with(ids=["drawer_none_meta"])


# ── Search Tool ─────────────────────────────────────────────────────────


class TestSearchTool:
    def test_search_basic(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_search

        result = tool_search(query="JWT authentication tokens")
        assert "results" in result
        assert len(result["results"]) > 0
        # Top result should be the auth drawer
        top = result["results"][0]
        assert "JWT" in top["text"] or "authentication" in top["text"].lower()

    def test_search_with_wing_filter(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_search

        result = tool_search(query="planning", wing="notes")
        assert all(r["wing"] == "notes" for r in result["results"])

    def test_search_with_room_filter(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_search

        result = tool_search(query="database", room="backend")
        assert all(r["room"] == "backend" for r in result["results"])

    def test_search_min_similarity_backwards_compat(
        self, monkeypatch, config, palace_path, seeded_collection, kg
    ):
        """Old min_similarity param still works via backwards-compat shim."""
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_search

        # Old name should work
        result = tool_search(query="JWT", min_similarity=1.5)
        assert "results" in result

        # Old name takes precedence when both provided
        result_strict = tool_search(query="JWT", max_distance=999.0, min_similarity=0.01)
        result_loose = tool_search(query="JWT", max_distance=0.01, min_similarity=999.0)
        assert len(result_strict["results"]) <= len(result_loose["results"])

    def test_list_rooms_rejects_invalid_wing(self, monkeypatch, config, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        monkeypatch.setattr(mcp_server, "_get_collection", lambda: pytest.fail())

        result = mcp_server.tool_list_rooms(wing="../etc/passwd")
        assert "error" in result

    def test_search_rejects_invalid_room(self, monkeypatch, config, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        monkeypatch.setattr(mcp_server, "search_memories", lambda: pytest.fail())

        result = mcp_server.tool_search(query="JWT", room="../backend")
        assert "error" in result

    def test_search_retries_once_on_hnsw_flush_transient(self, monkeypatch, config, kg):
        """Issue #1315: post-bulk-mine 'Error finding id' is retried once."""
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        calls = {"n": 0}
        reset_calls = {"n": 0}

        def fake_search(*args, **kwargs):
            calls["n"] += 1
            if calls["n"] == 1:
                return {
                    "error": "Search error: Error executing plan: Internal error: Error finding id"
                }
            return {"results": [{"text": "ok", "wing": "w", "room": "r"}]}

        def fake_reset():
            reset_calls["n"] += 1

        monkeypatch.setattr(mcp_server, "search_memories", fake_search)
        monkeypatch.setattr(mcp_server, "_force_chroma_cache_reset", fake_reset)
        monkeypatch.setattr(mcp_server.time, "sleep", lambda _: None)

        result = mcp_server.tool_search(query="anything")

        assert calls["n"] == 2
        assert reset_calls["n"] == 1
        assert "results" in result
        assert result.get("index_recovered") is True

    def test_search_does_not_retry_on_non_transient_error(self, monkeypatch, config, kg):
        """Validation / unrelated errors must not trigger the retry path."""
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        calls = {"n": 0}

        def fake_search(*args, **kwargs):
            calls["n"] += 1
            return {"error": "Search error: invalid query syntax"}

        monkeypatch.setattr(mcp_server, "search_memories", fake_search)

        result = mcp_server.tool_search(query="anything")

        assert calls["n"] == 1
        assert "error" in result
        assert "index_recovered" not in result

    def test_search_returns_second_error_if_retry_also_fails(self, monkeypatch, config, kg):
        """If the transient persists past the retry, surface the second error."""
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        calls = {"n": 0}

        def fake_search(*args, **kwargs):
            calls["n"] += 1
            return {"error": "Search error: Error executing plan: Internal error: Error finding id"}

        monkeypatch.setattr(mcp_server, "search_memories", fake_search)
        monkeypatch.setattr(mcp_server, "_force_chroma_cache_reset", lambda: None)
        monkeypatch.setattr(mcp_server.time, "sleep", lambda _: None)

        result = mcp_server.tool_search(query="anything")

        assert calls["n"] == 2
        assert "error" in result
        assert "index_recovered" not in result

    def test_list_drawers_rejects_invalid_wing(self, monkeypatch, config, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        monkeypatch.setattr(mcp_server, "_get_collection", lambda: pytest.fail())

        result = mcp_server.tool_list_drawers(wing="../notes")
        assert "error" in result

    def test_find_tunnels_rejects_invalid_wing(self, monkeypatch, config, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        monkeypatch.setattr(mcp_server, "_get_collection", lambda: pytest.fail())

        result = mcp_server.tool_find_tunnels(wing_a="../project")
        assert "error" in result

    def test_wal_redacts_sensitive_fields(self, monkeypatch, config, kg, tmp_path):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        wal_file = tmp_path / "write_log.jsonl"
        monkeypatch.setattr(mcp_server, "_WAL_FILE", wal_file)

        mcp_server._wal_log(
            "test",
            {"content": "secret note", "query": "private search", "safe": "ok"},
        )

        entry = json.loads(wal_file.read_text().strip())
        assert entry["params"]["content"].startswith("[REDACTED")
        assert entry["params"]["query"].startswith("[REDACTED")
        assert entry["params"]["safe"] == "ok"


# ── Write Tools ─────────────────────────────────────────────────────────


class TestWriteTools:
    def test_add_drawer(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_add_drawer

        result = tool_add_drawer(
            wing="test_wing",
            room="test_room",
            content="This is a test memory about Python decorators and metaclasses.",
        )
        assert result["success"] is True
        assert result["wing"] == "test_wing"
        assert result["room"] == "test_room"
        assert result["drawer_id"].startswith("drawer_test_wing_test_room_")

    def test_add_drawer_duplicate_detection(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_add_drawer

        content = "This is a unique test memory about Rust ownership and borrowing."
        result1 = tool_add_drawer(wing="w", room="r", content=content)
        assert result1["success"] is True

        result2 = tool_add_drawer(wing="w", room="r", content=content)
        assert result2["success"] is True
        assert result2["reason"] == "already_exists"

    def test_add_drawer_fails_when_readback_misses(self, monkeypatch, config, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        class _FakeGetResult:
            ids = []

        class _FakeCol:
            def get(self, **kwargs):
                return _FakeGetResult()

            def upsert(self, **kwargs):
                return None

        monkeypatch.setattr(mcp_server, "_get_collection", lambda create=False: _FakeCol())

        result = mcp_server.tool_add_drawer("w", "r", "content")
        assert result["success"] is False
        assert "not readable" in result["error"]

    def test_add_drawer_shared_header_no_collision(self, monkeypatch, config, palace_path, kg):
        """Documents sharing a >100-char header must get distinct IDs (full-content hash)."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_add_drawer

        header = "# ACME Corp Knowledge Base\n**Project:** Alpha | **Team:** Backend | **Status:** Active\n\n"
        doc1 = (
            header
            + "Decision: Use PostgreSQL for primary storage. Rationale: ACID compliance required."
        )
        doc2 = header + "Decision: Use Redis for session caching. Rationale: sub-ms latency needed."

        result1 = tool_add_drawer(wing="work", room="decisions", content=doc1)
        result2 = tool_add_drawer(wing="work", room="decisions", content=doc2)

        assert result1["success"] is True
        assert result2["success"] is True
        assert result1["drawer_id"] != result2["drawer_id"], (
            "Documents with shared header but different content must have distinct drawer IDs"
        )

    def test_delete_drawer(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_delete_drawer

        result = tool_delete_drawer("drawer_proj_backend_aaa")
        assert result["success"] is True
        assert seeded_collection.count() == 3

    def test_delete_drawer_not_found(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_delete_drawer

        result = tool_delete_drawer("nonexistent_drawer")
        assert result["success"] is False

    def test_check_duplicate_handles_none_metadata(self, monkeypatch, config, kg):
        """tool_check_duplicate must tolerate None entries in the result lists
        that ChromaDB 1.5.x returns for partially-flushed rows.

        Previously ``meta = results["metadatas"][0][i]`` was unguarded and
        raised ``AttributeError: 'NoneType' object has no attribute 'get'``
        the moment the first matching drawer came back with None metadata —
        surfacing to the MCP client as the uninformative
        ``"Duplicate check failed"`` because the broad ``except Exception``
        wrapper swallows the real cause.
        """
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        mock_col = MagicMock()
        mock_col.query.return_value = {
            "ids": [["d1", "d2"]],
            "distances": [[0.05, 0.05]],
            "metadatas": [[{"wing": "w", "room": "r"}, None]],
            "documents": [["first doc", None]],
        }
        monkeypatch.setattr(mcp_server, "_get_collection", lambda: mock_col)

        result = mcp_server.tool_check_duplicate("any content", threshold=0.5)

        # Both entries land in matches (above threshold), None ones rendered
        # with sentinel values rather than crashing the whole response.
        assert result.get("is_duplicate") is True
        assert len(result["matches"]) == 2
        # The None-metadata entry falls back to sentinels.
        none_entry = result["matches"][1]
        assert none_entry["wing"] == "?"
        assert none_entry["room"] == "?"
        assert none_entry["content"] == ""

    def test_check_duplicate(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_check_duplicate

        # Exact match text from seeded_collection should be flagged
        result = tool_check_duplicate(
            "The authentication module uses JWT tokens for session management. "
            "Tokens expire after 24 hours. Refresh tokens are stored in HttpOnly cookies.",
            threshold=0.5,
        )
        assert result["is_duplicate"] is True

        # Unrelated content should not be flagged
        result = tool_check_duplicate(
            "Black holes emit Hawking radiation at the event horizon.",
            threshold=0.99,
        )
        assert result["is_duplicate"] is False

    def test_check_duplicate_short_circuits_when_vector_disabled(self, monkeypatch):
        from mempalace import mcp_server

        monkeypatch.setattr(
            mcp_server,
            "hnsw_capacity_status",
            lambda *_args, **_kwargs: {"diverged": True, "message": "capacity mismatch"},
        )

        def fail_get_collection():
            raise AssertionError("_get_collection must not run when vector search is disabled")

        monkeypatch.setattr(mcp_server, "_get_collection", fail_get_collection)
        result = mcp_server.tool_check_duplicate("content")

        assert result["is_duplicate"] is False
        assert result["vector_disabled"] is True
        assert result["vector_disabled_reason"] == "capacity mismatch"

    def test_get_drawer(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_get_drawer

        result = tool_get_drawer("drawer_proj_backend_aaa")
        assert result["drawer_id"] == "drawer_proj_backend_aaa"
        assert result["wing"] == "project"
        assert result["room"] == "backend"
        assert "JWT tokens" in result["content"]

    def test_get_drawer_not_found(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_get_drawer

        result = tool_get_drawer("nonexistent_drawer")
        assert "error" in result

    def test_get_drawer_does_not_leak_absolute_source_file_path(
        self, monkeypatch, config, palace_path, collection, kg
    ):
        """tool_get_drawer must not expose the absolute filesystem path
        that the miners write into ``source_file``. Same threat class as
        the palace_path leak in mempalace_status: in nested-agent or
        multi-server MCP topologies the client is a separate trust
        domain, and the directory layout of the host has no documented
        client-side use. Basename is enough for citation."""
        _patch_mcp_server(monkeypatch, config, kg)

        secret_dir = "/private/home/alice/secret-research/2026"
        absolute_source = f"{secret_dir}/notes.md"
        collection.add(
            ids=["drawer_leak_probe"],
            documents=["verbatim drawer body for leak probe"],
            metadatas=[
                {
                    "wing": "research",
                    "room": "notes",
                    "source_file": absolute_source,
                    "chunk_index": 0,
                    "added_by": "miner",
                    "filed_at": "2026-05-03T00:00:00",
                }
            ],
        )

        from mempalace.mcp_server import tool_get_drawer

        result = tool_get_drawer("drawer_leak_probe")
        assert result["drawer_id"] == "drawer_leak_probe"
        assert result["metadata"]["source_file"] == "notes.md"
        # Defense-in-depth: no field anywhere in the response should
        # contain the absolute path or its parent directory.
        serialized = json.dumps(result)
        assert absolute_source not in serialized
        assert secret_dir not in serialized

    def test_list_drawers(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_list_drawers

        result = tool_list_drawers()
        assert result["count"] == 4
        assert len(result["drawers"]) == 4

    def test_list_drawers_with_wing_filter(
        self, monkeypatch, config, palace_path, seeded_collection, kg
    ):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_list_drawers

        result = tool_list_drawers(wing="project")
        assert result["count"] == 3
        assert all(d["wing"] == "project" for d in result["drawers"])

    def test_list_drawers_with_room_filter(
        self, monkeypatch, config, palace_path, seeded_collection, kg
    ):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_list_drawers

        result = tool_list_drawers(wing="project", room="backend")
        assert result["count"] == 2
        assert all(d["room"] == "backend" for d in result["drawers"])

    def test_list_drawers_pagination(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_list_drawers

        result = tool_list_drawers(limit=2, offset=0)
        assert result["count"] == 2
        assert result["limit"] == 2
        assert result["offset"] == 0

    def test_list_drawers_negative_offset_clamped(
        self, monkeypatch, config, palace_path, seeded_collection, kg
    ):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_list_drawers

        result = tool_list_drawers(offset=-5)
        assert result["offset"] == 0

    def test_update_drawer_content(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_update_drawer, tool_get_drawer

        result = tool_update_drawer(
            "drawer_proj_backend_aaa", content="Updated content about auth."
        )
        assert result["success"] is True

        fetched = tool_get_drawer("drawer_proj_backend_aaa")
        assert fetched["content"] == "Updated content about auth."

    def test_update_drawer_wing_and_room(
        self, monkeypatch, config, palace_path, seeded_collection, kg
    ):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_update_drawer

        result = tool_update_drawer("drawer_proj_backend_aaa", wing="new_wing", room="new_room")
        assert result["success"] is True
        assert result["wing"] == "new_wing"
        assert result["room"] == "new_room"

    def test_update_drawer_not_found(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_update_drawer

        result = tool_update_drawer("nonexistent_drawer", content="hello")
        assert result["success"] is False

    def test_update_drawer_noop(self, monkeypatch, config, palace_path, seeded_collection, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_update_drawer

        result = tool_update_drawer("drawer_proj_backend_aaa")
        assert result["success"] is True
        assert result.get("noop") is True

    def test_tool_create_tunnel_preserves_hyphenated_wings(self, monkeypatch, tmp_path):
        """Regression for #1504: ``tool_create_tunnel`` stores the wing slug
        verbatim, and both hyphen and underscore queries find the result."""
        from mempalace import mcp_server, palace_graph

        tunnel_file = tmp_path / "tunnels.json"
        monkeypatch.setattr(palace_graph, "_get_tunnel_file", lambda *a, **kw: str(tunnel_file))
        monkeypatch.setattr(
            palace_graph,
            "_legacy_tunnel_file",
            lambda: str(tmp_path / "legacy-tunnels.json"),
        )
        monkeypatch.setattr(palace_graph, "_get_collection", lambda *a, **kw: None)

        t = mcp_server.tool_create_tunnel(
            source_wing="other-wing",
            source_room="r1",
            target_wing="my-wing",
            target_room="r2",
            label="hyphen preservation",
        )

        assert t["source"]["wing"] == "other-wing"
        assert t["target"]["wing"] == "my-wing"
        assert len(mcp_server.tool_list_tunnels(wing="my-wing")) == 1
        assert len(mcp_server.tool_list_tunnels(wing="my_wing")) == 1

    def test_tool_create_tunnel_surfaces_value_error(self, monkeypatch):
        """Regression for #1473: a ValueError from create_tunnel (e.g. a
        missing room) must be returned to the caller as a clear error,
        not escape and get wrapped as the opaque 'Internal tool error'."""
        from mempalace import mcp_server

        msg = "Target room 'does-not-exist-probe' does not exist in wing 'wing_minerva'"

        def _raise(*args, **kwargs):
            raise ValueError(msg)

        monkeypatch.setattr(mcp_server, "create_tunnel", _raise)

        result = mcp_server.tool_create_tunnel(
            source_wing="wing_minerva",
            source_room="fx-invariants",
            target_wing="wing_minerva",
            target_room="does-not-exist-probe",
        )

        assert result == {"error": msg}

    def test_add_drawer_normal_content_single_drawer(self, monkeypatch, config, palace_path, kg):
        """Regression catch: content below CHUNK_SIZE produces exactly
        one drawer with ``chunks == 1``. Pre-#1539 contract preserved."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_add_drawer

        result = tool_add_drawer(wing="w", room="r", content="Short content well under chunk_size.")
        assert result["success"] is True
        assert result["chunks"] == 1
        assert "chunk_ids" not in result
        _client2, col = _get_collection(palace_path)
        del _client2
        assert col.count() == 1
        assert col.get()["ids"] == [result["drawer_id"]]

    def test_add_drawer_oversized_content_chunked(self, monkeypatch, config, palace_path, kg):
        """Regression for #1539: content far above chunk_size must be
        sliced into bounded per-chunk drawers, each linked by a
        ``parent_drawer_id`` metadata field. No stored document may
        exceed the configured chunk_size."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_add_drawer

        oversized = "X" * 10000
        result = tool_add_drawer(wing="w", room="r", content=oversized)
        assert result["success"] is True
        assert result["chunks"] > 1
        assert "chunk_ids" in result and len(result["chunk_ids"]) == result["chunks"]

        _client2, col = _get_collection(palace_path)
        del _client2
        stored = col.get()
        max_doc = max(len(d) for d in stored["documents"])
        assert max_doc <= config.chunk_size, (
            f"no stored document may exceed chunk_size={config.chunk_size}; got max={max_doc}"
        )
        # Chroma does not guarantee insertion order on a bare ``get()``;
        # sort by ``chunk_index`` before joining so the verbatim check
        # is deterministic.
        ordered = sorted(
            zip(stored["metadatas"], stored["documents"]),
            key=lambda pair: pair[0]["chunk_index"],
        )
        assert "".join(doc for _meta, doc in ordered) == oversized
        parent_ids = {m.get("parent_drawer_id") for m in stored["metadatas"]}
        assert parent_ids == {result["drawer_id"]}, (
            f"all chunks must share one parent_drawer_id; got {parent_ids}"
        )

    def test_add_drawer_oversized_idempotency_skips_duplicate_chunk_writes(
        self, monkeypatch, config, palace_path, kg
    ):
        """Re-calling with identical oversized content must not duplicate
        any drawer. Idempotency on the chunked path probes the last
        chunk id (its presence implies the whole batch committed) and
        also the legacy logical drawer_id so a pre-#1539 single-row
        write under the same logical id does not get co-resident chunk
        siblings on the next call."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_add_drawer

        oversized = "Y" * 5000
        r1 = tool_add_drawer(wing="w", room="r", content=oversized)
        assert r1["success"] is True and r1["chunks"] > 1
        r2 = tool_add_drawer(wing="w", room="r", content=oversized)
        assert r2["success"] is True
        assert r2.get("reason") == "already_exists"

        _client2, col = _get_collection(palace_path)
        del _client2
        assert col.count() == r1["chunks"]
        # The probe must succeed against the last chunk id (atomicity
        # signal), and no row must be stored under the logical id.
        last_chunk = r1["chunk_ids"][-1]
        assert col.get(ids=[last_chunk])["ids"] == [last_chunk]
        assert col.get(ids=[r1["drawer_id"]])["ids"] == []

    def test_add_drawer_chunk_metadata_carries_parent_link(
        self, monkeypatch, config, palace_path, kg
    ):
        """Every chunk produced from oversized content must carry both
        ``chunk_index`` (0..N-1) and ``parent_drawer_id`` matching the
        logical group handle returned to the caller."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_add_drawer

        result = tool_add_drawer(wing="w", room="r", content="Q" * 3500)
        assert result["success"] is True and result["chunks"] > 1

        _client2, col = _get_collection(palace_path)
        del _client2
        stored = col.get()
        indices = sorted(m["chunk_index"] for m in stored["metadatas"])
        assert indices == list(range(len(indices)))
        for meta in stored["metadatas"]:
            assert meta.get("parent_drawer_id") == result["drawer_id"]

    def test_add_drawer_boundary_exact_chunk_size_stays_single(
        self, monkeypatch, config, palace_path, kg
    ):
        """The ``<= chunk_size`` predicate must include the boundary:
        content of exactly chunk_size chars stays a single drawer, not
        an off-by-one chunked write."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_add_drawer

        boundary = "Z" * config.chunk_size
        result = tool_add_drawer(wing="w", room="r", content=boundary)
        assert result["success"] is True
        assert result["chunks"] == 1
        assert "chunk_ids" not in result

    def test_add_drawer_chunked_logical_id_not_fetchable_directly(
        self, monkeypatch, config, palace_path, kg
    ):
        """Documented contract on the chunked path: ``tool_get_drawer``
        and ``tool_delete_drawer`` against the returned logical
        ``drawer_id`` report ``not found`` because no row is stored
        under that id. Callers must iterate ``chunk_ids`` or query by
        ``parent_drawer_id`` metadata."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_add_drawer, tool_delete_drawer, tool_get_drawer

        result = tool_add_drawer(wing="w", room="r", content="P" * 4000)
        assert result["success"] is True and result["chunks"] > 1

        # tool_get_drawer against logical id: not found.
        got_logical = tool_get_drawer(result["drawer_id"])
        assert "error" in got_logical and "not found" in got_logical["error"].lower()

        # tool_get_drawer against the first chunk id: found, full content slice.
        got_chunk = tool_get_drawer(result["chunk_ids"][0])
        assert got_chunk["content"] == "P" * config.chunk_size
        assert got_chunk["metadata"]["parent_drawer_id"] == result["drawer_id"]

        # tool_delete_drawer against logical id: also not found.
        deleted_logical = tool_delete_drawer(result["drawer_id"])
        assert deleted_logical["success"] is False
        assert "not found" in deleted_logical["error"].lower()


# ── KG Tools ────────────────────────────────────────────────────────────


class TestKGTools:
    def test_kg_add(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_kg_add

        result = tool_kg_add(
            subject="Alice",
            predicate="likes",
            object="coffee",
            valid_from="2025-01-01",
        )
        assert result["success"] is True

    def test_kg_query(self, monkeypatch, config, palace_path, seeded_kg):
        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import tool_kg_query

        result = tool_kg_query(entity="Max")
        assert result["count"] > 0

    def test_kg_invalidate(self, monkeypatch, config, palace_path, seeded_kg):
        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import tool_kg_invalidate

        result = tool_kg_invalidate(
            subject="Max",
            predicate="does",
            object="chess",
            ended="2026-03-01",
        )
        assert result["success"] is True
        # Regression #1314: response must echo the actual ended date,
        # not silently drop it and return the literal string "today".
        assert result["ended"] == "2026-03-01"

    def test_kg_add_forwards_valid_to(self, monkeypatch, config, palace_path, kg):
        """Regression #1314 case 1: valid_to must round-trip through kg_add."""
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_kg_add

        result = tool_kg_add(
            subject="_test_temporal",
            predicate="had_value",
            object="probe",
            valid_from="2026-01-01",
            valid_to="2026-04-28",
        )
        assert result["success"] is True

        facts = kg.query_entity("_test_temporal")
        assert len(facts) == 1
        assert facts[0]["valid_from"] == "2026-01-01"
        assert facts[0]["valid_to"] == "2026-04-28"
        # An already-ended fact must not be reported as still current.
        assert facts[0]["current"] is False

    def test_kg_add_forwards_source_provenance(self, monkeypatch, config, palace_path, kg):
        """Regression #1314 case 3: source_file / source_drawer_id reach storage."""
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_kg_add

        result = tool_kg_add(
            subject="operating-verb",
            predicate="candidate",
            object="husbandry",
            valid_from="2026-04-28",
            source_closet="closet-42",
            source_file="docs/decisions.md",
            source_drawer_id="drawer_abc123",
        )
        assert result["success"] is True

        triple_id = result["triple_id"]
        # Read raw row to verify all provenance columns persisted.
        with kg._lock:
            row = (
                kg._conn()
                .execute(
                    "SELECT source_closet, source_file, source_drawer_id FROM triples WHERE id = ?",
                    (triple_id,),
                )
                .fetchone()
            )
        assert row is not None
        assert row["source_closet"] == "closet-42"
        assert row["source_file"] == "docs/decisions.md"
        assert row["source_drawer_id"] == "drawer_abc123"

    def test_kg_invalidate_returns_actual_ended_date(
        self, monkeypatch, config, palace_path, seeded_kg
    ):
        """Regression #1314 case 2: response reports the resolved date, not 'today'."""
        from datetime import date as _date

        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import tool_kg_invalidate

        # Caller-supplied date round-trips into the response.
        explicit = tool_kg_invalidate(
            subject="Max",
            predicate="does",
            object="swimming",
            ended="2026-04-28",
        )
        assert explicit["ended"] == "2026-04-28"

        # Caller-omitted date resolves to today's ISO date — never the
        # literal string "today" the buggy implementation used to return.
        implicit = tool_kg_invalidate(
            subject="Max",
            predicate="loves",
            object="Chess",
        )
        assert implicit["ended"] != "today"
        assert implicit["ended"] == _date.today().isoformat()

    def test_kg_timeline(self, monkeypatch, config, palace_path, seeded_kg):
        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import tool_kg_timeline

        result = tool_kg_timeline(entity="Alice")
        assert result["count"] > 0

    def test_kg_stats(self, monkeypatch, config, palace_path, seeded_kg):
        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import tool_kg_stats

        result = tool_kg_stats()
        assert result["entities"] >= 4

    # --- Date validation at the MCP boundary (issue #1164) ---

    def test_kg_add_rejects_invalid_valid_from(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace.mcp_server import tool_kg_add

        result = tool_kg_add(
            subject="Alice",
            predicate="likes",
            object="coffee",
            valid_from="Jan 2025",
        )
        assert result["success"] is False
        assert "valid_from" in result["error"]
        assert "ISO-8601" in result["error"]

    def test_kg_query_rejects_invalid_as_of(self, monkeypatch, config, palace_path, seeded_kg):
        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import tool_kg_query

        result = tool_kg_query(entity="Max", as_of="March 2026")
        assert "error" in result
        assert "as_of" in result["error"]

    def test_kg_invalidate_rejects_invalid_ended(self, monkeypatch, config, palace_path, seeded_kg):
        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import tool_kg_invalidate

        result = tool_kg_invalidate(
            subject="Max",
            predicate="does",
            object="chess",
            ended="yesterday",
        )
        assert result["success"] is False
        assert "ended" in result["error"]

    def test_kg_query_rejects_partial_iso_dates(self, monkeypatch, config, palace_path, seeded_kg):
        _patch_mcp_server(monkeypatch, config, seeded_kg)
        from mempalace.mcp_server import tool_kg_query

        # Partial ISO dates are rejected: KG queries compare TEXT dates
        # lexicographically, so "2026-01-01" <= "2026" is False, which
        # silently excludes facts. Reject at the boundary — only YYYY-MM-DD
        # produces correct results.
        for value in ("2026", "2026-03"):
            result = tool_kg_query(entity="Max", as_of=value)
            assert "error" in result, f"accepted partial date {value!r}: {result}"

        # Full ISO-8601 dates still pass.
        result = tool_kg_query(entity="Max", as_of="2026-03-15")
        assert "error" not in result, f"rejected valid date: {result}"

    def test_kg_add_accepts_datetime_valid_from(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)

        from mempalace import mcp_server

        result = mcp_server.tool_kg_add(
            "Alice",
            "works_at",
            "Acme",
            valid_from="2026-05-06T14:23:00Z",
        )

        assert result["success"] is True

        facts = kg.query_entity("Alice", direction="outgoing")
        fact = next(r for r in facts if r["predicate"] == "works_at" and r["object"] == "Acme")

        assert fact["valid_from"] == "2026-05-06T14:23:00Z"

    def test_kg_add_accepts_datetime_valid_to(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)

        from mempalace import mcp_server

        result = mcp_server.tool_kg_add(
            "Alice",
            "worked_at",
            "OldCo",
            valid_from="2026-05-06T14:00:00Z",
            valid_to="2026-05-06T15:00:00Z",
        )

        assert result["success"] is True

        facts = kg.query_entity("Alice", direction="outgoing")
        fact = next(r for r in facts if r["predicate"] == "worked_at" and r["object"] == "OldCo")

        assert fact["valid_from"] == "2026-05-06T14:00:00Z"
        assert fact["valid_to"] == "2026-05-06T15:00:00Z"

    def test_kg_query_accepts_datetime_as_of(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)

        kg.add_triple(
            "Alice",
            "works_at",
            "Acme",
            valid_from="2026-05-06T14:00:00Z",
        )

        from mempalace import mcp_server

        result = mcp_server.tool_kg_query(
            "Alice",
            as_of="2026-05-06T14:23:00Z",
            direction="outgoing",
        )

        assert "error" not in result
        assert result["as_of"] == "2026-05-06T14:23:00Z"
        assert result["count"] == 1
        assert result["facts"][0]["object"] == "Acme"

    def test_kg_invalidate_accepts_datetime_ended(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)

        kg.add_triple(
            "Alice",
            "works_at",
            "Acme",
            valid_from="2026-05-06T14:00:00Z",
        )

        from mempalace import mcp_server

        result = mcp_server.tool_kg_invalidate(
            "Alice",
            "works_at",
            "Acme",
            ended="2026-05-06T14:23:00Z",
        )

        assert result["success"] is True
        assert result["ended"] == "2026-05-06T14:23:00Z"

        facts = kg.query_entity("Alice", direction="outgoing")
        fact = next(r for r in facts if r["predicate"] == "works_at" and r["object"] == "Acme")

        assert fact["valid_to"] == "2026-05-06T14:23:00Z"

    def test_kg_add_rejects_non_canonical_datetimes(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)

        from mempalace import mcp_server

        invalid_values = [
            "2026-05-06T14:23:00+02:00",
            "2026-05-06T14:23:00-05:30",
            "2026-05-06T14:23:00.123Z",
            "2026-05-06 14:23:00",
            "2026-05-06T14:23:00",
        ]

        for value in invalid_values:
            result = mcp_server.tool_kg_add(
                "Alice",
                "works_at",
                "Acme",
                valid_from=value,
            )

            assert result["success"] is False, value
            assert "valid_from" in result["error"]
            assert "YYYY-MM-DDTHH:MM:SSZ" in result["error"]

    def test_kg_query_rejects_non_canonical_datetime_as_of(
        self, monkeypatch, config, palace_path, kg
    ):
        _patch_mcp_server(monkeypatch, config, kg)

        from mempalace import mcp_server

        invalid_values = [
            "2026-05-06T14:23:00+02:00",
            "2026-05-06T14:23:00-05:30",
            "2026-05-06T14:23:00.123Z",
            "2026-05-06 14:23:00",
            "2026-05-06T14:23:00",
        ]

        for value in invalid_values:
            result = mcp_server.tool_kg_query(
                "Alice",
                as_of=value,
                direction="outgoing",
            )

            assert "error" in result, value
            assert "as_of" in result["error"]
            assert "YYYY-MM-DDTHH:MM:SSZ" in result["error"]

    def test_kg_invalidate_rejects_non_canonical_ended(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)

        kg.add_triple(
            "Alice",
            "works_at",
            "Acme",
            valid_from="2026-05-06T14:00:00Z",
        )

        from mempalace import mcp_server

        invalid_values = [
            "2026-05-06T14:23:00+02:00",
            "2026-05-06T14:23:00-05:30",
            "2026-05-06T14:23:00.123Z",
            "2026-05-06 14:23:00",
            "2026-05-06T14:23:00",
        ]

        for value in invalid_values:
            result = mcp_server.tool_kg_invalidate(
                "Alice",
                "works_at",
                "Acme",
                ended=value,
            )

            assert result["success"] is False, value
            assert "ended" in result["error"]
            assert "YYYY-MM-DDTHH:MM:SSZ" in result["error"]

    def test_kg_add_rejects_timezone_offset_datetime(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)

        from mempalace import mcp_server

        result = mcp_server.tool_kg_add(
            "Alice",
            "works_at",
            "Acme",
            valid_from="2026-05-06T14:23:00+02:00",
        )

        assert result["success"] is False
        assert "valid_from" in result["error"]
        assert "YYYY-MM-DDTHH:MM:SSZ" in result["error"]


# ── Diary Tools ─────────────────────────────────────────────────────────


class TestDiaryTools:
    def test_diary_write_and_read(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_diary_write, tool_diary_read

        w = tool_diary_write(
            agent_name="TestAgent",
            entry="Today we discussed authentication patterns.",
            topic="architecture",
        )
        assert w["success"] is True
        # agent_name is normalized to lowercase on write (#1243).
        assert w["agent"] == "testagent"

        r = tool_diary_read(agent_name="TestAgent")
        assert r["total"] == 1
        assert r["entries"][0]["topic"] == "architecture"
        assert "authentication" in r["entries"][0]["content"]

    def test_diary_read_empty(self, monkeypatch, config, palace_path, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_diary_read

        r = tool_diary_read(agent_name="Nobody")
        assert r["entries"] == []

    def test_diary_write_same_second_shared_prefix_no_collision(
        self, monkeypatch, config, palace_path, kg
    ):
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client

        from mempalace import mcp_server

        class FrozenDateTime:
            calls = [
                datetime(2026, 4, 13, 22, 15, 30, 123456),
                datetime(2026, 4, 13, 22, 15, 30, 123457),
            ]
            fallback = datetime(2026, 4, 13, 22, 15, 30, 123457)

            @classmethod
            def now(cls):
                if cls.calls:
                    return cls.calls.pop(0)
                return cls.fallback

        monkeypatch.setattr(mcp_server, "datetime", FrozenDateTime)

        from mempalace.mcp_server import tool_diary_read, tool_diary_write

        entry1 = "A" * 50 + " entry one"
        entry2 = "A" * 50 + " entry two"

        result1 = tool_diary_write(agent_name="TestAgent", entry=entry1, topic="status")
        result2 = tool_diary_write(agent_name="TestAgent", entry=entry2, topic="status")

        assert result1["success"] is True
        assert result2["success"] is True
        assert result1["entry_id"] != result2["entry_id"]

        read_result = tool_diary_read(agent_name="TestAgent")
        contents = [entry["content"] for entry in read_result["entries"]]
        assert read_result["total"] == 2
        assert entry1 in contents
        assert entry2 in contents

    def test_diary_read_empty_wing_spans_all_wings(self, monkeypatch, config, palace_path, kg):
        """diary_read(wing='') must return entries from every wing this agent
        wrote to. Hooks write to project-derived wings (#659); a reader that
        silos by default wing would never see those entries."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_diary_read, tool_diary_write

        w1 = tool_diary_write(
            agent_name="TestAgent",
            entry="default-wing entry",
            topic="general",
        )
        w2 = tool_diary_write(
            agent_name="TestAgent",
            entry="project-wing entry",
            topic="general",
            wing="wing_someproject",
        )
        assert w1["success"] and w2["success"]

        # Empty wing → return both entries
        r = tool_diary_read(agent_name="TestAgent", wing="")
        assert r["total"] == 2
        contents = {e["content"] for e in r["entries"]}
        assert "default-wing entry" in contents
        assert "project-wing entry" in contents

        # Explicit wing → return only that wing's entries
        r_scoped = tool_diary_read(agent_name="TestAgent", wing="wing_someproject")
        assert r_scoped["total"] == 1
        assert r_scoped["entries"][0]["content"] == "project-wing entry"

    def test_diary_read_case_insensitive_agent(self, monkeypatch, config, palace_path, kg):
        """Regression for #1243: diary_read must be case-insensitive over
        agent_name. Writing as "Claude" and reading as "claude" (or vice
        versa) must surface the same entries — sanitize_name preserved
        case, which silently dropped reads when the agent name's casing
        differed from the write."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_diary_read, tool_diary_write

        # Write as "Claude" → read as "claude" should match.
        w1 = tool_diary_write(
            agent_name="Claude",
            entry="entry written as Claude",
            topic="general",
        )
        assert w1["success"]

        r1 = tool_diary_read(agent_name="claude")
        assert "entries" in r1, r1
        contents1 = {e["content"] for e in r1["entries"]}
        assert "entry written as Claude" in contents1

        # Write as "CLAUDE" → read as "Claude" should also match the
        # same agent. After normalization both writes target the same
        # lowercase agent identity, so both entries are returned.
        w2 = tool_diary_write(
            agent_name="CLAUDE",
            entry="entry written as CLAUDE",
            topic="general",
        )
        assert w2["success"]

        r2 = tool_diary_read(agent_name="Claude")
        contents2 = {e["content"] for e in r2["entries"]}
        assert "entry written as Claude" in contents2
        assert "entry written as CLAUDE" in contents2

        # The stored agent metadata is the lowercase form, and the
        # default wing is derived from that lowercase form too.
        assert w1["agent"] == "claude"
        assert w2["agent"] == "claude"

    # ── #1539: oversized-entry chunking ────────────────────────────

    def test_diary_write_normal_entry_single_drawer(self, monkeypatch, config, palace_path, kg):
        """Regression catch: a normal entry (< CHUNK_SIZE) must produce
        exactly one drawer with ``chunks == 1`` in the result. Existing
        pre-#1539 behaviour preserved for the common path."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_diary_write

        r = tool_diary_write(
            agent_name="TestAgent",
            entry="A normal-length entry that fits comfortably under chunk_size.",
            topic="general",
        )
        assert r["success"] is True
        assert r["chunks"] == 1
        _client2, col = _get_collection(palace_path)
        del _client2
        assert col.count() == 1

    def test_diary_write_oversized_entry_chunked(self, monkeypatch, config, palace_path, kg):
        """Regression for #1539: an entry far above CHUNK_SIZE must be
        sliced into bounded per-chunk drawers, each linked by a
        ``parent_entry_id`` metadata field. No single document stored
        may exceed CHUNK_SIZE."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_diary_write

        # 5000 chars: well above CHUNK_SIZE=800. Expected chunks: ceil(5000/800) = 7.
        oversized = "Z" * 5000
        r = tool_diary_write(agent_name="TestAgent", entry=oversized, topic="general")

        assert r["success"] is True
        assert r["chunks"] > 1, f"oversized entry must produce >1 chunks; got {r['chunks']}"
        assert "chunk_ids" in r and len(r["chunk_ids"]) == r["chunks"]

        _client2, col = _get_collection(palace_path)
        del _client2
        stored = col.get()
        assert all(len(d) <= 800 for d in stored["documents"]), (
            f"no stored document may exceed CHUNK_SIZE=800; "
            f"got max={max(len(d) for d in stored['documents'])}"
        )
        joined = "".join(stored["documents"])
        assert joined == oversized, "joined chunks must equal original entry verbatim"

        parent_ids = {m.get("parent_entry_id") for m in stored["metadatas"]}
        assert len(parent_ids) == 1 and None not in parent_ids, (
            f"all chunks must share one parent_entry_id; got {parent_ids}"
        )

    def test_diary_write_chunk_index_metadata(self, monkeypatch, config, palace_path, kg):
        """Regression for #1539: each oversized-entry chunk must carry a
        ``chunk_index`` metadata field that runs 0, 1, 2, ... in order."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace.mcp_server import tool_diary_write

        oversized = "Q" * 3500  # ~5 chunks at CHUNK_SIZE=800
        r = tool_diary_write(agent_name="TestAgent", entry=oversized, topic="general")
        assert r["success"] is True and r["chunks"] > 1

        _client2, col = _get_collection(palace_path)
        del _client2
        stored = col.get()
        indices = sorted(m["chunk_index"] for m in stored["metadatas"])
        assert indices == list(range(len(indices))), (
            f"chunk_index must be 0..N-1 contiguous; got {indices}"
        )


# ── Cache Invalidation (inode/mtime) ──────────────────────────────────


class TestCacheInvalidation:
    """Tests for _get_collection inode/mtime cache invalidation logic."""

    def test_mtime_change_invalidates_cache(self, monkeypatch, config, palace_path, kg):
        """When mtime changes, the cached collection should be replaced."""
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        # Create a real collection so _get_collection succeeds
        _client, _col = _get_collection(palace_path, create=True)
        del _client

        # Prime the cache
        col1 = mcp_server._get_collection()
        assert col1 is not None

        # Simulate an external write changing the mtime
        old_mtime = mcp_server._palace_db_mtime
        monkeypatch.setattr(mcp_server, "_palace_db_mtime", old_mtime - 10.0)

        # _get_collection should detect the mtime drift and reconnect
        col2 = mcp_server._get_collection()
        assert col2 is not None

    def test_inode_change_invalidates_cache(self, monkeypatch, config, palace_path, kg):
        """When inode changes (file replaced), the cached collection should be replaced."""
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        _client, _col = _get_collection(palace_path, create=True)
        del _client

        # Prime the cache
        col1 = mcp_server._get_collection()
        assert col1 is not None

        # Simulate a rebuild that changes the inode
        monkeypatch.setattr(mcp_server, "_palace_db_inode", 99999)

        col2 = mcp_server._get_collection()
        assert col2 is not None

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="Windows holds chroma.sqlite3 open while the client is cached, blocking os.remove",
    )
    def test_missing_db_invalidates_cache(self, monkeypatch, config, palace_path, kg):
        """When chroma.sqlite3 disappears, a cached collection should be invalidated."""
        _patch_mcp_server(monkeypatch, config, kg)
        import os
        from mempalace import mcp_server

        _client, _col = _get_collection(palace_path, create=True)
        del _client

        # Prime the cache
        col1 = mcp_server._get_collection()
        assert col1 is not None
        assert mcp_server._collection_cache is not None

        # Delete the DB file to simulate a rebuild in progress
        db_file = os.path.join(palace_path, "chroma.sqlite3")
        if os.path.isfile(db_file):
            os.remove(db_file)

        # Cache should be invalidated; _get_collection returns None
        # because the backend can't open a missing DB without create=True
        mcp_server._get_collection()
        # The key assertion: the old cached collection was dropped
        assert mcp_server._palace_db_inode == 0
        assert mcp_server._palace_db_mtime == 0.0

    def test_reconnect_reports_failure_when_no_palace(self, monkeypatch, config, kg):
        """tool_reconnect should report failure when no collection is available."""
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        # Make _get_collection always return None
        monkeypatch.setattr(mcp_server, "_get_collection", lambda create=False: None)

        result = mcp_server.tool_reconnect()
        assert result["success"] is False
        assert "No palace found" in result["message"]
        assert result["drawers"] == 0

    def test_reconnect_reports_success(self, monkeypatch, config, palace_path, kg):
        """tool_reconnect should report success with drawer count."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace import mcp_server

        result = mcp_server.tool_reconnect()
        assert result["success"] is True
        assert "Reconnected" in result["message"]
        assert isinstance(result["drawers"], int)

    def test_reconnect_closes_shared_backend(self, monkeypatch, config, kg):
        _patch_mcp_server(monkeypatch, config, kg)
        from unittest.mock import MagicMock

        from mempalace import mcp_server, palace

        close_palace = MagicMock()
        monkeypatch.setattr(palace._DEFAULT_BACKEND, "close_palace", close_palace)

        class _FakeCol:
            def count(self):
                return 7

        monkeypatch.setattr(mcp_server, "_get_collection", lambda create=False: _FakeCol())

        result = mcp_server.tool_reconnect()
        assert result["success"] is True
        close_palace.assert_called_once_with(config.palace_path)

    def test_get_collection_create_true_avoids_get_or_create_on_reopen(
        self, monkeypatch, config, palace_path, kg
    ):
        """Regression for the MCP-server half of #1262.

        ChromaDB 1.5.x's Rust bindings SIGSEGV when
        ``client.get_or_create_collection`` is called with metadata that
        differs from the collection's stored metadata. The Stop hook
        path (``tool_diary_write`` -> ``_get_collection(create=True)``)
        was reaching that codepath on every session-end; #1262 fixed
        the equivalent crash class in ``ChromaBackend`` but left this
        site untouched. ``_get_collection(create=True)`` must call
        ``client.get_collection`` first and only fall back to
        ``client.create_collection`` when the collection does not yet
        exist on disk.
        """
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        col1 = mcp_server._get_collection(create=True)
        assert col1 is not None

        client = mcp_server._client_cache
        assert client is not None

        # Patch at the class level — chromadb's mtime-change detection
        # may rebuild the client between calls, so an instance-level
        # spy would not survive.
        client_cls = type(client)
        calls: list[tuple] = []

        def _spy(self, *args, **kwargs):
            calls.append((args, kwargs))
            raise AssertionError(
                "get_or_create_collection must not be called on reopen "
                "(SIGSEGV path on metadata mismatch)"
            )

        monkeypatch.setattr(client_cls, "get_or_create_collection", _spy)
        mcp_server._collection_cache = None

        col2 = mcp_server._get_collection(create=True)
        assert col2 is not None
        assert calls == [], f"get_or_create_collection was called: {calls}"

    def test_get_collection_passes_embedding_function(self, monkeypatch, config, palace_path, kg):
        """Regression for #1299.

        ``mcp_server._get_collection`` must pass ``embedding_function=`` into
        both ``client.get_collection`` and ``client.create_collection``,
        mirroring ``ChromaBackend.get_collection``. Without it, ChromaDB 1.x
        falls back to its built-in ``DefaultEmbeddingFunction`` (whose lazy
        ONNX provider selection has SIGSEGV'd on python 3.14 + Apple Silicon),
        and writers/readers can disagree with the miner about which EF is
        bound to the collection. The miner / Stop hook ingest path routes
        through ``ChromaBackend.get_collection`` which does this correctly;
        the MCP server must match.
        """
        _patch_mcp_server(monkeypatch, config, kg)
        from mempalace import mcp_server

        client = mcp_server._get_client()
        client_cls = type(client)
        captured: dict[str, list[dict]] = {"get": [], "create": []}
        real_get = client_cls.get_collection
        real_create = client_cls.create_collection

        def _spy_get(self, name, **kwargs):
            captured["get"].append(dict(kwargs))
            return real_get(self, name, **kwargs)

        def _spy_create(self, name, **kwargs):
            captured["create"].append(dict(kwargs))
            return real_create(self, name, **kwargs)

        monkeypatch.setattr(client_cls, "get_collection", _spy_get)
        monkeypatch.setattr(client_cls, "create_collection", _spy_create)
        mcp_server._collection_cache = None

        col = mcp_server._get_collection(create=True)
        assert col is not None

        all_calls = captured["get"] + captured["create"]
        assert all_calls, "expected get_collection or create_collection to be called"
        for kwargs in all_calls:
            assert "embedding_function" in kwargs, (
                f"missing embedding_function= in chromadb call: {kwargs}"
            )
            assert kwargs["embedding_function"] is not None

        # Same expectation on the create=False (cache-miss) reopen path.
        mcp_server._collection_cache = None
        captured["get"].clear()
        captured["create"].clear()
        col2 = mcp_server._get_collection()
        assert col2 is not None
        assert captured["get"], "expected get_collection on cache-miss reopen"
        for kwargs in captured["get"]:
            assert "embedding_function" in kwargs
            assert kwargs["embedding_function"] is not None

    def test_get_collection_retries_once_on_exception(self, monkeypatch, config, palace_path, kg):
        """Regression: a transient failure inside _get_collection must trigger
        one retry after clearing the client/collection caches, not silently
        return None.

        Before this fix, a stale chromadb handle (e.g. the rust bindings
        invalidating after an out-of-band write) would raise inside the
        single ``try`` block, get swallowed by ``except Exception: return
        None``, and every subsequent tool call would hit the same poisoned
        cache returning None. The retry forces ``_get_client()`` to rebuild
        the client (which re-runs ``quarantine_stale_hnsw`` per #1322), so
        the second attempt heals the common stale-handle case.
        """
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace import mcp_server

        # Force a cold cache so the first call goes through the open path.
        mcp_server._client_cache = None
        mcp_server._collection_cache = None

        real_get_client = mcp_server._get_client
        attempts = {"count": 0}

        def flaky_get_client():
            attempts["count"] += 1
            if attempts["count"] == 1:
                raise RuntimeError("simulated transient chromadb failure")
            return real_get_client()

        monkeypatch.setattr(mcp_server, "_get_client", flaky_get_client)

        col = mcp_server._get_collection()

        # Both attempts ran and the second succeeded.
        assert attempts["count"] == 2
        assert col is not None

    def test_get_collection_returns_none_after_two_failures(
        self, monkeypatch, config, palace_path, kg
    ):
        """If both attempts fail, return None (matches the prior contract for
        permanent failures — only the transient case is now self-healing)."""
        _patch_mcp_server(monkeypatch, config, kg)
        _client, _col = _get_collection(palace_path, create=True)
        del _client
        from mempalace import mcp_server

        mcp_server._client_cache = None
        mcp_server._collection_cache = None

        attempts = {"count": 0}

        def always_fails():
            attempts["count"] += 1
            raise RuntimeError("permanent chromadb failure")

        monkeypatch.setattr(mcp_server, "_get_client", always_fails)

        col = mcp_server._get_collection()

        assert attempts["count"] == 2
        assert col is None


class TestKGLazyCache:
    """Lazy per-path KnowledgeGraph cache (issue #1136)."""

    def test_lazy_init_no_import_side_effect(self, tmp_path):
        """Importing mcp_server must not create knowledge_graph.sqlite3.

        Runs in a fresh subprocess with HOME pointed at tmp_path so the
        assertion targets a clean filesystem, independent of conftest's
        session-level HOME patch.
        """
        import subprocess
        import sys

        kg_file = tmp_path / ".mempalace" / "knowledge_graph.sqlite3"
        env = {k: v for k, v in os.environ.items() if not k.startswith("MEMPAL")}
        env["HOME"] = str(tmp_path)
        env["USERPROFILE"] = str(tmp_path)
        result = subprocess.run(
            [sys.executable, "-c", "import mempalace.mcp_server"],
            env=env,
            capture_output=True,
            text=True,
            timeout=30,
        )
        assert result.returncode == 0, f"import failed: {result.stderr}"
        assert not kg_file.exists(), f"import created sqlite file at {kg_file} as a side effect"

    def test_get_kg_returns_same_instance(self, tmp_path, monkeypatch):
        """Two calls with the same resolved path return the same KG."""
        from mempalace import mcp_server

        monkeypatch.setattr(mcp_server, "_kg_by_path", {})
        monkeypatch.setattr(mcp_server, "_palace_flag_given", True)
        monkeypatch.setenv("MEMPALACE_PALACE_PATH", str(tmp_path))

        kg1 = mcp_server._get_kg()
        kg2 = mcp_server._get_kg()
        assert kg1 is kg2
        assert len(mcp_server._kg_by_path) == 1

    def test_get_kg_different_paths_different_instances(self, tmp_path, monkeypatch):
        """Different palace paths map to different KG instances."""
        from mempalace import mcp_server

        tmp_a = tmp_path / "a"
        tmp_b = tmp_path / "b"
        tmp_a.mkdir()
        tmp_b.mkdir()

        monkeypatch.setattr(mcp_server, "_kg_by_path", {})
        monkeypatch.setattr(mcp_server, "_palace_flag_given", True)

        monkeypatch.setenv("MEMPALACE_PALACE_PATH", str(tmp_a))
        kg_a = mcp_server._get_kg()
        monkeypatch.setenv("MEMPALACE_PALACE_PATH", str(tmp_b))
        kg_b = mcp_server._get_kg()

        assert kg_a is not kg_b
        assert len(mcp_server._kg_by_path) == 2

    def test_multi_tenant_env_switch(self, tmp_path, monkeypatch):
        """The issue #1136 acceptance scenario.

        Rotating MEMPALACE_PALACE_PATH between MCP tool calls must route
        each call to the correct tenant's KG sqlite file.
        """
        from mempalace import mcp_server

        tmp_a = tmp_path / "tenant_a"
        tmp_b = tmp_path / "tenant_b"
        tmp_a.mkdir()
        tmp_b.mkdir()

        monkeypatch.setattr(mcp_server, "_kg_by_path", {})
        monkeypatch.setattr(mcp_server, "_palace_flag_given", True)

        monkeypatch.setenv("MEMPALACE_PALACE_PATH", str(tmp_a))
        add_result = mcp_server.tool_kg_add(
            subject="alice_secret",
            predicate="owns",
            object="repo_a",
        )
        assert add_result.get("success") is True, add_result

        monkeypatch.setenv("MEMPALACE_PALACE_PATH", str(tmp_b))
        query_b = mcp_server.tool_kg_query(entity="alice_secret")
        assert query_b.get("count", 0) == 0, f"tenant B leaked tenant A's fact: {query_b}"

        monkeypatch.setenv("MEMPALACE_PALACE_PATH", str(tmp_a))
        query_a = mcp_server.tool_kg_query(entity="alice_secret")
        assert query_a.get("count", 0) >= 1, f"tenant A lost its own fact: {query_a}"


# ── Structured error codes + MineAlreadyRunning (#1552) ─────────────────


class TestStructuredErrors:
    """Verify that _internal_tool_error and MineAlreadyRunning return
    machine-readable structured data (#1552)."""

    def test_internal_tool_error_without_exc_has_no_data_field(self):
        """Backward-compat: callers that omit exc still get a valid error dict."""
        from mempalace.mcp_server import _internal_tool_error

        try:
            raise ValueError("test error")
        except ValueError:
            resp = _internal_tool_error("req-1", "mempalace_search")

        assert resp["jsonrpc"] == "2.0"
        assert resp["id"] == "req-1"
        err = resp["error"]
        assert err["code"] == -32000
        assert err["message"] == "Internal tool error"
        assert "data" not in err

    def test_internal_tool_error_with_exc_includes_structured_data(self):
        """When exc is supplied, the error body must include data.error_class
        and data.message so callers can distinguish error types (#1552)."""
        from mempalace.mcp_server import _internal_tool_error

        exc = RuntimeError("chromadb cold init wedge")
        try:
            raise exc
        except RuntimeError:
            resp = _internal_tool_error("req-2", "mempalace_add_drawer", exc)

        err = resp["error"]
        assert err["code"] == -32000
        assert "data" in err
        assert err["data"]["error_class"] == "RuntimeError"
        assert "chromadb cold init wedge" in err["data"]["message"]

    def test_internal_tool_error_exception_dispatch_passes_exc(self, monkeypatch):
        """handle_request's Exception branch must pass exc to _internal_tool_error."""
        from mempalace import mcp_server

        captured = {}

        def fake_handler(**kwargs):
            raise OSError("fake disk error")

        fake_tool_entry = {
            "handler": fake_handler,
            "input_schema": {"type": "object", "properties": {}},
        }
        monkeypatch.setattr(
            mcp_server,
            "TOOLS",
            {"mempalace_fake": fake_tool_entry},
        )

        original = mcp_server._internal_tool_error

        def spy_error(req_id, tool_name, exc=None):
            captured["exc"] = exc
            return original(req_id, tool_name, exc)

        monkeypatch.setattr(mcp_server, "_internal_tool_error", spy_error)

        req = {
            "jsonrpc": "2.0",
            "id": "r1",
            "method": "tools/call",
            "params": {"name": "mempalace_fake", "arguments": {}},
        }
        resp = mcp_server.handle_request(req)
        assert resp["error"]["code"] == -32000
        assert isinstance(captured.get("exc"), OSError)
        assert "data" in resp["error"]
        assert resp["error"]["data"]["error_class"] == "OSError"

    def test_tool_sync_mine_already_running_returns_error_class(self, monkeypatch, tmp_path):
        """tool_sync MineAlreadyRunning path returns error_class: LockHeldByOtherProcess."""
        from mempalace import mcp_server
        from mempalace.palace import MineAlreadyRunning

        cfg = MagicMock()
        cfg.palace_path = str(tmp_path / "palace")
        monkeypatch.setattr(mcp_server, "_config", cfg)
        monkeypatch.setattr(mcp_server, "_get_kg", lambda *a, **kw: MagicMock())

        def _raise_locked(*args, **kwargs):
            raise MineAlreadyRunning("pid=12345")

        import mempalace.sync as sync_mod

        monkeypatch.setattr(sync_mod, "sync_palace", _raise_locked, raising=False)

        result = mcp_server.tool_sync()
        assert result["success"] is False
        assert "another mine is in progress" in result["error"]
        assert result.get("error_class") == "LockHeldByOtherProcess"

    def test_mcp_idle_timeout_invalid_env_disables_watchdog(self, monkeypatch):
        """Invalid MEMPALACE_MCP_IDLE_HOURS disables idle auto-exit."""
        from mempalace import mcp_server

        monkeypatch.setenv("MEMPALACE_MCP_IDLE_HOURS", "not-a-float")
        assert mcp_server._mcp_idle_timeout_secs() == 0.0

    def test_cache_thread_safe(self, tmp_path, monkeypatch):
        """Concurrent _get_kg() for the same path yields one instance."""
        import concurrent.futures
        from mempalace import mcp_server

        monkeypatch.setattr(mcp_server, "_kg_by_path", {})
        monkeypatch.setattr(mcp_server, "_palace_flag_given", True)
        monkeypatch.setenv("MEMPALACE_PALACE_PATH", str(tmp_path))

        with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool:
            results = list(pool.map(lambda _: mcp_server._get_kg(), range(16)))

        ids = {id(kg) for kg in results}
        assert len(ids) == 1, f"expected 1 unique instance, got {len(ids)}"
        assert len(mcp_server._kg_by_path) == 1

    def test_tool_reconnect_drains_kg_cache(self, monkeypatch):
        """``tool_reconnect`` must close cached KG instances and clear the dict.

        Without this, an external replacement of ``knowledge_graph.sqlite3``
        leaves the server pinned to a stale ``sqlite3.Connection``.
        """
        from mempalace import mcp_server

        class _FakeKG:
            def __init__(self):
                self.closed = False

            def close(self):
                self.closed = True

        fake_a = _FakeKG()
        fake_b = _FakeKG()
        monkeypatch.setattr(mcp_server, "_kg_by_path", {"/a": fake_a, "/b": fake_b})
        # Bypass real ChromaDB so the test isolates KG-cache behaviour.
        monkeypatch.setattr(mcp_server, "_get_collection", lambda: None)

        mcp_server.tool_reconnect()

        assert fake_a.closed is True
        assert fake_b.closed is True
        assert mcp_server._kg_by_path == {}

    def test_tool_reconnect_swallows_kg_close_errors(self, monkeypatch):
        """A failing ``close()`` on one cached KG must not block cache clearing."""
        from mempalace import mcp_server

        class _BoomKG:
            def close(self):
                raise RuntimeError("boom")

        monkeypatch.setattr(mcp_server, "_kg_by_path", {"/a": _BoomKG()})
        monkeypatch.setattr(mcp_server, "_get_collection", lambda: None)

        mcp_server.tool_reconnect()

        assert mcp_server._kg_by_path == {}

    def test_call_kg_retries_after_concurrent_close(self, monkeypatch):
        """A KG closed mid-handler must trigger a one-shot retry with a fresh
        instance — not surface a -32000 to the MCP client."""
        import sqlite3 as _sqlite3

        from mempalace import mcp_server

        path = "/fake/palace/knowledge_graph.sqlite3"
        monkeypatch.setattr(mcp_server, "_resolve_kg_path", lambda: path)

        class _ClosedKG:
            def query_entity(self, entity, **kwargs):
                raise _sqlite3.ProgrammingError("Cannot operate on a closed database")

        class _FreshKG:
            def query_entity(self, entity, **kwargs):
                return [{"entity": entity}]

        cache = {mcp_server._canonicalize_kg_path(path): _ClosedKG()}
        monkeypatch.setattr(mcp_server, "_kg_by_path", cache)

        # Second _get_kg() call (after the cache eviction) constructs a new
        # KG. Patch the constructor so we don't open a real sqlite file.
        monkeypatch.setattr(mcp_server, "KnowledgeGraph", lambda **_: _FreshKG())

        result = mcp_server._call_kg(lambda kg: kg.query_entity("Alice"))
        assert result == [{"entity": "Alice"}]
        # The closed instance must be evicted; the fresh one must be cached.
        assert isinstance(cache[mcp_server._canonicalize_kg_path(path)], _FreshKG)

    def test_call_kg_does_not_retry_on_other_errors(self, monkeypatch):
        """Non-ProgrammingError exceptions must propagate without retry —
        we don't want the retry guard masking real bugs."""
        from mempalace import mcp_server

        path = "/fake/palace/knowledge_graph.sqlite3"
        monkeypatch.setattr(mcp_server, "_resolve_kg_path", lambda: path)

        calls = {"count": 0}

        class _FailingKG:
            def query_entity(self, entity, **kwargs):
                calls["count"] += 1
                raise ValueError("bad input")

        monkeypatch.setattr(
            mcp_server, "_kg_by_path", {mcp_server._canonicalize_kg_path(path): _FailingKG()}
        )
        monkeypatch.setattr(mcp_server, "KnowledgeGraph", lambda **_: _FailingKG())

        with pytest.raises(ValueError, match="bad input"):
            mcp_server._call_kg(lambda kg: kg.query_entity("Alice"))
        assert calls["count"] == 1, "non-ProgrammingError must not trigger retry"

    def test_call_kg_gives_up_after_one_retry(self, monkeypatch):
        """If the second attempt also hits a closed DB, give up rather than
        loop forever — a sustained close-stream is a different bug."""
        import sqlite3 as _sqlite3

        from mempalace import mcp_server

        path = "/fake/palace/knowledge_graph.sqlite3"
        monkeypatch.setattr(mcp_server, "_resolve_kg_path", lambda: path)

        calls = {"count": 0}

        class _AlwaysClosedKG:
            def query_entity(self, entity, **kwargs):
                calls["count"] += 1
                raise _sqlite3.ProgrammingError("closed again")

        cache = {}
        monkeypatch.setattr(mcp_server, "_kg_by_path", cache)
        monkeypatch.setattr(mcp_server, "KnowledgeGraph", lambda **_: _AlwaysClosedKG())

        with pytest.raises(_sqlite3.ProgrammingError):
            mcp_server._call_kg(lambda kg: kg.query_entity("Alice"))
        assert calls["count"] == 2, "expected exactly one retry beyond the initial attempt"

    def test_call_kg_passes_captured_path_through_resolve_drift(self, monkeypatch):
        """``_call_kg`` must thread its captured canonical path through
        ``_get_kg`` so insertion and eviction agree on the cache key even
        when FS or env state would otherwise drift between attempts. The
        end-to-end invariant: after the retry, the closed handle that was
        cached under the captured path is gone (evicted) and the cache no
        longer holds it under the stale key.
        """
        import sqlite3 as _sqlite3
        from mempalace import mcp_server

        class _ClosedKG:
            def query_entity(self, entity, **kwargs):
                raise _sqlite3.ProgrammingError("Cannot operate on a closed database")

        class _FreshKG:
            def query_entity(self, entity, **kwargs):
                return [{"entity": entity}]

        # _resolve_kg_path returns shifting values (env rotation between
        # attempts). _canonicalize_kg_path is identity so paths flow
        # through verbatim.
        resolved_seq = iter(["/path/v1", "/path/v2", "/path/v3"])
        monkeypatch.setattr(mcp_server, "_resolve_kg_path", lambda: next(resolved_seq))
        monkeypatch.setattr(mcp_server, "_canonicalize_kg_path", lambda p: p)

        closed = _ClosedKG()
        cache = {"/path/v1": closed}
        monkeypatch.setattr(mcp_server, "_kg_by_path", cache)

        get_kg_args: list = []

        def spy_get_kg(canonical_path=None):
            get_kg_args.append(canonical_path)
            return cache.get(canonical_path) if canonical_path in cache else _FreshKG()

        monkeypatch.setattr(mcp_server, "_get_kg", spy_get_kg)

        result = mcp_server._call_kg(lambda kg: kg.query_entity("Alice"))

        assert result == [{"entity": "Alice"}]
        # Both _get_kg calls received the captured path "/path/v1" rather
        # than the drifted "/path/v2". Without pass-through, the second
        # call would have used "/path/v2" and the closed handle at
        # "/path/v1" would never have been evicted.
        assert get_kg_args == ["/path/v1", "/path/v1"], (
            f"expected both _get_kg calls to receive captured '/path/v1', "
            f"got {get_kg_args} -- captured-path pass-through broken"
        )
        # Eviction landed under the captured key: the closed handle is
        # gone from the cache. With drift the closed handle would still
        # be at "/path/v1" because eviction would have probed "/path/v2".
        assert "/path/v1" not in cache, (
            f"closed handle leaked under captured key after retry; "
            f"cache state: {[(k, type(v).__name__) for k, v in cache.items()]}"
        )

    def test_call_kg_oserror_at_top_propagates_unmasked(self, monkeypatch):
        """``OSError`` from ``_canonicalize_kg_path`` at the top of
        ``_call_kg`` (e.g. transient Windows realpath hiccup on a stale
        junction) must propagate unchanged. The fix-rationale invariant:
        capturing the canonical path before the retry loop means an FS
        error surfaces cleanly to the dispatcher's exception envelope
        instead of getting raised inside the ``except`` branch where it
        would mask a ``sqlite3.ProgrammingError``.
        """
        from mempalace import mcp_server

        monkeypatch.setattr(mcp_server, "_resolve_kg_path", lambda: "/fake/path")
        monkeypatch.setattr(
            mcp_server,
            "_canonicalize_kg_path",
            lambda p: (_ for _ in ()).throw(OSError("simulated realpath failure")),
        )

        op_calls = {"n": 0}

        def op(kg):
            op_calls["n"] += 1
            return None

        with pytest.raises(OSError, match="simulated realpath failure"):
            mcp_server._call_kg(op)
        assert op_calls["n"] == 0, "op must not run if canonicalize fails at top"

    def test_canonicalize_kg_path_collapses_symlink_alias(self, tmp_path):
        """A symlink layer over the palace directory must collapse to one
        cache key — otherwise two tenants pointing at /srv/A and
        /srv/link-to-A open duplicate sqlite3.Connections over the same
        file."""
        if sys.platform == "win32":
            pytest.skip("symlink creation requires admin privileges on Windows runners")

        from mempalace import mcp_server

        target = tmp_path / "real"
        target.mkdir()
        link = tmp_path / "link"
        link.symlink_to(target)

        real_db = str(target / "knowledge_graph.sqlite3")
        link_db = str(link / "knowledge_graph.sqlite3")

        assert mcp_server._canonicalize_kg_path(real_db) == mcp_server._canonicalize_kg_path(
            link_db
        )

    def test_canonicalize_kg_path_routes_through_normcase(self, monkeypatch):
        """``_canonicalize_kg_path`` must apply ``os.path.normcase`` so the
        cache key collapses Windows drive-letter casing
        (``C:\\palace`` vs ``c:\\palace``). On POSIX runners normcase is a
        no-op, so we patch both ``realpath`` and ``normcase`` with sentinel
        wrappers and assert the helper composes them as
        ``normcase(realpath(p))`` -- swapping the order would leave Windows
        symlinks under the original case, defeating the dedup.
        """
        from mempalace import mcp_server

        def fake_realpath(p: str) -> str:
            return f"<RP:{p}>"

        def fake_normcase(p: str) -> str:
            return f"<NC:{p}>"

        monkeypatch.setattr(os.path, "realpath", fake_realpath)
        monkeypatch.setattr(os.path, "normcase", fake_normcase)

        result = mcp_server._canonicalize_kg_path("/some/Path/KG.sqlite3")

        assert result == "<NC:<RP:/some/Path/KG.sqlite3>>", (
            f"expected normcase(realpath(p)) composition, got {result!r}"
        )

    def test_get_kg_dedupes_symlink_alias_end_to_end(self, tmp_path, monkeypatch):
        """End-to-end: two ``_get_kg()`` calls via different symlink layers
        return the same cached instance and construct only one
        ``KnowledgeGraph``."""
        if sys.platform == "win32":
            pytest.skip("symlink creation requires admin privileges on Windows runners")

        from mempalace import mcp_server

        target = tmp_path / "real"
        target.mkdir()
        link = tmp_path / "link"
        link.symlink_to(target)

        real_db = str(target / "knowledge_graph.sqlite3")
        link_db = str(link / "knowledge_graph.sqlite3")

        constructed: list = []

        class _StubKG:
            def __init__(self, db_path=None):
                constructed.append(db_path)

        monkeypatch.setattr(mcp_server, "_kg_by_path", {})
        monkeypatch.setattr(mcp_server, "KnowledgeGraph", _StubKG)

        paths = iter([real_db, link_db])
        monkeypatch.setattr(mcp_server, "_resolve_kg_path", lambda: next(paths))

        kg1 = mcp_server._get_kg()
        kg2 = mcp_server._get_kg()

        assert kg1 is kg2, "symlink alias must hit the cached KG, not construct a duplicate"
        assert len(constructed) == 1, f"expected 1 KG construction, got {len(constructed)}"
        assert len(mcp_server._kg_by_path) == 1


# ── Param-shape diagnostics on tools/call dispatch (#1351) ──────────────


class TestParamShapeDiagnostics:
    """Dispatch-level TypeError on tools/call should surface as JSON-RPC
    -32602 (Invalid params) with the offending parameter named, instead of
    the opaque -32000 Internal tool error. Handler-internal TypeError and
    non-TypeError exceptions stay generic -32000 (no internals leak).
    """

    def test_missing_required_returns_32602_with_param_name(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request(
            {
                "method": "tools/call",
                "id": 1,
                "params": {
                    "name": "mempalace_diary_write",
                    "arguments": {"agent_name": "test"},
                },
            }
        )
        assert resp["error"]["code"] == -32602
        assert "'entry'" in resp["error"]["message"]
        assert "mempalace_diary_write" in resp["error"]["message"]

    def test_handler_internal_typeerror_stays_generic_32000(self, monkeypatch):
        from mempalace import mcp_server

        def boom(**_kw):
            raise TypeError("unsupported operand type(s) for +: 'int' and 'str'")

        monkeypatch.setitem(mcp_server.TOOLS["mempalace_status"], "handler", boom)

        resp = mcp_server.handle_request(
            {
                "method": "tools/call",
                "id": 2,
                "params": {"name": "mempalace_status", "arguments": {}},
            }
        )
        assert resp["error"]["code"] == -32000
        assert resp["error"]["message"] == "Internal tool error"
        assert "unsupported operand" not in resp["error"]["message"]

    def test_chromadb_exception_stays_generic_32000(self, monkeypatch):
        from mempalace import mcp_server

        def boom(**_kw):
            raise RuntimeError("db schema mismatch at /private/path/chroma.sqlite3")

        monkeypatch.setitem(mcp_server.TOOLS["mempalace_status"], "handler", boom)

        resp = mcp_server.handle_request(
            {
                "method": "tools/call",
                "id": 3,
                "params": {"name": "mempalace_status", "arguments": {}},
            }
        )
        assert resp["error"]["code"] == -32000
        assert resp["error"]["message"] == "Internal tool error"
        assert "db schema" not in resp["error"]["message"]
        assert "/private/path" not in resp["error"]["message"]

    def test_two_missing_required_lists_both_names(self):
        """For 2+ missing args Python emits 'a' and 'b'; the response should
        list both quoted names, not return a syntactically broken string.
        """
        from mempalace.mcp_server import handle_request

        resp = handle_request(
            {
                "method": "tools/call",
                "id": 4,
                "params": {"name": "mempalace_diary_write", "arguments": {}},
            }
        )
        assert resp["error"]["code"] == -32602
        message = resp["error"]["message"]
        assert "parameters" in message
        assert "'agent_name'" in message
        assert "'entry'" in message
        assert " and " not in message.split("for tool")[0]

    def test_handler_internal_signature_shape_stays_generic(self, monkeypatch):
        """A TypeError whose function name does not match the dispatched
        handler — e.g. raised by a helper called inside the handler body —
        must fall through to generic -32000, otherwise we'd leak internal
        helper/parameter names as if they were public tool parameters.
        """
        from mempalace import mcp_server

        def calling_handler(**_kw):
            def helper(req):
                return req

            helper()

        monkeypatch.setitem(mcp_server.TOOLS["mempalace_status"], "handler", calling_handler)

        resp = mcp_server.handle_request(
            {
                "method": "tools/call",
                "id": 5,
                "params": {"name": "mempalace_status", "arguments": {}},
            }
        )
        assert resp["error"]["code"] == -32000
        assert resp["error"]["message"] == "Internal tool error"
        assert "'req'" not in resp["error"]["message"]
        assert "helper" not in resp["error"]["message"]

    def test_unexpected_kw_typeerror_inside_handler_stays_generic(self, monkeypatch):
        """The 'got an unexpected keyword argument' shape is unreachable from
        real dispatch (schema-filter on line 2236 drops unknown kwargs for
        normal handlers; **kwargs handlers per #684 accept anything). If a
        handler raises that shape manually, the qualname mismatch must keep
        it on the generic -32000 path so internal helper names cannot leak.
        """
        from mempalace import mcp_server

        def boom(**_kw):
            raise TypeError("some_helper() got an unexpected keyword argument 'foo'")

        monkeypatch.setitem(mcp_server.TOOLS["mempalace_status"], "handler", boom)

        resp = mcp_server.handle_request(
            {
                "method": "tools/call",
                "id": 6,
                "params": {"name": "mempalace_status", "arguments": {}},
            }
        )
        assert resp["error"]["code"] == -32000
        assert resp["error"]["message"] == "Internal tool error"
        assert "'foo'" not in resp["error"]["message"]
        assert "some_helper" not in resp["error"]["message"]


class TestUnknownParamName:
    """A kwarg not in the tool schema (wrong parameter *name*, e.g. text=
    instead of content=) should surface as JSON-RPC -32602 naming the
    offending kwarg, instead of being silently dropped and resurfacing
    indirectly as a later "Missing required 'X'". Symmetric with the
    missing-required path in TestParamShapeDiagnostics. The internal
    wait_for_previous transport kwarg must never be flagged, and
    **kwargs pass-through handlers must keep accepting unknown kwargs.
    """

    def test_unknown_param_returns_32602_naming_the_wrong_kwarg(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request(
            {
                "method": "tools/call",
                "id": 7,
                "params": {
                    "name": "mempalace_add_drawer",
                    "arguments": {"wing": "w", "room": "r", "text": "hello"},
                },
            }
        )
        assert resp["error"]["code"] == -32602
        message = resp["error"]["message"]
        assert "'text'" in message
        assert "Unknown parameter" in message
        assert "mempalace_add_drawer" in message
        # Names the actual wrong kwarg, not the indirect missing-required symptom.
        assert "Missing required" not in message

    def test_two_unknown_params_list_both_names(self):
        from mempalace.mcp_server import handle_request

        resp = handle_request(
            {
                "method": "tools/call",
                "id": 8,
                "params": {
                    "name": "mempalace_add_drawer",
                    "arguments": {"wing": "w", "room": "r", "text": "a", "bogus": "b"},
                },
            }
        )
        assert resp["error"]["code"] == -32602
        message = resp["error"]["message"]
        assert "parameters" in message
        assert "'text'" in message
        assert "'bogus'" in message

    def test_wait_for_previous_not_flagged_as_unknown(self, monkeypatch):
        """wait_for_previous is an internal transport kwarg in no tool schema;
        it is popped before dispatch and must not trip the unknown-param check
        for a normal (non-**kwargs) handler.
        """
        from mempalace import mcp_server

        def stub(agent_name, entry, topic="general"):
            return {"ok": True, "agent": agent_name}

        monkeypatch.setitem(mcp_server.TOOLS["mempalace_diary_write"], "handler", stub)

        resp = mcp_server.handle_request(
            {
                "method": "tools/call",
                "id": 9,
                "params": {
                    "name": "mempalace_diary_write",
                    "arguments": {
                        "agent_name": "x",
                        "entry": "y",
                        "wait_for_previous": True,
                    },
                },
            }
        )
        assert "error" not in resp
        assert "result" in resp

    def test_kwargs_passthrough_handler_keeps_accepting_unknown(self, monkeypatch):
        """Handlers that explicitly accept **kwargs (per #684) bypass the
        schema filter entirely, so an unknown kwarg must still pass through
        rather than being rejected as -32602.
        """
        from mempalace import mcp_server

        def passthrough(**kwargs):
            return {"ok": True, "got": sorted(kwargs)}

        monkeypatch.setitem(mcp_server.TOOLS["mempalace_status"], "handler", passthrough)

        resp = mcp_server.handle_request(
            {
                "method": "tools/call",
                "id": 10,
                "params": {"name": "mempalace_status", "arguments": {"bogus": 1}},
            }
        )
        assert "error" not in resp
        assert "result" in resp