Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,9 @@ jobs:
- name: hook e2e coverage gate (positive + negative test per detector)
run: python3 scripts/check_hook_test_coverage.py

- name: rule wiring gate (every phrase checker has a caller or a stated reason)
run: python3 scripts/check_rules_are_wired.py

- name: skills three-harness install gate
run: python3 scripts/check_skills_three_harnesses.py

Expand Down
3 changes: 2 additions & 1 deletion engine/hooks/llm-judge/phrases/example.json
Original file line number Diff line number Diff line change
Expand Up @@ -12,5 +12,6 @@
"You're right. Let's go with option B.",
"Nothing in my earlier count was wrong."
],
"on_hit": "example: the last reply took back an earlier claim."
"on_hit": "example: the last reply took back an earlier claim.",
"unwired_reason": "the worked example for llm-judge's README; it documents the phrase-file shape and is loaded only by that hook's own tests, so it must never be submitted against a real reply"
}
213 changes: 213 additions & 0 deletions scripts/check_rules_are_wired.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,213 @@
#!/usr/bin/env python3
"""Every phrase checker must be reachable from code, or say why it is not.

A rule that exists but nothing calls is worse than a missing rule: it reads as
covered in review and never fires in practice. Five `plain-words-*` checkers --
including the one banning jargon -- shipped with no caller at all, so the rule
was written five ways and enforced zero.

Wired means one of two things. Either a non-test source file names the checker
in quotes, or a non-test file in the owning hook enumerates that phrases
directory at runtime -- `diu-stop/plain_words.py` loads every `plain-words-*`
file by listing the directory, so it names none of them and wires all of them.
A gate that only understood the quoted form would fail the repo's own working
code.

The only other acceptable state is an explicit `unwired_reason` in the phrase
file, mirroring this repo's `subagent_stop.inherit: false` + `reason` pattern:
an opt-out has to be stated, not inferred from silence.

Fail-safe defaults, Saltzer & Schroeder 1975:
https://web.mit.edu/Saltzer/www/publications/protection/Basic.html -- absence of
a caller is a deny, never a pass.
"""
from __future__ import annotations

import contextlib
import io
import json
import os
import subprocess
import sys
import tempfile

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
HOOKS_ROOT = os.path.join(REPO_ROOT, "engine", "hooks")
SOURCE_SUFFIXES = (".py", ".sh", ".mjs", ".cjs", ".js", ".json")


def phrase_files() -> list[str]:
found = []
for hook in sorted(os.listdir(HOOKS_ROOT)):
phrases_dir = os.path.join(HOOKS_ROOT, hook, "phrases")
if not os.path.isdir(phrases_dir):
continue
for name in sorted(os.listdir(phrases_dir)):
if name.endswith(".json"):
found.append(os.path.join(phrases_dir, name))
return found


def checker_name(path: str) -> str | None:
try:
with open(path, encoding="utf-8") as handle:
data = json.load(handle)
except (OSError, json.JSONDecodeError) as exc:
print(f"check_rules_are_wired: cannot read {path}: {exc}", file=sys.stderr)
return None
name = data.get("checker")
return name if isinstance(name, str) and name else None


def unwired_reason(path: str) -> str | None:
try:
with open(path, encoding="utf-8") as handle:
data = json.load(handle)
except (OSError, json.JSONDecodeError):
return None
reason = data.get("unwired_reason")
return reason.strip() if isinstance(reason, str) and reason.strip() else None


def callers(name: str) -> list[str]:
"""Non-test source files under engine/ that name this checker.

The name must appear quoted. A bare substring match passes a dormant
checker whose name is an ordinary word: `example` matched three files that
merely used the word in prose.
"""
pattern = f"[\"']{name}[\"']"
try:
result = subprocess.run(
["grep", "-rlE", "--include=*.py", "--include=*.sh", "--include=*.mjs",
"--include=*.js", "--include=*.json", pattern, HOOKS_ROOT],
capture_output=True, text=True, timeout=60,
)
except (OSError, subprocess.SubprocessError) as exc:
raise SystemExit(f"check_rules_are_wired: grep failed: {exc}") from exc

hits = []
for line in result.stdout.splitlines():
path = line.strip()
if not path:
continue
parts = path.split(os.sep)
if "phrases" in parts or "tests" in parts or "__pycache__" in parts:
continue
if path.endswith(SOURCE_SUFFIXES):
hits.append(os.path.relpath(path, REPO_ROOT))
return hits


