"""Tests for ``mm uninstall`` — local state cleanup CLI.

Coverage spans the install-context inventory, flag combinations, server
liveness refusal, partial-deletion error path, and the ``RuntimeProfile``
private-import pin so any rename/move in ``cli.init_cmd`` breaks here
loud and immediate (MEDIUM 6 mitigation).
"""

from __future__ import annotations

import contextlib
import errno
import json
import os
import sqlite3
import sys
from collections.abc import Iterator
from pathlib import Path

import pytest
from click.testing import CliRunner

from memtomem.cli import cli
from .helpers import set_home


@contextlib.contextmanager
def _hold_pid_lock(pid_file: Path) -> Iterator[None]:
    """Hold an exclusive lock on ``pid_file`` for the duration of the block.

    Mirrors what ``server/__init__.py:main`` does at runtime so the
    portalocker-based liveness probe (#387, #817) sees a live writer.
    Cross-platform via ``portalocker``; on Windows ``"rb+"`` open is
    required by the ``MsvcrtLocker`` backend (see ``cli/_liveness.py:54``).
    """
    import portalocker

    fp = open(pid_file, "rb+")
    try:
        portalocker.lock(fp, portalocker.LOCK_EX | portalocker.LOCK_NB)
        try:
            yield
        finally:
            portalocker.unlock(fp)
    finally:
        fp.close()


# ---------------------------------------------------------------- fixtures


@pytest.fixture
def home(tmp_path, monkeypatch):
    """Tmp HOME with both env override and _bootstrap._CONFIG_PATH patched.

    Mirrors the isolation pattern from ``test_cli_index_noop_e2e.py``: the
    module-level ``_bootstrap._CONFIG_PATH = Path.home() / ...`` is bound
    at import time, so ``monkeypatch.setenv("HOME")`` alone leaves it
    pointing at the developer's real home. Patching it directly is
    required for hermetic tests.

    Also isolates ``$XDG_RUNTIME_DIR`` so the new runtime pid file
    location (#412) lives under ``tmp_path`` rather than the developer's
    real ``/run/user/{uid}/memtomem/`` or a shared ``/tmp`` subdir.

    On Windows, ``_runtime_paths.runtime_dir()`` skips the
    ``XDG_RUNTIME_DIR`` branch entirely (``XDG_RUNTIME_DIR`` is a Linux/
    systemd convention; the POSIX-mode-bit gate guarding it is
    meaningless on NTFS) and falls through to
    ``tempfile.gettempdir() / memtomem-0``. ``gettempdir()`` resolves to
    the user-shared ``%LOCALAPPDATA%\\Temp\\``, so leftover artifacts
    from prior pytest runs or concurrent tools blocked the prune
    assertion in ``TestRuntimePidCleanedWithOther`` (#759 failure 3).
    Pin ``tempfile.tempdir`` to a per-test path so the runtime dir is
    always test-scoped, regardless of platform.
    """
    import tempfile

    from memtomem.cli import _bootstrap
    from memtomem.cli import uninstall_cmd

    h = tmp_path / "home"
    h.mkdir()
    # The external-MCP probe also checks ``Path.cwd() / ".mcp.json"`` (the
    # project-local Claude config). Pin cwd into the tmp home so the repo's
    # own ``.mcp.json`` doesn't leak into the probe and trip the negative
    # "not reported" assertions — CI runs pytest from the repo root, which
    # ships a real ``.mcp.json``.
    monkeypatch.chdir(h)
    xdg = tmp_path / "xdg_runtime"
    xdg.mkdir()
    os.chmod(xdg, 0o700)  # _runtime_paths validator requires owner-only
    fake_tempdir = tmp_path / "tempdir"
    fake_tempdir.mkdir()
    monkeypatch.setattr(tempfile, "tempdir", str(fake_tempdir))
    set_home(monkeypatch, h)
    monkeypatch.setenv("XDG_RUNTIME_DIR", str(xdg))
    monkeypatch.setattr(_bootstrap, "_CONFIG_PATH", h / ".memtomem" / "config.json")
    monkeypatch.setattr(uninstall_cmd, "_DEFAULT_STATE_DIR", h / ".memtomem")
    return h


def _seed_state(home: Path, *, with_db: bool = True, with_fragments: bool = True) -> Path:
    """Populate ~/.memtomem/ with realistic state for inventory tests."""
    state = home / ".memtomem"
    state.mkdir(parents=True, exist_ok=True)

    (state / "config.json").write_text('{"embedding": {"provider": "none"}}', encoding="utf-8")
    (state / "config.json.bak-2026-04-22T00-00-00").write_text("{}", encoding="utf-8")
    if with_fragments:
        (state / "config.d").mkdir()
        (state / "config.d" / "claude.json").write_text("{}", encoding="utf-8")
    if with_db:
        (state / "memtomem.db").write_bytes(b"sqlite-fake")
        (state / "memtomem.db-wal").write_bytes(b"wal")
        (state / "memtomem.db-shm").write_bytes(b"shm")
    (state / "memories").mkdir()
    (state / "memories" / "x.md").write_text("# hello", encoding="utf-8")
    (state / ".current_session").write_text("sess-id", encoding="utf-8")
    return state


# -------------------------------------------------------------------- 1


class TestEmptyState:
    def test_no_state_directory_exits_cleanly(self, home):
        result = CliRunner().invoke(cli, ["uninstall"])
        assert result.exit_code == 0, result.output
        assert "No memtomem state to remove" in result.output
        assert "Binary install detected" in result.output


