"""Tests for ``memtomem.context.install`` update path.

Covers PR-D C2 commits 3 and 4:

- Commit 3: ``mm context update <type> <name>`` semantics — clean drift,
  dirty refuse, ``--force`` + ``.bak``, no-op invariant,
  ``NotInstalledError``, the flipped ``AlreadyInstalledError`` message.
- Commit 4: ``--all`` orchestration — 4-state classification,
  batch-once ``current_commit``, cache reuse, refuse-blocks-batch,
  ``--yes --force`` invariant, wiki-dirty warn timing.
"""

from __future__ import annotations

import json
import os
import subprocess
from datetime import datetime, timezone
from pathlib import Path

import pytest
from click.testing import CliRunner

import memtomem.context.install as install_module
from memtomem import privacy
from memtomem.cli.context_cmd import context as context_group
from memtomem.context._names import InvalidNameError
from memtomem.context.install import (
    AlreadyInstalledError,
    NotInstalledError,
    StaleInstallError,
    _classify_for_all_update,
    install_skill,
    update_agent,
    update_command,
    update_skill,
)
from memtomem.context.lockfile import LockfileCorruptError
from memtomem.context.privacy_scan import PrivacyBlockedError
from memtomem.wiki.store import WikiStore


# ── helpers ──────────────────────────────────────────────────────────────


def _initialized_wiki(wiki_root_path: Path) -> WikiStore:
    store = WikiStore.at_default()
    store.init()
    return store


def _seed_wiki_skill(wiki_root_path: Path, name: str, files: dict[str, bytes]) -> str:
    """Add ``skills/<name>/`` to wiki + git commit. Returns the commit SHA."""
    skill_dir = wiki_root_path / "skills" / name
    skill_dir.mkdir(parents=True, exist_ok=True)
    for relpath, data in files.items():
        target = skill_dir / relpath
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(data)
    subprocess.run(["git", "-C", str(wiki_root_path), "add", "."], check=True, capture_output=True)
    subprocess.run(
        ["git", "-C", str(wiki_root_path), "commit", "-m", f"add {name}"],
        check=True,
        capture_output=True,
    )
    return WikiStore.at_default().current_commit()


def _modify_wiki_skill(wiki_root_path: Path, name: str, files: dict[str, bytes]) -> str:
    """Modify wiki skill files + commit. Wiki HEAD advances. Returns new SHA."""
    skill_dir = wiki_root_path / "skills" / name
    for relpath, data in files.items():
        target = skill_dir / relpath
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(data)
    subprocess.run(["git", "-C", str(wiki_root_path), "add", "."], check=True, capture_output=True)
    subprocess.run(
        ["git", "-C", str(wiki_root_path), "commit", "-m", f"modify {name}"],
        check=True,
        capture_output=True,
    )
    return WikiStore.at_default().current_commit()


def _bump_mtime(path: Path, *, seconds_in_future: float = 1.0) -> None:
    """Force *path*'s mtime to ``now + seconds_in_future``.

    Tests need a deterministic strict ``>`` margin against ``installed_at``
    that single-second filesystem precision can't eat.
    """
    future = datetime.now(timezone.utc).timestamp() + seconds_in_future
    os.utime(path, (future, future))


# ── happy paths ──────────────────────────────────────────────────────────


def test_clean_drift_updates(wiki_root: Path, tmp_path: Path) -> None:
    """Wiki HEAD advances → update copies new bytes + bumps lockfile."""
    _initialized_wiki(wiki_root)
    old_commit = _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    project = tmp_path

    install_skill(project, "foo")

    new_commit = _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    result = update_skill(project, "foo")

    assert result.was_no_op is False
    assert result.old_wiki_commit == old_commit
    assert result.new_wiki_commit == new_commit
    assert old_commit != new_commit
    assert (project / ".memtomem" / "skills" / "foo" / "SKILL.md").read_bytes() == b"v2\n"

    lock_doc = json.loads((project / ".memtomem" / "lock.json").read_text())
    assert lock_doc["skills"]["foo"]["wiki_commit"] == new_commit
    assert lock_doc["skills"]["foo"]["installed_at"] == result.installed_at


def test_dirty_refuse_without_force(wiki_root: Path, tmp_path: Path) -> None:
    """Local edit + wiki advance + no --force → StaleInstallError."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    project = tmp_path
    install_skill(project, "foo")

    edited = project / ".memtomem" / "skills" / "foo" / "SKILL.md"
    edited.write_bytes(b"local edit\n")
    _bump_mtime(edited)
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    with pytest.raises(StaleInstallError) as excinfo:
        update_skill(project, "foo")
    msg = str(excinfo.value)
    assert "modified locally" in msg
    assert "--force" in msg
    # The user's edit MUST still be on disk (no partial write happened).
    assert edited.read_bytes() == b"local edit\n"


def test_dirty_force_writes_bak(wiki_root: Path, tmp_path: Path) -> None:
    """--force preserves the user edit's bytes in .bak before overwriting."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    project = tmp_path
    install_skill(project, "foo")

    edited = project / ".memtomem" / "skills" / "foo" / "SKILL.md"
    edited.write_bytes(b"my edit\n")
    _bump_mtime(edited)
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    result = update_skill(project, "foo", force=True)

    bak = edited.with_suffix(edited.suffix + ".bak")
    assert bak.is_file()
    assert bak.read_bytes() == b"my edit\n"
    # Wiki bytes won the overwrite.
    assert edited.read_bytes() == b"v2\n"
    assert result.bak_files_written == (bak,)
    assert result.was_no_op is False


def test_dirty_force_only_dirty_files_get_bak(wiki_root: Path, tmp_path: Path) -> None:
    """Clean files have no .bak sibling — only edited files get one."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(
        wiki_root,
        "foo",
        {"SKILL.md": b"v1\n", "scripts/run.sh": b"original\n"},
    )
    project = tmp_path
    install_skill(project, "foo")

    skill_dir = project / ".memtomem" / "skills" / "foo"
    edited = skill_dir / "scripts" / "run.sh"
    edited.write_bytes(b"my edit\n")
    _bump_mtime(edited)
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n", "scripts/run.sh": b"new\n"})

    result = update_skill(project, "foo", force=True)

    skill_md = skill_dir / "SKILL.md"
    assert not (skill_md.with_suffix(skill_md.suffix + ".bak")).exists()
    bak = edited.with_suffix(edited.suffix + ".bak")
    assert bak.is_file()
    assert bak.read_bytes() == b"my edit\n"
    assert result.bak_files_written == (bak,)


# ── failure paths ────────────────────────────────────────────────────────


def test_not_installed_raises(wiki_root: Path, tmp_path: Path) -> None:
    """No lockfile entry → NotInstalledError points at install."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    project = tmp_path

    with pytest.raises(NotInstalledError) as excinfo:
        update_skill(project, "foo")
    msg = str(excinfo.value)
    assert "no lockfile entry" in msg
    assert "mm context install skill foo" in msg


