diff --git a/engine/skills/make-pr/scripts/preflight.py b/engine/skills/make-pr/scripts/preflight.py index 3a2a5b5b..7a0ad0bf 100644 --- a/engine/skills/make-pr/scripts/preflight.py +++ b/engine/skills/make-pr/scripts/preflight.py @@ -103,6 +103,7 @@ def gates_for(paths: list[str], base: str | None = None) -> list[list[str]]: ["python3", "scripts/check_skill_test_coverage.py"], ["python3", "scripts/check_skill_trigger_mechanism.py"], ["python3", "scripts/check_skill_trigger_policy.py"], + ["python3", "scripts/check_subagent_scope_contract.py"], ] return cmds diff --git a/engine/skills/make-pr/tests/test_preflight.py b/engine/skills/make-pr/tests/test_preflight.py index f9b6d28f..6b180890 100644 --- a/engine/skills/make-pr/tests/test_preflight.py +++ b/engine/skills/make-pr/tests/test_preflight.py @@ -65,6 +65,16 @@ def test_gates_for_skill_slice_include_trigger_policy(self): cmds = pf.gates_for(["product/skills/how/SKILL.md"]) self.assertIn(["python3", "scripts/check_skill_trigger_policy.py"], cmds) + def test_gates_for_skill_slice_include_subagent_scope_contract(self): + """A skill slice must run the subagent-scope gate. + + It catches a new fan-out skill that never states the scope its + subagents inherit, which is the boundary a parent cannot review after + the fact. + """ + cmds = pf.gates_for(["product/skills/how/SKILL.md"]) + self.assertIn(["python3", "scripts/check_subagent_scope_contract.py"], cmds) + def test_gates_for_rule_prose_with_base_includes_dated_provenance_check(self): self.assertIn( ["python3", "scripts/check_no_dated_provenance.py", "--base", "origin/main"], diff --git a/scripts/check_subagent_scope_contract.py b/scripts/check_subagent_scope_contract.py new file mode 100755 index 00000000..2b568e68 --- /dev/null +++ b/scripts/check_subagent_scope_contract.py @@ -0,0 +1,138 @@ +#!/usr/bin/env python3 +"""A skill that tells the model to spawn subagents MUST state the scope contract. + +A parent delegates because it cannot hold the work itself, which is exactly +why it cannot review every action the subagent took -- it sees a summary. So +the boundary has to be in the prompt the skill tells the parent to write, or +it is nowhere. See corpus/skills/principle-subagent-inherits-scope. + +A skill qualifies as a spawner when its body instructs delegation (spawn / +fan out / subagent_type / one worker per ...). Such a skill must name +`principle-subagent-inherits-scope`, or cite the scope contract in +corpus/skills/principle-prove-it/references/finding-shape.md, which carries +it for the investigation products. + +Pre-existing spawners are grandfathered in +scripts/subagent_scope_debt_allowlist.txt, shrink-only in the same way as +scripts/skill_test_debt_allowlist.txt. A new spawner must state the contract +rather than land on the list. + + python3 scripts/check_subagent_scope_contract.py + python3 scripts/check_subagent_scope_contract.py --list # spawners found + +SPAWNER_RE is deliberately narrow: prose *about* fan-out is not an instruction +to fan out. "don't fan out delegates to hand-apply what a script can do" +(principle-build-the-lever) and "re-spawning the agent" +(principle-trace-token-burn-loop) are descriptions, and a looser pattern +flagged both. So the verb must take a concrete agent object, and NEGATED_RE +skips a line that negates or merely prices delegation. +""" +from __future__ import annotations + +import argparse +import re +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +SKILL_BUCKETS = ("engine/skills", "corpus/skills", "product/skills") +ALLOWLIST = REPO_ROOT / "scripts" / "subagent_scope_debt_allowlist.txt" + +PRINCIPLE = "principle-subagent-inherits-scope" +CONTRACT_REF = "finding-shape.md" + +SPAWNER_RE = re.compile( + r"spawn(?:s|ing)?\s+(?:\w+\s+){0,4}?(?:subagent|agent|explorer|investigator|" + r"reviewer|judge|worker|candidate|synthesizer|fork)s?\b|" + r"`?subagent_type`?\s*[:=]|" + r"fan(?:s|ning)?\s+(?:out|N)\b|" + r"one\s+(?:worker|reviewer|explorer|investigator|agent)\s+per\b|" + r"parallel\s+`?Agent`?\s+calls", + re.IGNORECASE, +) + +NEGATED_RE = re.compile( + r"don't|do not|never|beats|avoid|instead of|rather than|re-spawning|" + r"uncounted cost|worth knowing about", + re.IGNORECASE, +) + + +def body(text: str) -> str: + """Everything after the frontmatter block.""" + if text.startswith("---"): + end = text.find("\n---", 3) + if end != -1: + return text[end + 4:] + return text + + +def allowlisted(path: Path = ALLOWLIST) -> set[str]: + if not path.is_file(): + return set() + return { + line.strip() + for line in path.read_text(encoding="utf-8").splitlines() + if line.strip() and not line.lstrip().startswith("#") + } + + +def spawners(repo_root: Path) -> list[str]: + """'bucket/name' for every skill whose body instructs delegation.""" + out: list[str] = [] + for bucket in SKILL_BUCKETS: + root = repo_root / bucket + if not root.is_dir(): + continue + for skill_dir in sorted(p for p in root.iterdir() if p.is_dir()): + md = skill_dir / "SKILL.md" + if not md.is_file(): + continue + for line in body(md.read_text(encoding="utf-8")).splitlines(): + if SPAWNER_RE.search(line) and not NEGATED_RE.search(line): + out.append(f"{bucket}/{skill_dir.name}") + break + return out + + +def states_contract(repo_root: Path, rel: str) -> bool: + text = (repo_root / rel / "SKILL.md").read_text(encoding="utf-8") + return PRINCIPLE in text or CONTRACT_REF in text + + +def violations(repo_root: Path, allow: set[str]) -> list[str]: + out: list[str] = [] + for rel in spawners(repo_root): + name = rel.split("/")[-1] + if name == PRINCIPLE or rel in allow or name in allow: + continue + if not states_contract(repo_root, rel): + out.append( + f"{rel}: instructs spawning subagents but names neither " + f"{PRINCIPLE} nor the scope contract in {CONTRACT_REF}" + ) + return out + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--list", action="store_true", help="print the spawners found") + args = ap.parse_args() + + if args.list: + for rel in spawners(REPO_ROOT): + mark = "ok " if states_contract(REPO_ROOT, rel) else "BARE" + print(f"{mark}\t{rel}") + return 0 + + errors = violations(REPO_ROOT, allowlisted()) + if errors: + for e in errors: + print(f"fail\t{e}", file=sys.stderr) + return 1 + print("ok\tsubagent scope contract") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/subagent_scope_debt_allowlist.txt b/scripts/subagent_scope_debt_allowlist.txt new file mode 100644 index 00000000..944a5c44 --- /dev/null +++ b/scripts/subagent_scope_debt_allowlist.txt @@ -0,0 +1,16 @@ +# Skills grandfathered out of scripts/check_subagent_scope_contract.py because +# they instructed spawning subagents before corpus/skills/ +# principle-subagent-inherits-scope existed. +# +# This list is shrink-only, same rule as scripts/skill_test_debt_allowlist.txt: +# a skill graduates by stating the scope contract (naming the principle, or +# citing the contract in corpus/skills/principle-prove-it/references/ +# finding-shape.md) and removing its line. A NEW spawner must state the +# contract instead of landing here. +# +# Each of these genuinely delegates and genuinely needs the contract; they are +# listed rather than edited because rewriting three established skills is a +# different review than adding the rule. +engine/skills/reflect +product/skills/independent-judge-swarm +product/skills/show-me-your-work diff --git a/tests/test_subagent_scope_contract.py b/tests/test_subagent_scope_contract.py new file mode 100644 index 00000000..7afeb04f --- /dev/null +++ b/tests/test_subagent_scope_contract.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +"""Tests for scripts/check_subagent_scope_contract.py. + +The negation fixtures are the real false positives a looser pattern produced +on this repo: principle-build-the-lever tells you NOT to fan out delegates, +and principle-trace-token-burn-loop mentions re-spawning as a cost. Both were +flagged as spawners before the per-line negation guard existed. +""" +from __future__ import annotations + +import sys +import tempfile +import unittest +from pathlib import Path + +REPO = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO / "scripts")) +import check_subagent_scope_contract as sc # noqa: E402 + +BARE_SPAWNER = """--- +name: bare-fanout +description: "Fans out workers." +--- + +# Bare fanout + +Spawn all explorers in a single message, one worker per package. +""" + +SPAWNER_NAMING_PRINCIPLE = BARE_SPAWNER.replace( + "# Bare fanout", "# Bare fanout\n\nInherits principle-subagent-inherits-scope." +) + +SPAWNER_CITING_CONTRACT = BARE_SPAWNER.replace( + "# Bare fanout", + "# Bare fanout\n\nEach worker returns the shape in finding-shape.md.", +) + +NEGATED_FANOUT = """--- +name: principle-build-the-lever +description: "Build the script." +--- + +# Build the Lever + +- A deterministic lever beats fan-out. If the tool can process every unit in + one pass, run it yourself; don't fan out delegates to hand-apply what a + script can do. +""" + +COST_MENTION = """--- +name: principle-trace-token-burn-loop +description: "Trace the burn." +--- + +# Trace the loop + +The same context gets re-sent on every turn (including after +re-spawning the agent). +""" + +NOT_A_SPAWNER = """--- +name: spike-and-validate +description: "Build a throwaway." +--- + +# Spike + +Build the smallest thing that could fail, run it, paste the output. +""" + + +def _repo(skills: dict[str, str]) -> Path: + root = Path(tempfile.mkdtemp()) + for rel, text in skills.items(): + md = root / rel / "SKILL.md" + md.parent.mkdir(parents=True, exist_ok=True) + md.write_text(text, encoding="utf-8") + return root + + +class TestSpawnerDetection(unittest.TestCase): + def test_bare_spawner_is_a_violation(self): + root = _repo({"product/skills/bare-fanout": BARE_SPAWNER}) + errs = sc.violations(root, allow=set()) + self.assertEqual(len(errs), 1, errs) + self.assertIn("bare-fanout", errs[0]) + + def test_naming_the_principle_satisfies_the_gate(self): + root = _repo({"product/skills/bare-fanout": SPAWNER_NAMING_PRINCIPLE}) + self.assertEqual(sc.violations(root, allow=set()), []) + + def test_citing_the_contract_reference_satisfies_the_gate(self): + root = _repo({"product/skills/bare-fanout": SPAWNER_CITING_CONTRACT}) + self.assertEqual(sc.violations(root, allow=set()), []) + + def test_allowlist_grandfathers_a_bare_spawner(self): + root = _repo({"engine/skills/reflect": BARE_SPAWNER}) + self.assertEqual(sc.violations(root, allow={"engine/skills/reflect"}), []) + + def test_non_spawner_is_not_flagged(self): + root = _repo({"product/skills/spike-and-validate": NOT_A_SPAWNER}) + self.assertEqual(sc.spawners(root), []) + + +class TestNegationGuard(unittest.TestCase): + def test_telling_you_not_to_fan_out_is_not_a_spawn_instruction(self): + root = _repo({"corpus/skills/principle-build-the-lever": NEGATED_FANOUT}) + self.assertEqual(sc.spawners(root), []) + + def test_pricing_a_respawn_is_not_a_spawn_instruction(self): + root = _repo({"corpus/skills/principle-trace-token-burn-loop": COST_MENTION}) + self.assertEqual(sc.spawners(root), []) + + +class TestRealRepoState(unittest.TestCase): + def test_repo_passes_with_its_own_allowlist(self): + self.assertEqual(sc.violations(REPO, sc.allowlisted()), []) + + def test_the_four_investigation_products_state_the_contract(self): + found = set(sc.spawners(REPO)) + for rel in ( + "product/skills/how", + "product/skills/why", + "product/skills/alternatives-considered", + ): + self.assertIn(rel, found, f"{rel} should be detected as a spawner") + self.assertTrue(sc.states_contract(REPO, rel), f"{rel} lacks the contract") + + +if __name__ == "__main__": + unittest.main()