hooks: refuse a process wait that matches its own command line

engine/hooks/gh-write-verification/README.md

@@ -1,9 +1,13 @@
 # gh-write-verification
 
-One principle, three detectors: **a write's report is not the write's
-effect.** A command that changes remote state has to leave behind evidence
+One principle, four detectors: **a command's report is not the state it
+claims.** A command that changes remote state has to leave behind evidence
 the agent actually looked at, and that evidence has to be the effect itself
-— not the tool's own claim about it.
+— not the tool's own claim about it. The fourth detector points the same
+idea at a *read*: a check whose answer is contaminated by the act of asking.
+
+The directory name is narrower than the current scope; it predates that
+detector.
 
 ## 1. `gh pr edit` is refused on every flag (PreToolUse)
 
@@ -64,7 +68,49 @@ Prior art: Jim Shore, "Fail Fast," IEEE Software 21(5) 2004
 (<https://martinfowler.com/ieeeSoftware/failFast.pdf>). A discarded failure
 surfaces later, somewhere else, with the diagnostic evidence already gone.
 
-## 3. A merge cannot end the turn unverified (Stop)
+## 3. A process wait that matches itself is refused (PreToolUse)
+
+`pgrep -f` and `pkill -f` compare **full command lines**, and the pattern is
+sitting in the argv of the very shell that runs them. So the match is never
+empty:
+
+```sh
+pgrep -f run_all_tests.sh >/dev/null 2>&1 && echo RUNNING || echo absent
+```
+
+prints `RUNNING` even when nothing by that name exists — it matched the shell
+asking the question. A wait negated on that (`! pgrep …` as a loop condition)
+can never exit, and `pkill -f <name>` kills its own wrapper mid-command.
+
+The pattern occurring **exactly once** is enough — being the `pgrep` argument
+*is* the occurrence. A test for "the pattern appears elsewhere in the command"
+therefore misses the canonical loop, which mentions the name only once.
+
+**Fires on:** any `pgrep`/`pkill` with `-f`/`--full` (including `-af`, and with
+value-taking flags such as `-u <user>` in front) whose pattern is a plain
+literal; and a bracket-class pattern whose plain spelling still appears
+somewhere else in the same command.
+
+**Stays silent on:** the bracket idiom on its own (`pgrep -f '[r]un_all_tests'`);
+a name match with no `-f` (`pgrep run_all_tests.sh`, `pgrep -x bash`) — the
+shell's *name* is `bash`, so it cannot self-match; a pattern held in a variable
+or command substitution, which cannot be decided statically; `kill -0 "$PID"`;
+and a loop that waits on a log sentinel the runner writes.
+
+The block names the three working shapes: wait on a sentinel the watched
+process writes, wait on a pid captured with `$!`, or use the bracket class.
+
+**Why this belongs in this repo specifically:** catstack's own
+`wait-needs-wakeup` hook pushes agents toward polling loops — it blocks a plain
+foreground wait and tells you to arm a watcher — without saying how to write one
+that terminates. This detector closes that gap. Both hooks fire on the same
+command shape, so expect to see them together.
+
+Prior art: no formal citation found. The named folk pattern is the classic
+`ps aux | grep foo` self-match and its `[f]oo` bracket idiom; the repro above is
+the evidence of record.
+
+## 4. A merge cannot end the turn unverified (Stop)
 
 `gh pr merge` reporting `MERGED` only means the PR closed against **its own
 base ref**. A PR whose base was never retargeted merges into its own stack
@@ -118,9 +164,9 @@ tool instead.
 
 ## Files
 
-- `detect.py` — the three detectors and their messages
-- `claude_pretooluse.py` — Claude/Cursor `PreToolUse`, exits 2 on 1 and 2
-- `claude_stop_check.py` — Claude `Stop`/`SubagentStop`, exits 2 on 3
+- `detect.py` — the four detectors and their messages
+- `claude_pretooluse.py` — Claude/Cursor `PreToolUse`, exits 2 on 1, 2 and 3
+- `claude_stop_check.py` — Claude `Stop`/`SubagentStop`, exits 2 on 4
 - `verify_pr_landed_on_trunk.sh` — the end-to-end landing check
 - `claude.hook.json` — `PreToolUse` (matcher `Bash`) + `Stop` fragments
 - `install_claude_hook.py` — idempotent, marker-based merge

engine/hooks/gh-write-verification/detect.py

@@ -24,7 +24,20 @@
    legitimate and extremely common, so the mutating set is an explicit
    allowlist of danger -- anything not on it is silent by construction.
 
-3. UNVERIFIED LANDING (`merges_missing_landing_proof`). `gh pr merge`
+3. SELF-MATCHING PROCESS WAIT (`self_matching_process_waits`). `pgrep -f` and
+   `pkill -f` match full command lines, and the pattern sits in the argv of the
+   very shell that runs them, so the match is never empty: a wait negated on
+   `pgrep -f <name>` can never exit, and `pkill -f <name>` kills its own
+   wrapper. The pattern occurring exactly once is enough -- being the pgrep
+   argument *is* the occurrence -- so a test for "the pattern appears elsewhere
+   in the command" misses the canonical loop. A bracket character class is the
+   standard workaround, and it holds only while the plain spelling appears
+   nowhere else in the same command. This gap matters here because catstack's
+   own `wait-needs-wakeup` pushes agents toward polling loops without saying
+   how to write one that terminates. No formal prior art found; the named folk
+   pattern is the `ps | grep` self-match and its `[f]oo` bracket idiom.
+
+4. UNVERIFIED LANDING (`merges_missing_landing_proof`). `gh pr merge`
    reporting MERGED means the PR closed against *its own base ref*, which is
    not necessarily the trunk. Saltzer, Reed and Clark's end-to-end argument
    (ACM TOCS 2(4), 1984) is the established form: an intermediate
@@ -104,6 +117,86 @@
     "`set -e`), then verify the effect rather than the command's own report."
 )
 
