"""Tests for mempalace.cli — the main CLI dispatcher."""

import argparse
import errno
import os
import shlex
import socket
import sqlite3
import subprocess
import sys
from contextlib import closing
from pathlib import Path
from unittest.mock import MagicMock, call, patch

import pytest

from mempalace.cli import (
    cmd_compress,
    cmd_hook,
    cmd_init,
    cmd_instructions,
    cmd_daemon,
    cmd_mine,
    cmd_repair,
    cmd_search,
    cmd_split,
    cmd_status,
    cmd_wakeup,
    main,
)


# ── CLI entry point: PYTHONPATH stripping ────────────────────────────────


_LEAK_PREFIX = "/__mempalace_cli_leak_sentinel__"


def test_cli_main_strips_leaked_pythonpath_from_env():
    """mempalace.cli:main must drop PYTHONPATH from the process env so
    any subprocess the CLI 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: CLI entry-point env strip ran.

    SystemExit is caught with a narrowed exit-code check so a future
    argparse change that exits with a non-zero code (e.g. usage error)
    surfaces as a test failure instead of being swallowed."""
    expected_env = f"{_LEAK_PREFIX}/a{os.pathsep}{_LEAK_PREFIX}/b"
    env = os.environ.copy()
    env["PYTHONPATH"] = expected_env
    # Run main() with --version so it exits cleanly without entering any
    # subcommand. argparse raises SystemExit(0) on --version; the wrapper
    # asserts the exit code is clean and prints the post-main PYTHONPATH
    # so the assertion is observable.
    code = (
        "import os, sys\n"
        "from mempalace.cli import main\n"
        f"prefix = {_LEAK_PREFIX!r}\n"
        "print('ENV_MID:', repr(os.environ.get('PYTHONPATH')))\n"
        "print('SENTINEL_IN_PATH:', any(prefix in (p or '') for p in sys.path))\n"
        "sys.argv = ['mempalace', '--version']\n"
        "try:\n"
        "    main()\n"
        "except SystemExit as exc:\n"
        "    assert exc.code in (0, None), f'unexpected exit code: {exc.code!r}'\n"
        "print('ENV_AFTER:', repr(os.environ.get('PYTHONPATH')))\n"
    )
    result = subprocess.run(
        [sys.executable, "-c", code],
        env=env,
        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.stdout, (
        f"package import unexpectedly stripped env (regression in __init__.py): {diag}"
    )
    assert "SENTINEL_IN_PATH: False" in result.stdout, (
        f"package import did not filter sys.path (regression in __init__.py): {diag}"
    )
    assert "ENV_AFTER: None" in result.stdout, f"CLI did not strip PYTHONPATH: {diag}"


# ── cmd_status ─────────────────────────────────────────────────────────


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_status_default_palace(mock_config_cls):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(palace=None)
    mock_miner = MagicMock()
    with patch.dict("sys.modules", {"mempalace.miner": mock_miner}):
        cmd_status(args)
        mock_miner.status.assert_called_once_with(palace_path="/fake/palace")


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_status_custom_palace(mock_config_cls):
    args = argparse.Namespace(palace="~/my_palace")
    mock_miner = MagicMock()
    with patch.dict("sys.modules", {"mempalace.miner": mock_miner}):
        cmd_status(args)
        import os

        expected = os.path.expanduser("~/my_palace")
        mock_miner.status.assert_called_once_with(palace_path=expected)


# ── cmd_search ─────────────────────────────────────────────────────────


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_search_calls_search(mock_config_cls):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        palace=None,
        query="test query",
        wing="mywing",
        room="myroom",
        results=3,
        since="2026-04-01",
        before=None,
    )
    with patch("mempalace.searcher.search") as mock_search:
        cmd_search(args)
        mock_search.assert_called_once_with(
            query="test query",
            palace_path="/fake/palace",
            wing="mywing",
            room="myroom",
            n_results=3,
            since="2026-04-01",
            before=None,
        )


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_search_error_exits(mock_config_cls):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        palace=None, query="q", wing=None, room=None, results=5, since=None, before=None
    )
    from mempalace.searcher import SearchError

    with patch("mempalace.searcher.search", side_effect=SearchError("fail")):
        with pytest.raises(SystemExit) as exc_info:
            cmd_search(args)
        assert exc_info.value.code == 1


# ── cmd_instructions ───────────────────────────────────────────────────


def test_cmd_instructions_calls_run_instructions():
    args = argparse.Namespace(name="help")
    with patch("mempalace.instructions_cli.run_instructions") as mock_run:
        cmd_instructions(args)
        mock_run.assert_called_once_with(name="help")


# ── cmd_hook ───────────────────────────────────────────────────────────


def test_cmd_hook_calls_run_hook():
    args = argparse.Namespace(hook="session-start", harness="claude-code")
    with patch("mempalace.hooks_cli.run_hook") as mock_run:
        cmd_hook(args)
        mock_run.assert_called_once_with(hook_name="session-start", harness="claude-code")


def test_cmd_hook_session_end_calls_run_hook():
    args = argparse.Namespace(hook="session-end", harness="claude-code")
    with patch("mempalace.hooks_cli.run_hook") as mock_run:
        cmd_hook(args)
        mock_run.assert_called_once_with(hook_name="session-end", harness="claude-code")


# ── cmd_init ───────────────────────────────────────────────────────────


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_init_no_entities(mock_config_cls, tmp_path):
    args = argparse.Namespace(dir=str(tmp_path), yes=True)
    with (
        patch("mempalace.entity_detector.scan_for_detection", return_value=[]),
        patch("mempalace.room_detector_local.detect_rooms_local") as mock_rooms,
        patch("mempalace.cli._maybe_run_mine_after_init"),
    ):
        cmd_init(args)
        mock_rooms.assert_called_once_with(project_dir=str(tmp_path), yes=True)
        mock_config_cls.return_value.init.assert_called_once()


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_init_with_entities(mock_config_cls, tmp_path):
    fake_files = [tmp_path / "a.txt"]
    detected = {"people": [{"name": "Alice"}], "projects": [], "uncertain": []}
    confirmed = {"people": ["Alice"], "projects": []}
    args = argparse.Namespace(dir=str(tmp_path), yes=True)
    with (
        patch("mempalace.entity_detector.scan_for_detection", return_value=fake_files),
        patch("mempalace.entity_detector.detect_entities", return_value=detected),
        patch("mempalace.entity_detector.confirm_entities", return_value=confirmed),
        patch("mempalace.room_detector_local.detect_rooms_local"),
        # Pass 0 (corpus_origin) needs real file IO; this test mocks
        # builtins.open globally for the entities.json write, which would
        # break Pass 0's file-reading path. Patch Pass 0 out — a separate
        # suite (tests/test_corpus_origin_integration.py) covers it directly.
        patch("mempalace.cli._run_pass_zero", return_value=None),
        patch("builtins.open", MagicMock()),
        patch("mempalace.cli._maybe_run_mine_after_init"),
    ):
        cmd_init(args)


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_init_normalizes_wing_name_for_topics_registry(mock_config_cls, tmp_path):
    """Regression for #1194: hyphenated dir names must be normalized to the
    same slug ``mempalace.yaml`` uses, otherwise ``topics_by_wing`` keys
    miss the miner's lookup at mine time and tunnels are silently dropped.
    """
    project = tmp_path / "my-cool-app"
    project.mkdir()
    fake_files = [project / "a.txt"]
    detected = {
        "people": [{"name": "Alice"}],
        "projects": [],
        "topics": [{"name": "Bun"}],
        "uncertain": [],
    }
    confirmed = {"people": ["Alice"], "projects": [], "topics": ["Bun"]}
    args = argparse.Namespace(dir=str(project), yes=True)
    with (
        patch("mempalace.entity_detector.scan_for_detection", return_value=fake_files),
        patch("mempalace.entity_detector.detect_entities", return_value=detected),
        patch("mempalace.entity_detector.confirm_entities", return_value=confirmed),
        patch("mempalace.miner.add_to_known_entities") as mock_register,
        patch("mempalace.room_detector_local.detect_rooms_local"),
        patch("builtins.open", MagicMock()),
        patch("mempalace.cli._maybe_run_mine_after_init"),
        # Pass-zero corpus-origin detection runs unconditionally inside
        # cmd_init now (#1221 / #1223). It accesses MempalaceConfig fields
        # that don't survive MagicMock stringification, so stub it out —
        # this test only cares about the wing-slug write to the registry.
        patch("mempalace.cli._run_pass_zero", return_value=None),
    ):
        mock_register.return_value = "/tmp/known_entities.json"
        cmd_init(args)
        mock_register.assert_called_once()
        assert mock_register.call_args.kwargs["wing"] == "my_cool_app"