def test_update_over_corrupt_lockfile_names_the_lockfile(wiki_root: Path, tmp_path: Path) -> None:
    """A corrupt lock.json used to read as "no entry" → NotInstalledError
    pointing at install, which then raised AlreadyInstalled (dest exists) —
    the #1247 id 16 wedge. The strict read now names the real problem, and
    the update's lockfile upsert can no longer persist a silent reset."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    project = tmp_path

    install_skill(project, "foo")
    lock_json = project / ".memtomem" / "lock.json"
    corrupt = b"not valid json {{"
    lock_json.write_bytes(corrupt)

    with pytest.raises(LockfileCorruptError, match="lock.json"):
        update_skill(project, "foo")
    assert lock_json.read_bytes() == corrupt  # refusal left the file untouched


# ── no-op invariant pin (lockfile mtime + installed_at + flag) ──────────


def test_no_op_invariant_pin(wiki_root: Path, tmp_path: Path) -> None:
    """No-op update MUST NOT touch the lockfile bytes.

    Pin: lockfile mtime + bytes unchanged AND installed_at echoed AND
    was_no_op=True. The lockfile-mtime pin is the load-bearing assertion
    — any future regression that re-writes the lockfile on no-op
    (e.g. someone "refreshing" installed_at) breaks this test loudly.
    """
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    project = tmp_path
    install_result = install_skill(project, "foo")

    lock_path = project / ".memtomem" / "lock.json"
    pre_bytes = lock_path.read_bytes()
    pre_mtime = lock_path.stat().st_mtime

    result = update_skill(project, "foo")

    # Positive marker — bytes untouched.
    assert lock_path.read_bytes() == pre_bytes
    # Positive marker — mtime untouched (no rewrite happened).
    assert lock_path.stat().st_mtime == pre_mtime
    # Positive marker — installed_at echoed from prior install.
    assert result.installed_at == install_result.installed_at
    # Positive marker — flag set.
    assert result.was_no_op is True
    assert result.bak_files_written == ()
    assert result.files_written == 0


# ── pin-and-invert: AlreadyInstalledError points at update ──────────────


def test_already_installed_message_points_to_update(wiki_root: Path, tmp_path: Path) -> None:
    """``AlreadyInstalledError`` now points at ``mm context update``.

    Pin-and-invert pair (per ``feedback_pin_invert_symmetric_assertion``):

    - Positive marker: ``"mm context update"`` in the message; the
      asset-type-singular form (``skill``, not ``skills``) matches the
      CLI argument the user types.
    - Negative marker: ``"reserved for PR-D"`` is gone; this is the
      string that lived there before C2 and must not return.
    """
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    project = tmp_path

    install_skill(project, "foo")

    with pytest.raises(AlreadyInstalledError) as excinfo:
        install_skill(project, "foo")
    msg = str(excinfo.value)

    # Positive markers — current invariant.
    assert "mm context update skill foo" in msg
    assert "refresh from wiki HEAD" in msg
    # Negative marker — the would-be-old message must not be present.
    assert "reserved for PR-D" not in msg
    # The diagnostic prefix is preserved (composes with other tests).
    assert "lockfile_entry=yes" in msg
    assert "dest=yes" in msg


# ── CLI: no-op exit clean ────────────────────────────────────────────────


@pytest.fixture
def project_cwd(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    """Tmp project root (with sentinel ``.git``) wired as cwd for
    ``_find_project_root``. Same shape as the install CLI tests'
    fixture so behavior parity is easy to reason about."""
    project = tmp_path / "proj"
    project.mkdir()
    (project / ".git").mkdir()
    monkeypatch.chdir(project)
    return project


def test_cli_update_no_op_exits_clean(
    wiki_root: Path,
    project_cwd: Path,
) -> None:
    """``mm context update`` no-op prints ``unchanged`` with exit 0."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    install_skill(project_cwd, "foo")

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo"])

    assert result.exit_code == 0, result.output
    assert "unchanged" in result.output


# ── CLI: variant dispatch ────────────────────────────────────────────────