# -------------------------------------------------------------------- 2


class TestDefaultDeletion:
    def test_default_removes_everything(self, home):
        state = _seed_state(home)
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "Removed:" in result.output
        assert not state.exists(), f"state dir should be pruned, found: {list(state.iterdir())}"


# -------------------------------------------------------------------- 3


class TestKeepConfig:
    def test_keep_config_preserves_config_surface(self, home):
        state = _seed_state(home)
        result = CliRunner().invoke(cli, ["uninstall", "-y", "--keep-config"])
        assert result.exit_code == 0, result.output
        assert (state / "config.json").exists()
        assert (state / "config.d" / "claude.json").exists()
        assert (state / "config.json.bak-2026-04-22T00-00-00").exists()
        # data side wiped
        assert not (state / "memtomem.db").exists()
        assert not (state / "memories" / "x.md").exists()


# -------------------------------------------------------------------- 4


class TestKeepData:
    def test_keep_data_preserves_db_and_memories(self, home):
        state = _seed_state(home)
        result = CliRunner().invoke(cli, ["uninstall", "-y", "--keep-data"])
        assert result.exit_code == 0, result.output
        assert (state / "memtomem.db").exists()
        assert (state / "memtomem.db-wal").exists()
        assert (state / "memories" / "x.md").exists()
        # config side wiped
        assert not (state / "config.json").exists()
        assert not (state / "config.d").exists()


# -------------------------------------------------------------------- 5


class TestCustomStoragePath:
    def test_custom_storage_path_in_inventory_and_deleted(self, home, tmp_path, monkeypatch):
        """``storage.sqlite_path`` outside ~/.memtomem/ should still be cleaned."""
        custom_dir = tmp_path / "elsewhere"
        custom_dir.mkdir()
        custom_db = custom_dir / "foo.db"
        custom_db.write_bytes(b"sqlite-fake")
        (custom_dir / "foo.db-wal").write_bytes(b"wal")
        (custom_dir / "unrelated.txt").write_text("user file", encoding="utf-8")

        # Seed config that points to the custom path
        state = home / ".memtomem"
        state.mkdir()
        (state / "config.json").write_text(
            json.dumps({"storage": {"sqlite_path": str(custom_db)}}), encoding="utf-8"
        )

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "custom storage path" in result.output
        # custom DB siblings deleted
        assert not custom_db.exists()
        assert not (custom_dir / "foo.db-wal").exists()
        # unrelated sibling left alone
        assert (custom_dir / "unrelated.txt").exists(), "non-DB siblings must NOT be deleted"


# -------------------------------------------------------------------- 6


class TestUserMemoryDirsUntouched:
    def test_user_managed_memory_dirs_never_deleted(self, home, tmp_path):
        user_notes = tmp_path / "Documents" / "notes"
        user_notes.mkdir(parents=True)
        (user_notes / "important.md").write_text("# user data", encoding="utf-8")

        state = home / ".memtomem"
        state.mkdir()
        (state / "config.json").write_text(
            json.dumps({"indexing": {"memory_dirs": [str(user_notes)]}}), encoding="utf-8"
        )

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert (user_notes / "important.md").exists(), (
            "user-managed memory_dirs MUST stay — only ~/.memtomem/memories/ is in scope"
        )


# -------------------------------------------------------------------- 7


class TestExternalsDetectedNotModified:
    def test_external_mcp_files_detected_but_unmodified(self, home):
        claude_json = home / ".claude.json"
        original_text = json.dumps({"mcpServers": {"memtomem": {"command": "mm-server"}}})
        claude_json.write_text(original_text, encoding="utf-8")

        # state dir so we don't hit the empty-state fast path
        state = home / ".memtomem"
        state.mkdir()
        (state / "config.json").write_text("{}", encoding="utf-8")

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" in result.output
        assert ".claude.json" in result.output
        # external file untouched
        assert claude_json.exists()
        assert claude_json.read_text(encoding="utf-8") == original_text