def test_cmd_init_honors_palace_flag(tmp_path, monkeypatch):
    """Regression for #1313: ``cmd_init`` must honor ``--palace`` instead of
    silently writing to ``~/.mempalace``. Mirrors the env-var pattern used
    by ``cmd_mine`` / ``cmd_status`` / ``mcp_server`` so every downstream
    read of ``cfg.palace_path`` (Pass 0, ``cfg.init()``, post-init mine)
    routes to the user-specified location.
    """
    project = tmp_path / "project"
    project.mkdir()
    palace = tmp_path / "custom_palace"

    # Make sure no leftover env var from another test leaks in — we want to
    # verify that --palace ALONE drives the resolution. Prime monkeypatch's
    # undo list with setenv first so that the env var ``cmd_init`` writes
    # below is rolled back at teardown (``delenv(raising=False)`` on a
    # missing key registers no undo entry, which would leak into the next
    # test).
    monkeypatch.setenv("MEMPALACE_PALACE_PATH", "")
    monkeypatch.setenv("MEMPAL_PALACE_PATH", "")
    monkeypatch.delenv("MEMPALACE_PALACE_PATH")
    monkeypatch.delenv("MEMPAL_PALACE_PATH")

    args = argparse.Namespace(
        dir=str(project),
        palace=str(palace),
        yes=True,
        auto_mine=False,
    )

    captured = {}

    def fake_pass_zero(project_dir, palace_dir, llm_provider):
        # Capture the palace_dir Pass 0 sees — this is the smoking-gun
        # value for the bug. Pre-fix it was always ~/.mempalace.
        captured["pass_zero_palace_dir"] = palace_dir
        return None

    with (
        patch("mempalace.entity_detector.scan_for_detection", return_value=[]),
        patch("mempalace.room_detector_local.detect_rooms_local"),
        patch("mempalace.cli._run_pass_zero", side_effect=fake_pass_zero),
        patch("mempalace.cli._maybe_run_mine_after_init"),
    ):
        cmd_init(args)

    expected = str(palace)
    # Pass 0 must have been handed the --palace location, not ~/.mempalace.
    assert captured["pass_zero_palace_dir"] == expected
    # And the env var must point at the custom palace so any downstream
    # ``cfg.palace_path`` read in this process resolves correctly too.
    import os

    assert os.environ.get("MEMPALACE_PALACE_PATH") == os.path.abspath(expected)


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_init_with_entities_zero_total(mock_config_cls, tmp_path, capsys):
    """When entities detected but total is 0, prints 'No entities' message."""
    fake_files = [tmp_path / "a.txt"]
    detected = {"people": [], "projects": [], "uncertain": []}
    args = argparse.Namespace(dir=str(tmp_path), yes=False)
    with (
        patch("mempalace.entity_detector.scan_for_detection", return_value=fake_files),
        patch("mempalace.entity_detector.detect_entities", return_value=detected),
        patch("mempalace.room_detector_local.detect_rooms_local"),
        patch("mempalace.cli._maybe_run_mine_after_init"),
    ):
        cmd_init(args)
    out = capsys.readouterr().out
    assert "No entities detected" in out


# ── _maybe_run_mine_after_init (init → mine prompt, #1181) ─────────────


def _init_args(tmp_path, *, yes=False, auto_mine=False):
    return argparse.Namespace(dir=str(tmp_path), yes=yes, auto_mine=auto_mine)


def _fake_cfg(tmp_path):
    cfg = MagicMock()
    cfg.palace_path = str(tmp_path / "palace")
    return cfg


def _fake_scanned(tmp_path, n=3):
    """Build n real Path objects with stat()-able sizes for the scan estimate."""
    paths = []
    for i in range(n):
        p = tmp_path / f"f{i}.txt"
        p.write_text("x" * 1024)  # 1 KB each
        paths.append(p)
    return paths


def test_maybe_run_mine_prompt_accepted_runs_mine(tmp_path):
    """Empty / 'y' / 'yes' on the prompt triggers mine() in-process."""
    from mempalace.cli import _maybe_run_mine_after_init

    args = _init_args(tmp_path, yes=False, auto_mine=False)
    cfg = _fake_cfg(tmp_path)
    scanned = _fake_scanned(tmp_path, n=3)
    with (
        patch("mempalace.miner.mine") as mock_mine,
        patch("mempalace.miner.scan_project", return_value=scanned),
        patch("builtins.input", return_value=""),
    ):
        _maybe_run_mine_after_init(args, cfg)
        mock_mine.assert_called_once_with(
            project_dir=str(tmp_path),
            palace_path=cfg.palace_path,
            files=scanned,
        )


def test_maybe_run_mine_prompt_yes_accepted_runs_mine(tmp_path):
    """Explicit 'y' answer also runs mine()."""
    from mempalace.cli import _maybe_run_mine_after_init

    args = _init_args(tmp_path, yes=False, auto_mine=False)
    cfg = _fake_cfg(tmp_path)
    with (
        patch("mempalace.miner.mine") as mock_mine,
        patch("mempalace.miner.scan_project", return_value=[]),
        patch("builtins.input", return_value="Y"),
    ):
        _maybe_run_mine_after_init(args, cfg)
        mock_mine.assert_called_once()


def test_maybe_run_mine_prompt_declined_prints_hint(tmp_path, capsys):
    """'n' answer skips mine() and prints the resume hint."""
    from mempalace.cli import _maybe_run_mine_after_init

    args = _init_args(tmp_path, yes=False, auto_mine=False)
    cfg = _fake_cfg(tmp_path)
    with (
        patch("mempalace.miner.mine") as mock_mine,
        patch("mempalace.miner.scan_project", return_value=[]),
        patch("builtins.input", return_value="n"),
    ):
        _maybe_run_mine_after_init(args, cfg)
        mock_mine.assert_not_called()
    out = capsys.readouterr().out
    # shlex.quote is a no-op on POSIX-safe paths but wraps Windows paths
    # (which contain backslashes) in single quotes, so the assertion has
    # to mirror what the production code actually emits.
    assert f"mempalace mine {shlex.quote(str(tmp_path))}" in out
    assert "Skipped" in out


def test_maybe_run_mine_yes_alone_still_prompts(tmp_path):
    """`--yes` is scoped to entity auto-accept and MUST still prompt for mine.

    Regression guard for the flag-overload review feedback on #1183: extending
    `--yes` to also auto-mine would silently change behaviour for scripted
    callers and turn a fast command into a minutes-long ChromaDB write.
    """
    from mempalace.cli import _maybe_run_mine_after_init

    args = _init_args(tmp_path, yes=True, auto_mine=False)
    cfg = _fake_cfg(tmp_path)
    with (
        patch("mempalace.miner.mine") as mock_mine,
        patch("mempalace.miner.scan_project", return_value=[]),
        patch("builtins.input", return_value="n") as mock_input,
    ):
        _maybe_run_mine_after_init(args, cfg)
        mock_input.assert_called_once()  # the prompt MUST fire
        mock_mine.assert_not_called()


def test_maybe_run_mine_auto_mine_skips_prompt(tmp_path):
    """`--auto-mine` runs mine() automatically without calling input()."""
    from mempalace.cli import _maybe_run_mine_after_init

    args = _init_args(tmp_path, yes=False, auto_mine=True)
    cfg = _fake_cfg(tmp_path)
    scanned = _fake_scanned(tmp_path, n=2)
    with (
        patch("mempalace.miner.mine") as mock_mine,
        patch("mempalace.miner.scan_project", return_value=scanned),
        patch("builtins.input", side_effect=AssertionError("input() must not be called")),
    ):
        _maybe_run_mine_after_init(args, cfg)
        mock_mine.assert_called_once_with(
            project_dir=str(tmp_path),
            palace_path=cfg.palace_path,
            files=scanned,
        )


def test_maybe_run_mine_yes_and_auto_mine_fully_noninteractive(tmp_path):
    """`--yes --auto-mine` together: never call input(), always mine."""
    from mempalace.cli import _maybe_run_mine_after_init

    args = _init_args(tmp_path, yes=True, auto_mine=True)
    cfg = _fake_cfg(tmp_path)
    with (
        patch("mempalace.miner.mine") as mock_mine,
        patch("mempalace.miner.scan_project", return_value=[]),
        patch("builtins.input", side_effect=AssertionError("input() must not be called")),
    ):
        _maybe_run_mine_after_init(args, cfg)
        mock_mine.assert_called_once()


def test_maybe_run_mine_decline_quotes_path_with_spaces(tmp_path, capsys):
    """The resume hint must shell-quote the project dir so paths with
    spaces / metacharacters produce a copy-paste-safe command."""
    from mempalace.cli import _maybe_run_mine_after_init

    spaced_dir = tmp_path / "my project dir"
    spaced_dir.mkdir()
    args = argparse.Namespace(dir=str(spaced_dir), yes=False, auto_mine=False)
    cfg = _fake_cfg(tmp_path)
    with (
        patch("mempalace.miner.mine"),
        patch("mempalace.miner.scan_project", return_value=[]),
        patch("builtins.input", return_value="n"),
    ):
        _maybe_run_mine_after_init(args, cfg)
    out = capsys.readouterr().out
    # shlex.quote wraps paths with spaces (and Windows backslashes) in
    # single quotes — the assertion must use the same shlex form so the
    # test passes on every platform's tmp_path layout.
    assert f"mempalace mine {shlex.quote(str(spaced_dir))}" in out
    # Bare unquoted form must NOT appear — that's the bug we're guarding.
    assert f"mempalace mine {spaced_dir} " not in out
    assert f"mempalace mine {spaced_dir}`" not in out


def test_maybe_run_mine_eof_on_stdin_treated_as_decline(tmp_path, capsys):
    """Piped / non-interactive stdin (EOFError) declines without crashing."""
    from mempalace.cli import _maybe_run_mine_after_init

    args = _init_args(tmp_path, yes=False, auto_mine=False)
    cfg = _fake_cfg(tmp_path)
    with (
        patch("mempalace.miner.mine") as mock_mine,
        patch("mempalace.miner.scan_project", return_value=[]),
        patch("builtins.input", side_effect=EOFError),
    ):
        _maybe_run_mine_after_init(args, cfg)
        mock_mine.assert_not_called()
    assert "Skipped" in capsys.readouterr().out


