Done plan phases gate every PR

src/agent_comms/phase_evidence.py

@@ -36,6 +36,7 @@ class MainTree:
     commit: str
     branch: ClassVar[str] = "main"
     source: ClassVar[str] = ".py"
+    parsed: ClassVar[dict[str, tuple[str, ParsedModule]]] = {}
 
     @classmethod
     def of(cls, repo: Repo) -> MainTree:
@@ -97,18 +98,25 @@ def present_modules(self, paths: tuple[str, ...]) -> Iterator[ParsedModule]:
                 yield self.module(path)
 
     def modules(self, paths: tuple[str, ...]) -> Iterator[ParsedModule]:
-        """Every Python module under ``paths``, read in one ``git cat-file --batch``."""
-        listed = Repo.git(self.repo.checkout, "ls-tree", "-r", "--name-only", self.commit, "--",
-                          *paths)
-        names = [name for name in listed.splitlines() if name.endswith(self.source)]
+        """Every Python module under ``paths``. Each path keeps its last parsed blob, so every
+        evidence reading one commit parses it once, and the rest are read in one
+        ``git cat-file --batch``."""
+        listed = Repo.git(self.repo.checkout, "ls-tree", "-r", self.commit, "--", *paths)
+        blobs = [(entry.split()[2], name) for entry, _, name in
+                 (line.partition("\t") for line in listed.splitlines())
+                 if name.endswith(self.source)]
+        fresh = [(blob, name) for blob, name in blobs
+                 if self.parsed.get(name, ("", None))[0] != blob]
         batch = Repo.git_bytes(self.repo.checkout, "cat-file", "--batch",
-                               input="".join(f"{self.commit}:{name}\n" for name in names))
-        for name in names:
+                               input="".join(f"{blob}\n" for blob, _ in fresh)) if fresh else b""
+        for blob, name in fresh:
             header, _, batch = batch.partition(b"\n")
             size = int(header.rsplit(b" ", 1)[1])
             source = batch[:size].decode()
-            yield ParsedModule(name, source, ast.parse(source, filename=name), name)
+            self.parsed[name] = blob, ParsedModule(name, source,
+                                                   ast.parse(source, filename=name), name)
             batch = batch[size + 1 :]
+        yield from (self.parsed[name][1] for _, name in blobs)
 
     def contains(self, sha: str) -> bool:
         return self.repo.contains(sha, self.commit)
@@ -186,6 +194,11 @@ def holds(self, reading: PlanReading) -> bool: ...
     @abstractmethod
     def describe(self) -> str: ...
 
+    def sites(self, reading: PlanReading) -> tuple[str, ...]:
+        """Each ``PATH:LINE`` that keeps this evidence from holding; evidence naming no line
+        has none, and its description says what is missing."""
+        return ()
+
 
 class MainEvidence(PhaseEvidence):
     """Evidence that reads only main: one answer per main commit."""
@@ -206,6 +219,12 @@ def holds(self, reading: PlanReading) -> bool:
     @abstractmethod
     def read(self, main: MainTree) -> bool: ...
 
+    def sites(self, reading: PlanReading) -> tuple[str, ...]:
+        return self.located(reading.main)
+
+    def located(self, main: MainTree) -> tuple[str, ...]:
+        return ()
+
 
 @dataclass(frozen=True)
 class CommitMerged(MainEvidence):
@@ -237,17 +256,25 @@ class PatternEvidence(MainEvidence):
     paths: tuple[str, ...] = ("src",)
 
     def read(self, main: MainTree) -> bool:
+        return self.satisfied(main.count(self.pattern, self.searched(main)))
+
+    def searched(self, main: MainTree) -> tuple[str, ...]:
+        """``paths`` without the phase declarations, which spell every pattern they ban."""
         from .plan_phases import PlanPhase
 
         declaration = main.repo.module_file(PlanPhase.__module__)
-        return self.satisfied(main.count(self.pattern, (*self.paths, f":!{declaration.as_posix()}")))
+        return (*self.paths, f":!{declaration.as_posix()}")
 
     @abstractmethod
     def satisfied(self, count: int) -> bool: ...
 
 
 @dataclass(frozen=True)
 class PatternAbsent(PatternEvidence):
