merge queue: checking #307 on main (104f21f)

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
 

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"],

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())

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

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()