PROMISED_CATCH = (
"no-caller",
"prose-mention-only",
"empty-unwired-reason",
"no-checker-name",
)
PROMISED_ALLOW = (
"quoted-caller",
"stated-unwired-reason",
"runtime-discovery",
)


def flags_exemplar(exemplar: str) -> bool:
"""True when this gate fails on `exemplar`. One throwaway hooks tree each."""
payload = {
"checker": "demo-rule",
"meaning": "m",
"reads": "reply",
"match": ["a"],
"not_match": ["b"],
"on_hit": "h",
}
if exemplar == "no-checker-name":
payload.pop("checker")
if exemplar == "empty-unwired-reason":
payload["unwired_reason"] = " "
if exemplar == "stated-unwired-reason":
payload["unwired_reason"] = "sample only, never submitted"

caller = {
"quoted-caller": 'NAMES = ("demo-rule",)\n',
"prose-mention-only": '"""A demo-rule is described here in prose."""\n',
"runtime-discovery": 'import os\nnames = os.listdir("phrases")\n',
}.get(exemplar)

global HOOKS_ROOT, REPO_ROOT
saved = (HOOKS_ROOT, REPO_ROOT)
with tempfile.TemporaryDirectory() as tmp:
hooks = os.path.join(tmp, "engine", "hooks", "demo")
os.makedirs(os.path.join(hooks, "phrases"))
with open(os.path.join(hooks, "phrases", "demo.json"), "w", encoding="utf-8") as handle:
json.dump(payload, handle)
if caller:
with open(os.path.join(hooks, "detect.py"), "w", encoding="utf-8") as handle:
handle.write(caller)
HOOKS_ROOT = os.path.dirname(hooks)
REPO_ROOT = os.path.dirname(HOOKS_ROOT)
try:
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
return main() != 0
finally:
HOOKS_ROOT, REPO_ROOT = saved


def enumerates_dir(hook_dir: str) -> str | None:
"""A non-test file in this hook that lists its own phrases directory."""
for name in sorted(os.listdir(hook_dir)):
path = os.path.join(hook_dir, name)
if not os.path.isfile(path) or not name.endswith(".py"):
continue
if name.startswith("test"):
continue
try:
with open(path, encoding="utf-8") as handle:
text = handle.read()
except OSError as exc:
print(f"check_rules_are_wired: cannot read {path}: {exc}", file=sys.stderr)
continue
if "phrases" not in text:
continue
if any(call in text for call in ("os.listdir", "glob.glob", ".iterdir(", "scandir")):
return os.path.relpath(path, REPO_ROOT)
return None


def main() -> int:
problems = []
checked = 0
for path in phrase_files():
relative = os.path.relpath(path, REPO_ROOT)
name = checker_name(path)
if not name:
problems.append(f"{relative}: no `checker` name, so nothing can call it")
continue
checked += 1
if callers(name):
continue
dynamic = enumerates_dir(os.path.dirname(os.path.dirname(path)))
if dynamic:
print(f"dynamic {name}: loaded by {dynamic}")
continue
reason = unwired_reason(path)
if reason:
print(f"opt-out {name}: {reason}")
continue
problems.append(
f"{relative}: checker `{name}` has no caller under engine/ outside phrases/ "
"and tests/, and nothing in that hook enumerates the directory. Wire it, or add "
"an `unwired_reason` saying why it ships dormant.")

if problems:
print("check_rules_are_wired: FAIL", file=sys.stderr)
for problem in problems:
print(f" - {problem}", file=sys.stderr)
return 1
print(f"check_rules_are_wired: OK ({checked} checker(s) checked)")
return 0


if __name__ == "__main__":
sys.exit(main())
125 changes: 125 additions & 0 deletions tests/test_rules_are_wired.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,125 @@
#!/usr/bin/env python3
"""The bad case this gate exists for is real: five `plain-words-*` checkers,
including the jargon ban, shipped with no caller. The false-negative case is
real too -- `example` passed on a bare substring match against three files that
merely used the word in prose.
"""
from __future__ import annotations

import importlib.util
import json
import os
import subprocess
import sys
import tempfile
import unittest

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
GATE = os.path.join(REPO_ROOT, "scripts", "check_rules_are_wired.py")