+    def located(self, main: MainTree) -> tuple[str, ...]:
+        return tuple(f"{path}:{line}" for path, line in
+                     sorted(main.matching_lines(self.pattern, self.searched(main))))
+
     def satisfied(self, count: int) -> bool:
         return count == 0
 
@@ -333,6 +360,10 @@ class AllEvidence(PhaseEvidence):
     def holds(self, reading: PlanReading) -> bool:
         return all(reading.holds(part) for part in self.parts)
 
+    def sites(self, reading: PlanReading) -> tuple[str, ...]:
+        return tuple(site for part in self.parts if not reading.holds(part)
+                     for site in part.sites(reading))
+
     def describe(self) -> str:
         return "; ".join(part.describe() for part in self.parts)
 
@@ -448,22 +479,12 @@ def of(
         return functions.found
 
 
-class Classes(ast.NodeVisitor):
-    """Every class statement in the visited trees, nested ones included."""
-
-    def __init__(self) -> None:
-        self.found: list[ast.ClassDef] = []
+class Classes:
+    """Every class statement in the given trees, nested ones included."""
 
-    def visit_ClassDef(self, node: ast.ClassDef) -> None:
-        self.found.append(node)
-        self.generic_visit(node)
-
-    @classmethod
-    def of(cls, trees: Iterable[ast.AST]) -> list[ast.ClassDef]:
-        classes = cls()
-        for tree in trees:
-            classes.visit(tree)
-        return classes.found
+    @staticmethod
+    def of(trees: Iterable[ast.AST]) -> list[ast.ClassDef]:
+        return [node for tree in trees for node in ast.walk(tree) if isinstance(node, ast.ClassDef)]
 
 
 @dataclass(frozen=True)
@@ -654,7 +675,8 @@ class ClassChecks(ast.NodeVisitor):
 
     def __init__(self) -> None:
         self.declared: set[str] = set()
-        self.checked: set[str] = set()
+        self.path = ""
+        self.checked: list[tuple[str, int, frozenset[str]]] = []
         self._scopes: list[set[str]] = []
         self._methods: set[ast.FunctionDef | ast.AsyncFunctionDef] = set()
         self._comparison_guards: set[ast.Call] = set()
@@ -701,20 +723,32 @@ def visit_Call(self, node: ast.Call) -> None:
         match node:
             case ast.Call(func=ast.Name(id=isinstance.__name__), args=[_, target]) \
                     if node not in self._comparison_guards:
-                self.checked |= NamedClasses.of((target,))
+                self.checked.append((self.path, node.lineno, frozenset(NamedClasses.of((target,)))))
         self.generic_visit(node)
 
+    def owned(self) -> tuple[tuple[str, int], ...]:
+        """The path and line of each check naming a class the visited modules declare."""
+        return tuple((path, line) for path, line, named in self.checked if named & self.declared)
+
 
 @dataclass(frozen=True)
 class OwnClassChecksAbsent(MainEvidence):
     """No ``isinstance`` in ``src`` names a class that ``src`` declares; checks against AST,
     typing and SDK types stay."""
 
     def read(self, main: MainTree) -> bool:
+        return not self.located(main)
+
+    def located(self, main: MainTree) -> tuple[str, ...]:
+        return tuple(f"{path}:{line}" for path, line in self.checks(main).owned())
+
+    @staticmethod
+    def checks(main: MainTree) -> ClassChecks:
         checks = ClassChecks()
         for module in main.modules(("src",)):
+            checks.path = module.path
             checks.visit(module.tree)
-        return not checks.declared & checks.checked
+        return checks
 
     def describe(self) -> str:
         return "no isinstance against a class declared in src on main"
@@ -1013,24 +1047,24 @@ def visit_Call(self, node: ast.Call) -> None:
 
 
 class ClassAnswers(ast.NodeVisitor):
-    """Each member a class body declares by name, with its line and its whole declaration. A
-    field counts only on a dataclass, and a name ``type`` reads per class never counts."""
+    """Each member a class body declares by name, with its declaration. A field counts only on
+    a dataclass, and a name ``type`` reads per class never counts."""
 
     def __init__(self, declared: ast.ClassDef, per_class: frozenset[str]) -> None:
         self.fields = dataclass.__name__ in DecoratorNames.of(declared.decorator_list)
         self.per_class = per_class
-        self.found: dict[str, tuple[int, str]] = {}
+        self.found: dict[str, ast.stmt] = {}
 
     @classmethod
-    def of(cls, declared: ast.ClassDef, per_class: frozenset[str]) -> dict[str, tuple[int, str]]:
+    def of(cls, declared: ast.ClassDef, per_class: frozenset[str]) -> dict[str, ast.stmt]:
         answers = cls(declared, per_class)
         for member in declared.body:
             answers.visit(member)
         return answers.found
 
     def record(self, name: str, member: ast.stmt) -> None:
         if name not in self.per_class:
-            self.found[name] = (member.lineno, ast.dump(member))
+            self.found[name] = member
 
     def generic_visit(self, node: ast.AST) -> None:
         """Only the body's own statements declare members."""
@@ -1066,16 +1100,19 @@ class RestatedMembersAbsent(MainEvidence):
     def read(self, main: MainTree) -> bool:
         return not any(self.restatements(main))
 
+    def located(self, main: MainTree) -> tuple[str, ...]:
+        return tuple(self.restatements(main))
+
     def restatements(self, main: MainTree) -> Iterator[str]:
         """Each restated member, as ``PATH:LINE Class.member``."""
         scopes = DeclaredScopes.of(main, self.paths)
         for declared in scopes.classes.values():
             for parent in scopes.parents(declared):
                 inherited = ClassAnswers.of(parent.node, self.per_class)
-                for name, (line, answer) in ClassAnswers.of(declared.node, self.per_class).items():
-                    if name in inherited and inherited[name][1] == answer:
-                        member = QualifiedName((declared.node.name, name))
-                        yield f"{declared.path}:{line} {member}"
+                for name, member in ClassAnswers.of(declared.node, self.per_class).items():
+                    if name in inherited and ast.dump(inherited[name]) == ast.dump(member):
+                        restated = QualifiedName((declared.node.name, name))
+                        yield f"{declared.path}:{member.lineno} {restated}"
 
     def describe(self) -> str:
         return f"no member restating its base's in {' '.join(self.paths)} on main"
@@ -1095,6 +1132,9 @@ class ClassRostersAbsent(MainEvidence):
     def read(self, main: MainTree) -> bool:
         return not any(self.rosters(main))
 
+    def located(self, main: MainTree) -> tuple[str, ...]:
+        return tuple(self.rosters(main))
+
     def relations(self, scopes: DeclaredScopes) -> dict[DeclaredClass, frozenset[str]]:
         """Each family's declaration under the roots, found by its qualified name, with the class
         attributes it declares as a tuple of its own classes."""

tests/guards/test_debt_ratchet.py

@@ -23,6 +23,7 @@
     BroadExceptDefault,
     BroadSwallow,
     ContentCache,
+    DonePhaseBroken,
     ForeignFieldSpellings,
     FrozenFieldWrite,
     FunctionLocalImport,
@@ -328,6 +329,31 @@ def test_added_type_check_fails(repo: Repository) -> None:
     }
 
 
