ersatztv/scripts/tests/test_worktree_ownership_guard.py

"""The worktree-ownership mechanism is TWO files, and this drives both halves as one thing.

`pretooluse-worktree-guard.sh` denies a `git commit`/`git merge` inside a worktree another session
created. It can only do that because `posttooluse-worktree-marker.sh` wrote the
`.claude-worktree-owner` marker at `git worktree add` time. Neither file had a test, and — the part
that makes this rank second in ersatztv#785 — **the halves had never been exercised together**, so a
regression in either one is invisible: the marker hook silently writing nothing and the guard hook
silently reading nothing produce the identical outcome, which is *the commit is allowed*, which is
also what a correct fail-open looks like.

Both hooks are deliberately fail-open (`docs/decisions` — the main tree is never marked, and
pre-convention worktrees have no marker), and that is exactly why an absent mechanism is
indistinguishable from a working one from the outside. It is the shape
`testing.guard-ships-with-mutation-proof` was written for: "in every case a human had read the guard
and believed it worked. The guard was not subtly wrong, it was *absent*."

So this file:

  * drives the REAL pair end to end over a REAL `git worktree add`, in the real payload shape —
    marker hook first as the harness would fire it, then the guard hook;
  * carries a negative control (a non-mutating git command) and a fail-open control (an unmarked
    worktree), because a guard that denied everything would satisfy the deny assertions;
  * and performs FOUR clause-level mutations: the guard's marker read, the guard's ownership
    comparison, the command-detection alternation (`commit|merge`), and the *other file's* marker
    write. The last is the one that could not exist while the halves were tested apart.

These four are the clauses whose disarm this file detects. They are not every line in either hook —
the `git -C` / `cd` redirection extraction and the marker hook's argument parsing are exercised
behaviourally but not mutated, and the grade in `docs/guard-inventory.md` covers the clause its
cited case mutates, not the whole file.
"""

from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[2]
GUARD = REPO_ROOT / ".claude" / "hooks" / "pretooluse-worktree-guard.sh"
MARKER_HOOK = REPO_ROOT / ".claude" / "hooks" / "posttooluse-worktree-marker.sh"
MARKER_NAME = ".claude-worktree-owner"

SESSION_A = "session-aaaa-1111"
SESSION_B = "session-bbbb-2222"


def _env() -> dict:
    """The subprocess environment, built PER CALL — never snapshotted at import.

    Two things it must get right.

    `CLAUDE_PROJECT_DIR` is pinned because the hooks resolve `scripts/hook-fire-log.sh` from it,
    falling back to a path relative to their own location; a MUTATED copy lives in tmp_path, where
    that fallback finds nothing. Without the pin the mutant differs from the subject in a second way
    and the comparison stops being about the mutated clause.

    And it is a FUNCTION rather than a module-level dict because `conftest.py`'s autouse
    `isolate_hook_fire_log` fixture monkeypatches `ETV_HOOK_FIRE_LOG_DIR` into `os.environ` at test
    setup — which happens AFTER this module is imported. A `{**os.environ}` snapshot taken at import
    time captures the environment as it was before the fixture ran, so every hook subprocess writes
    to the REAL `$HOME/.cache/ersatztv/hook-fire/` log instead of the fixture's tmp dir. That is not
    untidiness: it is #776's defect reintroduced in the file that is meant to prove #776's hooks,
    and it corrupts the `hook-fire-log.sh report` surface this repo cites as the observability claim
    for every guard still graded NONE. The reproduction, rather than a figure whose evidence has
    since been deleted: reintroduce the snapshot and run this file, then count records for the two
    synthetic session ids below — 58 per run, on macOS and Linux alike.
    """
    return {**os.environ, "CLAUDE_PROJECT_DIR": str(REPO_ROOT)}


def _git(cwd: Path, *args: str) -> str:
    p = subprocess.run(
        ["git", *args],
        cwd=str(cwd),
        check=True,
        capture_output=True,
        text=True,
        env={
            **os.environ,
            "GIT_AUTHOR_NAME": "t",
            "GIT_AUTHOR_EMAIL": "t@e",
            "GIT_COMMITTER_NAME": "t",
            "GIT_COMMITTER_EMAIL": "t@e",
        },
    )
    return p.stdout


