Files
ersatztv/scripts/tests/test_hook_fire_log.py
T
timothyandClaude Opus 5 56ed794ff7 feat(776): every hook reports that it fired, and the report is measured
In this harness version only `Stop` hooks emit a structured transcript record.
`PreToolUse` and `PostToolUse` hooks leave no durable execution trace — which is
every guard this repo relies on. #773 §5.4 could therefore only count filename
mentions in settings dumps, i.e. inference, so a silently broken hook looked
identical to a working one indefinitely.

Each hook now sources one shared sink, `scripts/hook-fire-log.sh`, and records a
`fire` record on entry and an `exit` record carrying its status and decision. The
decision is parsed from the bytes the hook actually emitted rather than declared
by the author: Claude hooks always exit 0 and decide by printing JSON, so stdout
is diverted and replayed; git hooks decide by exit code and stream their output,
so theirs is left alone. That avoided ~40 edits inside pretooluse-merge-consent
and records what the harness sees rather than what the author meant.

The instrumentation sits in the stdin/stdout path of the most authoritative
guards in the repo, so transparency is the load-bearing claim, not coverage:
test_instrumentation_changes_NOTHING_the_harness_can_see drives every hook with
and without it over a payload matrix reaching each hook's deciding branch, and
demands byte-equal stdout and equal exit status. It shipped with two defects that
only that shape catches — `out=$(cat f)` stripping a trailing newline, and an
unconditional stdin slurp that hangs on a TTY and would have frozen every
interactive `git commit`. Both are now pinned, the second through a real pty.

fixes #776

Decisions-Edit: yes
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-14 18:52:20 +02:00

491 lines
23 KiB
Python