def test_cli_update_dispatches_to_correct_wrapper(
    project_cwd: Path,
    wiki_root: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``update_cmd`` routes ``skill / agent / command`` to the right wrapper.

    Mocks the 3 wrappers so the test pins the CLI dispatch shape without
    needing to seed wiki content for all 3 asset types. A regression
    that miswires (e.g. ``skill`` calling ``update_agent``) fails this
    test loudly.
    """
    # Need wiki initialized so the CLI doesn't error out before dispatch.
    _initialized_wiki(wiki_root)

    calls: list[tuple[str, str]] = []

    def _fake(asset_type: str):
        def _impl(root: Path, name: str, *, wiki: object = None, force: bool = False) -> object:
            calls.append((asset_type, name))
            # Return a sentinel UpdateResult — minimal fields the CLI prints.
            from memtomem.context.install import UpdateResult

            return UpdateResult(
                asset_type=f"{asset_type}s",  # type: ignore[arg-type]
                name=name,
                old_wiki_commit="0" * 40,
                new_wiki_commit="0" * 40,
                installed_at="2026-01-01T00:00:00.000000Z",
                was_no_op=True,
                bak_files_written=(),
                dest=root / ".memtomem" / f"{asset_type}s" / name,
                files_written=0,
            )

        return _impl

    monkeypatch.setattr("memtomem.cli.context_cmd.update_skill", _fake("skill"))
    monkeypatch.setattr("memtomem.cli.context_cmd.update_agent", _fake("agent"))
    monkeypatch.setattr("memtomem.cli.context_cmd.update_command", _fake("command"))

    runner = CliRunner()
    for asset_type, name in [("skill", "alpha"), ("agent", "beta"), ("command", "gamma")]:
        result = runner.invoke(context_group, ["update", asset_type, name])
        assert result.exit_code == 0, f"{asset_type}/{name}: {result.output}"

    assert calls == [("skill", "alpha"), ("agent", "beta"), ("command", "gamma")]


# ── CLI: stale refuse → exit non-zero with hint ─────────────────────────


def test_cli_update_stale_refuse_message(
    wiki_root: Path,
    project_cwd: Path,
) -> None:
    """CLI surfaces ``StaleInstallError`` text + non-zero exit."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    install_skill(project_cwd, "foo")

    edited = project_cwd / ".memtomem" / "skills" / "foo" / "SKILL.md"
    edited.write_bytes(b"local\n")
    _bump_mtime(edited)
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo"])

    assert result.exit_code != 0
    assert "modified locally" in result.output
    assert "--force" in result.output


# ── coverage: silence unused-import warnings on agent/command wrappers ──


def test_update_agent_command_wrappers_dispatch(monkeypatch: pytest.MonkeyPatch) -> None:
    """``update_agent`` / ``update_command`` route through ``_update_asset``
    with the right ``asset_type`` literal. Direct call sites (not via CLI)
    must keep the 3-wrapper API stable for downstream Python callers."""
    seen: list[str] = []

    def _fake_update_asset(
        project_root: object, asset_type: str, name: str, *, wiki: object, force: bool
    ) -> object:
        seen.append(asset_type)
        from memtomem.context.install import UpdateResult

        return UpdateResult(
            asset_type=asset_type,  # type: ignore[arg-type]
            name=name,
            old_wiki_commit="0" * 40,
            new_wiki_commit="0" * 40,
            installed_at="2026-01-01T00:00:00.000000Z",
            was_no_op=True,
            bak_files_written=(),
            dest=Path("/dev/null"),
            files_written=0,
        )

    monkeypatch.setattr("memtomem.context.install._update_asset", _fake_update_asset)

    update_agent(Path("/tmp/p"), "a")
    update_command(Path("/tmp/p"), "c")
    update_skill(Path("/tmp/p"), "s")

    assert seen == ["agents", "commands", "skills"]


# ════════════════════════════════════════════════════════════════════════
# Commit 4 — --all + wiki dirty warn
# ════════════════════════════════════════════════════════════════════════


# ── helpers for --all tests ─────────────────────────────────────────────


def _seed_known_projects(path: Path, project_roots: list[Path]) -> None:
    """Write a ``known_projects.json`` listing the given roots."""
    path.parent.mkdir(parents=True, exist_ok=True)
    doc = {
        "version": 1,
        "projects": [
            {"root": str(p), "added_at": "2026-01-01T00:00:00.000000Z", "label": None}
            for p in project_roots
        ],
    }
    path.write_text(json.dumps(doc), encoding="utf-8")


def _patch_known_projects_path(monkeypatch: pytest.MonkeyPatch, path: Path) -> None:
    """Make ``ContextGatewayConfig()`` in the CLI return *path*."""

    class _FakeCfg:
        known_projects_path = path

    monkeypatch.setattr(
        "memtomem.cli.context_cmd.ContextGatewayConfig",
        lambda: _FakeCfg(),
    )


# ── _classify_for_all_update: batch-once + 4-state coverage ─────────────


def test_classify_for_all_update_calls_current_commit_once(
    wiki_root: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """``current_commit`` runs ONCE regardless of how many projects are scanned."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    project_a = tmp_path / "proj_a"
    project_a.mkdir()
    project_b = tmp_path / "proj_b"
    project_b.mkdir()
    install_skill(project_a, "foo")
    install_skill(project_b, "foo")

    wiki = WikiStore.at_default()
    real_current_commit = wiki.current_commit
    call_count = {"n": 0}

    def _spy_current_commit() -> str:
        call_count["n"] += 1
        return real_current_commit()

    monkeypatch.setattr(WikiStore, "current_commit", lambda self: _spy_current_commit())

    new_commit, classifications = _classify_for_all_update(
        "skills", "foo", wiki=wiki, projects=[project_a, project_b]
    )

    # Wiki state read once for the whole batch.
    assert call_count["n"] == 1
    assert len(new_commit) == 40  # full SHA
    assert len(classifications) == 2


def test_classify_for_all_update_4_state_coverage(wiki_root: Path, tmp_path: Path) -> None:
    """One project per state — verifies all 4 states are reachable.

    Setup strategy: wiki has a single commit. All 4 projects install
    against that commit. Then 3 of them have their lockfile entries
    back-dated (or corrupted) to simulate drift / dirty / error states.
    The 4th stays at the live HEAD and classifies as ``unchanged``.
    """
    from memtomem.context.lockfile import Lockfile

    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    wiki = WikiStore.at_default()
    head_commit = wiki.current_commit()

    # State 1: "unchanged" — fresh install, lockfile pin == HEAD.
    proj_unchanged = tmp_path / "unchanged"
    proj_unchanged.mkdir()
    install_skill(proj_unchanged, "foo")

    # State 2: "update" — install, then back-date lockfile pin so HEAD
    # looks like a newer commit. Dest mtimes are ahead of installed_at
    # (just installed → fresh), so we also back-date installed_at to
    # ensure files are NOT classified dirty. Use a fixed past timestamp.
    # Wait — we WANT the dest tree to be clean (= reason="clean"), but
    # back-dating installed_at to far past makes ALL current files look
    # dirty. We need the OPPOSITE: future installed_at OR explicit clean
    # mtimes. Easiest: back-date both lockfile fields, then bump every
    # dest file's mtime to BEFORE installed_at_epoch via os.utime.
    proj_update = tmp_path / "update_clean"
    proj_update.mkdir()
    install_skill(proj_update, "foo")
    Lockfile.at(proj_update).upsert_entry(
        "skills",
        "foo",
        wiki_commit="0" * 40,  # ≠ HEAD → drift
        installed_at="2030-01-01T00:00:00.000000Z",  # future → all current files are clean
    )

    # State 3: "refuse" — drift + dirty.
    proj_refuse = tmp_path / "refuse"
    proj_refuse.mkdir()
    install_skill(proj_refuse, "foo")
    Lockfile.at(proj_refuse).upsert_entry(
        "skills",
        "foo",
        wiki_commit="0" * 40,
        # Past installed_at so every current file is dirty.
        installed_at="2020-01-01T00:00:00.000000Z",
    )

    # State 4: "error" — unknown lockfile version.
    proj_error = tmp_path / "error"
    proj_error.mkdir()
    (proj_error / ".memtomem").mkdir()
    (proj_error / ".memtomem" / "lock.json").write_text(
        json.dumps({"version": 99, "skills": {"foo": {"wiki_commit": "x", "installed_at": "y"}}}),
        encoding="utf-8",
    )

    new_commit, classifications = _classify_for_all_update(
        "skills",
        "foo",
        wiki=wiki,
        projects=[proj_unchanged, proj_update, proj_refuse, proj_error],
    )

    assert new_commit == head_commit
    states = {c.project_root.name: c.state for c in classifications}
    assert states == {
        "unchanged": "unchanged",
        "update_clean": "update",
        "refuse": "refuse",
        "error": "error",
    }
    # Cache pin: unchanged + error have no dirty_report; update + refuse do.
    by_name = {c.project_root.name: c for c in classifications}
    assert by_name["unchanged"].dirty_report is None
    assert by_name["error"].dirty_report is None
    assert by_name["update_clean"].dirty_report is not None
    assert by_name["update_clean"].dirty_report.reason == "clean"
    assert by_name["refuse"].dirty_report is not None
    assert by_name["refuse"].dirty_report.reason == "dirty"


def test_classify_rejects_path_traversal_name(wiki_root: Path, tmp_path: Path) -> None:
    """``_classify_for_all_update`` validates ``name`` at its boundary —
    a traversal name like ``../escape`` raises ``InvalidNameError``
    *before* any per-project loop runs.

    Defense in depth (`feedback_public_api_ship_time_validation`):
    even though the CLI ``update_cmd`` is the expected upstream caller
    and the single-asset path goes through ``_update_asset`` which
    already validates, the ``--all`` path reaches ``_classify_for_all_
    update`` directly and would feed ``name`` into ``Path`` joins
    (``src = wiki.root / asset_type / name``) without this check.
    """
    _initialized_wiki(wiki_root)
    proj = tmp_path / "p"
    proj.mkdir()

    wiki = WikiStore.at_default()
    with pytest.raises(InvalidNameError):
        _classify_for_all_update("skills", "../etc", wiki=wiki, projects=[proj])
    with pytest.raises(InvalidNameError):
        _classify_for_all_update("skills", "../../escape", wiki=wiki, projects=[proj])
    with pytest.raises(InvalidNameError):
        _classify_for_all_update("skills", "foo/bar", wiki=wiki, projects=[proj])


def test_classify_skips_projects_without_lockfile_entry(wiki_root: Path, tmp_path: Path) -> None:
    """Projects without a lockfile entry for this asset are silently
    skipped — no ``"skipped"`` row clutters the preview table."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    proj_with = tmp_path / "with_foo"
    proj_with.mkdir()
    install_skill(proj_with, "foo")

    proj_without = tmp_path / "without_foo"
    proj_without.mkdir()  # no lockfile, no install

    wiki = WikiStore.at_default()
    new_commit, classifications = _classify_for_all_update(
        "skills", "foo", wiki=wiki, projects=[proj_with, proj_without]
    )

    assert len(classifications) == 1
    assert classifications[0].project_root == proj_with


# ── CLI --all: empty store + no-projects-have-asset ─────────────────────


def test_cli_update_all_rejects_path_traversal_name(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The CLI surfaces ``InvalidNameError`` as a non-zero exit before
    touching any project. Pins that the validation gate is reachable
    through the user-facing path, not just from direct calls."""
    _initialized_wiki(wiki_root)
    proj = tmp_path / "p"
    proj.mkdir()
    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj])
    _patch_known_projects_path(monkeypatch, known)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "../etc", "--all", "--yes"])

    assert result.exit_code != 0
    assert "invalid" in result.output.lower()


def test_cli_update_all_empty_known_projects_exits_zero(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``--all`` with empty known_projects.json exits 0 with an info message.

    cron/CI safety: a first-run before any project registration must
    not fail with a non-zero exit.
    """
    _initialized_wiki(wiki_root)
    known = tmp_path / "known.json"
    _seed_known_projects(known, [])
    _patch_known_projects_path(monkeypatch, known)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all"])

    assert result.exit_code == 0, result.output
    assert "No known projects" in result.output


def test_cli_update_all_no_projects_have_asset_exits_zero(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """All registered projects exist on disk but none has the asset → exit 0."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    proj = tmp_path / "no_lock"
    proj.mkdir()  # exists but no install
    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj])
    _patch_known_projects_path(monkeypatch, known)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all"])

    assert result.exit_code == 0, result.output
    assert "No projects have skills/foo" in result.output


def test_cli_update_all_skips_disabled_projects(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A disabled (paused) known project is excluded from ``--all`` batch sync,
    matching the web Sync gate / ``sync_eligible``. The project exists on disk
    AND has the asset installed, so the *only* reason it is skipped is
    ``enabled=False`` — with it paused the batch finds nothing to update."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    paused = tmp_path / "paused"
    paused.mkdir()
    install_skill(paused, "foo")  # asset present, but the project is paused
    known = tmp_path / "known.json"
    known.write_text(
        json.dumps(
            {
                "version": 1,
                "projects": [
                    {
                        "root": str(paused),
                        "added_at": "2026-01-01T00:00:00Z",
                        "label": None,
                        "enabled": False,
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    _patch_known_projects_path(monkeypatch, known)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all"])

    assert result.exit_code == 0, result.output
    assert "nothing to update" in result.output


# ── CLI --all: refuse blocks batch ──────────────────────────────────────


def test_cli_update_all_refuse_without_force_blocks_batch(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """When any project is dirty and ``--force`` is absent, the entire
    batch refuses — no per-project writes happen."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    # Install in two projects; dirty one of them.
    proj_clean = tmp_path / "clean"
    proj_clean.mkdir()
    install_skill(proj_clean, "foo")

    proj_dirty = tmp_path / "dirty"
    proj_dirty.mkdir()
    install_skill(proj_dirty, "foo")
    edited = proj_dirty / ".memtomem" / "skills" / "foo" / "SKILL.md"
    edited.write_bytes(b"local\n")
    _bump_mtime(edited)

    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj_clean, proj_dirty])
    _patch_known_projects_path(monkeypatch, known)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all"])

    assert result.exit_code != 0
    assert "local edits" in result.output
    assert "--force" in result.output

    # Critical: the clean project's lockfile MUST NOT have been bumped.
    # If a partial write happened, this would catch the regression.
    lock_doc = json.loads((proj_clean / ".memtomem" / "lock.json").read_text())
    # wiki_commit on lockfile entry should still match the original install
    # (not the new wiki HEAD) — the entire batch refused before any write.
    assert (proj_clean / ".memtomem" / "skills" / "foo" / "SKILL.md").read_bytes() == b"v1\n"
    assert lock_doc["skills"]["foo"]["wiki_commit"] != WikiStore.at_default().current_commit()


# ── CLI --all: --yes invariants (no WARN unless --force) ────────────────


def test_cli_update_all_yes_skips_prompt_and_no_destructive_warning(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``--yes`` (without ``--force``) skips the confirm prompt and does
    NOT print the destructive WARNING. Plain ``--yes`` is non-destructive
    automation; the WARNING only fires for ``--force``-laden batches."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    proj_clean = tmp_path / "clean"
    proj_clean.mkdir()
    install_skill(proj_clean, "foo")
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj_clean])
    _patch_known_projects_path(monkeypatch, known)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all", "--yes"])

    assert result.exit_code == 0, result.output
    # Prompt-skip pin: input was empty but the command didn't hang.
    assert "Continue?" not in result.output
    # WARN-pin (negative): no destructive warning for plain --yes.
    assert "WARNING:" not in result.output
    # Update happened.
    assert "updated" in result.output
    assert (proj_clean / ".memtomem" / "skills" / "foo" / "SKILL.md").read_bytes() == b"v2\n"


def test_cli_update_all_yes_force_three_way_invariant(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``--yes --force`` invariant: WARNING printed AND prompt skipped AND
    batch executed (3-way conjunction). Tested as a single block so a
    regression in any one of the three is caught."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    proj_dirty = tmp_path / "dirty"
    proj_dirty.mkdir()
    install_skill(proj_dirty, "foo")
    edited = proj_dirty / ".memtomem" / "skills" / "foo" / "SKILL.md"
    edited.write_bytes(b"local\n")
    _bump_mtime(edited)
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj_dirty])
    _patch_known_projects_path(monkeypatch, known)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all", "--yes", "--force"])

    assert result.exit_code == 0, result.output
    # 1. WARNING printed.
    assert "WARNING:" in result.output
    # 2. Prompt skipped.
    assert "Continue?" not in result.output
    # 3. Batch executed — bytes changed and .bak survived the dirty edit.
    bak = edited.with_suffix(edited.suffix + ".bak")
    assert bak.is_file()
    assert bak.read_bytes() == b"local\n"
    assert edited.read_bytes() == b"v2\n"