def _scratch_repo(tmp_path: Path) -> Path:
    repo = tmp_path / "main-tree"
    repo.mkdir()
    _git(repo, "init", "-q", "-b", "main", ".")
    (repo / "f.txt").write_text("one\n")
    _git(repo, "add", "f.txt")
    _git(repo, "commit", "-qm", "init")
    return repo


def _run(hook: Path, payload: dict, cwd: Path) -> tuple[int, bytes]:
    p = subprocess.run(
        ["bash", str(hook)],
        input=json.dumps(payload).encode(),
        capture_output=True,
        cwd=str(cwd),
        env=_env(),
        timeout=60,
    )
    return p.returncode, p.stdout


def _add_worktree(repo: Path, name: str, marker_hook: Path | None, session: str) -> Path:
    """`git worktree add` exactly as a session does it, then fire the PostToolUse marker hook.

    The marker hook is driven with the payload the harness would hand it AFTER the command
    succeeded, which is when PostToolUse fires — not a hand-planted marker file. A hand-planted
    marker would make every deny below a test of the guard alone, and the untested seam is the
    handoff between the two files.
    """
    wt = repo.parent / name
    _git(repo, "worktree", "add", "-q", str(wt))
    if marker_hook is not None:
        rc, _ = _run(
            marker_hook,
            {
                "session_id": session,
                "hook_event_name": "PostToolUse",
                "tool_name": "Bash",
                "cwd": str(repo),
                "tool_input": {"command": f"git worktree add {wt}"},
            },
            repo,
        )
        assert rc == 0, "the marker hook must always exit 0"
    return wt


def _commit_payload(session: str, cwd: Path, command: str = "git commit -m x") -> dict:
    return {
        "session_id": session,
        "hook_event_name": "PreToolUse",
        "tool_name": "Bash",
        "cwd": str(cwd),
        "tool_input": {"command": command},
    }


def _denied(out: bytes) -> bool:
    return b'"permissionDecision": "deny"' in out or b'"permissionDecision":"deny"' in out


# ------------------------------------------------------------------------------------------------
# ANTI-VACUITY — if the fixture never produces a marked worktree, every deny below is meaningless
# ------------------------------------------------------------------------------------------------


def test_the_subprocess_env_CARRIES_the_isolated_hook_fire_log_dir():
    """The hooks these tests drive must log to the fixture's dir, never the production one.

    `conftest.py`'s autouse `isolate_hook_fire_log` monkeypatches `ETV_HOOK_FIRE_LOG_DIR` into
    `os.environ` at test setup. Anything that snapshots `os.environ` at IMPORT time captures the
    value from before the fixture ran and silently defeats it — the hooks then append to
    `$HOME/.cache/ersatztv/hook-fire/`, which is #776's defect reintroduced inside the file that
    proves #776's hooks, corrupting the one surface this repo cites as the observability claim for
    every guard still graded NONE.

    It is invisible from the outside: the tests pass either way, because the fire-log library is
    fail-open by design. So it needs its own assertion.
    """
    env = _env()
    isolated = os.environ.get("ETV_HOOK_FIRE_LOG_DIR")
    assert isolated, "the autouse isolation fixture did not run; conftest.py is not being loaded"
    assert env.get("ETV_HOOK_FIRE_LOG_DIR") == isolated, (
        "the subprocess environment does not carry the isolated log dir, so every hook driven by "
        "this file is writing into the production hook-fire log. Build the env per call; do not "
        f"snapshot os.environ at import time. env has {env.get('ETV_HOOK_FIRE_LOG_DIR')!r}"
    )
    # The path `hook-fire-log.sh` falls back to when ETV_HOOK_FIRE_LOG_DIR is unset, derived the
    # same way it derives it rather than restated as a literal.
    production = Path(os.environ.get("HOME", "/tmp")) / ".cache" / "ersatztv" / "hook-fire"  # noqa: S108 — mirrors hook-fire-log.sh's own ${HOME:-/tmp}
    assert Path(isolated).resolve() != production.resolve(), (
        f"the 'isolated' log dir IS the production one ({production}), so the fixture is isolating "
        "nothing and this test would pass while the leak continued"
    )