"""Every hook reports that it fired, and the reporting changes nothing else (ersatztv#776).
Two separable claims, and conflating them is how instrumentation ships as a regression:
1. COVERAGE — every hook script records its own execution. The population is DERIVED from
`.claude/hooks/*.sh`, never listed, per `testing.guard-derives-population-from-source`, so a
hook added tomorrow is uninstrumented-and-red rather than silently unobserved.
2. TRANSPARENCY — the wrapper is invisible to the harness. It slurps stdin and replays it, and it
diverts stdout and replays it, which puts it directly in the path of the most load-bearing
guards in this repo. If it drops a byte of a `deny` payload or swallows an exit code, it
disables a guard while reporting that the guard fired. That is strictly worse than the
blindness #776 set out to fix, so the differential test below drives EVERY hook, with and
without the instrumentation, over a payload matrix and demands byte-equal stdout and equal
exit status.
The mutation proof for claim 1 is the contrapositive form allowed by
`testing.guard-ships-with-mutation-proof` for a checker-guard: the defect the guard exists to catch
(an uninstrumented hook) is introduced into an isolated copy of the guarded artifact, and the named
check must go red. See `test_a_hook_that_LOSES_its_instrumentation_is_DETECTED`.
Claim 2's proof is not a mutation of an assertion but a real A/B over the real scripts, which is the
only thing that can establish it — a unit test of the wrapper in isolation would pass on a wrapper
that is fine alone and lethal in context.
"""
from __future__ import annotations
import os
import re
import subprocess
import pty
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parents[2]
HOOKS_DIR = REPO_ROOT / ".claude" / "hooks"
SINK = REPO_ROOT / "scripts" / "hook-fire-log.sh"
# The three lines a hook must carry. Matched structurally rather than as one frozen blob so that a
# comment reflow does not redden the suite, but the two things that matter — sourcing the single
# shared sink, and calling begin with this hook's own name — are both pinned.
_SOURCES_SINK = re.compile(r'^\[ -r "\$ETV_HOOK_FIRE_LIB" \] && \. "\$ETV_HOOK_FIRE_LIB" \|\| true$', re.M)
_BEGINS = re.compile(r"^etv_hook_fire_begin (\S+) .*\|\| true$", re.M)
def hook_scripts() -> list[Path]:
"""THE POPULATION, from the filesystem. Never a list."""
return sorted(HOOKS_DIR.glob("*.sh"))
def instrumentation_faults(text: str, name: str) -> list[str]:
"""The check itself, factored out so a mutation can be fed to the same code the suite runs."""
faults = []
if not _SOURCES_SINK.search(text):
faults.append(f"{name}: does not source scripts/hook-fire-log.sh")
m = _BEGINS.search(text)
if not m:
faults.append(f"{name}: never calls etv_hook_fire_begin")
elif m.group(1) != name:
faults.append(f"{name}: calls etv_hook_fire_begin for {m.group(1)!r}, not for itself")
else:
# Order is the whole point: 8 of the hooks slurp stdin with `input=$(cat)`, and a begin
# placed after that read would find the pipe already drained — recording a fire with no
# tool name, and replaying nothing into a hook that has already consumed its payload.
stdin_read = text.find("$(cat)")
if 0 <= stdin_read < m.start():
faults.append(f"{name}: reads stdin before etv_hook_fire_begin")
return faults
def strip_instrumentation(text: str) -> str:
"""Reconstruct the pre-#776 script. Used BOTH as the mutation and as the A/B control."""
out, skipping = [], False
for line in text.splitlines(keepends=True):
if line.startswith("# ersatztv#776 —"):
skipping = True
continue
if skipping:
if line.startswith(("#", "ETV_HOOK_FIRE_LIB=", "[ -r ", "etv_hook_fire_begin ")):
continue
skipping = False
out.append(line)
return "".join(out)
# ------------------------------------------------------------------------------------------------
# ANTI-VACUITY FIRST — a glob that stopped matching would make every assertion below iterate over
# nothing and report full coverage. That is the characteristic failure of a completeness check, and
# this repo has shipped it (#631, #751), so it is asserted before anything depends on it.
# ------------------------------------------------------------------------------------------------
def test_the_population_is_not_empty():
hooks = hook_scripts()
assert len(hooks) >= 10, (
f"only found {len(hooks)} hook scripts under {HOOKS_DIR} — the glob has stopped matching, "
"so every coverage assertion in this file is vacuous."
)
assert SINK.exists(), "the shared sink is missing; the instrumentation cannot work"
assert os.access(SINK, os.X_OK), "the sink is not executable, so its report side cannot be run"
def test_the_stripper_actually_strips():
"""The A/B control and the mutation are the same function. If it were a no-op, the differential
test would compare each hook against itself and pass on a wrapper that breaks everything."""
for hook in hook_scripts():
text = hook.read_text()
stripped = strip_instrumentation(text)
assert stripped != text, f"{hook.name}: strip_instrumentation() removed nothing"
assert "etv_hook_fire_begin" not in stripped, f"{hook.name}: begin survived the strip"
assert not instrumentation_faults(text, hook.stem), f"{hook.name} is not instrumented"
# ------------------------------------------------------------------------------------------------
# CLAIM 1 — COVERAGE
# ------------------------------------------------------------------------------------------------
def test_every_hook_reports_that_it_fired():
faults = []
for hook in hook_scripts():
faults += instrumentation_faults(hook.read_text(), hook.stem)
assert not faults, (
"these hooks do not report their own execution:\n " + "\n ".join(faults) +
"\n\nAdd the three-line preamble after the `set -` line and BEFORE any stdin read. A hook "
"that does not report is one whose firing we can only infer, which is ersatztv#776."
)
def test_a_hook_that_LOSES_its_instrumentation_is_DETECTED():
"""THE MUTATION PROOF (contrapositive form, for a checker-guard).
Introduce the defect this guard exists to catch — an uninstrumented hook — into an isolated copy
of each hook in turn, and the check must report it. Every hook, not a sample: the interesting
loss is whichever file someone actually edits.
"""
for hook in hook_scripts():
mutated = strip_instrumentation(hook.read_text())
faults = instrumentation_faults(mutated, hook.stem)
assert faults, (
f"stripping the instrumentation from {hook.name} left the check GREEN. The check is "
"not load-bearing — it would not notice a hook silently losing its reporting."
)
def test_begin_placed_AFTER_the_stdin_read_is_DETECTED():
"""The subtler mutation: present but too late. Ordering is the property that makes it work."""
late = 'set -euo pipefail\ninput=$(cat)\nETV_HOOK_FIRE_LIB="x"\n[ -r "$ETV_HOOK_FIRE_LIB" ] && . "$ETV_HOOK_FIRE_LIB" || true\netv_hook_fire_begin demo "" capture || true\n'
faults = instrumentation_faults(late, "demo")
assert any("reads stdin before" in f for f in faults), (
"a begin call placed after `input=$(cat)` was accepted. It would record a fire with no "
f"payload and replay nothing. faults={faults}"
)
# ------------------------------------------------------------------------------------------------
# CLAIM 2 — TRANSPARENCY. The wrapper must be invisible to the harness.
# ------------------------------------------------------------------------------------------------
# Payloads chosen to reach the DECIDING branch of each hook, not merely its early-exit. A matrix
# that only ever produced silent allows would prove the wrapper transparent on the one path where
# there is nothing to be transparent about.
PAYLOADS = {
"bash-deny": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"ETV_UPDATE_GOLDENS=1 dotnet test"}}',
"bash-allow": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"ls -la"}}',
"git-commit": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"git commit -m x"}}',
"nav-deny": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"mcp__x__navigate","cwd":"%(cwd)s","tool_input":{"url":"http://h/iptv/channels.m3u"}}',
"agent-no-model": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Agent","cwd":"%(cwd)s","tool_input":{"description":"d","prompt":"p"}}',
"agent-model": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Agent","cwd":"%(cwd)s","tool_input":{"description":"d","prompt":"p","model":"sonnet"}}',
"merge": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"mcp__gitea__pull_request_write","cwd":"%(cwd)s","tool_input":{"method":"merge","owner":"timothy","repo":"ersatztv","pull_number":1}}',
"worktree-add": '{"session_id":"s","hook_event_name":"PostToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"git worktree add /tmp/nope-%(nonce)s HEAD"}}',
"ui-edit": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Edit","cwd":"%(cwd)s","tool_input":{"file_path":"web/src/screens/Channels.tsx"}}',
"empty": "",
"garbage": "not json at all",
}
@pytest.fixture(scope="module")
def sandbox(tmp_path_factory):
"""A controlled environment so the A/B pair cannot differ for reasons unrelated to the wrapper.
`memory_pressure` is stubbed because `pretooluse-agent-ram.sh` reads live free RAM: a real
reading that crossed the 10% or 20% threshold between the two runs would make the A and B halves
disagree for a reason that has nothing to do with instrumentation, and a differential test that
flakes is one that gets rerun until green.
"""
root = tmp_path_factory.mktemp("hookfire")
bindir = root / "bin"
bindir.mkdir()
stub = bindir / "memory_pressure"
stub.write_text("#!/bin/sh\necho 'System-wide memory free percentage: 42%'\n")
stub.chmod(0o755)
env = dict(os.environ)
env["PATH"] = f"{bindir}:{env['PATH']}"
env["ETV_HOOK_FIRE_LOG_DIR"] = str(root / "log")
env["CLAUDE_PROJECT_DIR"] = str(REPO_ROOT)
# No creds: `pretooluse-merge-consent.sh` must take its documented "cannot derive state" path
# rather than making live Gitea calls from a unit test.
for k in ("ETV_GITEA_TOKEN", "ETV_GITEA_BASICAUTH"):
env.pop(k, None)
return root, env
def _run(script: Path, payload: str, env: dict, cwd: Path, arg: str | None = None):
cmd = ["bash", str(script)] + ([arg] if arg else [])
p = subprocess.run(
cmd, input=payload.encode(), capture_output=True, cwd=str(cwd), env=env, timeout=60
)
return p.returncode, p.stdout
@pytest.mark.parametrize("hook", hook_scripts(), ids=lambda h: h.stem)
def test_instrumentation_changes_NOTHING_the_harness_can_see(hook, sandbox):
"""THE REGRESSION NET. Every hook, every payload, byte-equal stdout and equal exit status.
This is what stands between #776 and having silently disarmed the guard set while adding a log
that cheerfully reports every hook firing.
"""
root, env = sandbox
control = root / f"control-{hook.name}"
control.write_text(strip_instrumentation(hook.read_text()))
# `design-sync-reminder.sh` dispatches on argv[1], so both of its branches are exercised.
args = [None, "start", "finish"] if hook.stem == "design-sync-reminder" else [None]
for name, template in PAYLOADS.items():
for arg in args:
payload = template % {"cwd": str(root), "nonce": name}
# A SEPARATE TMPDIR PER SIDE, not per pair. `design-sync-reminder.sh` throttles itself
# with a marker file under $TMPDIR, so a shared one let the control fire and then
# silently suppress the instrumented run — the A/B reported a swallowed decision that
# was really the hook's own one-shot behaviour working correctly. Sharing state between
# the two halves of a differential test makes it measure the state, not the difference.
def side(tag: str) -> dict:
e = dict(env)
e["TMPDIR"] = str(root / f"tmp-{name}-{arg}-{tag}")
Path(e["TMPDIR"]).mkdir(parents=True, exist_ok=True)
return e
rc_a, out_a = _run(control, payload, side("control"), root, arg)
rc_b, out_b = _run(hook, payload, side("instrumented"), root, arg)
label = f"{hook.name} payload={name} arg={arg}"
assert rc_a == rc_b, f"{label}: exit status changed {rc_a} -> {rc_b}"
assert out_a == out_b, (
f"{label}: stdout changed.\n without: {out_a!r}\n with: {out_b!r}"
)
def test_stdout_is_replayed_BYTE_EXACT(sandbox):
"""Pinned separately from the A/B because the first version of the sink lost exactly one byte.
`out=$(cat file)` strips trailing newlines, so the hook's JSON reached the harness a newline
short. Every JSON parser accepts it, so nothing downstream would ever have complained — the only
thing that catches it is comparing bytes.
"""
root, env = sandbox
hook = root / "trailing.sh"
hook.write_text(
"#!/usr/bin/env bash\nset -euo pipefail\n"
f'. "{SINK}"\n'
"etv_hook_fire_begin trailing \"\" capture || true\n"
"input=$(cat)\n"
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}\n\n"' + "\n"
)
rc, out = _run(hook, '{"tool_name":"Bash"}', env, root)
assert rc == 0
assert out == b'{"hookSpecificOutput":{"permissionDecision":"deny"}}\n\n', (
f"trailing bytes were altered: {out!r}"
)
@pytest.mark.parametrize("code", [0, 1, 2, 3])
def test_the_exit_code_is_preserved(sandbox, code):
"""Exit 2 is the harness's block channel and non-zero is git's. A wrapper that normalised the
status to 0 would turn every git hook into a rubber stamp while logging that it ran."""
root, env = sandbox
hook = root / f"exit{code}.sh"
hook.write_text(
"#!/usr/bin/env bash\nset -uo pipefail\n"
f'. "{SINK}"\n'
f'etv_hook_fire_begin exit{code} "" capture || true\n'
f"exit {code}\n"
)
rc, _ = _run(hook, "{}", env, root)
assert rc == code, f"exit {code} came back as {rc}"
def test_stdin_reaches_the_hook_intact(sandbox):
"""The wrapper reads stdin before the hook does. If it did not replay it faithfully, every
guard that parses `.tool_input` would see an empty payload and fail open, silently."""
root, env = sandbox
payload = '{"tool_name":"Bash","tool_input":{"command":"echo \\"quoted\\" && ls"},"session_id":"s"}'
hook = root / "echo-stdin.sh"
hook.write_text(
"#!/usr/bin/env bash\nset -euo pipefail\n"
f'. "{SINK}"\n'
'etv_hook_fire_begin echo-stdin "" capture || true\n'
"input=$(cat)\nprintf '%s' \"$input\"\n"
)
rc, out = _run(hook, payload, env, root)
assert rc == 0
assert out.decode() == payload, f"stdin was altered:\n sent: {payload}\n saw: {out.decode()}"
def test_a_TTY_stdin_is_not_slurped(sandbox):
"""The hang this nearly shipped with.
An interactive `git commit` hands its hooks a terminal on fd 0. `cat` on a tty blocks until the
user types EOF, so an unconditional slurp would freeze every commit — instrumentation hanging
the very thing it was added to observe. Driven through a real pty, because the `[ ! -t 0 ]`
guard is precisely what a pipe-based test cannot exercise.
"""
root, env = sandbox
hook = root / "tty.sh"
hook.write_text(
"#!/usr/bin/env bash\nset -uo pipefail\n"
f'. "{SINK}"\n'
'etv_hook_fire_begin ttyhook "" stream || true\n'
"echo ALIVE\n"
)
master, slave = pty.openpty()
try:
p = subprocess.Popen(
["bash", str(hook)], stdin=slave, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, cwd=str(root), env=env,
)
os.close(slave)
try:
out, _ = p.communicate(timeout=20)
except subprocess.TimeoutExpired:
p.kill()
pytest.fail(
"the hook hung with a tty on stdin — the instrumentation slurped a terminal, "
"which would freeze every interactive `git commit`."
)
finally:
os.close(master)
assert b"ALIVE" in out
def test_logging_failure_does_not_break_the_hook(sandbox):
"""Fail-open, in the one direction that is correct here. This file is observability; if it
cannot write, the guard must still guard. The inverse — a logging bug that swallows a `deny` —
is the failure mode the A/B test above exists to prevent."""
root, env = sandbox
bad = dict(env)
bad["ETV_HOOK_FIRE_LOG_DIR"] = "/dev/null/cannot-exist"
hook = root / "failopen.sh"
hook.write_text(
"#!/usr/bin/env bash\nset -euo pipefail\n"
f'. "{SINK}"\n'
'etv_hook_fire_begin failopen "" capture || true\n'
"input=$(cat)\n"
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}"' + "\n"
)
rc, out = _run(hook, '{"tool_name":"Bash"}', bad, root)
assert rc == 0
assert out == b'{"hookSpecificOutput":{"permissionDecision":"deny"}}', (
f"an unwritable log dir changed the hook's decision output: {out!r}"
)
# ------------------------------------------------------------------------------------------------
# THE RECORD ITSELF, and the report over it
# ------------------------------------------------------------------------------------------------
@pytest.mark.parametrize(
"emit,mode,expected",
[
(r'{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}', "capture", "deny"),
(r'{\"hookSpecificOutput\":{\"permissionDecision\":\"ask\"}}', "capture", "ask"),
(r'{\"hookSpecificOutput\":{\"permissionDecision\":\"allow\"}}', "capture", "allow"),
(r'{\"hookSpecificOutput\":{\"additionalContext\":\"hi\"}}', "capture", "context"),
(r'{\"decision\":\"block\",\"reason\":\"r\"}', "capture", "block"),
("", "capture", "no-op"),
("", "stream", "pass"),
],
)
def test_the_recorded_decision_is_the_one_the_hook_emitted(sandbox, emit, mode, expected):
"""The decision in the log is derived from the bytes the harness receives, never from something
the hook author remembered to declare. That is the whole difference between this and inference.
"""
root, env = sandbox
logdir = root / f"log-{expected}-{mode}"
runenv = dict(env)
runenv["ETV_HOOK_FIRE_LOG_DIR"] = str(logdir)
hook = root / f"decide-{expected}-{mode}.sh"
body = f'printf "{emit}"\n' if emit else ""
hook.write_text(
"#!/usr/bin/env bash\nset -uo pipefail\n"
f'. "{SINK}"\n'
f'etv_hook_fire_begin decider "" {mode} || true\n'
"input=$(cat)\n" + body
)
rc, _ = _run(hook, '{"session_id":"sess1","tool_name":"Bash"}', runenv, root)
assert rc == 0
records = (logdir / "sess1.jsonl").read_text().splitlines()
exits = [r for r in records if '"phase":"exit"' in r]
assert len(exits) == 1, f"expected one exit record, got {records}"
assert f'"decision":"{expected}"' in exits[0], f"expected {expected} in {exits[0]}"
assert any('"phase":"fire"' in r for r in records), "no fire record was written"
def test_a_stream_mode_hook_that_BLOCKS_is_recorded_as_blocked(sandbox):
"""A git hook decides by exit code; recording a non-zero exit as `pass` would put the report
back into the business of inferring, with the answer inverted on the only path that matters."""
root, env = sandbox
logdir = root / "log-blocked"
runenv = dict(env)
runenv["ETV_HOOK_FIRE_LOG_DIR"] = str(logdir)
hook = root / "blocker.sh"
hook.write_text(
"#!/usr/bin/env bash\nset -uo pipefail\n"
f'. "{SINK}"\n'
'etv_hook_fire_begin blocker "" stream || true\n'
"input=$(cat)\nexit 1\n"
)
rc, _ = _run(hook, '{"session_id":"sess1"}', runenv, root)
assert rc == 1
assert '"decision":"blocked"' in (logdir / "sess1.jsonl").read_text()
def test_the_report_names_hooks_that_NEVER_fired(sandbox):
"""The finding #776 asks for is the ZERO row, so the report must derive its rows from the hooks
that exist rather than from the hooks that happen to appear in the log. A report built from the
log alone can only ever show hooks that fired, and would print a clean table while the broken
hook is the one missing from it."""
root, env = sandbox
logdir = root / "log-report"
logdir.mkdir()
(logdir / "sessX.jsonl").write_text(
'{"ts":"t","session":"sessX","pid":"1","hook":"pretooluse-bash-guard","label":"",'
'"event":"PreToolUse","tool":"Bash","phase":"fire","code":"","decision":""}\n'
'{"ts":"t","session":"sessX","pid":"1","hook":"pretooluse-bash-guard","label":"",'
'"event":"PreToolUse","tool":"Bash","phase":"exit","code":"0","decision":"deny"}\n'
)
p = subprocess.run(
["bash", str(SINK), "report", "--all", "--dir", str(logdir)],
capture_output=True, cwd=str(REPO_ROOT), env=env, timeout=60,
)
out = p.stdout.decode()
assert p.returncode == 0, p.stderr.decode()
assert "pretooluse-bash-guard" in out and "deny=1" in out
for hook in hook_scripts():
assert hook.stem in out, (
f"{hook.stem} is absent from the report. Only hooks present in the log were listed, so "
"a hook that never fires — the finding this report exists to surface — is invisible."
)
zero_rows = [ln for ln in out.splitlines() if re.search(r"\s0\s", ln)]
assert zero_rows, "no never-fired rows shown despite only one hook appearing in the log"
def test_the_report_REFUSES_an_empty_population(tmp_path, sandbox):
"""Anti-vacuity on the read side. A report over zero hooks would print a tidy empty table and
read as 'all hooks accounted for' — the shape of #751's 6-second green.
The sink is COPIED into a hook-less tree rather than pointed at one via `CLAUDE_PROJECT_DIR`,
because the real script deliberately falls back to its own location when that variable names a
tree with no hooks. That fallback is correct — it is what makes `scripts/hook-fire-log.sh
report` work from anywhere — and the first version of this test drove the fallback straight back
into the real repo, so it was asserting against all 13 live hooks while claiming to test the
empty case.
"""
_, env = sandbox
fake_scripts = tmp_path / "scripts"
fake_scripts.mkdir()
copied = fake_scripts / "hook-fire-log.sh"
copied.write_text(SINK.read_text())
copied.chmod(0o755)
runenv = dict(env)
runenv.pop("CLAUDE_PROJECT_DIR", None)
p = subprocess.run(
["bash", str(copied), "report", "--all", "--dir", str(tmp_path / "log")],
capture_output=True, cwd=str(tmp_path), env=runenv, timeout=60,
)
assert p.returncode == 2, (
"the report exited 0 over an empty hook population. It must refuse rather than print a "
f"clean table of nothing. stdout={p.stdout.decode()!r}"
)
assert b"refusing to report" in p.stderr