def test_maybe_run_mine_failure_surfaces_via_exit(tmp_path, capsys):
    """Mine errors are not swallowed — they exit non-zero with an error line."""
    from mempalace.cli import _maybe_run_mine_after_init

    args = _init_args(tmp_path, yes=False, auto_mine=True)
    cfg = _fake_cfg(tmp_path)
    with (
        patch("mempalace.miner.mine", side_effect=RuntimeError("boom")),
        patch("mempalace.miner.scan_project", return_value=[]),
    ):
        with pytest.raises(SystemExit) as exc_info:
            _maybe_run_mine_after_init(args, cfg)
    assert exc_info.value.code == 1
    err = capsys.readouterr().err
    assert "boom" in err


def test_maybe_run_mine_estimate_appears_before_prompt(tmp_path, capsys):
    """The file-count + size estimate line MUST render BEFORE the prompt.

    Required by the spec: hitting Enter on a default-Y prompt with no size
    info is a footgun on a real corpus where mine takes minutes. The user
    must see scope before being asked to confirm.
    """
    from mempalace.cli import _maybe_run_mine_after_init

    args = _init_args(tmp_path, yes=False, auto_mine=False)
    cfg = _fake_cfg(tmp_path)
    scanned = _fake_scanned(tmp_path, n=4)  # 4 files * 1 KB each
    captured_when_prompted = {}

    def fake_input(prompt):
        # Snapshot what stdout looked like at the moment the prompt fires.
        captured_when_prompted["stdout"] = capsys.readouterr().out
        return "n"

    with (
        patch("mempalace.miner.mine"),
        patch("mempalace.miner.scan_project", return_value=scanned),
        patch("builtins.input", side_effect=fake_input),
    ):
        _maybe_run_mine_after_init(args, cfg)

    pre_prompt = captured_when_prompted["stdout"]
    assert "4 files" in pre_prompt, f"file count missing from pre-prompt output: {pre_prompt!r}"
    assert "MB" in pre_prompt, f"size estimate missing from pre-prompt output: {pre_prompt!r}"
    assert "would be mined" in pre_prompt


# ── cmd_mine ───────────────────────────────────────────────────────────


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_mine_projects_mode(mock_config_cls):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        dir="/src",
        palace=None,
        mode="projects",
        wing=None,
        agent="mempalace",
        limit=0,
        dry_run=False,
        no_gitignore=False,
        include_ignored=[],
        extract="exchange",
    )
    with patch("mempalace.miner.mine") as mock_mine:
        cmd_mine(args)
        mock_mine.assert_called_once_with(
            project_dir="/src",
            palace_path="/fake/palace",
            wing_override=None,
            agent="mempalace",
            limit=0,
            dry_run=False,
            respect_gitignore=True,
            include_ignored=[],
            max_chunks_per_file=None,
        )


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_mine_convos_mode(mock_config_cls):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        dir="/chats",
        palace=None,
        mode="convos",
        wing="mywing",
        agent="me",
        limit=10,
        dry_run=True,
        no_gitignore=False,
        include_ignored=[],
        extract="general",
        include_subagents=False,
    )
    with patch("mempalace.convo_miner.mine_convos") as mock_mine:
        cmd_mine(args)
        mock_mine.assert_called_once_with(
            convo_dir="/chats",
            palace_path="/fake/palace",
            wing="mywing",
            agent="me",
            limit=10,
            dry_run=True,
            extract_mode="general",
            include_subagents=False,
        )


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_mine_convos_mode_threads_include_subagents_flag(mock_config_cls):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        dir="/chats",
        palace=None,
        mode="convos",
        wing="mywing",
        agent="me",
        limit=10,
        dry_run=True,
        no_gitignore=False,
        include_ignored=[],
        extract="exchange",
        include_subagents=True,
    )
    with patch("mempalace.convo_miner.mine_convos") as mock_mine:
        cmd_mine(args)
        kwargs = mock_mine.call_args.kwargs
        assert kwargs["include_subagents"] is True


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_mine_include_ignored_comma_split(mock_config_cls):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        dir="/src",
        palace=None,
        mode="projects",
        wing=None,
        agent="mempalace",
        limit=0,
        dry_run=False,
        no_gitignore=False,
        include_ignored=["a.txt,b.txt", "c.txt"],
        extract="exchange",
    )
    with patch("mempalace.miner.mine") as mock_mine:
        cmd_mine(args)
        mock_mine.assert_called_once()
        call_kwargs = mock_mine.call_args[1]
        assert call_kwargs["include_ignored"] == ["a.txt", "b.txt", "c.txt"]


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_mine_daemon_background_submits_job(mock_config_cls, capsys):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        dir="/src",
        palace=None,
        mode="projects",
        wing=None,
        agent="mempalace",
        limit=0,
        dry_run=False,
        no_gitignore=False,
        include_ignored=["a.txt,b.txt"],
        extract="exchange",
        daemon=True,
        background=True,
        backend=None,
        global_backend=None,
        max_chunks_per_file=None,
        redetect_origin=False,
    )
    with patch("mempalace.daemon.submit_job", return_value={"id": "job-1"}) as mock_submit:
        with patch("mempalace.miner.mine") as mock_mine:
            cmd_mine(args)

    mock_mine.assert_not_called()
    mock_submit.assert_called_once()
    call_kwargs = mock_submit.call_args.kwargs
    assert call_kwargs["palace_path"] == "/fake/palace"
    assert call_kwargs["wait"] is False
    payload = mock_submit.call_args.args[1]
    assert payload["include_ignored"] == ["a.txt", "b.txt"]
    assert "job-1" in capsys.readouterr().out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_mine_daemon_lock_deferral_reports_a_runnable_command(mock_config_cls, capsys):
    """A foreground mine refused the palace lock must say so and hand back a
    command that actually works. --palace is global, so it has to be echoed back
    ahead of the subcommand: without it the suggestion silently lists the
    default palace's queue instead of the one the job is parked in."""
    mock_config_cls.return_value.palace_path = "/fake/palace"
    parked = {
        "id": "job-1",
        "state": "queued",
        "error": {
            "error_class": "LockHeldByOtherProcess",
            "message": "palace /my palace is held by PID 999",
        },
    }
    args = argparse.Namespace(
        dir="/src",
        palace="/my palace",  # non-default, and with a space to pin the quoting
        mode="projects",
        wing=None,
        agent="mempalace",
        limit=0,
        dry_run=False,
        no_gitignore=False,
        include_ignored=None,
        extract="exchange",
        daemon=True,
        background=False,
        backend=None,
        global_backend=None,
        max_chunks_per_file=None,
        redetect_origin=False,
    )
    with patch("mempalace.daemon.submit_job", return_value=parked) as mock_submit:
        with pytest.raises(SystemExit) as exc:
            cmd_mine(args)

    assert exc.value.code == 1
    assert mock_submit.call_args.kwargs["stop_on_lock_deferral"] is True
    err = capsys.readouterr().err
    assert "is held by PID 999" in err  # the holder, not a generic failure
    assert "daemon submission failed" not in err  # the submission did not fail
    assert f"mempalace --palace {shlex.quote('/my palace')} daemon jobs" in err
    # Not `daemon wait`: this branch exists because we would not wait out the
    # holder, so it must not hand back a command that does exactly that.
    assert "daemon wait" not in err


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_mine_background_requires_daemon(mock_config_cls, capsys):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        dir="/src",
        palace=None,
        mode="projects",
        wing=None,
        agent="mempalace",
        limit=0,
        dry_run=False,
        no_gitignore=False,
        include_ignored=[],
        extract="exchange",
        daemon=False,
        background=True,
    )
    with pytest.raises(SystemExit) as excinfo:
        cmd_mine(args)
    assert excinfo.value.code == 2
    assert "--background requires --daemon" in capsys.readouterr().err


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_mine_exits_nonzero_on_lock_holder(mock_config_cls, capsys):
    """Regression #1264: lock contention must exit non-zero with a clear message.

    Before this fix the CLI silently returned 0 when another writer held
    the palace lock — operators using nohup/scripts had no way to detect
    the contention. The new behavior raises MineAlreadyRunning out of
    miner.mine() and cmd_mine catches it, printing the holder identity
    to stderr and exiting non-zero.
    """
    from mempalace.palace import MineAlreadyRunning

    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        dir="/src",
        palace=None,
        mode="projects",
        wing=None,
        agent="mempalace",
        limit=0,
        dry_run=False,
        no_gitignore=False,
        include_ignored=[],
        extract="exchange",
    )
    with patch(
        "mempalace.miner.mine",
        side_effect=MineAlreadyRunning(
            "palace /fake/palace is held by PID 12345 (mempalace mcp_server); wait for it to finish"
        ),
    ):
        with pytest.raises(SystemExit) as excinfo:
            cmd_mine(args)
    assert excinfo.value.code == 1
    captured = capsys.readouterr()
    assert "PID 12345" in captured.err
    assert "mcp_server" in captured.err


# ── cmd_wakeup ─────────────────────────────────────────────────────────


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_wakeup(mock_config_cls, capsys):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(palace=None, wing=None)
    mock_stack = MagicMock()
    mock_stack.wake_up.return_value = "Hello world context"
    with patch("mempalace.layers.MemoryStack", return_value=mock_stack):
        cmd_wakeup(args)
    out = capsys.readouterr().out
    assert "Hello world context" in out
    assert "tokens" in out


# ── cmd_split ──────────────────────────────────────────────────────────