class TestProbeExternalParsedMCP:
    """#975: parsed MCP config check avoids substring false positives."""

    def _seed_state_dir(self, home: Path) -> Path:
        state = home / ".memtomem"
        state.mkdir()
        (state / "config.json").write_text("{}", encoding="utf-8")
        return state

    def test_json_mcp_servers_memtomem_reported(self, home):
        """Valid JSON with ``mcpServers.memtomem`` is still reported."""
        self._seed_state_dir(home)
        path = home / ".cursor" / "mcp.json"
        path.parent.mkdir(parents=True)
        path.write_text(
            json.dumps({"mcpServers": {"memtomem": {"command": "mm-server"}}}),
            encoding="utf-8",
        )
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" in result.output
        assert "mcp.json" in result.output

    def test_kimi_share_dir_mcp_servers_memtomem_reported(self, home, monkeypatch):
        """Kimi MCP config under ``$KIMI_SHARE_DIR`` is reported when present."""
        self._seed_state_dir(home)
        share_dir = home / "kimi-share"
        monkeypatch.setenv("KIMI_SHARE_DIR", str(share_dir))
        path = share_dir / "mcp.json"
        path.parent.mkdir(parents=True)
        path.write_text(
            json.dumps({"mcpServers": {"memtomem": {"command": "mm-server"}}}),
            encoding="utf-8",
        )
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" in result.output
        assert "kimi-share" in result.output
        assert "mcp.json" in result.output

    def test_json_unrelated_text_not_reported(self, home):
        """JSON that only mentions 'memtomem' in a comment/description is NOT reported."""
        self._seed_state_dir(home)
        path = home / ".cursor" / "mcp.json"
        path.parent.mkdir(parents=True)
        path.write_text(
            json.dumps(
                {
                    "mcpServers": {
                        "other-server": {"command": "some-tool"},
                    },
                    "description": "This config works with memtomem for context",
                }
            ),
            encoding="utf-8",
        )
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" not in result.output

    def test_json_no_mcp_servers_key_not_reported(self, home):
        """JSON without an ``mcpServers`` key is NOT reported even if
        'memtomem' appears elsewhere in the config."""
        self._seed_state_dir(home)
        path = home / ".gemini" / "settings.json"
        path.parent.mkdir(parents=True)
        path.write_text(
            json.dumps({"someOtherKey": {"memtomem": "mentioned-but-not-mcp"}}),
            encoding="utf-8",
        )
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" not in result.output

    def test_toml_mcp_servers_memtomem_reported(self, home):
        """TOML (Codex config) with ``mcp_servers.memtomem`` is reported."""
        import tomllib as _tl

        self._seed_state_dir(home)
        path = home / ".codex" / "config.toml"
        path.parent.mkdir(parents=True)
        toml_text = 'command = "mm-server"\n[mcp_servers.memtomem]\n'
        path.write_text(toml_text, encoding="utf-8")
        # Sanity: tomllib round-trips
        parsed = _tl.loads(toml_text)
        assert "mcp_servers" in parsed
        assert "memtomem" in parsed["mcp_servers"]

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" in result.output
        assert "config.toml" in result.output

    def test_toml_unrelated_text_not_reported(self, home):
        """TOML with 'memtomem' only in a comment or non-MCP section is NOT reported."""
        self._seed_state_dir(home)
        path = home / ".codex" / "config.toml"
        path.parent.mkdir(parents=True)
        # "memtomem" appears only in a comment and in an unrelated value
        path.write_text(
            "# This config used to reference memtomem\n"
            "[unrelated]\n"
            'note = "not about memtomem here"\n',
            encoding="utf-8",
        )
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        # Must NOT be reported — "memtomem" is only in comment / unrelated value,
        # not under a parsed mcp_servers key.
        assert "External integrations" not in result.output

    def test_toml_no_mcp_servers_section_not_reported(self, home):
        """TOML with ``memtomem = true`` at top level (not under mcp_servers)
        is NOT reported — the parsed check only looks under ``mcp_servers``."""
        self._seed_state_dir(home)
        path = home / ".codex" / "config.toml"
        path.parent.mkdir(parents=True)
        path.write_text("memtomem = true\n", encoding="utf-8")
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" not in result.output

    def test_invalid_json_ignored_no_crash(self, home):
        """Malformed JSON must NOT crash uninstall — uninstall is a recovery path."""
        self._seed_state_dir(home)
        path = home / ".claude.json"
        path.write_text("{this is not valid json //}", encoding="utf-8")

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        # The substring "memtomem" isn't in the bad json, so no external is reported
        assert "External integrations" not in result.output

    def test_invalid_toml_ignored_no_crash(self, home):
        """Malformed TOML must NOT crash uninstall — parsed check fails, no fallback."""
        self._seed_state_dir(home)
        path = home / ".codex" / "config.toml"
        path.parent.mkdir(parents=True)
        path.write_text("[mcp_servers\n  memtomem = bad\n", encoding="utf-8")

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        # Invalid TOML → parse fails → silently skipped. No crash, no false report.
        assert "External integrations" not in result.output

    def test_windsurf_mcp_config_json_parsed(self, home):
        """Windsurf ``mcp_config.json`` is a JSON file and uses the parsed path."""
        self._seed_state_dir(home)
        path = home / ".codeium" / "windsurf" / "mcp_config.json"
        path.parent.mkdir(parents=True)
        path.write_text(
            json.dumps({"mcpServers": {"memtomem": {"command": "mm-server"}}}),
            encoding="utf-8",
        )
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" in result.output
        assert "mcp_config.json" in result.output

    def test_gemini_settings_json_parsed(self, home):
        """Gemini ``settings.json`` uses the parsed JSON path.
        No ``mcpServers`` key → NOT reported, even though 'memtomem' appears in text."""
        self._seed_state_dir(home)
        path = home / ".gemini" / "settings.json"
        path.parent.mkdir(parents=True)
        path.write_text(
            json.dumps({"gemini": {"version": "1.0", "description": "uses memtomem"}}),
            encoding="utf-8",
        )
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" not in result.output

    def test_nonexistent_files_skipped_silently(self, home):
        """Files that don't exist are simply skipped — no error, no output."""
        self._seed_state_dir(home)
        # No external files created → _probe must return empty list.
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "External integrations" not in result.output


# -------------------------------------------------------------------- 8