# ── CLI --all: cache reuse pin ──────────────────────────────────────────


def test_cli_update_all_mid_loop_fs_error_continues(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A per-project ``OSError`` during execute marks that row ✗ and
    proceeds to the next project — the batch is not aborted on the
    first row's failure. The summary reflects the partial outcome.
    """
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    proj_a = tmp_path / "proj_a"
    proj_a.mkdir()
    install_skill(proj_a, "foo")

    proj_b = tmp_path / "proj_b"
    proj_b.mkdir()
    install_skill(proj_b, "foo")

    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj_a, proj_b])
    _patch_known_projects_path(monkeypatch, known)

    real_apply_update = install_module._apply_update
    call_count = {"n": 0}

    def _apply_update_first_fails(*args: object, **kwargs: object) -> object:
        call_count["n"] += 1
        if call_count["n"] == 1:
            raise OSError("simulated fs error on first project")
        return real_apply_update(*args, **kwargs)  # type: ignore[arg-type]

    monkeypatch.setattr("memtomem.cli.context_cmd._apply_update", _apply_update_first_fails)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all", "--yes"])

    # Row failures now exit 1 (mirrors install --all, #1247) — the point
    # pinned here is CONTINUATION, not a clean exit: the batch must not
    # abort on the first row's failure.
    assert result.exit_code == 1, result.output
    # First project marked ✗, second project marked ✓ — both rows reached.
    assert "✗" in result.output
    assert "simulated fs error" in result.output
    assert "✓" in result.output
    # Summary: 1 updated, 1 failed.
    assert "1 updated" in result.output
    assert "1 failed" in result.output


def test_cli_update_all_re_walks_dirty_at_apply_time(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Execute phase MUST re-classify the dest tree at apply time — exactly
    one extra ``is_asset_dirty`` call per actionable project (#1247 id 13).

    Inverted pin: this test previously asserted ``call_count == 1`` ("the
    execute phase consumes the cached dirty_report, no re-walk") — which
    pinned the bug. The confirm prompt between classify and execute is
    unbounded, so gating writes on the classify-time report silently
    clobbered edits made during the prompt (and ``--force`` baked too few
    files). Now: classify walks once for the preview table, and
    ``_apply_update`` walks once for the gate. Total = 2 per actionable
    project; a count of 1 means apply-time gating regressed to the stale
    snapshot, 3+ means an accidental extra walk crept in.
    """
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    proj_dirty = tmp_path / "dirty"
    proj_dirty.mkdir()
    install_skill(proj_dirty, "foo")
    edited = proj_dirty / ".memtomem" / "skills" / "foo" / "SKILL.md"
    edited.write_bytes(b"local\n")
    _bump_mtime(edited)
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj_dirty])
    _patch_known_projects_path(monkeypatch, known)

    real_is_asset_dirty = install_module.is_asset_dirty
    call_count = {"n": 0}

    def _spy(*args: object, **kwargs: object):
        call_count["n"] += 1
        return real_is_asset_dirty(*args, **kwargs)  # type: ignore[arg-type]

    monkeypatch.setattr(install_module, "is_asset_dirty", _spy)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all", "--yes", "--force"])

    assert result.exit_code == 0, result.output
    # 2 = classification's preview walk + _apply_update's gate walk.
    assert call_count["n"] == 2