+def test_a_done_phase_fails_the_change_that_breaks_it(tmp_path: Path) -> None:
+    """OwnTypeChecksPhase holds at the base, so re-adding an isinstance on our own class
+    fails at its line; the phases not done at the base (every merged-commit one, here)
+    block nothing."""
+    repo = Repository(tmp_path, "src")
+    # The merged-commit phases name this repository's commits; borrowing its objects lets
+    # them read "not an ancestor" of the fixture instead of fetching.
+    objects = subprocess.check_output(["git", "-C", str(SOURCE), "rev-parse", "--path-format=absolute",
+                                       "--git-common-dir"], text=True).strip()
+    (tmp_path / ".git/objects/info/alternates").write_text(f"{objects}/objects\n")
+    repo.git("remote", "add", "origin", str(SOURCE))
+    owned = "class Owned:\n    pass\n"
+    base = repo.commit({"agent_comms/owned.py": owned})
+    head = repo.commit({"agent_comms/owned.py": owned + "\n\ndef owns(value):\n"
+                        "    return isinstance(value, Owned)\n"})
+    status, report = repo.compare(base, head, DonePhaseBroken)
+    assert status == 1
+    assert report["delta"] == {
+        "DonePhaseBroken:OwnTypeChecksPhase::src/agent_comms/owned.py:6": 1,
+    }
+    clean = repo.commit({"agent_comms/owned.py": owned + "\n\ndef owns(value):\n"
+                         "    return value.owned()\n"})
+    assert repo.compare(base, clean, DonePhaseBroken)[0] == 0
+
+
 def test_moves_between_files_pass(repo: Repository) -> None:
     source = 'x = row = {"key": 1}\nvalue = type(x) is not int\nread = row["key"]\n'
     base = repo.commit({"from.py": source, "other.py": "a = 1\n"})