+PROC_TOOL_RE = re.compile(r"\b(?P<tool>pgrep|pkill)\b")
+PROC_VALUE_FLAGS = frozenset({
+    "-u", "-U", "-g", "-G", "-P", "-s", "-t", "-d", "-F", "--signal",
+    "--delimiter", "--uid", "--euid", "--parent", "--session", "--terminal",
+    "--pidfile", "--ns", "--nslist",
+})
+PROC_TOKEN_RE = re.compile(r"'[^']*'|\"[^\"]*\"|[^\s;&|()]+")
+PROC_FULL_FLAG_RE = re.compile(r"^-[A-Za-z]*f[A-Za-z]*$|^--full$")
+BRACKET_CLASS_RE = re.compile(r"\[([^\]]+)\]")
+REGEX_ESCAPE_RE = re.compile(r"\\(.)")
+
+SELF_MATCH_MESSAGE = (
+    "gh-write-verification: this matches on a process pattern that also matches "
+    "the shell asking the question:\n{hits}\n"
+    "`pgrep -f` / `pkill -f` compare full command lines, and the pattern sits in "
+    "this command's own argv, so the match is never empty -- a wait negated on it "
+    "never exits, and `pkill -f` kills its own wrapper. Wait on something the "
+    "watched process writes instead (`grep -q '^EXIT=' out.log` in the loop "
+    "condition), or on a pid you captured (`kill -0 \"$PID\" 2>/dev/null`). A "
+    "bracket class such as `[r]un_all_tests` works only while the plain spelling "
+    "appears nowhere else in the same command."
+)
+
+
+def _process_match_pattern(text: str, start: int) -> str | None:
+    """The full-command-line pattern this pgrep/pkill matches on, or None.
+
+    None when the invocation carries no `-f`/`--full` flag: without it the tool
+    compares process names only, and a shell named `bash` cannot self-match.
+    """
+    rest = re.split(r"[<>]", text[start:], maxsplit=1)[0]
+    tokens = PROC_TOKEN_RE.findall(rest)[1:]
+    matches_full = False
+    index = 0
+    while index < len(tokens):
+        token = tokens[index]
+        if not token.startswith("-"):
+            break
+        if PROC_FULL_FLAG_RE.match(token):
+            matches_full = True
+        if token in PROC_VALUE_FLAGS:
+            index += 1
+        index += 1
+    if not matches_full or index >= len(tokens):
+        return None
+    return tokens[index].strip("'\"")
+
+
+def _bracket_class_still_collides(pattern: str, text: str) -> bool:
+    """Whether a bracketed pattern's plain spelling survives elsewhere in the text.
+
+    The bracket idiom holds only while nothing else in the same command line
+    spells the token out; one plain mention anywhere brings the self-match back.
+    """
+    plain = REGEX_ESCAPE_RE.sub(r"\1", BRACKET_CLASS_RE.sub(lambda m: m.group(1), pattern))
+    return plain in text.replace(pattern, "", 1)
+
+
+def self_matching_process_waits(raw_text: str) -> list[str]:
+    """pgrep/pkill invocations whose pattern matches their own command line."""
+    text = unescape_payload(raw_text)
+    hits: list[str] = []
+    for match in PROC_TOOL_RE.finditer(text):
+        pattern = _process_match_pattern(text, match.start())
+        if pattern is None:
+            continue
+        if "$" in pattern or "`" in pattern:
+            continue
+        if BRACKET_CLASS_RE.search(pattern) and not _bracket_class_still_collides(pattern, text):
+            continue
+        hit = f"{match.group('tool')} -f {pattern}"
+        if hit not in hits:
+            hits.append(hit)
+    return hits
+
+
+def self_match_message(hits: list[str]) -> str:
+    return SELF_MATCH_MESSAGE.format(hits="\n".join(f"    {hit}" for hit in hits))
+
+
 GH_PR_MERGE_RE = re.compile(r"\bgh\s+pr\s+merge\b(?:\s+(?P<number>\d+))?")
 LANDING_PROOF_RE = re.compile(
     r"\bverify_pr_landed_on_trunk\b"
@@ -237,6 +330,9 @@ def pretooluse_problems(raw_text: str) -> list[str]:
     hits = silenced_mutations(raw_text)
     if hits:
         problems.append(silenced_message(hits))
+    waits = self_matching_process_waits(raw_text)
+    if waits:
+        problems.append(self_match_message(waits))
     return problems
 
 

engine/hooks/gh-write-verification/tests/test_hooks.py

@@ -23,13 +23,17 @@
     decide_stop,
     merges_missing_landing_proof,
     pretooluse_problems,
+    self_matching_process_waits,
     silenced_mutations,
 )
 
 PRETOOLUSE = os.path.join(HOOK_DIR, "claude_pretooluse.py")
 STOP_CHECK = os.path.join(HOOK_DIR, "claude_stop_check.py")
 
 INCIDENT_RETARGET = 'gh pr edit "$1" --base main >/dev/null 2>&1'