def test_single_asset_update_walks_dirty_exactly_once(
    wiki_root: Path,
    project_cwd: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The non-batch path has no confirm prompt between classify and apply,
    so moving the walk into ``_apply_update`` (#1247 id 13) must not add a
    second walk there: ``_update_asset`` no longer pre-computes.
    """
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    install_skill(project_cwd, "foo")
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    real_is_asset_dirty = install_module.is_asset_dirty
    call_count = {"n": 0}

    def _spy(*args: object, **kwargs: object):
        call_count["n"] += 1
        return real_is_asset_dirty(*args, **kwargs)  # type: ignore[arg-type]

    monkeypatch.setattr(install_module, "is_asset_dirty", _spy)

    result = update_skill(project_cwd, "foo")

    assert result.was_no_op is False
    assert call_count["n"] == 1


# ── #1247 id 13: edits during the --all confirm prompt ──────────────────


def test_cli_update_all_refuses_edit_made_during_confirm_prompt(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A project clean at classify whose dest is edited DURING the confirm
    prompt must be refused at apply time, not silently overwritten (#1247
    id 13).

    The confirm window is unbounded — a human sits on the multi-project
    preview table while other sessions write. Pre-fix the execute phase
    consumed the stale clean ``dirty_report`` from classification, so
    ``copy_tree_atomic`` replaced the mid-prompt edit with wiki bytes:
    exit 0, no refusal, no .bak.
    """
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})

    proj = tmp_path / "race_proj"
    proj.mkdir()
    install_skill(proj, "foo")
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj])
    _patch_known_projects_path(monkeypatch, known)

    target = proj / ".memtomem" / "skills" / "foo" / "SKILL.md"

    def _edit_then_confirm(*args: object, **kwargs: object) -> bool:
        target.write_bytes(b"edited during prompt\n")
        _bump_mtime(target)
        return True

    monkeypatch.setattr("memtomem.cli.context_cmd.click.confirm", _edit_then_confirm)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all"])

    assert result.exit_code == 1, result.output
    assert target.read_bytes() == b"edited during prompt\n"
    assert not target.with_suffix(".md.bak").exists()