def test_cmd_split_basic():
    args = argparse.Namespace(dir="/chats", output_dir=None, dry_run=False, min_sessions=2)
    with patch("mempalace.split_mega_files.main") as mock_main:
        cmd_split(args)
        mock_main.assert_called_once()


def test_cmd_split_all_options():
    args = argparse.Namespace(dir="/chats", output_dir="/out", dry_run=True, min_sessions=5)
    with patch("mempalace.split_mega_files.main") as mock_main:
        cmd_split(args)
        mock_main.assert_called_once()
    # sys.argv should be restored
    assert sys.argv[0] != "mempalace split"


# ── main() argparse dispatch ──────────────────────────────────────────


def test_main_no_args_prints_help(capsys):
    with patch("sys.argv", ["mempalace"]):
        main()
    out = capsys.readouterr().out
    assert "MemPalace" in out


def test_main_status_dispatches():
    with (
        patch("sys.argv", ["mempalace", "status"]),
        patch("mempalace.cli.cmd_status") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_main_backend_flag_sets_explicit_backend(monkeypatch):
    monkeypatch.delenv("MEMPALACE_BACKEND_EXPLICIT", raising=False)
    monkeypatch.delenv("MEMPALACE_BACKEND", raising=False)
    with (
        patch("sys.argv", ["mempalace", "status", "--backend", "sqlite_exact"]),
        patch("mempalace.cli.cmd_status") as mock_cmd,
    ):
        main()

    mock_cmd.assert_called_once()
    args = mock_cmd.call_args.args[0]
    assert args.backend == "sqlite_exact"
    assert os.environ["MEMPALACE_BACKEND_EXPLICIT"] == "sqlite_exact"
    os.environ.pop("MEMPALACE_BACKEND_EXPLICIT", None)
    os.environ.pop("MEMPALACE_BACKEND", None)


def test_main_backend_flag_accepts_qdrant(monkeypatch):
    monkeypatch.delenv("MEMPALACE_BACKEND_EXPLICIT", raising=False)
    monkeypatch.delenv("MEMPALACE_BACKEND", raising=False)
    with (
        patch("sys.argv", ["mempalace", "status", "--backend", "qdrant"]),
        patch("mempalace.cli.cmd_status") as mock_cmd,
    ):
        main()

    mock_cmd.assert_called_once()
    args = mock_cmd.call_args.args[0]
    assert args.backend == "qdrant"
    assert os.environ["MEMPALACE_BACKEND_EXPLICIT"] == "qdrant"
    os.environ.pop("MEMPALACE_BACKEND_EXPLICIT", None)
    os.environ.pop("MEMPALACE_BACKEND", None)


def test_main_search_dispatches():
    with (
        patch("sys.argv", ["mempalace", "search", "my query"]),
        patch("mempalace.cli.cmd_search") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_main_init_dispatches():
    with (
        patch("sys.argv", ["mempalace", "init", "/some/dir"]),
        patch("mempalace.cli.cmd_init") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_main_mine_dispatches():
    with (
        patch("sys.argv", ["mempalace", "mine", "/some/dir"]),
        patch("mempalace.cli.cmd_mine") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_main_wakeup_dispatches():
    with (
        patch("sys.argv", ["mempalace", "wake-up"]),
        patch("mempalace.cli.cmd_wakeup") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_main_split_dispatches():
    with (
        patch("sys.argv", ["mempalace", "split", "/chats"]),
        patch("mempalace.cli.cmd_split") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_mcp_command_prints_setup_guidance(monkeypatch, capsys):
    monkeypatch.setattr(sys, "argv", ["mempalace", "mcp"])

    main()

    captured = capsys.readouterr()
    assert "MemPalace MCP quick setup:" in captured.out
    assert "claude mcp add mempalace -- mempalace-mcp" in captured.out
    assert "codex mcp add mempalace -- mempalace-mcp" in captured.out
    assert "\nOptional custom palace:\n" in captured.out
    assert "mempalace-mcp --palace /path/to/palace" in captured.out
    assert "[--palace /path/to/palace]" not in captured.out
    assert captured.err == ""


def test_mcp_command_uses_custom_palace_path_when_provided(monkeypatch, capsys):
    monkeypatch.setattr(sys, "argv", ["mempalace", "--palace", "~/tmp/my palace", "mcp"])

    main()

    captured = capsys.readouterr()
    expanded = str(Path("~/tmp/my palace").expanduser())

    assert "mempalace-mcp --palace" in captured.out
    assert expanded in captured.out
    assert "claude mcp add mempalace -- mempalace-mcp --palace" in captured.out
    assert "codex mcp add mempalace -- mempalace-mcp --palace" in captured.out
    assert "Optional custom palace:" not in captured.out
    assert "[--palace /path/to/palace]" not in captured.out
    assert captured.err == ""


def test_mcp_command_includes_backend_when_provided(monkeypatch, capsys):
    monkeypatch.delenv("MEMPALACE_BACKEND_EXPLICIT", raising=False)
    monkeypatch.delenv("MEMPALACE_BACKEND", raising=False)
    monkeypatch.setattr(sys, "argv", ["mempalace", "mcp", "--backend", "sqlite_exact"])

    main()

    captured = capsys.readouterr()
    assert "mempalace-mcp --backend sqlite_exact" in captured.out
    assert captured.err == ""
    os.environ.pop("MEMPALACE_BACKEND_EXPLICIT", None)
    os.environ.pop("MEMPALACE_BACKEND", None)


def test_mcp_command_includes_qdrant_backend(monkeypatch, capsys):
    monkeypatch.delenv("MEMPALACE_BACKEND_EXPLICIT", raising=False)
    monkeypatch.delenv("MEMPALACE_BACKEND", raising=False)
    monkeypatch.setattr(sys, "argv", ["mempalace", "mcp", "--backend", "qdrant"])

    main()

    captured = capsys.readouterr()
    assert "mempalace-mcp --backend qdrant" in captured.out
    assert captured.err == ""
    os.environ.pop("MEMPALACE_BACKEND_EXPLICIT", None)
    os.environ.pop("MEMPALACE_BACKEND", None)


def test_main_hook_no_subcommand_prints_help(capsys):
    with patch("sys.argv", ["mempalace", "hook"]):
        main()
    out = capsys.readouterr().out
    assert "hook" in out.lower() or "run" in out.lower()


def test_main_hook_run_dispatches():
    with (
        patch(
            "sys.argv",
            ["mempalace", "hook", "run", "--hook", "session-start", "--harness", "claude-code"],
        ),
        patch("mempalace.cli.cmd_hook") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_main_hook_run_dispatches_session_end():
    with (
        patch(
            "sys.argv",
            ["mempalace", "hook", "run", "--hook", "session-end", "--harness", "claude-code"],
        ),
        patch("mempalace.cli.cmd_hook") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_main_instructions_no_subcommand_prints_help(capsys):
    with patch("sys.argv", ["mempalace", "instructions"]):
        main()
    out = capsys.readouterr().out
    assert "instructions" in out.lower() or "init" in out.lower()


def test_main_instructions_dispatches():
    with (
        patch("sys.argv", ["mempalace", "instructions", "help"]),
        patch("mempalace.cli.cmd_instructions") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_main_repair_dispatches():
    with (
        patch("sys.argv", ["mempalace", "repair"]),
        patch("mempalace.cli.cmd_repair") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


def test_main_repair_rebuild_index_dispatches():
    with (
        patch("sys.argv", ["mempalace", "repair", "rebuild-index"]),
        patch("mempalace.cli.cmd_repair") as mock_cmd,
    ):
        main()
        args = mock_cmd.call_args.args[0]
        assert args.repair_action == "rebuild-index"


def test_main_compress_dispatches():
    with (
        patch("sys.argv", ["mempalace", "compress"]),
        patch("mempalace.cli.cmd_compress") as mock_cmd,
    ):
        main()
        mock_cmd.assert_called_once()


# ── cmd_repair ─────────────────────────────────────────────────────────


def _mock_backend_for(col=None, new_col=None):
    """Build a mock ChromaBackend whose get_collection/create_collection return *col* / *new_col*."""
    mock_backend = MagicMock()
    if col is not None:
        mock_backend.get_collection.return_value = col
    if new_col is not None:
        mock_backend.create_collection.return_value = new_col
    return mock_backend


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_no_palace(mock_config_cls, tmp_path, capsys):
    mock_config_cls.return_value.palace_path = str(tmp_path / "nonexistent")
    args = argparse.Namespace(palace=None)
    with patch("mempalace.backends.chroma.ChromaBackend"):
        cmd_repair(args)
    out = capsys.readouterr().out
    assert "No palace found" in out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_requires_palace_database(mock_config_cls, tmp_path, capsys):
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    args = argparse.Namespace(palace=None)
    with patch("mempalace.backends.chroma.ChromaBackend"):
        cmd_repair(args)
    out = capsys.readouterr().out
    assert "No palace database found" in out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_error_reading(mock_config_cls, tmp_path, capsys):
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None)
    mock_backend = MagicMock()
    mock_backend.get_collection.side_effect = Exception("corrupt db")
    with patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend):
        cmd_repair(args)
    out = capsys.readouterr().out
    assert "Error reading palace" in out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_error_reading_points_to_from_sqlite_not_remine(
    mock_config_cls, tmp_path, capsys
):
    """When the drawer-index read fails (the chromadb HNSW compactor cannot
    apply WAL logs to the segment), legacy repair must point the user at
    ``repair --mode from-sqlite`` — the rows are intact in chroma.sqlite3 —
    and must NOT advise re-mining from source files, which silently drops
    drawers added via the MCP server and diary entries (#1843)."""
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None)
    mock_col = MagicMock()
    mock_col.count.side_effect = Exception(
        "Error executing plan: Error sending backfill request to compactor: "
        "Failed to apply logs to the hnsw segment writer"
    )
    mock_backend = MagicMock()
    mock_backend.get_collection.return_value = mock_col
    with patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend):
        cmd_repair(args)
    out = capsys.readouterr().out
    assert "mempalace repair --mode from-sqlite --archive-existing" in out
    assert "may need to be re-mined" not in out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_zero_drawers(mock_config_cls, tmp_path, capsys):
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None)
    mock_col = MagicMock()
    mock_col.count.return_value = 0
    mock_backend = _mock_backend_for(col=mock_col)
    with patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend):
        cmd_repair(args)
    out = capsys.readouterr().out
    assert "Nothing to repair" in out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_success(mock_config_cls, tmp_path, capsys):
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None, yes=True)
    mock_col = MagicMock()
    mock_col.count.return_value = 2
    mock_col.get.return_value = {
        "ids": ["id1", "id2"],
        "documents": ["doc1", "doc2"],
        "metadatas": [{"wing": "a"}, {"wing": "b"}],
    }
    mock_temp_col = MagicMock()
    mock_temp_col.count.return_value = 2
    mock_new_col = MagicMock()
    mock_new_col.count.return_value = 2
    mock_backend = _mock_backend_for(col=mock_col, new_col=mock_new_col)
    mock_backend.create_collection.side_effect = [mock_temp_col, mock_new_col]
    with patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend):
        cmd_repair(args)
    out = capsys.readouterr().out
    assert "Repair complete" in out
    assert "2 drawers rebuilt" in out
    assert mock_backend.delete_collection.call_args_list == [
        call(str(palace_dir), "mempalace_drawers__repair_tmp"),
        call(str(palace_dir), "mempalace_drawers"),
        call(str(palace_dir), "mempalace_drawers__repair_tmp"),
    ]
    mock_temp_col.upsert.assert_called_once()
    mock_new_col.upsert.assert_called_once()
    mock_new_col.add.assert_not_called()


@pytest.mark.skipif(
    os.name == "nt" or not hasattr(socket, "AF_UNIX"),
    reason="Unix domain socket files are POSIX-only",
)
@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_survives_a_socket_in_the_palace_directory(
    mock_config_cls, tmp_path, monkeypatch, capsys
):
    """#2207: a leftover daemon socket aborted repair at the backup step.

    ``shutil.copytree`` raises on a Unix domain socket, so the command died
    with a traceback before ``_rebuild_collection_via_temp`` ever ran, and
    re-running only repeated it.
    """
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    # Bound by relative name: the absolute tmp path can exceed the sun_path
    # limit, which is tight on macOS.
    monkeypatch.chdir(palace_dir)
    leftover = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        leftover.bind("mcp.sock")
    finally:
        leftover.close()
    # A link out of the palace, to pin that this backup still DEREFERENCES
    # links. Preserving them instead would put a bare symlink in the backup,
    # and the rebuild that follows can leave it pointing at nothing.
    outside = tmp_path / "outside.json"
    outside.write_text("tunnel payload", encoding="utf-8")
    try:
        (palace_dir / "tunnels.json").symlink_to(outside)
    except OSError as exc:
        # Only a permission refusal is a skip; anything else is a bug here.
        if os.name != "nt" and exc.errno not in (errno.EPERM, errno.EACCES):
            raise
        pytest.skip(f"symlink creation not permitted for this user: {exc}")

    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None, yes=True)
    mock_col = MagicMock()
    mock_col.count.return_value = 2
    mock_col.get.return_value = {
        "ids": ["id1", "id2"],
        "documents": ["doc1", "doc2"],
        "metadatas": [{"wing": "a"}, {"wing": "b"}],
    }
    mock_temp_col = MagicMock()
    mock_temp_col.count.return_value = 2
    mock_new_col = MagicMock()
    mock_new_col.count.return_value = 2
    mock_backend = _mock_backend_for(col=mock_col, new_col=mock_new_col)
    mock_backend.create_collection.side_effect = [mock_temp_col, mock_new_col]

    with patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend):
        cmd_repair(args)

    out = capsys.readouterr().out
    assert "Repair complete" in out
    assert "    mcp.sock (socket)" in out.splitlines()
    backup_dir = tmp_path / "palace.backup"
    assert (backup_dir / "chroma.sqlite3").is_file()
    assert not (backup_dir / "mcp.sock").exists()
    # The socket is skipped, never removed from the live palace.
    assert (palace_dir / "mcp.sock").exists()
    # Dereferenced, so the backup survives losing what the link pointed at.
    backed_up_link = backup_dir / "tunnels.json"
    assert not backed_up_link.is_symlink()
    outside.unlink()
    assert backed_up_link.read_text(encoding="utf-8") == "tunnel payload"


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_uses_configured_collection(mock_config_cls, tmp_path, capsys):
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "custom_drawers"
    args = argparse.Namespace(palace=None, yes=True)
    mock_col = MagicMock()
    mock_col.count.return_value = 2
    mock_col.get.return_value = {
        "ids": ["id1", "id2"],
        "documents": ["doc1", "doc2"],
        "metadatas": [{"wing": "a"}, {"wing": "b"}],
    }
    mock_temp_col = MagicMock()
    mock_temp_col.count.return_value = 2
    mock_new_col = MagicMock()
    mock_new_col.count.return_value = 2
    mock_backend = _mock_backend_for(col=mock_col, new_col=mock_new_col)
    mock_backend.create_collection.side_effect = [mock_temp_col, mock_new_col]

    with patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend):
        cmd_repair(args)

    out = capsys.readouterr().out
    assert "Repair complete" in out
    mock_backend.get_collection.assert_called_once_with(str(palace_dir), "custom_drawers")
    assert mock_backend.create_collection.call_args_list == [
        call(str(palace_dir), "custom_drawers__repair_tmp"),
        call(str(palace_dir), "custom_drawers"),
    ]
    assert mock_backend.delete_collection.call_args_list == [
        call(str(palace_dir), "custom_drawers__repair_tmp"),
        call(str(palace_dir), "custom_drawers"),
        call(str(palace_dir), "custom_drawers__repair_tmp"),
    ]


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_default_mode_dry_run_writes_nothing(mock_config_cls, tmp_path, capsys):
    """``repair --dry-run`` with no --mode must print a plan and leave the palace alone.

    #2095 / #2133 fixed this for ``--mode from-sqlite``; the default (legacy)
    path still ran the backup copy and the full rebuild.
    """
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None, yes=True, dry_run=True)
    mock_col = MagicMock()
    mock_col.count.return_value = 2
    # A concrete extract payload, so that if the dry-run guard ever regresses
    # this test fails on its assertions instead of spinning in _extract_drawers.
    mock_col.get.return_value = {
        "ids": ["id1", "id2"],
        "documents": ["doc1", "doc2"],
        "metadatas": [{"wing": "a"}, {"wing": "b"}],
    }
    mock_backend = _mock_backend_for(col=mock_col)

    with (
        patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend),
        patch("mempalace.repair.sqlite_drawer_count", return_value=2) as mock_count,
        patch("mempalace.migrate.confirm_destructive_action") as mock_confirm,
    ):
        cmd_repair(args)

    out = capsys.readouterr().out
    assert "DRY RUN -- no changes will be made." in out
    assert "holds 2 rows" in out
    assert "Repair complete" not in out
    # The count comes from read-only SQLite, never from a chromadb client:
    # opening one is itself a write to chroma.sqlite3.
    mock_count.assert_called_once_with(str(palace_dir), "mempalace_drawers")
    mock_backend.get_collection.assert_not_called()
    mock_col.get.assert_not_called()
    mock_backend.create_collection.assert_not_called()
    mock_backend.delete_collection.assert_not_called()
    mock_confirm.assert_not_called()
    assert not (tmp_path / "palace.backup").exists()


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_default_mode_dry_run_honours_custom_collection(
    mock_config_cls, tmp_path, capsys
):
    """The preview must target the configured collection, not a hardcoded name."""
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "custom_drawers"
    args = argparse.Namespace(palace=None, yes=True, dry_run=True)

    with (
        patch("mempalace.backends.chroma.ChromaBackend", return_value=_mock_backend_for()),
        patch("mempalace.repair.sqlite_drawer_count", return_value=7) as mock_count,
    ):
        cmd_repair(args)

    assert "[custom_drawers]" in capsys.readouterr().out
    mock_count.assert_called_once_with(str(palace_dir), "custom_drawers")


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_default_mode_dry_run_reports_healable_fts5_and_continues(
    mock_config_cls, tmp_path, capsys
):
    """An isolated FTS5 error is auto-healed by a real run, so the preview must not abort.

    Aborting here would hand the operator the manual ``sqlite3 .recover``
    recipe for a palace ``maybe_autoheal_fts5_index`` repairs by itself (#1596).
    """
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None, yes=True, dry_run=True)

    with (
        patch(
            "mempalace.repair.sqlite_integrity_errors",
            return_value=["malformed inverted index for FTS5 table x"],
        ),
        patch("mempalace.repair.maybe_autoheal_fts5_index") as mock_autoheal,
        patch("mempalace.repair.sqlite_drawer_count", return_value=4),
        patch("mempalace.backends.chroma.ChromaBackend", return_value=_mock_backend_for()),
    ):
        cmd_repair(args)

    out = capsys.readouterr().out
    mock_autoheal.assert_not_called()
    assert "isolated FTS5 inverted-index error" in out
    assert "ABORT" not in out
    assert "holds 4 rows" in out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_default_mode_dry_run_still_aborts_on_broad_corruption(
    mock_config_cls, tmp_path, capsys
):
    """Corruption a real run cannot heal must still abort the preview with exit 1."""
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None, yes=True, dry_run=True)

    with (
        patch(
            "mempalace.repair.sqlite_integrity_errors",
            return_value=["*** in database main *** Page 42 is never used"],
        ),
        patch("mempalace.repair.maybe_autoheal_fts5_index") as mock_autoheal,
        patch("mempalace.backends.chroma.ChromaBackend"),
        pytest.raises(SystemExit) as excinfo,
    ):
        cmd_repair(args)

    assert excinfo.value.code == 1
    mock_autoheal.assert_not_called()
    assert "ABORT" in capsys.readouterr().out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_non_dry_run_still_autoheals_fts5(mock_config_cls, tmp_path, capsys):
    """The real path must keep calling the autoheal — the patch restructured this branch."""
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None, yes=True)
    mock_col = MagicMock()
    mock_col.count.return_value = 0

    with (
        patch(
            "mempalace.repair.sqlite_integrity_errors",
            return_value=["malformed inverted index for FTS5 table x"],
        ),
        patch("mempalace.repair.maybe_autoheal_fts5_index", return_value=[]) as mock_autoheal,
        patch(
            "mempalace.backends.chroma.ChromaBackend",
            return_value=_mock_backend_for(col=mock_col),
        ),
    ):
        cmd_repair(args)

    mock_autoheal.assert_called_once()
    assert "Nothing to repair" in capsys.readouterr().out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_default_mode_dry_run_refuses_to_invent_zero(mock_config_cls, tmp_path, capsys):
    """An unreadable count must refuse AND exit non-zero, like the from-sqlite preview.

    Otherwise ``repair --dry-run && repair --yes`` walks into the destructive
    run after a preview that could not be produced.
    """
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None, yes=True, dry_run=True)
    mock_backend = _mock_backend_for(col=MagicMock())

    with (
        patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend),
        patch("mempalace.repair.sqlite_drawer_count", return_value=None),
        pytest.raises(SystemExit) as excinfo,
    ):
        cmd_repair(args)

    out = capsys.readouterr().out
    assert excinfo.value.code == 1
    assert "Cannot preview [mempalace_drawers]" in out
    assert "refusing to invent zero counts" in out
    assert "would extract" not in out
    mock_backend.get_collection.assert_not_called()


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_restores_backup_on_live_rebuild_failure(mock_config_cls, tmp_path, capsys):
    """When the live swap fails after the delete, recovery must PROMOTE the
    verified temp copy instead of restoring a sqlite-only file backup."""
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None, yes=True)
    mock_col = MagicMock()
    mock_col.count.return_value = 2
    mock_col.get.return_value = {
        "ids": ["id1", "id2"],
        "documents": ["doc1", "doc2"],
        "metadatas": [{"wing": "a"}, {"wing": "b"}],
    }
    mock_temp_col = MagicMock()
    mock_temp_col.count.return_value = 2
    mock_promoted_col = MagicMock()
    mock_promoted_col.count.return_value = 2
    mock_backend = _mock_backend_for(col=mock_col)
    mock_backend.create_collection.side_effect = [
        mock_temp_col,
        RuntimeError("live build failed"),
        mock_promoted_col,
    ]
    with patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend):
        with pytest.raises(SystemExit) as excinfo:
            cmd_repair(args)
    out = capsys.readouterr().out
    assert excinfo.value.code == 1
    assert "Repair failed" in out
    assert "Attempting recovery: promoting verified copy" in out
    assert "Recovery succeeded" in out
    mock_backend.close_palace.assert_called_once_with(str(palace_dir))
    assert mock_backend.delete_collection.call_args_list == [
        call(str(palace_dir), "mempalace_drawers__repair_tmp"),  # pre-clean stale temp
        call(str(palace_dir), "mempalace_drawers"),  # live delete before re-upload
        call(str(palace_dir), "mempalace_drawers"),  # delete broken live before promotion
        call(str(palace_dir), "mempalace_drawers__repair_tmp"),  # temp cleaned after promotion
    ]
    assert mock_promoted_col.upsert.called