class TestBinaryHintPerOrigin:
    @pytest.mark.parametrize(
        "origin,expected_substring",
        [
            ("uv-tool", "uv tool uninstall memtomem"),
            ("uvx", "ephemeral"),
            ("venv-relative", "uv pip uninstall memtomem"),
            ("system", "pip uninstall memtomem"),
            ("unknown", "which mm"),
        ],
    )
    def test_hint_text_per_origin(self, home, monkeypatch, origin, expected_substring):
        from memtomem.cli import init_cmd
        from memtomem.cli import uninstall_cmd

        fake_profile = init_cmd.RuntimeProfile(
            cwd_install_type="pypi",
            cwd_install_dir=None,
            runtime_interpreter=Path("python3"),
            workspace_venv_path=Path("foo/.venv") if origin == "venv-relative" else None,
            mm_binary_origin=origin,
            runtime_matches_workspace=(origin == "venv-relative"),
        )
        monkeypatch.setattr(uninstall_cmd, "_runtime_profile", lambda: fake_profile)

        result = CliRunner().invoke(cli, ["uninstall"])
        assert result.exit_code == 0, result.output
        assert expected_substring in result.output


# -------------------------------------------------------------------- 9


class TestNonTtyAbort:
    def test_non_tty_without_yes_aborts(self, home):
        _seed_state(home)
        # CliRunner's default input is a non-TTY StringIO → isatty() is False.
        result = CliRunner().invoke(cli, ["uninstall"], input="")
        assert result.exit_code != 0
        assert "non-interactive shell" in result.output
        # state should be untouched
        assert (home / ".memtomem" / "memtomem.db").exists()


# -------------------------------------------------------------------- 10


class TestInteractiveCancellation:
    def test_interactive_no_cancels_without_changes(self, home, monkeypatch):
        """Interactive 'n' must produce a distinct cancellation message
        from the non-TTY abort path so users + tests can tell them apart.

        ``CliRunner`` substitutes ``sys.stdin`` with a ``StringIO`` whose
        ``isatty()`` returns False, so we patch the ``_isatty`` indirection
        in ``uninstall_cmd`` directly to flip the TTY check to True.
        """
        from memtomem.cli import uninstall_cmd

        _seed_state(home)
        monkeypatch.setattr(uninstall_cmd, "_isatty", lambda: True)

        result = CliRunner().invoke(cli, ["uninstall"], input="n\n")
        assert result.exit_code == 1
        assert "Cancelled" in result.output  # distinct from non-TTY's "non-interactive shell"
        assert "non-interactive shell" not in result.output
        # untouched
        assert (home / ".memtomem" / "memtomem.db").exists()


# -------------------------------------------------------------------- 11


class TestRuntimeProfileImportPin:
    """If init_cmd renames or moves _runtime_profile / RuntimeProfile this
    test breaks immediately. The follow-up is either to update
    uninstall_cmd.py to track the move, or extract the runtime profile to a
    shared module — see plan MEDIUM 6."""

    def test_runtime_profile_symbols_importable(self):
        import dataclasses

        from memtomem.cli.init_cmd import RuntimeProfile, _runtime_profile

        assert callable(_runtime_profile)
        # Frozen dataclass — fields live on the dataclass spec, not the class
        # __dict__, so use dataclasses.fields() rather than hasattr.
        field_names = {f.name for f in dataclasses.fields(RuntimeProfile)}
        assert "mm_binary_origin" in field_names
        assert "cwd_install_type" in field_names
        # Actually buildable (no args, returns the dataclass).
        prof = _runtime_profile()
        assert isinstance(prof, RuntimeProfile)


# -------------------------------------------------------------------- 12


