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

from scripts.tests.hook_fire_isolation import resolved_log_dir

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 the 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, so a `{**os.environ}` snapshot taken at
    import time carries the SESSION value and never any per-test one. `conftest.py`'s
    `pytest_configure` now also sets the variable before collection, so such a snapshot no longer
    reaches the REAL `$HOME/.cache/ersatztv/hook-fire/` log (ersatztv#809) — but it would still pin
    every subprocess to one directory for the whole session, which is not what this file wants.

    WHAT THAT USED TO COST, kept because it is why the helper is shaped this way and not as a live
    warning: before the pre-collection isolation, such a snapshot predated the fixture entirely and
    its hooks appended to the REAL `$HOME/.cache/ersatztv/hook-fire/` — #776's defect reintroduced
    in the file meant to prove #776's hooks, corrupting the `hook-fire-log.sh report` surface this
    repo cites as the observability claim for every guard still graded NONE. Reproduce by
    reintroducing the snapshot and counting records for the two synthetic session ids below: 58 per
    run, on macOS and Linux alike. Those 58 now land in the session's own temporary directory
    instead, so the figure is a measure of this file's hook traffic and no longer of a leak.
    """
    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, and its `pytest_configure` sets the same variable before collection.
    Since ersatztv#809 a snapshot taken at IMPORT time therefore carries the session directory, not
    a stale pre-isolation value, so it no longer reaches `$HOME/.cache/ersatztv/hook-fire/` — the
    shape that was #776's defect reintroduced inside the file meant to prove #776's hooks. What it
    would still do is pin every subprocess in this file to one directory for the whole session,
    which is why `_env()` remains a function.

    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 not logging where the fixture put it. 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 — taken from the
    # SHARED resolver rather than transliterated again here. The hand-written version this replaces
    # used `os.environ.get("HOME", "/tmp")`, which returns `""` for an empty HOME where the shell's
    # `${HOME:-/tmp}` takes `/tmp`: the exact mistake `resolved_log_dir`'s docstring exists to warn
    # about, sitting two files from the warning.
    production = resolved_log_dir({"HOME": os.environ.get("HOME", "")})
    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 every test in this file passes while 54 records go somewhere it did not
    choose. Before ersatztv#809's pre-collection isolation that somewhere was the REAL log; it is now
    the session directory, so the guard still catches the mistake but the consequence is contained.
    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 go to the session dir rather than the fixture's, because
    the hooks IT drives still log correctly. (Before ersatztv#809 that somewhere was the REAL log.)
    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 now lives in `conftest.py` (ersatztv#809):
    `pytest_configure` isolates before collection, so an import-time snapshot cannot be stale, and a
    `Popen` wrapper rejects any launch that does not CARRY an isolated log dir — the whitelist, not a
    comparison against the production path, which would model only one of the sink's two defaults. The obvious
    version — diff the production log around each test — was tried and removed: it races against a
    real session's hooks firing concurrently, and blames the suite when it loses (ersatztv#822).
    """
    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 — since "
        "ersatztv#809 that is the session dir rather than the production log, but it is still not "
        "where this file put it"
    )


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}"
    )