def test_driving_a_hook_LANDS_its_records_in_the_ISOLATED_dir(tmp_path):
    """The invariant, asserted at the EFFECT rather than at the helper that is supposed to produce it.

    `test_the_subprocess_env_CARRIES_...` above checks `_env()`'s return value, and that is not the
    same claim: `_env()` can be perfectly correct while a call site passes something else. Cold
    review demonstrated exactly that — restore the module-level snapshot and change one `env=_env()`
    back to `env=_ENV`, and all thirteen tests pass while 54 records leak into the real log. The
    guard was pinned to the shape of the fix instead of to the property, which is
    `verify-against-the-REAL-predecessor`: a hand-written revert is not the code a future tidy-up
    produces.

    So this drives a real hook through the real `_run()` and asserts the records landed where the
    fixture put them.

    ITS SCOPE, stated because the first version of this docstring claimed more than it delivers: it
    guards THE LAUNCH PATH IT DRIVES, not the file. Cold review demonstrated the gap — add a second
    launcher alongside `_run()` that passes a stale snapshot and point the mutation tests at it, and
    this test stays green while 18 records leak, because the hooks IT drives still log correctly.
    Every hook in this file goes through `_run()` today, which is what makes the guard sufficient
    HERE and not a general property. The general form is a suite-level check, tracked in #809; the
    reason it is hard is that the obvious version — diff the production log around each test — races
    against a real session's hooks firing concurrently.
    """
    isolated = Path(os.environ["ETV_HOOK_FIRE_LOG_DIR"])
    before = {p.name for p in isolated.glob("*.jsonl")} if isolated.exists() else set()

    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)
    rc, out = _run(GUARD, _commit_payload(SESSION_B, wt), wt)

    # Anti-vacuity: if the hook decided nothing, it may simply have had nothing to log.
    assert rc == 0 and _denied(out), f"the hook reached no decision, so 'records landed' would prove nothing: {out!r}"

    after = {p.name for p in isolated.glob("*.jsonl")} if isolated.exists() else set()
    assert after > before, (
        f"driving two hooks added no record to the isolated log dir {isolated}. Either the "
        "instrumentation stopped firing, or these hooks are logging somewhere else — and the only "
        "somewhere else is the production log this file must never touch"
    )


def test_the_marker_hook_really_marks_the_worktree_it_was_told_about(tmp_path):
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)

    marker = wt / MARKER_NAME
    assert marker.is_file(), (
        f"no {MARKER_NAME} in {wt}. Without it the guard has nothing to read and every 'denied' "
        "assertion in this file would be testing a mechanism that is not there"
    )
    assert marker.read_text().strip() == SESSION_A, (
        f"the marker names {marker.read_text().strip()!r}, not the session that created the "
        "worktree — ownership would be attributed to the wrong session"
    )


def test_the_marker_hook_ignores_a_command_that_is_not_a_worktree_add(tmp_path):
    """The write side's own negative control: a hook that marked on any command would pass above."""
    repo = _scratch_repo(tmp_path)
    wt = repo.parent / "wt-unrelated"
    _git(repo, "worktree", "add", "-q", str(wt))
    rc, _ = _run(
        MARKER_HOOK,
        {
            "session_id": SESSION_A,
            "hook_event_name": "PostToolUse",
            "tool_name": "Bash",
            "cwd": str(repo),
            "tool_input": {"command": f"ls {wt}"},
        },
        repo,
    )
    assert rc == 0
    assert not (wt / MARKER_NAME).exists(), "the marker hook stamped a worktree it never created"


# ------------------------------------------------------------------------------------------------
# THE PAIR DECIDES — both halves, in sequence, as the harness fires them
# ------------------------------------------------------------------------------------------------


def test_a_commit_in_ANOTHER_sessions_worktree_is_DENIED(tmp_path):
    """The #289 case the mechanism exists for, end to end across both files."""
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)

    rc, out = _run(GUARD, _commit_payload(SESSION_B, wt), wt)
    assert rc == 0, "the hook communicates by printing, and must always exit 0"
    assert _denied(out), f"a commit into session A's worktree was not denied from session B: {out!r}"
    assert SESSION_A.encode() in out, (
        "the deny reason must name the owning session — without it the operator cannot tell whether "
        f"the marker is stale or the worktree is genuinely foreign: {out!r}"
    )


def test_a_commit_in_MY_OWN_worktree_is_ALLOWED(tmp_path):
    """Negative control. A guard that denied every marked worktree would pass the test above."""
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)

    rc, out = _run(GUARD, _commit_payload(SESSION_A, wt), wt)
    assert rc == 0
    assert out == b"", f"the owning session was blocked from committing in its own worktree: {out!r}"


