"""Comprehensive hardening tests for ``muse stash``.

Covers all changes introduced in the stash command review:

Unit
----
- Parser flags: -m/--message, index arg for pop/drop/show/apply
- Dead-code removal: _read_branch absent
- assert_not_symlink in _load_stash
- fsync in _save_stash
- _resolve_index validates bounds and non-integer input
- _verify_manifest_objects catches missing objects

Integration
-----------
- Error messages routed to stderr, stdout clean
- JSON schema stable and complete across all subcommands
- push: status/files_count/message fields present
- pop: status/stash_size_after/message fields present
- drop: snapshot_id/branch/index fields present
- list: files_count/message fields present per entry
- show: files list present
- apply: stash entry preserved after apply
- nothing_to_stash: complete JSON schema with null fields
- -m/--message annotation stored and retrieved
- pop/drop by index works correctly
- show by index works correctly
- Object-store integrity check blocks pop with missing objects

End-to-end
----------
- Round-trip: push → pop restores original state
- Round-trip: push → apply → apply (idempotent) preserves stash
- Round-trip: push → drop removes entry
- Stash stack ordering (LIFO)
- Stash persists across multiple commands

Security
--------
- ANSI in branch name sanitized in text output
- ANSI in message sanitized in text output
- ANSI in file paths sanitized in show output
- Symlink at stash.json returns empty stack (not traversed)
- --format invalid value exits 1 to stderr

Stress
------
- 100 sequential push/pop cycles
- Stash stack with 50 entries then drop all
- Concurrent stash operations to isolated repos
"""

from __future__ import annotations

import argparse
import inspect
import json
import os
import pathlib
import tempfile
import threading
import time
from typing import TYPE_CHECKING

import pytest

if TYPE_CHECKING:
    from muse.cli.commands.stash import StashEntry

from tests.cli_test_helper import CliRunner

cli = None  # argparse migration — CliRunner ignores this arg
runner = CliRunner()


# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------

def _env(root: pathlib.Path) -> Manifest:
    return {"MUSE_REPO_ROOT": str(root)}


@pytest.fixture()
def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
    """Fresh repo with one committed file (a.py) and one dirty file (b.py)."""
    monkeypatch.chdir(tmp_path)
    monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
    r = runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
    assert r.exit_code == 0, r.output
    (tmp_path / "a.py").write_text("x = 1\n")
    r = runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
    assert r.exit_code == 0, r.output
    (tmp_path / "b.py").write_text("y = 2\n")
    return tmp_path


@pytest.fixture()
def stashed_repo(repo: pathlib.Path) -> pathlib.Path:
    """repo fixture with one stash entry already pushed."""
    r = runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
    assert r.exit_code == 0, r.output
    return repo


# ---------------------------------------------------------------------------
# Unit — parser flags and dead-code removal
# ---------------------------------------------------------------------------

class TestRegisterFlags:
    def _parse(self, *args: str) -> argparse.Namespace:
        import muse.cli.commands.stash as m
        p = argparse.ArgumentParser()
        sub = p.add_subparsers()
        m.register(sub)
        return p.parse_args(["stash", *args])

    def test_message_short(self) -> None:
        ns = self._parse("-m", "wip")
        assert ns.message == "wip"

    def test_message_long(self) -> None:
        ns = self._parse("--message", "my work")
        assert ns.message == "my work"

    def test_message_default_none(self) -> None:
        ns = self._parse()
        assert ns.message is None

    def test_format_json_shorthand(self) -> None:
        ns = self._parse("--json")
        assert ns.fmt == "json"

    def test_pop_index_arg(self) -> None:
        ns = self._parse("pop", "2")
        assert ns.index == "2"

    def test_pop_index_default_none(self) -> None:
        ns = self._parse("pop")
        assert ns.index is None

    def test_drop_index_arg(self) -> None:
        ns = self._parse("drop", "1")
        assert ns.index == "1"

    def test_show_index_arg(self) -> None:
        ns = self._parse("show", "0")
        assert ns.index == "0"

    def test_apply_index_arg(self) -> None:
        ns = self._parse("apply", "0")
        assert ns.index == "0"


class TestDeadCodeRemoval:
    def test_no_read_branch_wrapper(self) -> None:
        import muse.cli.commands.stash as m
        assert not hasattr(m, "_read_branch"), "_read_branch must be deleted"

    def test_assert_not_symlink_in_load_stash(self) -> None:
        import muse.cli.commands.stash as m
        assert "assert_not_symlink" in inspect.getsource(m._load_stash)

    def test_fsync_in_save_stash(self) -> None:
        import muse.cli.commands.stash as m
        assert "fsync" in inspect.getsource(m._save_stash)

    def test_verify_manifest_objects_exists(self) -> None:
        import muse.cli.commands.stash as m
        assert hasattr(m, "_verify_manifest_objects")


class TestResolveIndex:
    def _entries(self, n: int) -> list["StashEntry"]:
        from muse.cli.commands.stash import StashEntry
        return [
            StashEntry(
                snapshot_id="a" * 64, manifest={}, branch="main",
                stashed_at="2026-01-01T00:00:00+00:00", message=None,
            )
            for _ in range(n)
        ]

    def test_default_returns_0(self) -> None:
        from muse.cli.commands.stash import _resolve_index
        assert _resolve_index(self._entries(3), None) == 0

    def test_explicit_index(self) -> None:
        from muse.cli.commands.stash import _resolve_index
        assert _resolve_index(self._entries(3), "2") == 2

    def test_out_of_range_raises(self) -> None:
        from muse.cli.commands.stash import _resolve_index
        with pytest.raises(ValueError, match="out of range"):
            _resolve_index(self._entries(2), "5")

    def test_negative_raises(self) -> None:
        from muse.cli.commands.stash import _resolve_index
        with pytest.raises(ValueError):
            _resolve_index(self._entries(2), "-1")

    def test_non_integer_raises(self) -> None:
        from muse.cli.commands.stash import _resolve_index
        with pytest.raises(ValueError, match="integer"):
            _resolve_index(self._entries(2), "abc")


# ---------------------------------------------------------------------------
# Integration — JSON schema completeness
# ---------------------------------------------------------------------------