def _repair_backend_mocks(mock_config_cls, palace_dir, create_collection_results=None):
    """Config + backend mocks for a 2-drawer legacy repair run.

    ``create_collection_results`` overrides the ``create_collection``
    side_effect sequence; the default is a temp + live collection pair
    that succeeds.
    """
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    mock_col = MagicMock()
    mock_col.count.return_value = 2
    mock_col.get.return_value = {
        "ids": ["id1", "id2"],
        "documents": ["doc1", "doc2"],
        "metadatas": [{"wing": "a"}, {"wing": "b"}],
    }
    if create_collection_results is None:
        mock_temp_col = MagicMock()
        mock_temp_col.count.return_value = 2
        mock_new_col = MagicMock()
        mock_new_col.count.return_value = 2
        create_collection_results = [mock_temp_col, mock_new_col]
    mock_backend = _mock_backend_for(col=mock_col)
    mock_backend.create_collection.side_effect = create_collection_results
    return mock_backend


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_closes_handles_then_rebuilds_fts5(mock_config_cls, tmp_path):
    """cmd_repair must close chroma handles, then run _vacuum_and_rebuild_fts5.

    Mirrors test_rebuild_index_calls_vacuum in test_repair.py: ChromaDB's
    PersistentClient holds an open connection to chroma.sqlite3 and VACUUM
    requires exclusive access, so the handles must be released first. See
    #1747: the legacy path skipped this cleanup entirely.
    """
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    args = argparse.Namespace(palace=None, yes=True)
    mock_backend = _repair_backend_mocks(mock_config_cls, palace_dir)

    call_order = []
    with (
        patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend),
        patch(
            "mempalace.repair._close_chroma_handles",
            side_effect=lambda *a, **kw: call_order.append("close"),
        ) as mock_close,
        patch(
            "mempalace.repair._vacuum_and_rebuild_fts5",
            side_effect=lambda *a, **kw: call_order.append("vacuum"),
        ) as mock_vacuum,
    ):
        cmd_repair(args)

    mock_close.assert_called_once()
    mock_vacuum.assert_called_once()
    assert call_order == ["close", "vacuum"], "handles must be closed before VACUUM"
    vacuum_args, _ = mock_vacuum.call_args
    assert vacuum_args[0] == str(palace_dir)


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_success_rebuilds_fts5_and_vacuums(mock_config_cls, tmp_path, capsys):
    """A clean legacy repair leaves the FTS5 index rebuilt and the file vacuumed.

    Regression test for #1747: cmd_repair printed "Repair complete" without
    ever running _vacuum_and_rebuild_fts5, so the bulk delete + re-upsert
    cycle left the FTS5 inverted index inconsistent and the next repair run
    aborted at the integrity preflight. The two banners are the user-visible
    contract that the cleanup ran.
    """
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    with closing(sqlite3.connect(str(palace_dir / "chroma.sqlite3"))) as conn:
        conn.execute(
            "CREATE VIRTUAL TABLE embedding_fulltext_search"
            " USING fts5(string_value, tokenize='unicode61')"
        )
        conn.execute("INSERT INTO embedding_fulltext_search(string_value) VALUES('hello world')")
        conn.commit()
    args = argparse.Namespace(palace=None, yes=True)
    mock_backend = _repair_backend_mocks(mock_config_cls, palace_dir)

    with patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend):
        cmd_repair(args)

    out = capsys.readouterr().out
    assert "Repair complete" in out
    assert "FTS5 index rebuilt." in out
    assert "SQLite VACUUM complete." in out
    assert "post-repair cleanup failed" not in out
    with closing(sqlite3.connect(str(palace_dir / "chroma.sqlite3"))) as conn:
        result = conn.execute("PRAGMA quick_check").fetchall()
    assert result == [("ok",)]


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_does_not_vacuum_when_rebuild_fails(mock_config_cls, tmp_path, capsys):
    """Post-run FTS5 cleanup must not fire when the rebuild itself failed."""
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    args = argparse.Namespace(palace=None, yes=True)
    mock_temp_col = MagicMock()
    mock_temp_col.count.return_value = 2
    mock_backend = _repair_backend_mocks(
        mock_config_cls,
        palace_dir,
        create_collection_results=[mock_temp_col, RuntimeError("live build failed")],
    )

    with (
        patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend),
        patch("mempalace.repair._vacuum_and_rebuild_fts5") as mock_vacuum,
    ):
        with pytest.raises(SystemExit) as excinfo:
            cmd_repair(args)

    assert "Repair failed" in capsys.readouterr().out
    assert excinfo.value.code == 1
    mock_vacuum.assert_not_called()


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_aborts_without_confirmation(mock_config_cls, tmp_path, capsys):
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    sqlite3.connect(str(palace_dir / "chroma.sqlite3")).close()
    mock_config_cls.return_value.palace_path = str(palace_dir)
    mock_config_cls.return_value.collection_name = "mempalace_drawers"
    args = argparse.Namespace(palace=None)
    mock_col = MagicMock()
    mock_col.count.return_value = 1
    mock_backend = _mock_backend_for(col=mock_col)
    with (
        patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend),
        patch("builtins.input", return_value="n"),
    ):
        cmd_repair(args)
    out = capsys.readouterr().out
    assert "Aborted." in out
    mock_backend.create_collection.assert_not_called()