def test_an_UNMARKED_worktree_is_ALLOWED(tmp_path):
    """The deliberate fail-open: pre-convention worktrees and the main tree carry no marker."""
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-none", None, SESSION_A)

    rc, out = _run(GUARD, _commit_payload(SESSION_B, wt), wt)
    assert rc == 0
    assert out == b"", f"an unmarked worktree was blocked, which breaks the main tree too: {out!r}"


def test_a_NON_MUTATING_git_command_in_a_foreign_worktree_is_ALLOWED(tmp_path):
    """Only `commit`/`merge` are guarded; `git status` in a sibling worktree is normal work."""
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)

    rc, out = _run(GUARD, _commit_payload(SESSION_B, wt, "git status"), wt)
    assert rc == 0
    assert out == b"", f"a read-only git command was denied: {out!r}"


def test_a_MERGE_in_a_foreign_worktree_is_DENIED(tmp_path):
    """The other half of the guarded alternation.

    Every other deny case here uses `git commit`, so `merge` could be dropped from the detection
    regex and this file would stay green — the mechanism guards the plumbing-merge path
    (`process.foreign-worktree-plumbing-merge`) specifically, which makes that the more damaging
    half to lose.
    """
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)

    rc, out = _run(GUARD, _commit_payload(SESSION_B, wt, "git merge --no-ff topic"), wt)
    assert rc == 0
    assert _denied(out), f"a merge into session A's worktree was not denied: {out!r}"


def test_a_git_C_into_a_foreign_worktree_is_DENIED_from_the_main_tree(tmp_path):
    """The redirection that makes the guard non-trivial.

    The session's cwd is its OWN tree — where committing is fine — and only the `-C` argument moves
    the operation into the foreign worktree. A guard that looked at `cwd` alone would allow this,
    and `git -C` is how the sibling-worktree commit actually gets typed.
    """
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)

    rc, out = _run(GUARD, _commit_payload(SESSION_B, repo, f"git -C {wt} commit -m x"), repo)
    assert rc == 0
    assert _denied(out), f"`git -C <foreign worktree> commit` was not denied: {out!r}"


# ------------------------------------------------------------------------------------------------
# MUTATION PROOFS — four clauses, one per thing the mechanism hangs on
#
# Each mutant is an isolated copy with ONE clause disarmed, and each test asserts the UNMUTATED pair
# reaches the opposite decision on the same fixture FIRST. Without that positive control a mutation
# proof passes when the mechanism detects nothing at all, which is how the BOM guard sat fail-open
# for months while reading as covered.
# ------------------------------------------------------------------------------------------------


def _mutate(src: Path, tmp_path: Path, old: str, new: str, why: str) -> Path:
    assert old in src.read_text(), (
        f"the clause {old!r} has moved in {src.name}; RETARGET this mutation rather than loosening "
        f"it — a mutation that silently stops mutating is the failure this file is about ({why})"
    )
    dst = tmp_path / f"mutated-{src.name}"
    dst.write_text(src.read_text().replace(old, new, 1))
    return dst


def test_MUTATION_disarming_the_guards_MARKER_READ_stops_the_deny(tmp_path):
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)
    payload = _commit_payload(SESSION_B, wt)

    rc_live, out_live = _run(GUARD, payload, wt)
    assert rc_live == 0 and _denied(out_live), (
        "the UNMUTATED guard did not deny, so 'the mutant is silent' would prove nothing about the "
        f"marker read: {out_live!r}"
    )

    mutant = _mutate(
        GUARD,
        tmp_path,
        'marker="$root/.claude-worktree-owner"',
        'marker="$root/.claude-worktree-owner-NOTHING-WRITES-THIS"',
        "the guard's marker read",
    )
    rc, out = _run(mutant, payload, wt)
    assert rc == 0
    assert out == b"", (
        "the guard still denied with its marker read pointed at a file nothing writes, so the deny "
        f"is not coming from ownership at all: {out!r}"
    )