class TestServerAliveRefuses:
    def test_refuses_when_server_alive_at_legacy_path(self, home):
        """Pre-#412 servers still write ``~/.memtomem/.server.pid``. The
        mixed-version upgrade path (old server running, new uninstall)
        must still refuse — the flock probe checks both locations.

        Cross-platform via portalocker (#817/#819). On Windows
        ``LockFileEx`` blocks ``read`` from other handles too, so
        ``probe_pid_file`` cannot read the pid number out of the locked
        file and the message takes the unknown-pid branch with a
        Sysinternals/Resource Monitor hint instead of the POSIX
        ``lsof`` + recorded-pid path. Both branches must surface
        "Server still running" + ``exit_code == 2`` + state preserved.
        """
        state = _seed_state(home)
        pid_file = state / ".server.pid"
        # The PID inside the file is just for the user-facing message now;
        # the flock probe (#387) is what decides alive/dead.
        pid_file.write_text(str(os.getpid()), encoding="utf-8")

        with _hold_pid_lock(pid_file):
            result = CliRunner().invoke(cli, ["uninstall", "-y"])

        assert result.exit_code == 2
        assert "Server still running" in result.output
        if sys.platform == "win32":
            # Windows: pid is None (read blocked by LockFileEx), so the
            # message points at handle.exe / Resource Monitor, not lsof.
            assert "handle.exe" in result.output or "Resource Monitor" in result.output, (
                f"Windows refusal must point at a holder-finder; got: {result.output!r}"
            )
        else:
            assert str(os.getpid()) in result.output
        # nothing deleted
        assert (state / "memtomem.db").exists()
        assert (state / "config.json").exists()

    def test_refuses_when_server_alive_at_runtime_path(self, home):
        """Post-#412 servers hold the flock at
        ``$XDG_RUNTIME_DIR/memtomem/server.pid``. The probe must see it
        even though the pid file lives outside ``~/.memtomem/``.

        Same Windows caveat as ``test_refuses_when_server_alive_at_legacy_path``:
        the recorded pid is unreachable behind ``LockFileEx``, so the
        message routes through the unknown-pid branch with a
        Sysinternals hint.
        """
        from memtomem._runtime_paths import ensure_runtime_dir

        _seed_state(home)
        pid_file = ensure_runtime_dir() / "server.pid"
        pid_file.write_text(str(os.getpid()), encoding="utf-8")

        with _hold_pid_lock(pid_file):
            result = CliRunner().invoke(cli, ["uninstall", "-y"])

        assert result.exit_code == 2
        assert "Server still running" in result.output
        if sys.platform == "win32":
            assert "handle.exe" in result.output or "Resource Monitor" in result.output, (
                f"Windows refusal must point at a holder-finder; got: {result.output!r}"
            )
        else:
            assert str(os.getpid()) in result.output
        assert (home / ".memtomem" / "memtomem.db").exists()

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason=(
            "POSIX-only contract: --force unlinks the locked pid file via "
            "unlink-while-open. The Windows mirror of this contract lives in "
            "test_force_refuses_on_windows_when_pid_locked (#730 + #819)."
        ),
    )
    def test_force_overrides_liveness(self, home):
        state = _seed_state(home)
        pid_file = state / ".server.pid"
        pid_file.write_text(str(os.getpid()), encoding="utf-8")

        with _hold_pid_lock(pid_file):
            result = CliRunner().invoke(cli, ["uninstall", "-y", "--force"])

        assert result.exit_code == 0, result.output
        assert not state.exists()

    @pytest.mark.skipif(
        sys.platform != "win32",
        reason=(
            "Windows-only contract (#730 + #819): --force cannot wipe a "
            "locked pid file because NTFS refuses unlink-while-open "
            "(WinError 32). The refusal must be clean — exit 2 + actionable "
            "hint + state preserved — never a half-wiped state dir."
        ),
    )
    def test_force_refuses_on_windows_when_pid_locked(self, home):
        """Windows mirror of ``test_force_overrides_liveness``.

        POSIX ``--force`` succeeds via ``unlink-while-open``; Windows
        cannot, so the same invocation must refuse cleanly rather than
        partially wipe state. This pins the #730 destructive-CLI
        contract on the pid-lock side, complementing
        ``test_force_refuses_on_windows_when_writer_alive`` which pins
        the same contract on the DB-lock side.
        """
        state = _seed_state(home)
        pid_file = state / ".server.pid"
        pid_file.write_text(str(os.getpid()), encoding="utf-8")

        with _hold_pid_lock(pid_file):
            result = CliRunner().invoke(cli, ["uninstall", "-y", "--force"])

        assert result.exit_code == 2, result.output
        # State preserved — refusal fires before any deletion runs.
        assert state.exists()
        assert (state / "memtomem.db").exists()
        assert (state / "config.json").exists()
        assert pid_file.exists()
        # Actionable Windows-native hint must surface; the refusal text
        # came from #730 and uses Sysinternals/Task Manager wording.
        assert (
            "handle.exe" in result.output
            or "WinError 32" in result.output
            or "cannot wipe" in result.output
        ), f"Windows --force refusal must point at a holder-finder; got: {result.output!r}"

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="POSIX-specific 'lsof' wording (#819 follow-up); Windows uses handle.exe / Resource Monitor",
    )
    def test_refuses_with_unknown_pid_branch_when_pid_file_empty(self, home):
        """Empty pid file + flock held = the truncate-race fingerprint.

        When a pre-fix concurrent server start truncated the live
        server's pid file, the recorded pid is gone but the flock is
        still held. The user-facing message must distinguish this from
        the normal case so the user can run ``lsof <pidfile>`` to find
        the holder — falling back to the generic ``pid None`` was
        confusing enough to file in this PR.
        """
        from memtomem._runtime_paths import ensure_runtime_dir

        _seed_state(home)
        pid_file = ensure_runtime_dir() / "server.pid"
        pid_file.write_text("", encoding="utf-8")  # empty content, exists

        with _hold_pid_lock(pid_file):
            result = CliRunner().invoke(cli, ["uninstall", "-y"])

        assert result.exit_code == 2
        assert "pid unknown" in result.output, (
            "empty pid + held flock must surface the 'pid unknown' branch, "
            f"not the generic 'pid None' message; got: {result.output!r}"
        )
        assert "lsof" in result.output, (
            "empty-pid branch must point at lsof so the user can identify "
            "the flock holder without another diagnostic round-trip"
        )
        # Refusal still protects state — same WAL-corruption invariant.
        assert (home / ".memtomem" / "memtomem.db").exists()


class TestPidRecyclingDoesNotFalsePositive:
    """#387: a recorded PID that happens to point at a live unrelated process
    must not trip the liveness probe. With the old ``os.kill(pid, 0)`` probe
    this returned alive → uninstall refused. With the flock probe the absence
    of a lock holder is the sole signal."""

    def test_pid_alive_but_no_lock_means_dead(self, home):
        state = _seed_state(home)
        # Our own PID — definitely alive — but nobody is holding the flock.
        (state / ".server.pid").write_text(str(os.getpid()), encoding="utf-8")

        result = CliRunner().invoke(cli, ["uninstall", "-y"])

        assert result.exit_code == 0, result.output
        assert "Server still running" not in result.output
        assert not state.exists()

    def test_runtime_pid_alive_but_no_lock_means_dead(self, home):
        """Same as above at the new runtime location — a stale
        ``server.pid`` with a recycled live PID but no flock holder must
        not refuse the uninstall."""
        from memtomem._runtime_paths import ensure_runtime_dir

        _seed_state(home)
        (ensure_runtime_dir() / "server.pid").write_text(str(os.getpid()), encoding="utf-8")

        result = CliRunner().invoke(cli, ["uninstall", "-y"])

        assert result.exit_code == 0, result.output
        assert "Server still running" not in result.output
        assert not (home / ".memtomem").exists()