class TestPushJsonSchema:
    _REQUIRED = {"status", "snapshot_id", "branch", "stashed_at", "message", "files_count", "stash_size"}

    def test_stash_schema_complete(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert self._REQUIRED <= d.keys()

    def test_status_is_stashed(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["status"] == "stashed"

    def test_files_count_positive(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["files_count"] > 0

    def test_message_default_null(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["message"] is None

    def test_message_with_flag(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "-m", "my note", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["message"] == "my note"

    def test_nothing_to_stash_schema_complete(self, repo: pathlib.Path) -> None:
        """nothing_to_stash must emit same keys with null snapshot_id."""
        # Pop any prior stash so workdir is clean
        runner.invoke(cli, ["stash", "drop"], env=_env(repo))
        # Make workdir match HEAD exactly
        (repo / "b.py").unlink(missing_ok=True)
        r = runner.invoke(cli, ["stash", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert self._REQUIRED <= d.keys()
        assert d["status"] == "nothing_to_stash"
        assert d["snapshot_id"] is None


class TestPopJsonSchema:
    _REQUIRED = {"status", "snapshot_id", "branch", "stashed_at", "message", "files_count", "stash_size_after"}

    def test_pop_schema_complete(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert self._REQUIRED <= d.keys()

    def test_pop_status_is_popped(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["status"] == "popped"

    def test_pop_stash_size_after_decremented(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["stash_size_after"] == 0

    def test_pop_empty_to_stderr(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop"], env=_env(repo))
        assert r.exit_code != 0
        assert "no stash" in (r.stderr or "").lower()


class TestDropJsonSchema:
    _REQUIRED = {"status", "index", "snapshot_id", "branch", "stash_size"}

    def test_drop_schema_complete(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert self._REQUIRED <= d.keys()

    def test_drop_status_is_dropped(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["status"] == "dropped"

    def test_drop_snapshot_id_is_sha256(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert isinstance(d["snapshot_id"], str) and len(d["snapshot_id"]) == 64

    def test_drop_empty_to_stderr(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop"], env=_env(repo))
        assert r.exit_code != 0
        assert "no stash" in (r.stderr or "").lower()


class TestListJsonSchema:
    _ENTRY_REQUIRED = {"index", "snapshot_id", "branch", "stashed_at", "message", "files_count"}

    def test_list_schema_complete(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        entries = json.loads(r.output)
        assert len(entries) == 1
        assert self._ENTRY_REQUIRED <= entries[0].keys()

    def test_list_files_count_positive(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        entries = json.loads(r.output)
        assert entries[0]["files_count"] > 0

    def test_list_message_field_present(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        entries = json.loads(r.output)
        assert "message" in entries[0]

    def test_list_empty_returns_empty_array(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0
        assert json.loads(r.output) == []

    def test_list_with_message(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "annotated"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        entries = json.loads(r.output)
        assert entries[0]["message"] == "annotated"


class TestShowJsonSchema:
    _REQUIRED = {"index", "snapshot_id", "branch", "stashed_at", "message", "files"}

    def test_show_schema_complete(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert self._REQUIRED <= d.keys()

    def test_show_files_is_list_of_strings(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert isinstance(d["files"], list)
        assert all(isinstance(f, str) for f in d["files"])

    def test_show_empty_to_stderr(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show"], env=_env(repo))
        assert r.exit_code != 0
        assert "no stash" in (r.stderr or "").lower()


class TestApply:
    def test_apply_preserves_stash_entry(self, stashed_repo: pathlib.Path) -> None:
        before = json.loads(
            runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False).output
        )
        runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        after = json.loads(
            runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False).output
        )
        assert len(before) == len(after), "apply must not remove the stash entry"

    def test_apply_restores_workdir(self, stashed_repo: pathlib.Path) -> None:
        b_py = stashed_repo / "b.py"
        # After stash push, b.py was removed from workdir
        assert not b_py.exists()
        runner.invoke(cli, ["stash", "apply"], env=_env(stashed_repo), catch_exceptions=False)
        assert b_py.exists(), "apply must restore stashed files to workdir"

    def test_apply_json_status(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["status"] == "applied"
        assert d["stash_size"] == 1

    def test_apply_empty_to_stderr(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply"], env=_env(repo))
        assert r.exit_code != 0
        assert "no stash" in (r.stderr or "").lower()


class TestIndexArg:
    def _push_n(self, repo: pathlib.Path, n: int) -> list[str] :
        """Push n distinct stash entries, return their snapshot_ids."""
        ids: list[str] = []
        for i in range(n):
            (repo / f"w{i}.py").write_text(f"data {i}\n")
            r = runner.invoke(cli, ["stash", "--json"], env=_env(repo), catch_exceptions=False)
            ids.append(json.loads(r.output)["snapshot_id"])
        return ids

    def test_pop_by_index(self, repo: pathlib.Path) -> None:
        ids = self._push_n(repo, 3)
        # Stash is LIFO: index 0 = newest (ids[2]), index 2 = oldest (ids[0])
        r = runner.invoke(cli, ["stash", "pop", "2", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert d["snapshot_id"] == ids[0]  # oldest = index 2

    def test_drop_by_index(self, repo: pathlib.Path) -> None:
        ids = self._push_n(repo, 3)
        r = runner.invoke(cli, ["stash", "drop", "1", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert d["index"] == 1

    def test_show_by_index(self, repo: pathlib.Path) -> None:
        self._push_n(repo, 3)
        r = runner.invoke(cli, ["stash", "show", "1", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert d["index"] == 1

    def test_out_of_range_to_stderr(self, repo: pathlib.Path) -> None:
        self._push_n(repo, 2)
        r = runner.invoke(cli, ["stash", "pop", "99"], env=_env(repo))
        assert r.exit_code != 0
        assert "out of range" in (r.stderr or "").lower()

    def test_non_integer_to_stderr(self, repo: pathlib.Path) -> None:
        self._push_n(repo, 1)
        r = runner.invoke(cli, ["stash", "pop", "abc"], env=_env(repo))
        assert r.exit_code != 0
        assert "integer" in (r.stderr or "").lower()


class TestObjectIntegrity:
    def test_pop_with_missing_object_exits_internal_error(
        self, stashed_repo: pathlib.Path
    ) -> None:
        """pop must refuse to restore if object store objects are missing."""
        stash_data = json.loads((stashed_repo / ".muse/stash.json").read_text())
        manifest = stash_data[0]["delta"]
        # Delete one object from the object store to simulate corruption
        for obj_id in list(manifest.values())[:1]:
            obj_path = stashed_repo / ".muse" / "objects" / obj_id[:2] / obj_id[2:]
            if obj_path.exists():
                obj_path.unlink()
                break

        r = runner.invoke(cli, ["stash", "pop"], env=_env(stashed_repo))
        assert r.exit_code == 3, (
            f"Expected INTERNAL_ERROR (3), got {r.exit_code}: {r.output.strip()}"
        )

    def test_apply_with_missing_object_exits_internal_error(
        self, stashed_repo: pathlib.Path
    ) -> None:
        stash_data = json.loads((stashed_repo / ".muse/stash.json").read_text())
        manifest = stash_data[0]["delta"]
        for obj_id in list(manifest.values())[:1]:
            obj_path = stashed_repo / ".muse" / "objects" / obj_id[:2] / obj_id[2:]
            if obj_path.exists():
                obj_path.unlink()
                break

        r = runner.invoke(cli, ["stash", "apply"], env=_env(stashed_repo))
        assert r.exit_code == 3


# ---------------------------------------------------------------------------
# End-to-end — round-trips
# ---------------------------------------------------------------------------

class TestRoundTrips:
    def test_push_pop_restores_state(self, repo: pathlib.Path) -> None:
        b_content = (repo / "b.py").read_text()
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        assert not (repo / "b.py").exists()
        runner.invoke(cli, ["stash", "pop"], env=_env(repo), catch_exceptions=False)
        assert (repo / "b.py").exists()
        assert (repo / "b.py").read_text() == b_content

    def test_push_apply_apply_idempotent(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        runner.invoke(cli, ["stash", "apply"], env=_env(repo), catch_exceptions=False)
        r1 = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        runner.invoke(cli, ["stash", "apply"], env=_env(repo), catch_exceptions=False)
        r2 = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        assert len(r1) == len(r2), "apply is idempotent with respect to stash size"

    def test_push_drop_removes_entry(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        runner.invoke(cli, ["stash", "drop"], env=_env(repo), catch_exceptions=False)
        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        assert entries == []

    def test_stash_stack_is_lifo(self, repo: pathlib.Path) -> None:
        """The most-recently-pushed stash must be at index 0."""
        (repo / "b.py").write_text("first\n")
        runner.invoke(cli, ["stash", "-m", "first"], env=_env(repo), catch_exceptions=False)
        (repo / "b.py").write_text("second\n")
        runner.invoke(cli, ["stash", "-m", "second"], env=_env(repo), catch_exceptions=False)
        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        assert entries[0]["message"] == "second"
        assert entries[1]["message"] == "first"

    def test_message_survives_round_trip(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "wip: auth changes"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["message"] == "wip: auth changes"

    def test_stash_persists_across_invocations(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        # Simulate a new CLI invocation by calling list separately
        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        assert len(entries) == 1

    def test_nothing_to_stash_does_not_push(self, repo: pathlib.Path) -> None:
        (repo / "b.py").unlink(missing_ok=True)
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        assert len(entries) == 0


# ---------------------------------------------------------------------------
# Security
# ---------------------------------------------------------------------------

class TestSecurity:
    def test_ansi_in_branch_sanitized(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """Branch name with ANSI codes must not appear raw in stash list text."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
        (tmp_path / "x.py").write_text("data\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
        (tmp_path / "y.py").write_text("dirty\n")
        runner.invoke(cli, ["stash"], env=_env(tmp_path), catch_exceptions=False)

        # Manually inject ANSI into the stash.json branch field
        stash_path = tmp_path / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["branch"] = "\x1b[31mmain\x1b[0m"
        stash_path.write_text(json.dumps(raw))

        r = runner.invoke(cli, ["stash", "list"], env=_env(tmp_path), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    def test_ansi_in_message_sanitized(self, repo: pathlib.Path) -> None:
        """Message with ANSI codes must not appear raw in text output."""
        runner.invoke(cli, ["stash", "-m", "\x1b[31mwip\x1b[0m"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    def test_ansi_in_file_paths_sanitized_in_show(self, repo: pathlib.Path) -> None:
        """File paths in stash show must be sanitized."""
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["delta"]["\x1b[31mevil.py\x1b[0m"] = "a" * 64
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "show", "0"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    @pytest.mark.skipif(os.name == "nt", reason="symlinks require elevated privs on Windows")
    def test_symlink_stash_returns_empty_stack(
        self, repo: pathlib.Path
    ) -> None:
        """A symlink planted at .muse/stash.json must not be followed."""
        stash_path = repo / ".muse/stash.json"
        target = repo / "sensitive.txt"
        target.write_text("secret")
        stash_path.symlink_to(target)

        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0
        assert json.loads(r.output) == []

    def test_invalid_format_to_stderr(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "--format", "xml"], env=_env(repo))
        assert r.exit_code == 1
        assert "xml" in (r.stderr or "").lower()

    def test_pop_invalid_format_to_stderr(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--format", "html"], env=_env(stashed_repo))
        assert r.exit_code == 1
        assert "html" in (r.stderr or "").lower()

    def test_error_messages_in_stderr(self, repo: pathlib.Path) -> None:
        """All 'no stash entries' error messages must go to stderr."""
        for sub in ["pop", "apply", "drop", "show"]:
            r = runner.invoke(cli, ["stash", sub], env=_env(repo))
            assert r.exit_code != 0, f"stash {sub} should fail on empty stack"
            assert "no stash" in (r.stderr or "").lower(), (
                f"stash {sub}: error should be in stderr, got stdout={r.stdout!r} stderr={r.stderr!r}"
            )


# ---------------------------------------------------------------------------
# Stress
# ---------------------------------------------------------------------------

class TestStress:
    @pytest.mark.slow
    def test_sequential_push_pop_cycles(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """100 sequential push+pop cycles must all succeed with correct state."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)

        failures: list[str] = []
        for i in range(100):
            (tmp_path / "work.py").write_text(f"iteration {i}\n")
            r_push = runner.invoke(cli, ["stash", "--json"], env=env)
            if r_push.exit_code != 0:
                failures.append(f"push {i}: {r_push.output.strip()[:60]}")
                continue
            r_pop = runner.invoke(cli, ["stash", "pop", "--json"], env=env)
            if r_pop.exit_code != 0:
                failures.append(f"pop {i}: {r_pop.output.strip()[:60]}")

        assert not failures, f"Failures: {failures}"

    @pytest.mark.slow
    def test_large_stash_stack(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """Stack of 50 entries then clear all via drop loop."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)

        # Push 50 entries
        for i in range(50):
            (tmp_path / f"f{i}.py").write_text(f"file {i}\n")
            runner.invoke(cli, ["stash", "-m", f"entry {i}"], env=env, catch_exceptions=False)

        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=env, catch_exceptions=False).output)
        assert len(entries) == 50

        # Drop all — always drop index 0 (newest)
        for _ in range(50):
            r = runner.invoke(cli, ["stash", "drop", "0"], env=env)
            assert r.exit_code == 0

        entries_after = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=env, catch_exceptions=False).output)
        assert entries_after == []

    @pytest.mark.slow
    def test_concurrent_stash_isolated_repos(self, tmp_path: pathlib.Path) -> None:
        """Parallel stash operations to distinct repos must not interfere.

        Uses subprocess.run for true env isolation — CliRunner shares os.environ
        globally and is not safe for concurrent use across threads.
        """
        import subprocess
        import sys
        errors: list[str] = []

        muse_bin = pathlib.Path(sys.executable).parent / "muse"

        def do_stash(idx: int) -> None:
            repo_dir = tmp_path / f"repo_{idx}"
            repo_dir.mkdir()
            env = {**os.environ, "MUSE_REPO_ROOT": str(repo_dir)}

            def run(*args: str) -> subprocess.CompletedProcess[str]:
                return subprocess.run(
                    [str(muse_bin), *args],
                    capture_output=True, text=True, env=env, cwd=str(repo_dir),
                )

            run("init")
            (repo_dir / "base.py").write_text("base\n")
            run("commit", "-m", "base")
            (repo_dir / "work.py").write_text(f"work {idx}\n")
            r_push = run("stash", "--json")
            if r_push.returncode != 0:
                errors.append(f"repo {idx} push: {(r_push.stdout + r_push.stderr).strip()[:80]}")
                return
            r_pop = run("stash", "pop", "--json")
            if r_pop.returncode != 0:
                errors.append(f"repo {idx} pop: {(r_pop.stdout + r_pop.stderr).strip()[:80]}")

        threads = [threading.Thread(target=do_stash, args=(i,)) for i in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert not errors, f"Concurrent stash errors: {errors}"

    @pytest.mark.slow
    def test_show_performance(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """stash show on a 50-entry stack must complete in < 1 s."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        for i in range(50):
            (tmp_path / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        start = time.perf_counter()
        r = runner.invoke(cli, ["stash", "show", "49", "--json"], env=env, catch_exceptions=False)
        elapsed = time.perf_counter() - start
        assert r.exit_code == 0, r.output
        assert elapsed < 1.0, f"stash show took {elapsed:.2f}s"


# ---------------------------------------------------------------------------
# Extended — muse stash pop
# ---------------------------------------------------------------------------

class TestStashPopExtended:
    """Extended integration tests for ``muse stash pop``."""

    def test_pop_j_alias(self, stashed_repo: pathlib.Path) -> None:
        """-j must be an alias for --json."""
        r = runner.invoke(cli, ["stash", "pop", "-j"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert d["status"] == "popped"

    def test_pop_decrements_stash_size(self, repo: pathlib.Path) -> None:
        """Stash size must decrease by exactly 1 after pop."""
        # Push two entries
        (repo / "c.py").write_text("c\n")
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        (repo / "d.py").write_text("d\n")
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        before = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["stash_size_after"] == len(before) - 1

    def test_pop_removes_specific_index(self, repo: pathlib.Path) -> None:
        """Popping index 1 must leave index 0 and index 2 entries intact."""
        for i in range(3):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash", "-m", f"entry{i}"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "pop", "1", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        assert len(entries) == 2
        assert entries[0]["message"] == "entry2"
        assert entries[1]["message"] == "entry0"

    def test_pop_last_entry_leaves_empty_stack(self, stashed_repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "pop"], env=_env(stashed_repo), catch_exceptions=False)
        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False).output)
        assert entries == []

    def test_pop_restores_file_content(self, repo: pathlib.Path) -> None:
        """Content of restored file must be byte-for-byte identical."""
        content = "exact content 12345\n"
        (repo / "b.py").write_text(content)
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        runner.invoke(cli, ["stash", "pop"], env=_env(repo), catch_exceptions=False)
        assert (repo / "b.py").read_text() == content

    def test_pop_json_message_null_when_none(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["message"] is None

    def test_pop_json_message_present_when_annotated(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "my work"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["message"] == "my work"

    def test_pop_text_output_contains_branch(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop"], env=_env(stashed_repo), catch_exceptions=False)
        assert "main" in r.output

    def test_pop_text_output_contains_stash_at(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop"], env=_env(stashed_repo), catch_exceptions=False)
        assert "stash@{" in r.output

    def test_pop_text_output_contains_annotation(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "annotated-work"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "pop"], env=_env(repo), catch_exceptions=False)
        assert "annotated-work" in r.output

    def test_pop_json_snapshot_id_is_sha256(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert isinstance(d["snapshot_id"], str) and len(d["snapshot_id"]) == 64

    def test_pop_json_files_count_positive(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["files_count"] > 0

    def test_pop_json_stash_size_after_non_negative(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["stash_size_after"] >= 0

    def test_pop_removes_entry_from_disk(self, stashed_repo: pathlib.Path) -> None:
        """After pop, stash.json must not contain the popped entry."""
        before = json.loads((stashed_repo / ".muse/stash.json").read_text())
        snap_id = before[0]["snapshot_id"]
        runner.invoke(cli, ["stash", "pop"], env=_env(stashed_repo), catch_exceptions=False)
        stash_path = stashed_repo / ".muse/stash.json"
        if stash_path.exists():
            after = json.loads(stash_path.read_text())
            assert all(e["snapshot_id"] != snap_id for e in after)

    def test_pop_default_format_is_text(self, stashed_repo: pathlib.Path) -> None:
        """Default output must not be valid JSON (i.e., must be text)."""
        r = runner.invoke(cli, ["stash", "pop"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0
        try:
            json.loads(r.output)
            assert False, "default output should be text, not JSON"
        except (json.JSONDecodeError, ValueError):
            pass  # expected: text output

    def test_pop_index_zero_explicit(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "0", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        assert json.loads(r.output)["status"] == "popped"

    def test_pop_error_no_entries_exits_1(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop"], env=_env(repo))
        assert r.exit_code == 1

    def test_pop_error_out_of_range_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "99"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_pop_error_non_integer_index_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "not_a_number"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_pop_json_empty_error_schema(self, repo: pathlib.Path) -> None:
        """pop --json on empty stash must emit a JSON error object on stdout."""
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(repo))
        assert r.exit_code != 0
        # CliRunner combines stdout+stderr; find the JSON line in output.
        json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")]
        assert json_lines, f"No JSON line in output: {r.output!r}"
        d = json.loads(json_lines[0])
        assert "error" in d
        assert d["stash_size"] == 0

    def test_pop_branch_field_matches_current_branch(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["branch"] == "main"

    def test_pop_stashed_at_is_iso8601(self, stashed_repo: pathlib.Path) -> None:
        import datetime as _dt
        r = runner.invoke(cli, ["stash", "pop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        ts = _dt.datetime.fromisoformat(d["stashed_at"])
        assert ts.tzinfo is not None, "stashed_at must be timezone-aware"

    def test_pop_description_in_help(self) -> None:
        """pop --help must include description text."""
        r = runner.invoke(cli, ["stash", "pop", "--help"], env={})
        assert r.exit_code == 0
        assert "Index rules" in r.output or "stash@{0}" in r.output

    def test_pop_outside_repo_exits_2(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.chdir(tmp_path)
        monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
        r = runner.invoke(cli, ["stash", "pop"])
        assert r.exit_code == 2


# ---------------------------------------------------------------------------
# Security — muse stash pop
# ---------------------------------------------------------------------------

class TestStashPopSecurity:
    """Security tests for ``muse stash pop``."""

    def test_ansi_in_branch_sanitized_in_pop_text(self, repo: pathlib.Path) -> None:
        """Branch name with ANSI codes must not appear raw in pop text output."""
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["branch"] = "\x1b[31mmain\x1b[0m"
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "pop"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    def test_ansi_in_message_sanitized_in_pop_text(self, repo: pathlib.Path) -> None:
        """Message with ANSI codes must not appear raw in pop text output."""
        runner.invoke(cli, ["stash", "-m", "\x1b[32mwip\x1b[0m"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "pop"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[32m" not in r.output

    def test_invalid_format_exits_1_to_stderr(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "pop", "--format", "xml"], env=_env(stashed_repo))
        assert r.exit_code == 1
        assert "xml" in (r.stderr or r.output).lower()

    def test_missing_object_error_to_stderr(self, stashed_repo: pathlib.Path) -> None:
        """Missing-object error message must go to stderr."""
        stash_data = json.loads((stashed_repo / ".muse/stash.json").read_text())
        manifest = stash_data[0]["delta"]
        for obj_id in list(manifest.values())[:1]:
            obj_path = stashed_repo / ".muse" / "objects" / obj_id[:2] / obj_id[2:]
            if obj_path.exists():
                obj_path.unlink()
                break
        r = runner.invoke(cli, ["stash", "pop"], env=_env(stashed_repo))
        assert r.exit_code == 3
        assert "missing" in (r.stderr or "").lower() or "missing" in r.output.lower()

    def test_pop_index_zero_string_not_special(self, stashed_repo: pathlib.Path) -> None:
        """'0' as index must be treated as integer 0, not a path or injection."""
        r = runner.invoke(cli, ["stash", "pop", "0", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0

    def test_pop_index_with_whitespace_rejected(self, stashed_repo: pathlib.Path) -> None:
        """Index with leading space must not crash — either rejected or treated as integer."""
        r = runner.invoke(cli, ["stash", "pop", " 0"], env=_env(stashed_repo))
        # argparse may strip whitespace or pass " 0" verbatim; _resolve_index
        # must handle it without an unhandled exception (no traceback in output).
        assert "Traceback" not in r.output

    def test_pop_large_index_bounded(self, stashed_repo: pathlib.Path) -> None:
        """Arbitrarily large index must exit with USER_ERROR, not an exception."""
        r = runner.invoke(cli, ["stash", "pop", "999999999999"])
        assert r.exit_code == 1

    def test_pop_negative_index_rejected(self, stashed_repo: pathlib.Path) -> None:
        """Negative index must be rejected cleanly, not used as Python slice."""
        r = runner.invoke(cli, ["stash", "pop", "-1"], env=_env(stashed_repo))
        # -1 may be parsed as a flag by argparse; either way, must not silently
        # pop the last entry using Python negative indexing.
        if r.exit_code == 0:
            # If argparse accepted it, verify it did NOT use Python negative indexing
            # (i.e., must not have popped the last entry silently as stash@{-1})
            pass  # argparse-level rejection is also acceptable
        else:
            assert r.exit_code == 1


# ---------------------------------------------------------------------------
# Stress — muse stash pop
# ---------------------------------------------------------------------------

class TestStashPopStress:
    """Stress tests for ``muse stash pop``."""

    @pytest.mark.slow
    def test_pop_performance_large_stash(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """Pop from a 50-entry stash stack must complete in < 2 s."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        for i in range(50):
            (tmp_path / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        start = time.perf_counter()
        r = runner.invoke(cli, ["stash", "pop", "49", "--json"], env=env, catch_exceptions=False)
        elapsed = time.perf_counter() - start
        assert r.exit_code == 0, r.output
        assert elapsed < 2.0, f"pop took {elapsed:.2f}s"

    @pytest.mark.slow
    def test_pop_all_entries_sequential(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """Pop all 20 entries one-by-one; stash must be empty after."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)

        for i in range(20):
            (tmp_path / f"w{i}.py").write_text(f"data {i}\n")
            runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        failures: list[str] = []
        for i in range(20):
            r = runner.invoke(cli, ["stash", "pop", "--json"], env=env)
            if r.exit_code != 0:
                failures.append(f"pop {i}: exit={r.exit_code} out={r.output.strip()[:60]}")
        assert not failures, f"Pop failures: {failures}"

        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=env, catch_exceptions=False).output)
        assert entries == []

    @pytest.mark.slow
    def test_pop_concurrent_isolated_repos(self, tmp_path: pathlib.Path) -> None:
        """Concurrent pops in isolated repos must not interfere.

        Uses subprocess.run for true env isolation — CliRunner shares os.environ
        globally and is not safe for concurrent use across threads.
        """
        import subprocess
        import sys
        errors: list[str] = []

        muse_bin = pathlib.Path(sys.executable).parent / "muse"

        def do_push_pop(idx: int) -> None:
            repo_dir = tmp_path / f"repo_{idx}"
            repo_dir.mkdir()
            env = {**os.environ, "MUSE_REPO_ROOT": str(repo_dir)}

            def run(*args: str) -> subprocess.CompletedProcess[str]:
                return subprocess.run(
                    [str(muse_bin), *args],
                    capture_output=True, text=True, env=env, cwd=str(repo_dir),
                )

            run("init")
            (repo_dir / "base.py").write_text("base\n")
            run("commit", "-m", "base")
            (repo_dir / "work.py").write_text(f"work {idx}\n")
            r_push = run("stash", "--json")
            if r_push.returncode != 0:
                errors.append(f"repo {idx} push failed: {(r_push.stdout + r_push.stderr).strip()[:80]}")
                return
            r_pop = run("stash", "pop", "--json")
            if r_pop.returncode != 0:
                errors.append(f"repo {idx} pop failed: {(r_pop.stdout + r_pop.stderr).strip()[:80]}")
                return
            try:
                d = json.loads(r_pop.stdout)
            except json.JSONDecodeError:
                errors.append(f"repo {idx} non-JSON pop output: {r_pop.stdout.strip()[:60]}")
                return
            if d.get("status") != "popped":
                errors.append(f"repo {idx} wrong status: {d.get('status')}")

        threads = [threading.Thread(target=do_push_pop, args=(i,)) for i in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert not errors, f"Concurrent pop errors: {errors}"


# ---------------------------------------------------------------------------
# Extended — muse stash apply
# ---------------------------------------------------------------------------

class TestStashApplyExtended:
    """Extended integration tests for ``muse stash apply``."""

    def test_apply_j_alias(self, stashed_repo: pathlib.Path) -> None:
        """-j must be an alias for --json."""
        r = runner.invoke(cli, ["stash", "apply", "-j"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert d["status"] == "applied"

    def test_apply_preserves_all_entries(self, repo: pathlib.Path) -> None:
        """apply must not remove any stash entry."""
        for i in range(3):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash", "-m", f"e{i}"], env=_env(repo), catch_exceptions=False)
        before = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        runner.invoke(cli, ["stash", "apply", "1", "--json"], env=_env(repo), catch_exceptions=False)
        after = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        assert len(before) == len(after)

    def test_apply_stash_size_unchanged(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["stash_size"] == 1

    def test_apply_restores_file_content(self, repo: pathlib.Path) -> None:
        content = "precise content 99\n"
        (repo / "b.py").write_text(content)
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        runner.invoke(cli, ["stash", "apply"], env=_env(repo), catch_exceptions=False)
        assert (repo / "b.py").read_text() == content

    def test_apply_by_index(self, repo: pathlib.Path) -> None:
        """Applying entry at index 1 must restore that entry's files."""
        (repo / "b.py").write_text("first\n")
        r0 = runner.invoke(cli, ["stash", "-m", "first", "--json"], env=_env(repo), catch_exceptions=False)
        snap0 = json.loads(r0.output)["snapshot_id"]
        (repo / "b.py").write_text("second\n")
        runner.invoke(cli, ["stash", "-m", "second"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "apply", "1", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["snapshot_id"] == snap0

    def test_apply_json_status_is_applied(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["status"] == "applied"

    def test_apply_json_message_null_when_none(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["message"] is None

    def test_apply_json_message_when_annotated(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "wip stuff"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(repo), catch_exceptions=False)
        assert json.loads(r.output)["message"] == "wip stuff"

    def test_apply_json_snapshot_id_is_sha256(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        snap = json.loads(r.output)["snapshot_id"]
        assert isinstance(snap, str) and len(snap) == 64

    def test_apply_json_files_count_positive(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["files_count"] > 0

    def test_apply_json_branch_field(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["branch"] == "main"

    def test_apply_json_stashed_at_is_iso8601(self, stashed_repo: pathlib.Path) -> None:
        import datetime as _dt
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        ts = _dt.datetime.fromisoformat(json.loads(r.output)["stashed_at"])
        assert ts.tzinfo is not None

    def test_apply_text_contains_branch(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply"], env=_env(stashed_repo), catch_exceptions=False)
        assert "main" in r.output

    def test_apply_text_contains_stash_at(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply"], env=_env(stashed_repo), catch_exceptions=False)
        assert "stash@{" in r.output

    def test_apply_text_says_preserved(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply"], env=_env(stashed_repo), catch_exceptions=False)
        assert "preserved" in r.output.lower()

    def test_apply_text_contains_annotation(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "keep-this"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "apply"], env=_env(repo), catch_exceptions=False)
        assert "keep-this" in r.output

    def test_apply_default_format_is_text(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply"], env=_env(stashed_repo), catch_exceptions=False)
        try:
            json.loads(r.output)
            assert False, "default output should be text, not JSON"
        except (json.JSONDecodeError, ValueError):
            pass

    def test_apply_empty_exits_1(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply"], env=_env(repo))
        assert r.exit_code == 1

    def test_apply_out_of_range_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "99"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_apply_non_integer_index_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "abc"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_apply_json_empty_error_schema(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(repo))
        assert r.exit_code != 0
        json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")]
        assert json_lines, f"No JSON in output: {r.output!r}"
        d = json.loads(json_lines[0])
        assert "error" in d
        assert d["stash_size"] == 0

    def test_apply_twice_idempotent_stash_size(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        runner.invoke(cli, ["stash", "apply"], env=_env(repo), catch_exceptions=False)
        runner.invoke(cli, ["stash", "apply"], env=_env(repo), catch_exceptions=False)
        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        assert len(entries) == 1

    def test_apply_description_in_help(self) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--help"], env={})
        assert r.exit_code == 0
        assert "Index rules" in r.output or "stash@{0}" in r.output

    def test_apply_outside_repo_exits_2(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.chdir(tmp_path)
        monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
        r = runner.invoke(cli, ["stash", "apply"])
        assert r.exit_code == 2

    def test_apply_schema_complete(self, stashed_repo: pathlib.Path) -> None:
        _REQUIRED = {"status", "snapshot_id", "branch", "stashed_at", "message", "files_count", "stash_size"}
        r = runner.invoke(cli, ["stash", "apply", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert _REQUIRED <= json.loads(r.output).keys()


# ---------------------------------------------------------------------------
# Security — muse stash apply
# ---------------------------------------------------------------------------

class TestStashApplySecurity:
    """Security tests for ``muse stash apply``."""

    def test_ansi_in_branch_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["branch"] = "\x1b[31mmain\x1b[0m"
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "apply"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    def test_ansi_in_message_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "\x1b[32mwip\x1b[0m"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "apply"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[32m" not in r.output

    def test_invalid_format_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "--format", "xml"], env=_env(stashed_repo))
        assert r.exit_code == 1
        assert "xml" in (r.stderr or r.output).lower()

    def test_missing_object_listed_in_stderr(self, stashed_repo: pathlib.Path) -> None:
        """Each missing path must be listed in stderr output."""
        stash_data = json.loads((stashed_repo / ".muse/stash.json").read_text())
        manifest = stash_data[0]["delta"]
        for obj_id in list(manifest.values())[:1]:
            obj_path = stashed_repo / ".muse" / "objects" / obj_id[:2] / obj_id[2:]
            if obj_path.exists():
                obj_path.unlink()
                break
        r = runner.invoke(cli, ["stash", "apply"], env=_env(stashed_repo))
        assert r.exit_code == 3
        assert "missing" in (r.stderr or "").lower() or "missing" in r.output.lower()

    def test_missing_object_exits_3(self, stashed_repo: pathlib.Path) -> None:
        stash_data = json.loads((stashed_repo / ".muse/stash.json").read_text())
        for obj_id in list(stash_data[0]["delta"].values())[:1]:
            obj_path = stashed_repo / ".muse" / "objects" / obj_id[:2] / obj_id[2:]
            if obj_path.exists():
                obj_path.unlink()
                break
        r = runner.invoke(cli, ["stash", "apply"], env=_env(stashed_repo))
        assert r.exit_code == 3

    def test_apply_does_not_modify_stash_on_success(self, stashed_repo: pathlib.Path) -> None:
        """apply must not write stash.json."""
        stash_path = stashed_repo / ".muse/stash.json"
        before_mtime = stash_path.stat().st_mtime
        runner.invoke(cli, ["stash", "apply"], env=_env(stashed_repo), catch_exceptions=False)
        after_mtime = stash_path.stat().st_mtime
        assert before_mtime == after_mtime, "apply must not modify stash.json"

    def test_large_index_rejected(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "999999999"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_negative_index_rejected(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "apply", "-1"], env=_env(stashed_repo))
        assert r.exit_code != 0 or "Traceback" not in r.output


# ---------------------------------------------------------------------------
# Stress — muse stash apply
# ---------------------------------------------------------------------------

class TestStashApplyStress:
    """Stress tests for ``muse stash apply``."""

    @pytest.mark.slow
    def test_apply_performance_large_stash(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """apply on a 50-entry stack must complete in < 2 s."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        for i in range(50):
            (tmp_path / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        start = time.perf_counter()
        r = runner.invoke(cli, ["stash", "apply", "49", "--json"], env=env, catch_exceptions=False)
        elapsed = time.perf_counter() - start
        assert r.exit_code == 0, r.output
        assert elapsed < 2.0, f"apply took {elapsed:.2f}s"

    @pytest.mark.slow
    def test_apply_repeated_idempotent(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """Applying the same entry 20 times must leave stash size unchanged."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        (tmp_path / "work.py").write_text("data\n")
        runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        failures: list[str] = []
        for i in range(20):
            r = runner.invoke(cli, ["stash", "apply", "--json"], env=env)
            if r.exit_code != 0:
                failures.append(f"apply {i}: {r.output.strip()[:60]}")
        assert not failures

        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=env, catch_exceptions=False).output)
        assert len(entries) == 1

    @pytest.mark.slow
    def test_apply_concurrent_isolated_repos(self, tmp_path: pathlib.Path) -> None:
        """Concurrent applies in isolated repos must not interfere."""
        import subprocess
        import sys
        errors: list[str] = []
        muse_bin = pathlib.Path(sys.executable).parent / "muse"

        def do_apply(idx: int) -> None:
            repo_dir = tmp_path / f"repo_{idx}"
            repo_dir.mkdir()
            env = {**os.environ, "MUSE_REPO_ROOT": str(repo_dir)}

            def run(*args: str) -> subprocess.CompletedProcess[str]:
                return subprocess.run(
                    [str(muse_bin), *args],
                    capture_output=True, text=True, env=env, cwd=str(repo_dir),
                )

            run("init")
            (repo_dir / "base.py").write_text("base\n")
            run("commit", "-m", "base")
            (repo_dir / "work.py").write_text(f"work {idx}\n")
            run("stash")
            r = run("stash", "apply", "--json")
            if r.returncode != 0:
                errors.append(f"repo {idx} apply failed: {(r.stdout + r.stderr).strip()[:80]}")
                return
            try:
                d = json.loads(r.stdout)
            except json.JSONDecodeError:
                errors.append(f"repo {idx} non-JSON: {r.stdout.strip()[:60]}")
                return
            if d.get("status") != "applied":
                errors.append(f"repo {idx} wrong status: {d.get('status')}")

        threads = [threading.Thread(target=do_apply, args=(i,)) for i in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert not errors, f"Concurrent apply errors: {errors}"


# ---------------------------------------------------------------------------
# Extended — muse stash show
# ---------------------------------------------------------------------------

class TestStashShowExtended:
    """Extended integration tests for ``muse stash show``."""

    def test_show_j_alias(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "-j"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0
        assert json.loads(r.output)["index"] == 0

    def test_show_schema_includes_files_count(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert "files_count" in d
        assert d["files_count"] == len(d["files"])

    def test_show_schema_complete(self, stashed_repo: pathlib.Path) -> None:
        _REQUIRED = {"index", "snapshot_id", "branch", "stashed_at", "message", "files_count", "files"}
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert _REQUIRED <= json.loads(r.output).keys()

    def test_show_files_is_sorted(self, repo: pathlib.Path) -> None:
        for name in ["z.py", "a.py", "m.py"]:
            (repo / name).write_text("x\n")
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(repo), catch_exceptions=False)
        files = json.loads(r.output)["files"]
        assert files == sorted(files)

    def test_show_files_count_matches_files(self, repo: pathlib.Path) -> None:
        for name in ["a.py", "b.py", "c.py"]:
            (repo / name).write_text("x\n")
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["files_count"] == len(d["files"])

    def test_show_by_index(self, repo: pathlib.Path) -> None:
        (repo / "b.py").write_text("first\n")
        runner.invoke(cli, ["stash", "-m", "first"], env=_env(repo), catch_exceptions=False)
        (repo / "b.py").write_text("second\n")
        runner.invoke(cli, ["stash", "-m", "second"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "show", "1", "--json"], env=_env(repo), catch_exceptions=False)
        assert json.loads(r.output)["index"] == 1

    def test_show_index_zero_default(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["index"] == 0

    def test_show_does_not_modify_workdir(self, stashed_repo: pathlib.Path) -> None:
        before = set(p.name for p in stashed_repo.iterdir() if not p.name.startswith("."))
        runner.invoke(cli, ["stash", "show"], env=_env(stashed_repo), catch_exceptions=False)
        after = set(p.name for p in stashed_repo.iterdir() if not p.name.startswith("."))
        assert before == after

    def test_show_does_not_modify_stash_json(self, stashed_repo: pathlib.Path) -> None:
        stash_path = stashed_repo / ".muse/stash.json"
        mtime_before = stash_path.stat().st_mtime
        runner.invoke(cli, ["stash", "show"], env=_env(stashed_repo), catch_exceptions=False)
        assert stash_path.stat().st_mtime == mtime_before

    def test_show_message_null_when_none(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["message"] is None

    def test_show_message_when_annotated(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "review me"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(repo), catch_exceptions=False)
        assert json.loads(r.output)["message"] == "review me"

    def test_show_snapshot_id_sha256(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        snap = json.loads(r.output)["snapshot_id"]
        assert isinstance(snap, str) and len(snap) == 64

    def test_show_stashed_at_iso8601(self, stashed_repo: pathlib.Path) -> None:
        import datetime as _dt
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        ts = _dt.datetime.fromisoformat(json.loads(r.output)["stashed_at"])
        assert ts.tzinfo is not None

    def test_show_branch_field(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["branch"] == "main"

    def test_show_text_header(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show"], env=_env(stashed_repo), catch_exceptions=False)
        assert "stash@{0}" in r.output
        assert "main" in r.output

    def test_show_text_lists_files(self, repo: pathlib.Path) -> None:
        (repo / "b.py").write_text("data\n")
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "show"], env=_env(repo), catch_exceptions=False)
        assert "b.py" in r.output

    def test_show_text_annotation(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "my-note"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "show"], env=_env(repo), catch_exceptions=False)
        assert "my-note" in r.output

    def test_show_default_is_text(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show"], env=_env(stashed_repo), catch_exceptions=False)
        try:
            json.loads(r.output)
            assert False, "default output should be text"
        except (json.JSONDecodeError, ValueError):
            pass

    def test_show_empty_exits_1(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show"], env=_env(repo))
        assert r.exit_code == 1

    def test_show_out_of_range_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "99"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_show_non_integer_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "xyz"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_show_empty_json_error_schema(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(repo))
        assert r.exit_code != 0
        json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")]
        assert json_lines
        assert "error" in json.loads(json_lines[0])

    def test_show_description_in_help(self) -> None:
        r = runner.invoke(cli, ["stash", "show", "--help"], env={})
        assert r.exit_code == 0
        assert "Index rules" in r.output or "stash@{0}" in r.output

    def test_show_outside_repo_exits_2(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.chdir(tmp_path)
        monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
        r = runner.invoke(cli, ["stash", "show"])
        assert r.exit_code == 2

    def test_show_files_count_zero_for_empty_manifest(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["delta"] = {}
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(repo), catch_exceptions=False)
        d = json.loads(r.output)
        assert d["files_count"] == 0
        assert d["files"] == []


# ---------------------------------------------------------------------------
# Security — muse stash show
# ---------------------------------------------------------------------------

class TestStashShowSecurity:
    """Security tests for ``muse stash show``."""

    def test_ansi_in_branch_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["branch"] = "\x1b[31mmain\x1b[0m"
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "show"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    def test_ansi_in_message_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "\x1b[33mwip\x1b[0m"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "show"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[33m" not in r.output

    def test_ansi_in_file_path_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["delta"]["\x1b[31mevil.py\x1b[0m"] = "a" * 64
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "show"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    def test_invalid_format_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "--format", "csv"], env=_env(stashed_repo))
        assert r.exit_code == 1
        assert "csv" in (r.stderr or r.output).lower()

    def test_large_index_rejected(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "999999999"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_negative_index_no_crash(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "show", "-1"], env=_env(stashed_repo))
        assert "Traceback" not in r.output

    def test_read_only_no_side_effects(self, stashed_repo: pathlib.Path) -> None:
        stash_path = stashed_repo / ".muse/stash.json"
        before_stash = stash_path.read_text()
        before_files = set(p.name for p in stashed_repo.iterdir())
        runner.invoke(cli, ["stash", "show"], env=_env(stashed_repo), catch_exceptions=False)
        assert stash_path.read_text() == before_stash
        assert set(p.name for p in stashed_repo.iterdir()) == before_files

    def test_path_traversal_in_manifest_no_crash(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["delta"]["../etc/passwd"] = "b" * 64
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "show"], env=_env(repo), catch_exceptions=False)
        assert "Traceback" not in r.output


# ---------------------------------------------------------------------------
# Stress — muse stash show
# ---------------------------------------------------------------------------

class TestStashShowStress:
    """Stress tests for ``muse stash show``."""

    @pytest.mark.slow
    def test_show_large_manifest(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """show on an entry with 200 files must complete in < 1 s."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        for i in range(200):
            (tmp_path / f"f{i:03d}.py").write_text(f"x={i}\n")
        runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        start = time.perf_counter()
        r = runner.invoke(cli, ["stash", "show", "--json"], env=env, catch_exceptions=False)
        elapsed = time.perf_counter() - start
        assert r.exit_code == 0, r.output
        d = json.loads(r.output)
        assert d["files_count"] >= 200
        assert elapsed < 1.0, f"show took {elapsed:.2f}s"

    @pytest.mark.slow
    def test_show_all_indices_on_50_entry_stack(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        for i in range(50):
            (tmp_path / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        failures: list[str] = []
        for i in range(50):
            r = runner.invoke(cli, ["stash", "show", str(i), "--json"], env=env)
            if r.exit_code != 0:
                failures.append(f"show {i}: {r.output.strip()[:60]}")
        assert not failures

    @pytest.mark.slow
    def test_show_repeated_deterministic(self, stashed_repo: pathlib.Path) -> None:
        """Calling show 50 times must return identical output every time."""
        outputs: list[str] = []
        for _ in range(50):
            r = runner.invoke(cli, ["stash", "show", "--json"], env=_env(stashed_repo), catch_exceptions=False)
            assert r.exit_code == 0
            outputs.append(r.output.strip())
        assert len(set(outputs)) == 1, "show output must be deterministic"


# ---------------------------------------------------------------------------
# Extended — muse stash list
# ---------------------------------------------------------------------------

class TestStashListExtended:
    def test_list_j_alias(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "-j"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0
        assert isinstance(json.loads(r.output), list)

    def test_list_empty_exits_0(self, repo: pathlib.Path) -> None:
        assert runner.invoke(cli, ["stash", "list"], env=_env(repo)).exit_code == 0

    def test_list_empty_json_is_empty_array(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert json.loads(r.output) == []

    def test_list_empty_text_message(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list"], env=_env(repo), catch_exceptions=False)
        assert "No stash" in r.output

    def test_list_schema_complete(self, stashed_repo: pathlib.Path) -> None:
        _REQUIRED = {"index", "snapshot_id", "branch", "stashed_at", "message", "files_count"}
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        entries = json.loads(r.output)
        assert len(entries) == 1
        assert _REQUIRED <= entries[0].keys()

    def test_list_index_zero_based(self, repo: pathlib.Path) -> None:
        for i in range(3):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert [e["index"] for e in json.loads(r.output)] == [0, 1, 2]

    def test_list_lifo_order(self, repo: pathlib.Path) -> None:
        (repo / "b.py").write_text("first\n")
        runner.invoke(cli, ["stash", "-m", "first"], env=_env(repo), catch_exceptions=False)
        (repo / "b.py").write_text("second\n")
        runner.invoke(cli, ["stash", "-m", "second"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        entries = json.loads(r.output)
        assert entries[0]["message"] == "second"
        assert entries[1]["message"] == "first"

    def test_list_files_count_accurate(self, repo: pathlib.Path) -> None:
        for name in ["a.py", "b.py", "c.py"]:
            (repo / name).write_text("x\n")
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert json.loads(r.output)[0]["files_count"] >= 3

    def test_list_message_null_when_none(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)[0]["message"] is None

    def test_list_message_when_annotated(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "todo auth"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert json.loads(r.output)[0]["message"] == "todo auth"

    def test_list_snapshot_id_sha256(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        snap = json.loads(r.output)[0]["snapshot_id"]
        assert isinstance(snap, str) and len(snap) == 64

    def test_list_stashed_at_iso8601(self, stashed_repo: pathlib.Path) -> None:
        import datetime as _dt
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        ts = _dt.datetime.fromisoformat(json.loads(r.output)[0]["stashed_at"])
        assert ts.tzinfo is not None

    def test_list_branch_field(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)[0]["branch"] == "main"

    def test_list_text_stash_at_notation(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list"], env=_env(stashed_repo), catch_exceptions=False)
        assert "stash@{0}" in r.output

    def test_list_text_shows_branch(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list"], env=_env(stashed_repo), catch_exceptions=False)
        assert "main" in r.output

    def test_list_text_shows_annotation(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "in-progress"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list"], env=_env(repo), catch_exceptions=False)
        assert "in-progress" in r.output

    def test_list_default_is_text(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list"], env=_env(stashed_repo), catch_exceptions=False)
        try:
            json.loads(r.output)
            assert False, "default should be text"
        except (json.JSONDecodeError, ValueError):
            pass

    def test_list_does_not_modify_stash_json(self, stashed_repo: pathlib.Path) -> None:
        stash_path = stashed_repo / ".muse/stash.json"
        mtime_before = stash_path.stat().st_mtime
        runner.invoke(cli, ["stash", "list"], env=_env(stashed_repo), catch_exceptions=False)
        assert stash_path.stat().st_mtime == mtime_before

    def test_list_count_matches_pushes(self, repo: pathlib.Path) -> None:
        for i in range(5):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert len(json.loads(r.output)) == 5

    def test_list_description_in_help(self) -> None:
        r = runner.invoke(cli, ["stash", "list", "--help"], env={})
        assert r.exit_code == 0
        assert "jq" in r.output or "Agent quickstart" in r.output

    def test_list_outside_repo_exits_2(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.chdir(tmp_path)
        monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
        r = runner.invoke(cli, ["stash", "list"])
        assert r.exit_code == 2

    def test_list_indices_contiguous(self, repo: pathlib.Path) -> None:
        for i in range(4):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        indices = [e["index"] for e in json.loads(r.output)]
        assert indices == list(range(len(indices)))

    def test_list_after_pop_decrements(self, repo: pathlib.Path) -> None:
        for i in range(3):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        runner.invoke(cli, ["stash", "pop"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert len(json.loads(r.output)) == 2

    def test_list_after_drop_decrements(self, repo: pathlib.Path) -> None:
        for i in range(3):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        runner.invoke(cli, ["stash", "drop", "1"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert len(json.loads(r.output)) == 2

    def test_list_after_apply_unchanged(self, repo: pathlib.Path) -> None:
        for i in range(3):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        runner.invoke(cli, ["stash", "apply"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert len(json.loads(r.output)) == 3


# ---------------------------------------------------------------------------
# Security — muse stash list
# ---------------------------------------------------------------------------

class TestStashListSecurity:
    def test_ansi_in_branch_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["branch"] = "\x1b[31mmain\x1b[0m"
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "list"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    def test_ansi_in_message_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "\x1b[33mwip\x1b[0m"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[33m" not in r.output

    def test_ansi_in_timestamp_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["stashed_at"] = "\x1b[32m2026-01-01T00:00:00+00:00\x1b[0m"
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "list"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[32m" not in r.output

    def test_invalid_format_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "list", "--format", "yaml"], env=_env(stashed_repo))
        assert r.exit_code == 1
        assert "yaml" in (r.stderr or r.output).lower()

    def test_read_only_no_side_effects(self, stashed_repo: pathlib.Path) -> None:
        stash_path = stashed_repo / ".muse/stash.json"
        before = stash_path.read_text()
        runner.invoke(cli, ["stash", "list"], env=_env(stashed_repo), catch_exceptions=False)
        assert stash_path.read_text() == before

    def test_malformed_stash_json_empty_list(self, repo: pathlib.Path) -> None:
        (repo / ".muse/stash.json").write_text("{not valid json")
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0
        assert json.loads(r.output) == []

    def test_oversized_stash_json_empty_list(self, repo: pathlib.Path) -> None:
        (repo / ".muse/stash.json").write_bytes(b"x" * (65 * 1024 * 1024))
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0
        assert json.loads(r.output) == []

    @pytest.mark.skipif(os.name == "nt", reason="symlinks require elevated privs on Windows")
    def test_symlink_stash_json_empty_list(self, repo: pathlib.Path) -> None:
        target = repo / "sensitive.txt"
        target.write_text("secret")
        (repo / ".muse/stash.json").symlink_to(target)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False)
        assert r.exit_code == 0
        assert json.loads(r.output) == []


# ---------------------------------------------------------------------------
# Stress — muse stash list
# ---------------------------------------------------------------------------

class TestStashListStress:
    @pytest.mark.slow
    def test_list_performance_50_entries(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        for i in range(50):
            (tmp_path / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        start = time.perf_counter()
        r = runner.invoke(cli, ["stash", "list", "--json"], env=env, catch_exceptions=False)
        elapsed = time.perf_counter() - start
        assert r.exit_code == 0
        assert len(json.loads(r.output)) == 50
        assert elapsed < 1.0, f"list took {elapsed:.2f}s"

    @pytest.mark.slow
    def test_list_repeated_deterministic(self, stashed_repo: pathlib.Path) -> None:
        outputs = [
            runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False).output.strip()
            for _ in range(50)
        ]
        assert len(set(outputs)) == 1

    @pytest.mark.slow
    def test_list_push_pop_count_accurate(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        for i in range(20):
            (tmp_path / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)
        for _ in range(7):
            runner.invoke(cli, ["stash", "pop"], env=env, catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=env, catch_exceptions=False)
        assert len(json.loads(r.output)) == 13


# ---------------------------------------------------------------------------
# Extended — muse stash drop
# ---------------------------------------------------------------------------

class TestStashDropExtended:
    def test_drop_j_alias(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "-j"], env=_env(stashed_repo), catch_exceptions=False)
        assert r.exit_code == 0
        assert json.loads(r.output)["status"] == "dropped"

    def test_drop_schema_complete(self, stashed_repo: pathlib.Path) -> None:
        _REQUIRED = {"status", "index", "snapshot_id", "branch", "stashed_at", "message", "stash_size"}
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert _REQUIRED <= json.loads(r.output).keys()

    def test_drop_stashed_at_in_json(self, stashed_repo: pathlib.Path) -> None:
        import datetime as _dt
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        ts = _dt.datetime.fromisoformat(json.loads(r.output)["stashed_at"])
        assert ts.tzinfo is not None

    def test_drop_message_null_when_none(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["message"] is None

    def test_drop_message_when_annotated(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "wip notes"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(repo), catch_exceptions=False)
        assert json.loads(r.output)["message"] == "wip notes"

    def test_drop_stash_size_decrements(self, repo: pathlib.Path) -> None:
        for i in range(3):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(repo), catch_exceptions=False)
        assert json.loads(r.output)["stash_size"] == 2

    def test_drop_removes_entry_from_disk(self, stashed_repo: pathlib.Path) -> None:
        before = json.loads((stashed_repo / ".muse/stash.json").read_text())
        snap_id = before[0]["snapshot_id"]
        runner.invoke(cli, ["stash", "drop"], env=_env(stashed_repo), catch_exceptions=False)
        stash_path = stashed_repo / ".muse/stash.json"
        if stash_path.exists():
            after = json.loads(stash_path.read_text())
            assert all(e["snapshot_id"] != snap_id for e in after)

    def test_drop_by_index(self, repo: pathlib.Path) -> None:
        for i in range(3):
            (repo / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash", "-m", f"e{i}"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "drop", "1", "--json"], env=_env(repo), catch_exceptions=False)
        assert json.loads(r.output)["index"] == 1
        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=_env(repo), catch_exceptions=False).output)
        assert len(entries) == 2
        assert entries[0]["message"] == "e2"
        assert entries[1]["message"] == "e0"

    def test_drop_last_entry_leaves_empty(self, stashed_repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "drop"], env=_env(stashed_repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "list", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output) == []

    def test_drop_snapshot_id_sha256(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        snap = json.loads(r.output)["snapshot_id"]
        assert isinstance(snap, str) and len(snap) == 64

    def test_drop_branch_field(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["branch"] == "main"

    def test_drop_text_shows_index(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop"], env=_env(stashed_repo), catch_exceptions=False)
        assert "stash@{0}" in r.output

    def test_drop_text_shows_branch(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop"], env=_env(stashed_repo), catch_exceptions=False)
        assert "main" in r.output

    def test_drop_text_shows_annotation(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "cleanup"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "drop"], env=_env(repo), catch_exceptions=False)
        assert "cleanup" in r.output

    def test_drop_default_is_text(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop"], env=_env(stashed_repo), catch_exceptions=False)
        try:
            json.loads(r.output)
            assert False, "default should be text"
        except (json.JSONDecodeError, ValueError):
            pass

    def test_drop_does_not_restore_workdir(self, stashed_repo: pathlib.Path) -> None:
        """drop must not restore stashed files to the working tree."""
        assert not (stashed_repo / "b.py").exists()
        runner.invoke(cli, ["stash", "drop"], env=_env(stashed_repo), catch_exceptions=False)
        assert not (stashed_repo / "b.py").exists()

    def test_drop_empty_exits_1(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop"], env=_env(repo))
        assert r.exit_code == 1

    def test_drop_out_of_range_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "99"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_drop_non_integer_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "abc"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_drop_empty_json_error_schema(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(repo))
        assert r.exit_code != 0
        json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")]
        assert json_lines
        d = json.loads(json_lines[0])
        assert "error" in d
        assert d["stash_size"] == 0

    def test_drop_description_in_help(self) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--help"], env={})
        assert r.exit_code == 0
        assert "Index rules" in r.output or "stash@{0}" in r.output

    def test_drop_outside_repo_exits_2(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.chdir(tmp_path)
        monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
        r = runner.invoke(cli, ["stash", "drop"])
        assert r.exit_code == 2

    def test_drop_index_zero_explicit(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "0", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["index"] == 0

    def test_drop_stash_size_after_zero_when_last(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--json"], env=_env(stashed_repo), catch_exceptions=False)
        assert json.loads(r.output)["stash_size"] == 0


# ---------------------------------------------------------------------------
# Security — muse stash drop
# ---------------------------------------------------------------------------

class TestStashDropSecurity:
    def test_ansi_in_branch_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash"], env=_env(repo), catch_exceptions=False)
        stash_path = repo / ".muse/stash.json"
        raw = json.loads(stash_path.read_text())
        raw[0]["branch"] = "\x1b[31mmain\x1b[0m"
        stash_path.write_text(json.dumps(raw))
        r = runner.invoke(cli, ["stash", "drop"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    def test_ansi_in_message_sanitized(self, repo: pathlib.Path) -> None:
        runner.invoke(cli, ["stash", "-m", "\x1b[31mwip\x1b[0m"], env=_env(repo), catch_exceptions=False)
        r = runner.invoke(cli, ["stash", "drop"], env=_env(repo), catch_exceptions=False)
        assert "\x1b[31m" not in r.output

    def test_invalid_format_exits_1(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "--format", "xml"], env=_env(stashed_repo))
        assert r.exit_code == 1
        assert "xml" in (r.stderr or r.output).lower()

    def test_large_index_rejected(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "999999999"], env=_env(stashed_repo))
        assert r.exit_code == 1

    def test_negative_index_no_crash(self, stashed_repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop", "-1"], env=_env(stashed_repo))
        assert "Traceback" not in r.output

    def test_error_to_stderr(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["stash", "drop"], env=_env(repo))
        assert r.exit_code != 0
        assert "no stash" in (r.stderr or "").lower()

    def test_drop_does_not_leave_partial_on_crash(self, stashed_repo: pathlib.Path) -> None:
        """_save_stash uses atomic rename — stash.json is never left partial."""
        # Verify stash.json exists and is valid JSON before and after drop
        before = json.loads((stashed_repo / ".muse/stash.json").read_text())
        runner.invoke(cli, ["stash", "drop"], env=_env(stashed_repo), catch_exceptions=False)
        stash_path = stashed_repo / ".muse/stash.json"
        if stash_path.exists():
            after = json.loads(stash_path.read_text())  # must be valid JSON
            assert len(after) == len(before) - 1


# ---------------------------------------------------------------------------
# Stress — muse stash drop
# ---------------------------------------------------------------------------

class TestStashDropStress:
    @pytest.mark.slow
    def test_drop_all_sequential(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """Drop all 20 entries sequentially; stash must be empty after."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        for i in range(20):
            (tmp_path / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        failures: list[str] = []
        for i in range(20):
            r = runner.invoke(cli, ["stash", "drop", "--json"], env=env)
            if r.exit_code != 0:
                failures.append(f"drop {i}: {r.output.strip()[:60]}")
        assert not failures
        entries = json.loads(runner.invoke(cli, ["stash", "list", "--json"], env=env, catch_exceptions=False).output)
        assert entries == []

    @pytest.mark.slow
    def test_drop_performance_large_stash(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """drop on a 50-entry stack must complete in < 2 s."""
        monkeypatch.chdir(tmp_path)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
        env = _env(tmp_path)
        runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
        (tmp_path / "base.py").write_text("base\n")
        runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
        for i in range(50):
            (tmp_path / f"f{i}.py").write_text(f"{i}\n")
            runner.invoke(cli, ["stash"], env=env, catch_exceptions=False)

        start = time.perf_counter()
        r = runner.invoke(cli, ["stash", "drop", "49", "--json"], env=env, catch_exceptions=False)
        elapsed = time.perf_counter() - start
        assert r.exit_code == 0, r.output
        assert elapsed < 2.0, f"drop took {elapsed:.2f}s"

    @pytest.mark.slow
    def test_drop_concurrent_isolated_repos(self, tmp_path: pathlib.Path) -> None:
        """Concurrent drops in isolated repos must not interfere."""
        import subprocess
        import sys
        errors: list[str] = []
        muse_bin = pathlib.Path(sys.executable).parent / "muse"

        def do_drop(idx: int) -> None:
            repo_dir = tmp_path / f"repo_{idx}"
            repo_dir.mkdir()
            env = {**os.environ, "MUSE_REPO_ROOT": str(repo_dir)}

            def run(*args: str) -> subprocess.CompletedProcess[str]:
                return subprocess.run([str(muse_bin), *args], capture_output=True, text=True, env=env, cwd=str(repo_dir))

            run("init")
            (repo_dir / "base.py").write_text("base\n")
            run("commit", "-m", "base")
            (repo_dir / "work.py").write_text(f"work {idx}\n")
            run("stash")
            r = run("stash", "drop", "--json")
            if r.returncode != 0:
                errors.append(f"repo {idx} drop failed: {(r.stdout + r.stderr).strip()[:80]}")
                return
            try:
                d = json.loads(r.stdout)
            except json.JSONDecodeError:
                errors.append(f"repo {idx} non-JSON: {r.stdout.strip()[:60]}")
                return
            if d.get("status") != "dropped":
                errors.append(f"repo {idx} wrong status: {d.get('status')}")

        threads = [threading.Thread(target=do_drop, args=(i,)) for i in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert not errors, f"Concurrent drop errors: {errors}"