def test_MUTATION_inverting_the_OWNERSHIP_COMPARISON_blocks_the_owner(tmp_path):
    """The allow direction, which the deny mutation above cannot reach.

    Disarming the comparison the other way would only make the guard deny more, and every deny
    assertion in this file would stay green. Inverting it is what shows the comparison — rather than
    the mere presence of a marker — is what decides.
    """
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)
    own_payload = _commit_payload(SESSION_A, wt)

    rc_live, out_live = _run(GUARD, own_payload, wt)
    assert rc_live == 0 and out_live == b"", (
        f"the UNMUTATED guard already blocked the owner, so the inversion below proves nothing: {out_live!r}"
    )

    mutant = _mutate(
        GUARD,
        tmp_path,
        '[ "$owner" = "$me" ] && exit 0',
        '[ "$owner" != "$me" ] && exit 0',
        "the guard's ownership comparison",
    )
    rc, out = _run(mutant, own_payload, wt)
    assert rc == 0
    assert _denied(out), (
        "inverting the ownership comparison did not change the decision for the OWNING session, so "
        f"the comparison is not what allows it through: {out!r}"
    )


def test_MUTATION_a_marker_hook_that_stops_WRITING_makes_the_guard_go_quiet(tmp_path):
    """THE CROSS-FILE PROOF — the one that could not exist while the halves were tested apart.

    The clause disarmed here is in `posttooluse-worktree-marker.sh`; the assertion is about
    `pretooluse-worktree-guard.sh`. A regression in the write half is otherwise completely silent:
    the marker hook exits 0 either way, and the guard's fail-open turns a missing marker into an
    allowed commit that looks exactly like a correctly allowed one.
    """
    repo = _scratch_repo(tmp_path)

    wt_live = _add_worktree(repo, "wt-live", MARKER_HOOK, SESSION_A)
    rc_live, out_live = _run(GUARD, _commit_payload(SESSION_B, wt_live), wt_live)
    assert rc_live == 0 and _denied(out_live), (
        f"the UNMUTATED pair did not deny, so a silent mutant proves nothing about the marker write: {out_live!r}"
    )

    mutant_marker = _mutate(
        MARKER_HOOK,
        tmp_path,
        'printf \'%s\\n\' "$me" > "$abs/.claude-worktree-owner" 2>/dev/null || true',
        "true",
        "the marker hook's write",
    )
    wt_dead = _add_worktree(repo, "wt-dead", mutant_marker, SESSION_A)
    assert not (wt_dead / MARKER_NAME).exists(), (
        "the mutated marker hook wrote a marker anyway — the mutation did not disarm the write, so "
        "the assertion below would be about nothing"
    )

    rc, out = _run(GUARD, _commit_payload(SESSION_B, wt_dead), wt_dead)
    assert rc == 0
    assert out == b"", (
        "the guard denied a commit in a worktree that carries NO marker, which means the deny in "
        f"the live case above is not evidence that the two halves are connected: {out!r}"
    )


def test_MUTATION_dropping_MERGE_from_the_detection_clause_stops_denying_a_merge(tmp_path):
    """The alternation is two guarded operations, and losing one of them is silent.

    This mutation is deliberately narrow: it must stop the guard denying a `merge` while leaving it
    denying a `commit`. Asserting both is what distinguishes "the alternation is load-bearing" from
    "the mutant broke the regex", which would redden everything and prove nothing about `merge`.
    """
    repo = _scratch_repo(tmp_path)
    wt = _add_worktree(repo, "wt-a", MARKER_HOOK, SESSION_A)
    merge_payload = _commit_payload(SESSION_B, wt, "git merge --no-ff topic")
    commit_payload = _commit_payload(SESSION_B, wt)

    rc_live, out_live = _run(GUARD, merge_payload, wt)
    assert rc_live == 0 and _denied(out_live), (
        f"the UNMUTATED guard did not deny a merge, so a silent mutant proves nothing: {out_live!r}"
    )

    mutant = _mutate(
        GUARD,
        tmp_path,
        "(commit|merge)\\b",
        "(commit)\\b",
        "the command-detection alternation",
    )
    rc, out = _run(mutant, merge_payload, wt)
    assert rc == 0
    assert out == b"", f"dropping `merge` from the alternation did not stop the merge being denied: {out!r}"

    rc_c, out_c = _run(mutant, commit_payload, wt)
    assert rc_c == 0 and _denied(out_c), (
        "the mutant stopped denying COMMITS too, so it broke detection wholesale rather than "
        f"removing the merge alternative — this proves nothing about `merge`: {out_c!r}"
    )