class TestRuntimePidCleanedWithOther:
    """The runtime pid file lives outside ``~/.memtomem/`` but is still
    transient server state — uninstall must clean it up so a reinstall
    starts fresh. The runtime subdir is rmdir'd if we empty it."""

    def test_runtime_pid_deleted_and_subdir_pruned(self, home):
        from memtomem._runtime_paths import ensure_runtime_dir, runtime_dir

        _seed_state(home)
        rt = ensure_runtime_dir()
        pid_file = rt / "server.pid"
        pid_file.write_text("0", encoding="utf-8")

        result = CliRunner().invoke(cli, ["uninstall", "-y"])

        assert result.exit_code == 0, result.output
        assert not pid_file.exists(), "runtime pid file must be deleted"
        # subdir should be gone too — we own it
        assert not runtime_dir().exists(), "empty runtime subdir must be pruned"

    def test_runtime_subdir_preserved_when_unrelated_files_present(self, home):
        """Pin the empty-check precondition on the ``rmdir`` call so a
        future condition invert (``not any(iterdir())`` → ``any(...)``)
        wouldn't silently wipe unrelated files someone else parked in
        the runtime subdir. ``mm uninstall`` is scoped; other memtomem
        entry points (or a future #384 expansion) may legitimately
        register more pid files in the same dir."""
        from memtomem._runtime_paths import ensure_runtime_dir, runtime_dir

        _seed_state(home)
        rt = ensure_runtime_dir()
        # Our own pid file is cleaned, but a sibling registered by
        # another memtomem tool must survive.
        (rt / "server.pid").write_text("0", encoding="utf-8")
        sibling = rt / "someone-elses.pid"
        sibling.write_text("42", encoding="utf-8")

        result = CliRunner().invoke(cli, ["uninstall", "-y"])

        assert result.exit_code == 0, result.output
        # server.pid must be gone; sibling must remain; subdir must NOT rmdir.
        assert not (rt / "server.pid").exists()
        assert sibling.exists(), "unrelated file in runtime subdir must survive"
        assert runtime_dir().exists(), "runtime subdir must not be pruned when other files remain"


# -------------------------------------------------------------------- 13


class TestDbWriterLockRefuses:
    """Active SQLite writer without a .server.pid (mm web / watchdog / ad-hoc)
    must also block the uninstall — the gap #384 called out.

    The probe relies on ``BEGIN IMMEDIATE`` raising ``SQLITE_BUSY`` when
    another connection holds RESERVED-or-above. Holding an open
    ``BEGIN IMMEDIATE`` transaction in the test process reproduces this
    cross-process lock contention within a single pytest run.
    """

    def _make_real_db(self, home: Path) -> tuple[Path, sqlite3.Connection]:
        state = home / ".memtomem"
        state.mkdir(parents=True, exist_ok=True)
        # Keep parity with _seed_state for non-DB files so the inventory
        # path is exercised end-to-end, but seed a *real* SQLite DB.
        (state / "config.json").write_text('{"embedding": {"provider": "none"}}', encoding="utf-8")
        (state / "memories").mkdir(exist_ok=True)
        db_path = state / "memtomem.db"
        conn = sqlite3.connect(db_path)
        conn.execute("CREATE TABLE _probe (id INTEGER)")
        conn.commit()
        return db_path, conn

    def test_refuses_when_writer_holds_lock(self, home):
        db_path, conn = self._make_real_db(home)
        try:
            conn.execute("BEGIN IMMEDIATE")
            result = CliRunner().invoke(cli, ["uninstall", "-y"])
            assert result.exit_code == 2, result.output
            assert "holds a write lock" in result.output
            assert str(db_path) in result.output
            # "Server still running" path must NOT trigger — no .server.pid here.
            assert "Server still running" not in result.output
            # Hint contents differ per platform (#730): POSIX advertises
            # `lsof` + `--force`; Windows can't override --force, so it must
            # NOT advertise it and must point at Windows-native tools.
            if sys.platform == "win32":
                assert "lsof" not in result.output
                assert "pass --force" not in result.output
                assert (
                    "handle.exe" in result.output
                    or "Task Manager" in result.output
                    or "Get-Process" in result.output
                )
            else:
                assert "lsof" in result.output
                assert "pass --force" in result.output
            # Nothing deleted while the writer is alive.
            assert db_path.exists()
        finally:
            conn.rollback()
            conn.close()

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason=(
            "POSIX-only contract: --force wipes via unlink-while-open even "
            "though the writer still holds the inode. Windows variant lives "
            "in test_force_refuses_on_windows_when_writer_alive (#730)."
        ),
    )
    def test_force_overrides_db_lock_posix(self, home):
        db_path, conn = self._make_real_db(home)
        try:
            conn.execute("BEGIN IMMEDIATE")
            result = CliRunner().invoke(cli, ["uninstall", "-y", "--force"])
            assert result.exit_code == 0, result.output
            # State dir gets wiped — even if the lock-holding connection
            # still has the inode, the directory entry is gone.
            assert not db_path.exists()
        finally:
            try:
                conn.rollback()
            except sqlite3.ProgrammingError:
                pass  # connection may be invalidated after the file vanished
            conn.close()

    @pytest.mark.skipif(
        sys.platform != "win32",
        reason=(
            "Windows-only contract (#730): --force cannot wipe an open SQLite "
            "DB on Windows (WinError 32 on unlink-while-open), so --force "
            "must refuse cleanly instead of producing a half-wiped state dir."
        ),
    )
    def test_force_refuses_on_windows_when_writer_alive(self, home):
        db_path, conn = self._make_real_db(home)
        # _make_real_db seeds config.json + memories/ alongside the DB.
        # _delete_inventory wipes those BEFORE the DB, so checking they
        # survive proves the refusal fired before any deletion ran (#730).
        config_path = home / ".memtomem" / "config.json"
        memories_dir = home / ".memtomem" / "memories"
        assert config_path.exists()
        assert memories_dir.exists()
        try:
            conn.execute("BEGIN IMMEDIATE")
            result = CliRunner().invoke(cli, ["uninstall", "-y", "--force"])
            assert result.exit_code == 2, result.output
            assert "Windows" in result.output
            assert "stop the writer" in result.output.lower()
            # Refusal fired BEFORE _delete_inventory: nothing wiped.
            assert db_path.exists()
            assert config_path.exists()
            assert memories_dir.exists()
            # Partial-wipe message must NOT appear — that would mean
            # _delete_inventory ran and crashed mid-way.
            assert "Deletion failed at" not in result.output
        finally:
            try:
                conn.rollback()
            except sqlite3.ProgrammingError:
                pass
            conn.close()

    def test_proceeds_when_db_exists_but_no_writer(self, home):
        db_path, conn = self._make_real_db(home)
        conn.close()  # release before probing — no writer held.
        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "holds a write lock" not in result.output
        assert not db_path.exists()