# ── cmd_compress ───────────────────────────────────────────────────────


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_sync_no_palace_dir(mock_config_cls, tmp_path, capsys):
    """cmd_sync on a missing palace dir prints the State A message (#1498)."""
    from mempalace.cli import cmd_sync

    palace_path = tmp_path / "nonexistent"
    mock_config_cls.return_value.palace_path = str(palace_path)
    args = argparse.Namespace(palace=None, dir=None, root=[], wing=None, dry_run=False)
    cmd_sync(args)
    captured = capsys.readouterr()
    assert "No palace found" in captured.out + captured.err


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_sync_palace_dir_no_db(mock_config_cls, tmp_path, capsys):
    """cmd_sync on a palace dir without chroma.sqlite3 prints the State B
    message and does NOT trigger chromadb's lazy DB creation (#1498)."""
    from mempalace.cli import cmd_sync

    mock_config_cls.return_value.palace_path = str(tmp_path)
    args = argparse.Namespace(palace=None, dir=None, root=[], wing=None, dry_run=False)
    cmd_sync(args)
    captured = capsys.readouterr()
    assert "has no chroma.sqlite3 yet" in captured.out + captured.err
    # Side-effect-free: backend not invoked.
    assert list(tmp_path.iterdir()) == []


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_sync_daemon_background_submits_job(mock_config_cls, capsys):
    from mempalace.cli import cmd_sync

    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(
        palace=None,
        dir="/project",
        root=["/extra"],
        wing="wing_a",
        dry_run=False,
        daemon=True,
        background=True,
        backend=None,
        global_backend=None,
    )
    with patch("mempalace.daemon.submit_job", return_value={"id": "sync-job"}) as mock_submit:
        cmd_sync(args)

    mock_submit.assert_called_once()
    assert mock_submit.call_args.args[0] == "sync"
    payload = mock_submit.call_args.args[1]
    assert payload == {"dir": "/project", "root": ["/extra"], "wing": "wing_a", "dry_run": False}
    assert mock_submit.call_args.kwargs["wait"] is False
    assert "sync-job" in capsys.readouterr().out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_daemon_jobs_reads_durable_queue_when_stopped(
    mock_config_cls, tmp_path, monkeypatch, capsys
):
    from mempalace.daemon import QueueStore, queue_path

    palace_dir = tmp_path / "palace"
    state_root = tmp_path / "state"
    palace_dir.mkdir()
    monkeypatch.setenv("MEMPALACE_DAEMON_STATE_ROOT", str(state_root))
    mock_config_cls.return_value.palace_path = str(palace_dir)
    job = QueueStore(queue_path(str(palace_dir))).enqueue("mine", {"source": "/src"})

    args = argparse.Namespace(
        palace=None,
        backend=None,
        global_backend=None,
        daemon_action="jobs",
        limit=20,
    )
    with patch("mempalace.daemon.get_client_if_running", return_value=None):
        cmd_daemon(args)

    out = capsys.readouterr().out
    assert job.id in out
    assert "queued" in out
    assert "mine" in out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_daemon_wait_reads_finished_job_when_stopped(
    mock_config_cls, tmp_path, monkeypatch, capsys
):
    from mempalace.daemon import QueueStore, queue_path

    palace_dir = tmp_path / "palace"
    state_root = tmp_path / "state"
    palace_dir.mkdir()
    monkeypatch.setenv("MEMPALACE_DAEMON_STATE_ROOT", str(state_root))
    mock_config_cls.return_value.palace_path = str(palace_dir)
    store = QueueStore(queue_path(str(palace_dir)))
    queued = store.enqueue("mine", {"source": "/src"})
    store.finish(
        queued.id,
        state="succeeded",
        result={"success": True, "stdout": "done\n", "exit_code": 0},
    )

    args = argparse.Namespace(
        palace=None,
        backend=None,
        global_backend=None,
        daemon_action="wait",
        job_id=queued.id,
    )
    with patch("mempalace.daemon.get_client_if_running", return_value=None):
        cmd_daemon(args)

    assert "done" in capsys.readouterr().out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_compress_no_palace(mock_config_cls, tmp_path, capsys):
    """cmd_compress exits non-zero with a 'No palace found' message on a missing dir.

    Uses a real non-existent tmp_path so the stratified state helper (#1498)
    walks the State A branch instead of hitting the chromadb backend.
    """
    mock_config_cls.return_value.palace_path = str(tmp_path / "nonexistent")
    args = argparse.Namespace(palace=None, wing=None, dry_run=False, config=None)
    with pytest.raises(SystemExit):
        cmd_compress(args)
    assert "No palace found" in capsys.readouterr().out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_compress_no_drawers(mock_config_cls, capsys):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(palace=None, wing="mywing", dry_run=False, config=None)
    mock_col = MagicMock()
    mock_col.get.return_value = {"documents": [], "metadatas": [], "ids": []}
    mock_backend = _mock_backend_for(col=mock_col)
    with (
        patch("mempalace.palace._open_collection_or_explain", return_value=mock_col),
        patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend),
    ):
        cmd_compress(args)
    out = capsys.readouterr().out
    assert "No drawers found" in out


