Share the leaf filter in work and the parent walk in impact, and inline two skeleton checks

autoform_cli/impact.py

@@ -14,7 +14,7 @@
 import hashlib
 import json
 import re
-from collections.abc import Callable, Collection, Iterable, Mapping, Sequence
+from collections.abc import Callable, Collection, Iterable, Iterator, Mapping, Sequence
 from dataclasses import asdict, dataclass
 from pathlib import Path
 
@@ -709,29 +709,22 @@ def _owners(
 ) -> tuple[str, ...]:
     """Every article naming the nearest ``parent`` ancestor of ``record``."""
 
-    seen: set[str] = set()
-    parent = record.parent
-    while parent is not None and parent not in seen:
+    for parent in _ancestors(record, records):
         if parent in named:
             return tuple(sorted(set(named[parent])))
-        seen.add(parent)
-        ancestor = records.get(parent)
-        parent = ancestor.parent if ancestor is not None else None
     return ()
 
 
-def _descends_from(record: ConstantRecord, ancestor: str, records: Mapping[str, ConstantRecord]) -> bool:
-    """Whether ``ancestor`` is on the ``parent`` chain of ``record``."""
+def _ancestors(record: ConstantRecord, records: Mapping[str, ConstantRecord]) -> Iterator[str]:
+    """The ``parent`` chain of ``record``, nearest first, stopping where it repeats."""
 
     seen: set[str] = set()
     parent = record.parent
     while parent is not None and parent not in seen:
-        if parent == ancestor:
-            return True
+        yield parent
         seen.add(parent)
         ancestor_record = records.get(parent)
         parent = ancestor_record.parent if ancestor_record is not None else None
-    return False
 
 
 def _reaches(
@@ -776,7 +769,7 @@ def _users_through_internal(
             continue
         seen.add(user)
         further = users.get(user, set()) - {user}
-        if records[user].internal and (further or _descends_from(records[user], name, records)):
+        if records[user].internal and (further or name in _ancestors(records[user], records)):
             work.extend(sorted(further))
         else:
             found.add(user)

autoform_cli/skeleton.py

@@ -674,13 +674,17 @@ def _trusted_from_dict(item: object, *, root: str) -> TrustedDeclaration:
 def _report_entry(item: dict[str, object], name: str) -> TrustedDeclaration:
     """Read the fields every reported declaration has, with their source invariants."""
 
-    kind = _report_kind(item.get("kind"), name)
+    kind = item.get("kind")
+    if not isinstance(kind, str) or kind not in _DECLARATION_KINDS:
+        raise SkeletonError([f"invalid declaration kind for {name} in skeleton report"])
     semantic = _report_value(item.get("semantic"), f"semantic material for {name}")
     _validate_semantic_material(semantic, context=name, kind=kind)
     source = _report_value(item.get("source"), f"source for {name}", optional=True)
     if kind in {"theorem", "axiom"} and source is not None:
         raise SkeletonError([f"proof-bearing source is forbidden for {kind} {name}"])
-    withheld = _report_withheld(item.get("source_withheld"), source, name)
+    withheld = item.get("source_withheld")
+    if type(withheld) is not bool or (withheld and source is not None):
+        raise SkeletonError([f"invalid withheld source flag for {name} in skeleton report"])
     start = _report_value(item.get("start_line"), f"start line for {name}", int, optional=True)
     end = _report_value(item.get("end_line"), f"end line for {name}", int, optional=True)
     if (start is None) != (end is None) or (start is not None and end is not None and (start < 1 or end < start)):
@@ -712,12 +716,6 @@ def _report_value(value: object, context: str, kind: type = str, *, optional: bo
     return value
 
 
-def _report_withheld(value: object, source: str | None, name: str) -> bool:
-    if type(value) is not bool or (value and source is not None):
-        raise SkeletonError([f"invalid withheld source flag for {name} in skeleton report"])
-    return value
-
-
 def _report_string_tuple(value: object, context: str) -> tuple[str, ...]:
     if not isinstance(value, list) or not all(isinstance(item, str) and item for item in value):
         raise SkeletonError([f"invalid {context} in skeleton report"])
@@ -726,12 +724,6 @@ def _report_string_tuple(value: object, context: str) -> tuple[str, ...]:
     return tuple(value)
 
 
-def _report_kind(value: object, context: str) -> str:
-    if not isinstance(value, str) or value not in _DECLARATION_KINDS:
-        raise SkeletonError([f"invalid declaration kind for {context} in skeleton report"])
-    return value
-
-
 def _sha256_id(content: bytes) -> str:
     return f"sha256:{hashlib.sha256(content).hexdigest()}"
 

autoform_cli/work.py

@@ -151,29 +151,20 @@ def list_ready_work(
     lean_root: str | Path | None = None,
 ) -> WorkFrontier:
     runtime = load_runtime_graph(project_or_blueprint, lean_root=lean_root)
-    missing = tuple(
-        node.id
+    unfinished = [
+        node
         for node in runtime.nodes
-        if node.dispatchable
-        and not node.mathlib
-        and not node.status.proved
-        and node.article_id is None
-    )
+        if node.dispatchable and not node.mathlib and not node.status.proved
+    ]
+    missing = tuple(node.id for node in unfinished if node.article_id is None)
     if missing:
         raise WorkError(
             "formalizable leaves need durable article_id metadata: "
             + ", ".join(missing)
             + " (plan IDs with `autoform migrate article-ids <blueprint> --json`, then"
             " add each article_id to its article's frontmatter)"
         )
-    unversioned = tuple(
-        node.id
-        for node in runtime.nodes
-        if node.dispatchable
-        and not node.mathlib
-        and not node.status.proved
-        and node.source_sha256 is None
-    )
+    unversioned = tuple(node.id for node in unfinished if node.source_sha256 is None)
     if unversioned:
         raise WorkError(
             "formalizable leaves need durable article revision metadata: "