class TestPidStaleProceeds:
    def test_proceeds_when_pid_stale(self, home, monkeypatch):
        """Pick a pid that's guaranteed dead (high number, not currently
        in use). os.kill(stale_pid, 0) raises ProcessLookupError → not alive.
        """
        state = _seed_state(home)

        # Find a stale pid — start at a high number and bump until os.kill
        # raises ProcessLookupError. Skip if PermissionError (alive but
        # not ours, can happen at low PIDs).
        stale_pid = 999_999
        for candidate in range(999_999, 999_900, -1):
            try:
                os.kill(candidate, 0)
            except ProcessLookupError:
                stale_pid = candidate
                break
            except (PermissionError, OSError):
                continue
        else:
            pytest.skip("could not find a stale pid for testing")

        (state / ".server.pid").write_text(str(stale_pid), encoding="utf-8")

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "Server still running" not in result.output
        assert not state.exists()


# -------------------------------------------------------------------- 14


class TestConfigFallback:
    def test_falls_back_when_config_load_raises(self, home, monkeypatch):
        """``_load_config_safely`` must catch any exception escaping
        ``load_config_overrides`` and fall back to the default DB path.

        ``load_config_overrides`` already swallows malformed JSON itself
        (logs WARNING, returns), so we monkeypatch it to raise outright —
        that's the failure mode the safety net was added for.
        """
        state = home / ".memtomem"
        state.mkdir()
        (state / "config.json").write_text("{}", encoding="utf-8")
        (state / "memtomem.db").write_bytes(b"sqlite-fake")

        def _boom(_cfg):
            raise PermissionError("fake permission denied on config.json")

        monkeypatch.setattr("memtomem.config.load_config_overrides", _boom)

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 0, result.output
        assert "config unreadable" in result.output
        assert "fake permission denied" in result.output
        # default DB path used as fallback → DB still gets cleaned up
        assert "Removed:" in result.output


# -------------------------------------------------------------------- 15
#
# Transactional staging (#757): the wipe is staged via os.replace into a
# sibling .uninstall-staging-<pid>/ dir, then rmtree'd on success. A
# mid-stage failure rolls back so the user's state dir is never left
# half-gone, and a cross-FS layout is refused with a clean message
# instead of being half-staged.