def _make_mock_dialect_module(dialect_instance):
    """Create a mock dialect module with a Dialect class that returns the given instance."""
    mock_mod = MagicMock()
    mock_mod.Dialect.return_value = dialect_instance
    mock_mod.Dialect.from_config.return_value = dialect_instance
    mock_mod.Dialect.count_tokens = MagicMock(side_effect=lambda x: len(x) // 4)
    return mock_mod


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_compress_dry_run(mock_config_cls, capsys):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(palace=None, wing=None, dry_run=True, config=None)
    mock_col = MagicMock()
    mock_col.get.side_effect = [
        {
            "documents": ["some long text here for testing"],
            "metadatas": [{"wing": "test", "room": "general", "source_file": "test.txt"}],
            "ids": ["id1"],
        },
        {"documents": [], "metadatas": [], "ids": []},
    ]
    mock_backend = _mock_backend_for(col=mock_col)

    mock_dialect = MagicMock()
    mock_dialect.compress.return_value = "compressed"
    mock_dialect.compression_stats.return_value = {
        "original_chars": 100,
        "summary_chars": 30,
        "original_tokens_est": 25,
        "summary_tokens_est": 8,
        "size_ratio": 3.3,
        "note": "Estimates only.",
    }
    mock_dialect_mod = _make_mock_dialect_module(mock_dialect)

    with (
        patch("mempalace.palace._open_collection_or_explain", return_value=mock_col),
        patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend),
        patch.dict("sys.modules", {"mempalace.dialect": mock_dialect_mod}),
    ):
        cmd_compress(args)
    out = capsys.readouterr().out
    assert "dry run" in out.lower()
    assert "Compressing" in out
    assert "Total:" in out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_compress_with_config(mock_config_cls, tmp_path, capsys):
    mock_config_cls.return_value.palace_path = "/fake/palace"
    config_file = tmp_path / "entities.json"
    config_file.write_text('{"people": [], "projects": []}')
    args = argparse.Namespace(palace=None, wing=None, dry_run=True, config=str(config_file))
    mock_col = MagicMock()
    mock_col.get.return_value = {"documents": [], "metadatas": [], "ids": []}
    mock_backend = _mock_backend_for(col=mock_col)

    mock_dialect = MagicMock()
    mock_dialect_mod = _make_mock_dialect_module(mock_dialect)

    with (
        patch("mempalace.palace._open_collection_or_explain", return_value=mock_col),
        patch("mempalace.backends.chroma.ChromaBackend", return_value=mock_backend),
        patch.dict("sys.modules", {"mempalace.dialect": mock_dialect_mod}),
    ):
        cmd_compress(args)
    out = capsys.readouterr().out
    assert "Loaded entity config" in out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_compress_stores_results(mock_config_cls, capsys):
    """Non-dry-run compress stores to mempalace_closets collection (#1244)."""
    mock_config_cls.return_value.palace_path = "/fake/palace"
    args = argparse.Namespace(palace=None, wing=None, dry_run=False, config=None)
    mock_col = MagicMock()
    mock_col.get.side_effect = [
        {
            "documents": ["text"],
            "metadatas": [{"wing": "w", "room": "r", "source_file": "f.txt"}],
            "ids": ["id1"],
        },
        {"documents": [], "metadatas": [], "ids": []},
    ]
    mock_comp_col = MagicMock()
    mock_backend = MagicMock()
    mock_backend.get_collection.return_value = mock_col
    mock_backend.get_or_create_collection.return_value = mock_comp_col

    mock_dialect = MagicMock()
    mock_dialect.compress.return_value = "compressed"
    mock_dialect.compression_stats.return_value = {
        "original_chars": 100,
        "summary_chars": 30,
        "original_tokens_est": 25,
        "summary_tokens_est": 8,
        "size_ratio": 3.3,
        "note": "Estimates only.",
    }
    mock_dialect_mod = _make_mock_dialect_module(mock_dialect)

    with (
        patch("mempalace.palace._open_collection_or_explain", return_value=mock_col),
        patch("mempalace.palace.get_closets_collection", return_value=mock_comp_col),
        patch.dict("sys.modules", {"mempalace.dialect": mock_dialect_mod}),
    ):
        cmd_compress(args)
    out = capsys.readouterr().out
    assert "Stored" in out
    assert "Total:" in out
    mock_comp_col.upsert.assert_called_once()
    assert "mempalace_closets" in out