def test_cli_update_all_force_baks_file_dirtied_during_confirm_prompt(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``--force`` must .bak files that became dirty DURING the confirm
    prompt, not just the classify-time dirty set (#1247 id 13).

    Pre-fix the .bak loop iterated the stale ``dirty_files`` — a file edited
    mid-prompt was clobbered bak-less even though --force promises "each
    modified file gets a .bak sibling".
    """
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n", "EXTRA.md": b"e1\n"})

    proj = tmp_path / "race_proj"
    proj.mkdir()
    install_skill(proj, "foo")

    dest = proj / ".memtomem" / "skills" / "foo"
    first = dest / "SKILL.md"
    second = dest / "EXTRA.md"
    first.write_bytes(b"edited before classify\n")
    _bump_mtime(first)

    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n", "EXTRA.md": b"e2\n"})

    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj])
    _patch_known_projects_path(monkeypatch, known)

    def _edit_then_confirm(*args: object, **kwargs: object) -> bool:
        second.write_bytes(b"edited during prompt\n")
        _bump_mtime(second)
        return True

    monkeypatch.setattr("memtomem.cli.context_cmd.click.confirm", _edit_then_confirm)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all", "--force"])

    assert result.exit_code == 0, result.output
    first_bak = first.with_suffix(".md.bak")
    second_bak = second.with_suffix(".md.bak")
    assert first_bak.exists()
    assert first_bak.read_bytes() == b"edited before classify\n"
    assert second_bak.exists(), result.output
    assert second_bak.read_bytes() == b"edited during prompt\n"
    assert second.read_bytes() == b"e2\n"


# ── CLI: wiki dirty warn timing (single-asset & --all) ──────────────────


def test_cli_update_wiki_dirty_warn_single_asset(
    wiki_root: Path,
    project_cwd: Path,
) -> None:
    """Single-asset path: wiki dirty → warn on stderr at update entry."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    install_skill(project_cwd, "foo")

    # Make the wiki dirty.
    (wiki_root / "untracked_marker.txt").write_text("wip", encoding="utf-8")

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo"])

    assert result.exit_code == 0, result.output
    assert "wiki has uncommitted changes" in result.output


def test_cli_update_wiki_dirty_warn_all(
    wiki_root: Path,
    project_cwd: Path,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """--all path: wiki dirty → warn on stderr BEFORE classification."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    proj = tmp_path / "p"
    proj.mkdir()
    install_skill(proj, "foo")

    (wiki_root / "untracked_marker.txt").write_text("wip", encoding="utf-8")

    known = tmp_path / "known.json"
    _seed_known_projects(known, [proj])
    _patch_known_projects_path(monkeypatch, known)

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo", "--all", "--yes"])

    assert result.exit_code == 0, result.output
    assert "wiki has uncommitted changes" in result.output


def test_cli_update_no_wiki_dirty_warn_when_clean(
    wiki_root: Path,
    project_cwd: Path,
) -> None:
    """Negative pin: clean wiki → no dirty warn line."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    install_skill(project_cwd, "foo")

    runner = CliRunner()
    result = runner.invoke(context_group, ["update", "skill", "foo"])

    assert result.exit_code == 0, result.output
    assert "wiki has uncommitted changes" not in result.output


# ── B1: upstream-deletion reconcile + deletion-dirty (#1247) ─────────────


def _delete_wiki_skill_file(wiki_root_path: Path, name: str, relpath: str) -> str:
    """``git rm`` one file from a wiki skill + commit. Returns the new SHA.

    Removes the file from the wiki working tree too (``git rm``), so the
    live-tree copy source ``_update_asset`` reads matches the new commit.
    """
    subprocess.run(
        ["git", "-C", str(wiki_root_path), "rm", "-q", f"skills/{name}/{relpath}"],
        check=True,
        capture_output=True,
    )
    subprocess.run(
        ["git", "-C", str(wiki_root_path), "commit", "-m", f"delete {name}/{relpath}"],
        check=True,
        capture_output=True,
    )
    return WikiStore.at_default().current_commit()


def _lock_entry(project: Path, asset_type: str, name: str) -> dict:
    doc = json.loads((project / ".memtomem" / "lock.json").read_text(encoding="utf-8"))
    return doc[asset_type][name]


def _rewrite_lock_entry(project: Path, asset_type: str, name: str, entry: dict) -> None:
    lock_path = project / ".memtomem" / "lock.json"
    doc = json.loads(lock_path.read_text(encoding="utf-8"))
    doc[asset_type][name] = entry
    lock_path.write_text(json.dumps(doc, indent=2), encoding="utf-8")


def test_update_removes_upstream_deleted_file(wiki_root: Path, tmp_path: Path) -> None:
    """Core #1247 reported-critical: update must delete files removed upstream.

    Pre-B1 the additive copy left ``scripts/run-old.py`` on disk forever
    while the lockfile claimed the new commit and status said ``ok``.
    """
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "web", {"SKILL.md": b"v1\n", "scripts/run-old.py": b"old\n"})
    install_skill(tmp_path, "web")

    new_sha = _delete_wiki_skill_file(wiki_root, "web", "scripts/run-old.py")
    result = update_skill(tmp_path, "web")

    dest = tmp_path / ".memtomem" / "skills" / "web"
    assert not (dest / "scripts" / "run-old.py").exists()
    # The directory the removal emptied is pruned too.
    assert not (dest / "scripts").exists()
    assert (dest / "SKILL.md").read_bytes() == b"v1\n"

    assert [p.name for p in result.files_removed] == ["run-old.py"]

    entry = _lock_entry(tmp_path, "skills", "web")
    assert entry["wiki_commit"] == new_sha
    assert entry["files"] == ["SKILL.md"]
    assert entry["files_commit"] == new_sha


def test_update_force_removes_deleted_and_edited_file_with_bak(
    wiki_root: Path, tmp_path: Path
) -> None:
    """Upstream-deleted + locally-edited: refuse without --force; with
    --force the edit is preserved as ``.bak`` and the live file is removed."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "web", {"SKILL.md": b"v1\n", "notes.md": b"wiki\n"})
    install_skill(tmp_path, "web")

    dest = tmp_path / ".memtomem" / "skills" / "web"
    (dest / "notes.md").write_bytes(b"user edit\n")
    _bump_mtime(dest / "notes.md")

    _delete_wiki_skill_file(wiki_root, "web", "notes.md")

    with pytest.raises(StaleInstallError):
        update_skill(tmp_path, "web")

    result = update_skill(tmp_path, "web", force=True)

    assert (dest / "notes.md.bak").read_bytes() == b"user edit\n"
    assert not (dest / "notes.md").exists()
    assert [p.name for p in result.files_removed] == ["notes.md"]


def test_update_force_keeps_user_added_file_when_manifest_present(
    wiki_root: Path, tmp_path: Path
) -> None:
    """Manifest rule 1: a dest-only file NOT in the recorded manifest is
    user-authored content — never silently deleted, even under --force."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "web", {"SKILL.md": b"v1\n"})
    install_skill(tmp_path, "web")

    dest = tmp_path / ".memtomem" / "skills" / "web"
    (dest / "extra.md").write_bytes(b"mine\n")
    _bump_mtime(dest / "extra.md")

    _modify_wiki_skill(wiki_root, "web", {"SKILL.md": b"v2\n"})

    result = update_skill(tmp_path, "web", force=True)

    assert (dest / "extra.md").read_bytes() == b"mine\n"
    assert result.files_removed == ()


def test_update_legacy_entry_removes_stale_dest_only_file(wiki_root: Path, tmp_path: Path) -> None:
    """Legacy lock entry (no manifest): a dest-only file whose mtime is
    ``<= installed_at`` is provably old wiki bytes — removed."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "web", {"SKILL.md": b"v1\n", "old.md": b"old\n"})
    install_skill(tmp_path, "web")

    # Simulate a pre-B1 entry: strip the manifest keys.
    entry = _lock_entry(tmp_path, "skills", "web")
    entry.pop("files", None)
    entry.pop("files_commit", None)
    _rewrite_lock_entry(tmp_path, "skills", "web", entry)

    _delete_wiki_skill_file(wiki_root, "web", "old.md")
    result = update_skill(tmp_path, "web")

    dest = tmp_path / ".memtomem" / "skills" / "web"
    assert not (dest / "old.md").exists()
    assert [p.name for p in result.files_removed] == ["old.md"]


def test_update_noop_with_user_deleted_file_does_not_resurrect(
    wiki_root: Path, tmp_path: Path
) -> None:
    """Codex design-gate M1 regression: unchanged wiki HEAD short-circuits
    to a no-op BEFORE the dirty walk — nothing is written, so a user-deleted
    file must stay deleted (diagnosis surface is ``mm context status``)."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "web", {"SKILL.md": b"v1\n", "extra.md": b"x\n"})
    install_skill(tmp_path, "web")

    dest = tmp_path / ".memtomem" / "skills" / "web"
    (dest / "extra.md").unlink()

    result = update_skill(tmp_path, "web")

    assert result.was_no_op is True
    assert not (dest / "extra.md").exists()


def test_update_refuses_on_user_deleted_file_after_wiki_advance(
    wiki_root: Path, tmp_path: Path
) -> None:
    """Deletion-dirty (#1247 item 12): after the wiki advances, a user-deleted
    file counts as a local edit — refuse without --force; --force restores it
    as an explicit, informed action."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "web", {"SKILL.md": b"v1\n", "keep.md": b"k\n"})
    install_skill(tmp_path, "web")

    dest = tmp_path / ".memtomem" / "skills" / "web"
    (dest / "keep.md").unlink()

    _modify_wiki_skill(wiki_root, "web", {"SKILL.md": b"v2\n"})

    with pytest.raises(StaleInstallError) as excinfo:
        update_skill(tmp_path, "web")
    assert "deleted" in str(excinfo.value)

    result = update_skill(tmp_path, "web", force=True)
    assert (dest / "keep.md").read_bytes() == b"k\n"
    assert result.was_no_op is False


def test_update_stale_manifest_guard_ignores_mismatched_files_commit(
    wiki_root: Path, tmp_path: Path
) -> None:
    """An entry whose ``files_commit`` doesn't match ``wiki_commit`` (older
    tool rewrote the entry, or a hand-merged lock.json) must behave as if no
    manifest exists: no missing-file dirty, legacy mtime rule applies.

    The digest keys are stripped too — install records them since #1247
    id 15, and a valid digest pairing would (correctly) catch the deletion
    on the digest branch before the manifest guard is ever consulted; the
    digest-branch deletion behavior is pinned in test_dirty_digest.py.
    """
    from memtomem.context.dirty import is_asset_dirty

    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "web", {"SKILL.md": b"v1\n", "a.md": b"a\n"})
    install_skill(tmp_path, "web")

    entry = _lock_entry(tmp_path, "skills", "web")
    entry["files_commit"] = "0" * 40
    entry.pop("digests", None)
    entry.pop("digests_installed_at", None)
    _rewrite_lock_entry(tmp_path, "skills", "web", entry)

    (tmp_path / ".memtomem" / "skills" / "web" / "a.md").unlink()

    report = is_asset_dirty(tmp_path, "skills", "web")
    assert report.reason == "clean"
    assert report.missing_files == ()


def test_update_removes_file_replaced_by_symlink_upstream(wiki_root: Path, tmp_path: Path) -> None:
    """Codex implementation-gate M1: ``copy_tree_atomic`` skips symlinks, so
    a wiki file replaced by a symlink is NOT written to dest — reconcile
    membership must use the copier's effective file set, or the old regular
    file survives and is recorded clean in the manifest."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "web", {"SKILL.md": b"v1\n", "link.md": b"regular\n"})
    install_skill(tmp_path, "web")

    src_link = wiki_root / "skills" / "web" / "link.md"
    src_link.unlink()
    src_link.symlink_to("SKILL.md")
    subprocess.run(["git", "-C", str(wiki_root), "add", "."], check=True, capture_output=True)
    subprocess.run(
        ["git", "-C", str(wiki_root), "commit", "-m", "replace link.md with symlink"],
        check=True,
        capture_output=True,
    )

    result = update_skill(tmp_path, "web")

    dest = tmp_path / ".memtomem" / "skills" / "web"
    assert not (dest / "link.md").exists()
    assert [p.name for p in result.files_removed] == ["link.md"]
    entry = _lock_entry(tmp_path, "skills", "web")
    assert entry["files"] == ["SKILL.md"]


def test_wiki_shipped_bak_never_installed_or_stranded(wiki_root: Path, tmp_path: Path) -> None:
    """#1247 item 10 (re-review major): the wiki-install copies share the
    walker's suffix skip, so a wiki-shipped ``*.bak`` is never installed —
    it can't dodge dirty tracking, collide with the ``--force`` preservation
    namespace, or strand stale bytes when upstream later replaces it."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "web", {"SKILL.md": b"v1\n", "leftover.md.bak": b"wiki bak\n"})
    result = install_skill(tmp_path, "web")

    dest = tmp_path / ".memtomem" / "skills" / "web"
    assert not (dest / "leftover.md.bak").exists()
    assert result.files_written == 1
    assert _lock_entry(tmp_path, "skills", "web")["files"] == ["SKILL.md"]

    # Upstream replaces the .bak with a symlink (the re-review repro shape):
    # update stays clean — nothing was installed, so nothing can go stale.
    src_bak = wiki_root / "skills" / "web" / "leftover.md.bak"
    src_bak.unlink()
    src_bak.symlink_to("SKILL.md")
    subprocess.run(["git", "-C", str(wiki_root), "add", "."], check=True, capture_output=True)
    subprocess.run(
        ["git", "-C", str(wiki_root), "commit", "-m", "replace bak with symlink"],
        check=True,
        capture_output=True,
    )

    upd = update_skill(tmp_path, "web")
    assert upd.files_removed == ()
    assert not (dest / "leftover.md.bak").exists()


# ── B2: Gate A on update ingress (#1247 id 3) ────────────────────────────


# AKIA fixture per feedback_force_unsafe_redaction_valve_only.md — clean
# strings would pass the scan and false-negative every block assertion.
SECRET = "api_key=AKIA1234567890ABCDEF"


class TestUpdatePrivacyGate:
    """Gate A on the update ingress (ADR-0011 §5, design tests 2 / 3 / 9)."""

    @pytest.fixture(autouse=True)
    def _reset_privacy_counters(self):
        privacy.reset_for_tests()
        yield
        privacy.reset_for_tests()

    def test_update_blocks_on_poisoned_wiki_src(self, wiki_root: Path, tmp_path: Path) -> None:
        """Design test 2: secret lands in the wiki → update hard-refuses
        BEFORE any dest write — zero residue (old bytes, old pin, no .bak)."""
        _initialized_wiki(wiki_root)
        old_commit = _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
        project = tmp_path
        install_skill(project, "foo")

        lock_path = project / ".memtomem" / "lock.json"
        pre_lock_bytes = lock_path.read_bytes()

        _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": f"v2\n{SECRET}\n".encode()})

        with pytest.raises(PrivacyBlockedError):
            update_skill(project, "foo")

        dest = project / ".memtomem" / "skills" / "foo"
        # Zero-residue: dest still carries the OLD clean bytes.
        assert (dest / "SKILL.md").read_bytes() == b"v1\n"
        # Lockfile untouched — wiki_commit is still the install-time pin.
        assert _lock_entry(project, "skills", "foo")["wiki_commit"] == old_commit
        assert lock_path.read_bytes() == pre_lock_bytes
        # No .bak anywhere under dest (no preservation pass ran).
        assert list(dest.rglob("*.bak")) == []

    def test_force_update_blocks_on_secret_in_dirty_dest_file(
        self, wiki_root: Path, tmp_path: Path
    ) -> None:
        """Design test 3: --force scans each dirty file BEFORE its .bak
        copy2 — the secret never gets a second tracked copy."""
        _initialized_wiki(wiki_root)
        _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
        project = tmp_path
        install_skill(project, "foo")

        dest = project / ".memtomem" / "skills" / "foo"
        edited = dest / "SKILL.md"
        dirty_bytes = f"local edit\n{SECRET}\n".encode()
        edited.write_bytes(dirty_bytes)
        _bump_mtime(edited)

        # Wiki advance is CLEAN — the block must come from the dirty-file
        # scan, not the src scan.
        _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

        lock_path = project / ".memtomem" / "lock.json"
        pre_lock_bytes = lock_path.read_bytes()

        with pytest.raises(PrivacyBlockedError):
            update_skill(project, "foo", force=True)

        # NO .bak landed; the dirty bytes are intact; lockfile untouched.
        assert list(dest.rglob("*.bak")) == []
        assert edited.read_bytes() == dirty_bytes
        assert lock_path.read_bytes() == pre_lock_bytes
        # Surface attribution (#1246/#1248 real-ingress rule): single-asset
        # update audits as cli_context_update, never the batch surface.
        by_tool = privacy.snapshot()["by_tool"]
        assert by_tool["cli_context_update"]["blocked"] >= 1
        assert "cli_context_update_all" not in by_tool

    def test_cli_update_maps_block_to_exit_1_without_secret_echo(
        self, wiki_root: Path, project_cwd: Path
    ) -> None:
        """Design test 9 (update leg): PrivacyBlockedError → ClickException,
        exit 1, Gate A message shown, no secret echo, no traceback dump."""
        _initialized_wiki(wiki_root)
        _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
        install_skill(project_cwd, "foo")
        _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": f"v2\n{SECRET}\n".encode()})

        runner = CliRunner()
        result = runner.invoke(context_group, ["update", "skill", "foo"])

        assert result.exit_code == 1, result.output
        # ClickException mapping pin: the Gate A message reaches stdout
        # (an unhandled PrivacyBlockedError would leave output empty).
        assert "Gate A" in result.output
        # The block message must never echo the secret back (privacy.py
        # contract: path + hit count only).
        assert "AKIA1234567890ABCDEF" not in result.output
        assert "Traceback" not in result.output


# ── flat-layout guard (#1247 id 0) ───────────────────────────────────────


def _seed_wiki_agent(wiki_root_path: Path, name: str, files: dict[str, bytes]) -> str:
    """Add ``agents/<name>/`` to wiki + commit. Returns the commit SHA."""
    agent_dir = wiki_root_path / "agents" / name
    agent_dir.mkdir(parents=True, exist_ok=True)
    for relpath, data in files.items():
        target = agent_dir / relpath
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(data)
    subprocess.run(["git", "-C", str(wiki_root_path), "add", "."], check=True, capture_output=True)
    subprocess.run(
        ["git", "-C", str(wiki_root_path), "commit", "-m", f"agents/{name}"],
        check=True,
        capture_output=True,
    )
    return WikiStore.at_default().current_commit()


def _seed_flat_installed_agent(
    project: Path, name: str, pin: str, *, installed_at: str | None = None
) -> Path:
    """Seed the #1247 id 0 population: flat ``agents/<name>.md`` + lock entry, no dir.

    By default the flat file's mtime is backdated 10s before
    ``installed_at`` so the strict ``>`` rule lands deterministically on
    the clean side (float↔ns mtime round-trips make exact equality
    unreliable). Pass an explicit ``installed_at`` to simulate corrupt
    entries (mtime is left alone — the unprovable arm doesn't compare).
    """
    from memtomem.context.lockfile import Lockfile

    flat = project / ".memtomem" / "agents" / f"{name}.md"
    flat.parent.mkdir(parents=True, exist_ok=True)
    flat.write_bytes(b"# flat agent\n")
    if installed_at is None:
        installed_at = datetime.now(timezone.utc).isoformat()
        past = datetime.now(timezone.utc).timestamp() - 10.0
        os.utime(flat, (past, past))
    Lockfile.at(project).upsert_entry("agents", name, wiki_commit=pin, installed_at=installed_at)
    return flat


def test_update_refuses_dirty_flat_layout(wiki_root: Path, tmp_path: Path) -> None:
    """#1247 id 0: flat file + lock entry + local edit → refuse, not silent shadow.

    Pre-fix: the flat population classified ``missing_dest``, sailed past
    the dirty gate, the dir layout landed next to the flat file, and
    dir-wins fan-out silently stopped serving the user's edited bytes —
    no refusal, no --force, no .bak."""
    _initialized_wiki(wiki_root)
    pin = _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v1\n"})
    flat = _seed_flat_installed_agent(tmp_path, "foo", pin)
    flat.write_bytes(b"# local edit\n")
    _bump_mtime(flat)
    _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v2\n"})  # HEAD advances

    with pytest.raises(StaleInstallError) as excinfo:
        update_agent(tmp_path, "foo")

    msg = str(excinfo.value)
    assert "flat-layout" in msg
    assert "mm context migrate agent foo" in msg
    assert "--force" in msg
    # Refusal left zero residue: no dir layout, flat bytes untouched.
    assert not (tmp_path / ".memtomem" / "agents" / "foo").exists()
    assert flat.read_bytes() == b"# local edit\n"


def test_update_force_shadows_flat_but_preserves_it(wiki_root: Path, tmp_path: Path) -> None:
    """--force writes the dir layout; the flat file stays on disk, no .bak.

    Nothing overwrites the flat bytes (they are merely shadowed by
    dir-wins), so a .bak would be a redundant copy — design-gate Q1."""
    _initialized_wiki(wiki_root)
    pin = _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v1\n"})
    flat = _seed_flat_installed_agent(tmp_path, "foo", pin)
    flat.write_bytes(b"# local edit\n")
    _bump_mtime(flat)
    _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v2\n"})

    result = update_agent(tmp_path, "foo", force=True)

    assert result.was_no_op is False
    dir_file = tmp_path / ".memtomem" / "agents" / "foo" / "agent.md"
    assert dir_file.read_bytes() == b"wiki v2\n"
    assert flat.read_bytes() == b"# local edit\n"
    assert not flat.with_suffix(".md.bak").exists()
    assert result.bak_files_written == ()


def test_update_clean_flat_proceeds_and_warns(
    wiki_root: Path, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    """A provably-clean flat file doesn't block (no user-authored bytes at
    risk) — but the shadowing is logged with a migrate hint."""
    _initialized_wiki(wiki_root)
    pin = _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v1\n"})
    flat = _seed_flat_installed_agent(tmp_path, "foo", pin)
    _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v2\n"})

    with caplog.at_level("WARNING", logger="memtomem.context.install"):
        result = update_agent(tmp_path, "foo")

    assert result.was_no_op is False
    assert (tmp_path / ".memtomem" / "agents" / "foo" / "agent.md").read_bytes() == b"wiki v2\n"
    assert flat.is_file()
    messages = [rec.getMessage() for rec in caplog.records]
    assert any("shadow" in m for m in messages)
    assert any("mm context migrate" in m for m in messages)


def test_update_refuses_flat_with_unprovable_installed_at(wiki_root: Path, tmp_path: Path) -> None:
    """Design-gate M1: a malformed installed_at classifies never_installed
    and must NOT skip the flat guard — unprovable counts as dirty."""
    _initialized_wiki(wiki_root)
    pin = _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v1\n"})
    flat = _seed_flat_installed_agent(tmp_path, "foo", pin, installed_at="yesterday")
    _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v2\n"})

    with pytest.raises(StaleInstallError) as excinfo:
        update_agent(tmp_path, "foo")

    assert "flat-layout" in str(excinfo.value)
    assert not (tmp_path / ".memtomem" / "agents" / "foo").exists()
    assert flat.is_file()


def test_classify_for_all_update_flat_dirty_is_refuse(wiki_root: Path, tmp_path: Path) -> None:
    """The --all preview shows the same refusal the execute gate raises
    (#1247 id 0 preview/execute parity)."""
    _initialized_wiki(wiki_root)
    pin = _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v1\n"})
    flat = _seed_flat_installed_agent(tmp_path, "foo", pin)
    flat.write_bytes(b"# local edit\n")
    _bump_mtime(flat)
    _seed_wiki_agent(wiki_root, "foo", {"agent.md": b"wiki v2\n"})

    _new_commit, classifications = _classify_for_all_update(
        "agents", "foo", wiki=WikiStore.at_default(), projects=[tmp_path]
    )

    [c] = classifications
    assert c.state == "refuse"
    assert "flat layout" in (c.reason or "")
    assert "migrate" in (c.reason or "")


# ── unprovable install record (#1247 impl gate) ──────────────────────────


def _corrupt_installed_at(project: Path, asset_type: str, name: str, value: str) -> None:
    lock_path = project / ".memtomem" / "lock.json"
    doc = json.loads(lock_path.read_text(encoding="utf-8"))
    doc[asset_type][name]["installed_at"] = value
    lock_path.write_text(json.dumps(doc), encoding="utf-8")


def test_update_refuses_unusable_install_record_over_existing_dest(
    wiki_root: Path, tmp_path: Path
) -> None:
    """Impl-gate: malformed installed_at over an EXISTING dest must refuse,
    not silently overwrite. Pre-#1247 this crashed with a ValueError; the
    id-1 degrade (crash → never_installed) must not turn the crash into
    data loss by sailing past the dirty gate."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    install_skill(tmp_path, "foo")
    edited = tmp_path / ".memtomem" / "skills" / "foo" / "SKILL.md"
    edited.write_bytes(b"local edit\n")
    _bump_mtime(edited)
    _corrupt_installed_at(tmp_path, "skills", "foo", "yesterday")
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    with pytest.raises(StaleInstallError) as excinfo:
        update_skill(tmp_path, "foo")

    msg = str(excinfo.value)
    assert "install record unusable" in msg
    assert "--force" in msg
    assert edited.read_bytes() == b"local edit\n"  # nothing overwritten
    assert not edited.with_suffix(".md.bak").exists()  # refusal leaves zero residue


def test_update_force_unusable_record_baks_every_file(wiki_root: Path, tmp_path: Path) -> None:
    """--force over an unprovable record preserves EVERY current dest file
    as .bak — with no epoch there is no dirty subset to limit to."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n", "extra.md": b"e1\n"})
    install_skill(tmp_path, "foo")
    edited = tmp_path / ".memtomem" / "skills" / "foo" / "SKILL.md"
    edited.write_bytes(b"local edit\n")
    _bump_mtime(edited)
    _corrupt_installed_at(tmp_path, "skills", "foo", "yesterday")
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    result = update_skill(tmp_path, "foo", force=True)

    dest = tmp_path / ".memtomem" / "skills" / "foo"
    assert (dest / "SKILL.md").read_bytes() == b"v2\n"
    assert (dest / "SKILL.md.bak").read_bytes() == b"local edit\n"
    assert (dest / "extra.md.bak").read_bytes() == b"e1\n"
    assert len(result.bak_files_written) == 2


def test_classify_for_all_update_unusable_record_is_refuse(wiki_root: Path, tmp_path: Path) -> None:
    """Batch preview parity for the unprovable gate."""
    _initialized_wiki(wiki_root)
    _seed_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v1\n"})
    install_skill(tmp_path, "foo")
    _corrupt_installed_at(tmp_path, "skills", "foo", "yesterday")
    _modify_wiki_skill(wiki_root, "foo", {"SKILL.md": b"v2\n"})

    _new_commit, classifications = _classify_for_all_update(
        "skills", "foo", wiki=WikiStore.at_default(), projects=[tmp_path]
    )

    [c] = classifications
    assert c.state == "refuse"
    assert "install record unusable" in (c.reason or "")