tests/test_phase_evidence.py

@@ -260,7 +260,7 @@ def test_source_shape_evidence_reads_declarations_on_main(tmp_path):
 def test_only_foreign_comparison_refusal_exempts_an_owned_type_check(source, allowed):
     checks = ClassChecks()
     checks.visit(ast.parse(source))
-    assert (not checks.declared & checks.checked) is allowed
+    assert (not checks.owned()) is allowed
 
 
 @pytest.mark.parametrize(("required", "unskipped", "expected"), (

tools/debt_ratchet.py

@@ -53,6 +53,9 @@
 from stored_formats import StoredForm
 from agent_comms.field_codec import FieldCodec, boundary_decoder, projected
 from agent_comms.mro_dispatch import MroDispatch, handles
+from agent_comms.phase_evidence import MainTree, PlanReading
+from agent_comms.plan_phases import PlanPhase
+from agent_comms.repo import Repo
 
 AUDIT_SCRIPTS = Path(audit.__path__[0]).parent
 REPOSITORY = Path(__file__).resolve().parents[1]
@@ -2492,6 +2495,56 @@ def counted(cls, repo: Path, base: str, head: str, changed: set[str], root: str)
         return {cls.__name__: 0}, {cls.__name__: uncarried}
 
 
+class DonePhaseBroken(Measure):
+    """A plan phase whose evidence holds at the base and fails at the head: a done phase is a
+    standing guard, so a change re-adding what the phase deleted fails here, one key per site.
+
+    The phases are the Plan family's registry and ``PlanPhase.done`` is their only reader;
+    a phase not yet done at the base is plan work in progress and blocks nothing. Evidence
+    that cannot be read here (GitHub's rules, with no token) holds at neither revision. Each
+    revision is read in one process, so its modules are parsed once for every phase, and kept
+    by the commit and the ratchet's code, which holds the phases' declarations."""
+
+    @classmethod
+    def parts(cls, repo: Path, base: str, head: str, changed: set[str],
+              root: str) -> tuple[Callable[[], object], ...]:
+        if root != f"{Repo.of(repo).source}/":
+            return ()
+        return tuple(partial(cls.failing, repo, revision(repo, ref)) for ref in (base, head))
+
+    @classmethod
+    def combine(cls, results: list, repo: Path, base: str, head: str, changed: set[str],
+                root: str) -> tuple[dict[str, int], dict[str, int]]:
+        """A failing pull request pays for reading its sites; a passing one never reads them."""
+        if not results:
+            return {}, {}
+        before, after = results
+        reading = PlanReading((), MainTree(Repo.of(repo), revision(repo, head)))
+        broken = {
+            cls.key(f"{phase.__name__}::{site}"): 1
+            for phase in PlanPhase.members_with(PlanPhase)
+            if phase.__name__ in after - before
+            for site in phase.evidence.sites(reading) or (reading.describe(phase.evidence),)
+        }
+        return dict.fromkeys(broken, 0), broken
+
+    @classmethod
+    def grew(cls, key: str, grown: int) -> str:
+        phase, _, site = key.partition(":")[2].partition("::")
+        return f"done plan phase {phase} no longer holds: {site}"
+
+    @classmethod
+    def failing(cls, repo: Path, commit: str) -> frozenset[str]:
+        """The name of each phase not done at a commit."""
+        return ContentCache.memo(partial(cls.read, repo, commit), cls.__name__, commit)
+
+    @staticmethod
+    def read(repo: Path, commit: str) -> frozenset[str]:
+        reading = PlanReading((), MainTree(Repo.of(repo), commit))
+        return frozenset(phase.__name__ for phase in PlanPhase.members_with(PlanPhase)
+                         if not phase.done(reading))
+
+
 def git(repo: Path, *args: str, input: bytes = b"") -> bytes:
     return subprocess.check_output(["git", "-C", str(repo), *args], input=input, timeout=20)