def test_cmd_compress_output_readable_via_get_closets_collection(tmp_path, capsys):
    """End-to-end: cmd_compress output must be readable via the same code
    path palace.py uses (`get_closets_collection`). Regression for #1244."""
    from mempalace.backends.chroma import ChromaBackend
    from mempalace.palace import get_closets_collection, get_collection

    palace_path = str(tmp_path / "palace")

    # Seed a drawer in the palace so cmd_compress has something to compress.
    drawers = get_collection(palace_path, "mempalace_drawers", create=True)
    drawers.upsert(
        ids=["drawer-1"],
        documents=["The quick brown fox jumps over the lazy dog."],
        metadatas=[{"wing": "test", "room": "demo", "source_file": "fox.txt"}],
    )

    args = argparse.Namespace(palace=palace_path, wing=None, dry_run=False, config=None)
    with patch("mempalace.cli.MempalaceConfig") as mock_config_cls:
        mock_config_cls.return_value.palace_path = palace_path
        # Use a real ChromaBackend so the write actually lands on disk and
        # the read-side helper can find it.
        with patch("mempalace.backends.chroma.ChromaBackend", side_effect=ChromaBackend):
            cmd_compress(args)

    out = capsys.readouterr().out
    assert "Stored" in out

    # Now read via the *same* code path palace.py / searcher uses.
    closets = get_closets_collection(palace_path, create=False)
    got = closets.get(ids=["drawer-1"], include=["documents", "metadatas"])
    assert got["ids"] == ["drawer-1"], (
        "compressed drawer not found in mempalace_closets — "
        "cmd_compress wrote to the wrong collection (#1244)"
    )
    assert got["documents"] and got["documents"][0], "empty compressed doc"
    meta = got["metadatas"][0]
    assert meta.get("wing") == "test"
    assert "compression_ratio" in meta


def test_cmd_repair_trailing_slash_does_not_recurse():
    """Repair with trailing slash should put backup outside palace dir (#395)."""
    import os

    args = argparse.Namespace(palace="/tmp/fake_palace/")
    with patch("mempalace.cli.os.path.isdir", return_value=False):
        cmd_repair(args)
    # Verify the rstrip logic: palace_path should not end with separator
    palace_path = os.path.expanduser(args.palace).rstrip(os.sep)
    backup_path = palace_path + ".backup"
    assert not backup_path.startswith(palace_path + os.sep)


# ── stdio reconfigure on Windows ─────────────────────────────────────


class _ReconfigurableStringIO:
    def __init__(self):
        self.reconfigure_calls = []

    def reconfigure(self, **kwargs):
        self.reconfigure_calls.append(kwargs)


def test_reconfigures_stdio_to_utf8_on_windows():
    """Windows `mempalace` CLI must decode/encode stdio as UTF-8.

    Without this, piped non-ASCII input (`mempalace search ... < q.txt`)
    or piped non-ASCII output (`mempalace search "..." > out.txt`) is
    mojibaked through the system ANSI codepage on non-Latin Windows
    locales (cp1252/cp1251/cp950).
    """
    from mempalace.cli import _reconfigure_stdio_utf8_on_windows

    stdin = _ReconfigurableStringIO()
    stdout = _ReconfigurableStringIO()
    stderr = _ReconfigurableStringIO()
    with (
        patch.object(sys, "platform", "win32"),
        patch.object(sys, "stdin", stdin),
        patch.object(sys, "stdout", stdout),
        patch.object(sys, "stderr", stderr),
    ):
        _reconfigure_stdio_utf8_on_windows()

    # Per-stream errors policy: stdin survives bad bytes via
    # surrogateescape so a redirected non-UTF-8 file does not crash
    # the read; stdout/stderr use replace so a drawer carrying a
    # round-tripped surrogate half does not crash mid-print.
    assert stdin.reconfigure_calls == [{"encoding": "utf-8", "errors": "surrogateescape"}]
    assert stdout.reconfigure_calls == [{"encoding": "utf-8", "errors": "replace"}]
    assert stderr.reconfigure_calls == [{"encoding": "utf-8", "errors": "replace"}]


def test_reconfigure_stdio_is_noop_off_windows():
    """Linux/macOS already default to UTF-8 stdio -- helper must not touch streams."""
    from mempalace.cli import _reconfigure_stdio_utf8_on_windows

    stdin = _ReconfigurableStringIO()
    with (
        patch.object(sys, "platform", "linux"),
        patch.object(sys, "stdin", stdin),
    ):
        _reconfigure_stdio_utf8_on_windows()

    assert stdin.reconfigure_calls == []


# ── cmd_repair: from-sqlite mode exit codes ──────────────────────────


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_from_sqlite_validation_refusal_exits_nonzero(mock_config_cls, tmp_path, capsys):
    """When ``rebuild_from_sqlite`` returns ``{}`` for a validation
    refusal (missing source DB, in-place without --archive-existing,
    refusing to overwrite an existing dest), the CLI must surface a
    non-zero exit so unattended scripts and CI distinguish "invalid
    inputs" from "successful recovery that found zero rows."

    Catches: a regression where the CLI treats the validation-refusal
    sentinel as success, leaving CI green on a no-op repair that should
    have alerted an operator.
    """
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    mock_config_cls.return_value.palace_path = str(palace_dir)

    args = argparse.Namespace(
        palace=str(palace_dir),
        mode="from-sqlite",
        source=None,
        archive_existing=False,
        yes=True,
    )
    with patch("mempalace.repair.rebuild_from_sqlite", return_value={}):
        with pytest.raises(SystemExit) as excinfo:
            cmd_repair(args)
    assert excinfo.value.code == 1


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_from_sqlite_success_does_not_exit(mock_config_cls, tmp_path):
    """A successful from-sqlite rebuild — even one that finds zero rows
    in a legitimately empty source palace — must NOT call ``sys.exit``.
    A populated counts dict (with ``0`` values) is the success signal;
    only the empty dict ``{}`` is reserved for validation refusal.

    Catches: a regression where ``if not counts`` is replaced by
    ``if not sum(counts.values())`` or similar, conflating "empty source"
    with "validation refused" and breaking idempotent recovery scripts.
    """
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    mock_config_cls.return_value.palace_path = str(palace_dir)

    args = argparse.Namespace(
        palace=str(palace_dir),
        mode="from-sqlite",
        source=None,
        archive_existing=False,
        yes=True,
    )
    # Zero rows but per-collection keys present → success, no exit.
    fake_counts = {"mempalace_drawers": 0, "mempalace_closets": 0}
    with patch("mempalace.repair.rebuild_from_sqlite", return_value=fake_counts):
        # Should return cleanly; no SystemExit raised.
        cmd_repair(args)


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_from_sqlite_dry_run_passes_through_and_skips_confirm(mock_config_cls, tmp_path):
    """``repair --mode from-sqlite --dry-run`` must forward dry_run and skip
    the destructive confirmation prompt (#2095, #2133)."""
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    mock_config_cls.return_value.palace_path = str(palace_dir)

    args = argparse.Namespace(
        palace=str(palace_dir),
        mode="from-sqlite",
        source=None,
        archive_existing=False,
        yes=False,
        dry_run=True,
    )
    with (
        patch(
            "mempalace.repair.rebuild_from_sqlite",
            return_value={"mempalace_drawers": 0, "mempalace_closets": 0},
        ) as mock_rebuild,
        patch("mempalace.migrate.confirm_destructive_action") as mock_confirm,
    ):
        cmd_repair(args)

    mock_confirm.assert_not_called()
    assert mock_rebuild.call_args.kwargs["dry_run"] is True


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_from_sqlite_cleanup_failure_exits_nonzero(mock_config_cls, tmp_path, capsys):
    from mempalace.repair import RebuildCleanupError

    palace_dir = tmp_path / "palace"
    source_dir = tmp_path / "source"
    mock_config_cls.return_value.palace_path = str(palace_dir)
    args = argparse.Namespace(
        palace=str(palace_dir),
        mode="from-sqlite",
        source=str(source_dir),
        archive_existing=False,
        yes=True,
    )
    failure = RebuildCleanupError(
        "cleanup failed",
        counts={"mempalace_drawers": 1},
        dest_palace=str(palace_dir),
        archive_path=None,
    )
    with patch("mempalace.repair.rebuild_from_sqlite", side_effect=failure):
        with pytest.raises(SystemExit) as excinfo:
            cmd_repair(args)

    assert excinfo.value.code == 1
    assert "Rebuild cleanup failed" in capsys.readouterr().out


@patch("mempalace.cli.MempalaceConfig")
def test_cmd_repair_rebuild_index_alias_uses_sqlite_archive(mock_config_cls, tmp_path):
    """``repair rebuild-index`` must bypass Chroma reads and rebuild from SQLite."""
    palace_dir = tmp_path / "palace"
    palace_dir.mkdir()
    mock_config_cls.return_value.palace_path = str(palace_dir)

    args = argparse.Namespace(
        palace=str(palace_dir),
        repair_action="rebuild-index",
        mode="legacy",
        source=None,
        archive_existing=False,
        yes=True,
        dry_run=False,
    )
    fake_counts = {"mempalace_drawers": 1, "mempalace_closets": 0}
    with patch("mempalace.repair.rebuild_from_sqlite", return_value=fake_counts) as rebuild:
        cmd_repair(args)

    rebuild.assert_called_once_with(
        source_palace=str(palace_dir),
        dest_palace=str(palace_dir),
        archive_existing_dest=True,
        dry_run=False,
    )