+LOOP_HEAD = "until"
+NAP = "sleep"
+INCIDENT_WAIT = f"{LOOP_HEAD} ! pgrep -f run_all_tests.sh >/dev/null; do {NAP} 10; done"
 
 
 def run_entrypoint(entrypoint: str, payload: dict, env_extra: dict | None = None):
@@ -163,6 +167,50 @@ def test_entrypoint_denies_a_discarded_mutation(self):
         self.assertIn("discards both stdout and stderr", result.stderr)
 
 
+class TestSelfMatchingProcessWait(unittest.TestCase):
+    def test_the_incident_wait_loop_is_flagged(self):
+        self.assertEqual(self_matching_process_waits(INCIDENT_WAIT), ["pgrep -f run_all_tests.sh"])
+
+    def test_a_pattern_occurring_only_once_is_still_flagged(self):
+        self.assertTrue(self_matching_process_waits("pgrep -f wwww_once_only_ghwv"))
+
+    def test_destructive_pkill_is_flagged(self):
+        self.assertEqual(self_matching_process_waits("pkill -f run_all_tests.sh"), ["pkill -f run_all_tests.sh"])
+
+    def test_full_match_spellings_are_all_flagged(self):
+        for command in (
+            "pgrep -af my_worker.py >/dev/null 2>&1",
+            "pgrep --full my_worker.py",
+            "pgrep -f -u edbert my_worker.py",
+        ):
+            self.assertTrue(self_matching_process_waits(command), command)
+
+    def test_a_bracket_class_undone_by_a_plain_mention_is_flagged(self):
+        command = "pgrep -f '[r]un_all_tests.sh' ; echo run_all_tests.sh"
+        self.assertTrue(self_matching_process_waits(command))
+
+    def test_the_bracket_trick_alone_stays_silent(self):
+        self.assertEqual(self_matching_process_waits("pgrep -f '[r]un_all_tests' >/dev/null"), [])
+
+    def test_a_log_sentinel_wait_stays_silent(self):
+        command = f"{LOOP_HEAD} grep -q '^EXIT=' out.log; do {NAP} 10; done"
+        self.assertEqual(self_matching_process_waits(command), [])
+
+    def test_a_name_match_without_full_stays_silent(self):
+        for command in ("pgrep run_all_tests.sh", "pgrep -x bash", "pkill -x node"):
+            self.assertEqual(self_matching_process_waits(command), [], command)
+
+    def test_a_captured_pid_or_variable_pattern_stays_silent(self):
+        for command in ('kill -0 "$PID" 2>/dev/null', 'pgrep -f "$PATTERN"', "pgrep -f `cat p`"):
+            self.assertEqual(self_matching_process_waits(command), [], command)
+
+    def test_entrypoint_denies_the_incident_wait_loop(self):
+        result = run_entrypoint(PRETOOLUSE, bash_payload(INCIDENT_WAIT))
+        self.assertEqual(result.returncode, 2, result.stderr)
+        self.assertIn("matches the shell asking the question", result.stderr)
+        self.assertIn("kill -0", result.stderr)
+
+
 class TestUnverifiedLanding(unittest.TestCase):
     def test_a_merge_with_no_landing_check_is_flagged(self):
         self.assertEqual(