def load_gate(hooks_root: str):
spec = importlib.util.spec_from_file_location("gate_under_test", GATE)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
module.HOOKS_ROOT = hooks_root
module.REPO_ROOT = os.path.dirname(hooks_root)
return module


class WiringGate(unittest.TestCase):
def setUp(self) -> None:
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.hooks = os.path.join(self.tmp.name, "engine", "hooks")
os.makedirs(self.hooks)

def _phrase(self, hook: str, name: str, **extra) -> str:
directory = os.path.join(self.hooks, hook, "phrases")
os.makedirs(directory, exist_ok=True)
path = os.path.join(directory, f"{name}.json")
payload = {
"checker": name,
"meaning": "whatever",
"reads": "reply",
"match": ["a"],
"not_match": ["b"],
"on_hit": "hit",
}
payload.update(extra)
with open(path, "w", encoding="utf-8") as handle:
json.dump(payload, handle)
return path

def _caller(self, hook: str, body: str) -> None:
directory = os.path.join(self.hooks, hook)
os.makedirs(directory, exist_ok=True)
with open(os.path.join(directory, "detect.py"), "w", encoding="utf-8") as handle:
handle.write(body)

def test_a_checker_with_no_caller_fails(self) -> None:
self._phrase("diu-stop", "plain-words-tech-jargon")
gate = load_gate(self.hooks)
self.assertEqual(gate.main(), 1)

def test_a_quoted_caller_passes(self) -> None:
self._phrase("diu-stop", "plain-words-tech-jargon")
self._caller("diu-stop", 'CATEGORIES = ("plain-words-tech-jargon",)\n')
gate = load_gate(self.hooks)
self.assertEqual(gate.main(), 0)

def test_an_unquoted_prose_mention_is_not_a_caller(self) -> None:
self._phrase("llm-judge", "example")
self._caller("llm-judge", '"""For example, this is prose."""\n')
gate = load_gate(self.hooks)
self.assertEqual(gate.main(), 1)

def test_an_explicit_unwired_reason_passes(self) -> None:
self._phrase("llm-judge", "example", unwired_reason="sample only, never submitted")
gate = load_gate(self.hooks)
self.assertEqual(gate.main(), 0)

def test_an_empty_unwired_reason_does_not_count(self) -> None:
self._phrase("llm-judge", "example", unwired_reason=" ")
gate = load_gate(self.hooks)
self.assertEqual(gate.main(), 1)

def test_a_phrase_file_with_no_checker_name_fails(self) -> None:
directory = os.path.join(self.hooks, "diu-stop", "phrases")
os.makedirs(directory)
with open(os.path.join(directory, "broken.json"), "w", encoding="utf-8") as handle:
json.dump({"meaning": "no checker key"}, handle)
gate = load_gate(self.hooks)
self.assertEqual(gate.main(), 1)

def test_a_caller_inside_tests_does_not_count(self) -> None:
self._phrase("diu-stop", "plain-words-tech-jargon")
tests_dir = os.path.join(self.hooks, "diu-stop", "tests")
os.makedirs(tests_dir)
with open(os.path.join(tests_dir, "test_x.py"), "w", encoding="utf-8") as handle:
handle.write('load("plain-words-tech-jargon")\n')
gate = load_gate(self.hooks)
self.assertEqual(gate.main(), 1)

def test_the_real_repo_passes(self) -> None:
result = subprocess.run([sys.executable, GATE], capture_output=True, text=True, timeout=120)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)


class RuntimeDiscovery(WiringGate):
def test_a_hook_that_lists_its_phrases_dir_wires_all_of_them(self) -> None:
self._phrase("diu-stop", "plain-words-tech-jargon")
self._phrase("diu-stop", "plain-words-code-names")
self._caller("diu-stop", 'import os\nnames = os.listdir("phrases")\n')
gate = load_gate(self.hooks)
self.assertEqual(gate.main(), 0)

def test_mentioning_phrases_without_listing_it_is_not_wiring(self) -> None:
self._phrase("diu-stop", "plain-words-tech-jargon")
self._caller("diu-stop", '"""The phrases directory holds the word lists."""\n')
gate = load_gate(self.hooks)
self.assertEqual(gate.main(), 1)


if __name__ == "__main__":
unittest.main()
Loading