class TestTransactionalStaging:
    """Acceptance criteria from the issue:

    - Mid-stage failure → original state dir intact (no half-state).
    - --keep-config / --keep-data still work (covered by sibling tests
      above; this class just adds the failure-injection cases).
    - Cross-FS layout → refusal-or-fallback per the chosen design.
    """

    def _snapshot(self, root: Path) -> dict[str, bytes | str]:
        """Return ``{relpath: contents}`` for every file under ``root``,
        excluding any ``.uninstall-staging-*`` subtree (transient state
        we don't want included in the equality check).

        Keys use ``as_posix()`` so the snapshot is comparable across
        platforms — ``str(Path("memories/x.md"))`` is ``memories/x.md``
        on POSIX but ``memories\\x.md`` on Windows, which would make
        the hardcoded ``"memories/x.md" in before`` sanity check fail
        on the Windows lane (caught in CI).

        ``config.json`` is recorded as a presence sentinel rather than
        its bytes because ``_load_config_safely`` may legitimately
        rewrite it (e.g. the ``auto_discover`` migration) before
        staging begins — that mutation is orthogonal to whether
        staging is transactional. We still verify the file *exists*
        post-rollback, which is the staging invariant."""
        from memtomem.cli.uninstall_cmd import _STAGING_PREFIX

        snap: dict[str, bytes | str] = {}
        for path in sorted(root.rglob("*")):
            if not path.is_file():
                continue
            rel = path.relative_to(root)
            if any(part.startswith(_STAGING_PREFIX) for part in rel.parts):
                continue
            key = rel.as_posix()
            if key == "config.json":
                snap[key] = "<exists>"
            else:
                snap[key] = path.read_bytes()
        return snap

    def test_rollback_restores_original_state_on_mid_stage_failure(self, home, monkeypatch):
        """Inject an OSError on the Nth staging-direction os.replace and
        assert every original file is back where it started.

        Filters on the staging-prefix in ``dst`` so unrelated os.replace
        calls (e.g. config-load's atomic rewrite of ``config.json``)
        don't shift the call counter.

        N=2 leaves at least one earlier move that needs to be rolled
        back, so the test exercises the rollback path, not just the
        first-call-fails edge case."""
        from memtomem.cli import uninstall_cmd

        state = _seed_state(home)
        before = self._snapshot(state)
        # Sanity: snapshot is non-trivial (catches a future _seed_state shrink).
        assert "config.json" in before
        assert "memtomem.db" in before
        assert "memories/x.md" in before

        real_replace = os.replace
        stage_calls = {"n": 0}

        def _flaky_replace(src, dst, **kwargs):
            if uninstall_cmd._STAGING_PREFIX in str(dst):
                stage_calls["n"] += 1
                if stage_calls["n"] == 2:
                    raise OSError(errno.EACCES, "fake permission denied during stage")
            return real_replace(src, dst, **kwargs)

        monkeypatch.setattr(uninstall_cmd.os, "replace", _flaky_replace)

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 2, result.output
        assert "Deletion failed" in result.output
        assert "Rolled back" in result.output

        after = self._snapshot(state)
        assert after == before, (
            f"rollback did not restore original layout.\n"
            f"missing: {set(before) - set(after)}\n"
            f"extra:   {set(after) - set(before)}"
        )
        # No staging dir leftovers (rollback cleans up empty roots).
        leftover = sorted(
            p.name for p in state.iterdir() if p.name.startswith(uninstall_cmd._STAGING_PREFIX)
        )
        assert leftover == [], f"unexpected staging leftovers: {leftover}"

    def test_rollback_failure_surfaces_recovery_path(self, home, monkeypatch):
        """When BOTH stage AND rollback fail, the user must see the
        staging dir path so they can recover manually. Without this,
        partial-state recovery is impractical (the issue's whole
        motivation).

        Setup: forward staging move #1 succeeds, #2 trips the failure
        path, and the rollback move (recognized by ``_STAGING_PREFIX``
        in ``src``, not ``dst``) ALSO fails — the disaster scenario."""
        from memtomem.cli import uninstall_cmd

        _seed_state(home)
        real_replace = os.replace
        stage_calls = {"n": 0}

        def _broken_replace(src, dst, **kwargs):
            src_str, dst_str = str(src), str(dst)
            if uninstall_cmd._STAGING_PREFIX in dst_str:
                # Forward stage move.
                stage_calls["n"] += 1
                if stage_calls["n"] == 2:
                    raise OSError(errno.EACCES, "fake stage failure")
                return real_replace(src, dst, **kwargs)
            if uninstall_cmd._STAGING_PREFIX in src_str:
                # Rollback move — refuse so the recovery path activates.
                raise OSError(errno.EACCES, "fake rollback failure")
            return real_replace(src, dst, **kwargs)

        monkeypatch.setattr(uninstall_cmd.os, "replace", _broken_replace)

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 2, result.output
        assert "Deletion failed" in result.output
        assert "Rollback also failed" in result.output
        assert uninstall_cmd._STAGING_PREFIX in result.output, (
            "user must be told the staging dir path for manual recovery"
        )

    def test_cross_fs_layout_refused_cleanly(self, home, monkeypatch):
        """Simulate EXDEV from os.replace and assert the user sees a
        cross-filesystem refusal — not a generic stage-failed error —
        with the original state dir untouched.

        Real cross-FS layouts (bind mount inside ~/.memtomem) are too
        environment-dependent to set up in CI, so we monkeypatch the
        EXDEV path instead. Same code path as a real cross-FS layout
        would take."""
        from memtomem.cli import uninstall_cmd

        state = _seed_state(home)
        before = self._snapshot(state)

        def _exdev_replace(src, dst, **kwargs):
            raise OSError(errno.EXDEV, "fake cross-device link")

        monkeypatch.setattr(uninstall_cmd.os, "replace", _exdev_replace)

        result = CliRunner().invoke(cli, ["uninstall", "-y"])
        assert result.exit_code == 2, result.output
        assert "different" in result.output.lower() and "filesystem" in result.output.lower()
        # Original state intact — refusal fires before any partial wipe.
        after = self._snapshot(state)
        assert after == before


# (#448 / #625) The single-file regex pin that used to live here is
# superseded by ``test_no_posix_only_imports.py``, which AST-scans every
# module under ``packages/memtomem/src/memtomem/`` for module-level
# POSIX-only imports — fcntl, pwd, grp, termios, resource. The earlier
# regex covered only ``cli/uninstall_cmd.py``, which is exactly why
# PR #623's regression in ``context/_atomic.py`` slipped past it.
