Open statements and a revision contract for shared Lean declarations

.github/workflows/tests.yml

@@ -63,6 +63,10 @@ jobs:
           tests/test_project_inspect.py::test_stale_inherited_mathlib_is_ignored_by_real_lake
         env:
           AUTOFORM_REQUIRE_REAL_LEAN_TESTS: "1"
+      - name: Run the open-statement audit and impact probe against real Lean
+        run: timeout --signal=TERM --kill-after=30s 10m uv run pytest -q tests/test_lake_artifact_audit.py tests/test_impact.py tests/test_contract.py
+        env:
+          AUTOFORM_REQUIRE_REAL_LEAN_TESTS: "1"
 
   windows-hardening:
     name: Windows inspection, subprocess, and publication hardening

autoform_cli/__main__.py

@@ -20,6 +20,7 @@
 from .doctor import diagnose_project
 from .dashboard import publication_bound_live_state, serve_dashboard
 from .graph import GraphValidationError, load_graph
+from .impact import ImpactError, format_impact, revision_impact
 from .lean import build_linker, declaration_names
 from .project import ProjectCatalogError, inspect_project, load_release_catalog
 from .render import PublicationError, render_site
@@ -34,7 +35,7 @@
     write_packets,
     write_skeleton_report,
 )
-from .work import WORK_SCHEMA, WorkError, list_ready_work, work_context
+from .work import WORK_SCHEMA, WorkError, assumption_contract, list_ready_work, work_context
 
 
 def main(argv: Sequence[str] | None = None) -> int:
@@ -132,11 +133,42 @@ def main(argv: Sequence[str] | None = None) -> int:
     work_context_parser.add_argument(
         "--json", action="store_true", help="write stable machine-readable output"
     )
+    work_assumptions = work_subparsers.add_parser(
+        "assumptions", help="list open statements and the open statements each stated article rests on"
+    )
+    work_assumptions.add_argument(
+        "target", nargs="?", default=".", help="project root or blueprint directory"
+    )
+    work_assumptions.add_argument("--json", action="store_true", help="write stable machine-readable output")
+    work_impact = work_subparsers.add_parser(
+        "impact", help="show which articles and helpers a revision of an article's Lean declarations affects"
+    )
+    work_impact.add_argument("selector", help="path-derived node id or durable article_id")
+    work_impact.add_argument("target", nargs="?", default=".", help="project root or blueprint directory")
+    work_impact.add_argument(
+        "--lean-root", type=Path, required=True, metavar="PATH", help="the built Lean project"
+    )
+    work_impact.add_argument(
+        "--declaration",
+        action="append",
+        default=[],
+        dest="declarations",
+        metavar="NAME",
+        help="revise this project-local constant instead of the article's lean: declarations (repeatable)",
+    )
+    work_impact.add_argument("--json", action="store_true", help="write stable machine-readable output")
+    work_impact.add_argument(
+        "--timeout",
+        type=_positive_seconds,
+        metavar="SECONDS",
+        help=f"seconds the Lean probe may run (default {DEFAULT_PROBE_TIMEOUT:g}); "
+        "the Lake freshness check before it has its own budget",
+    )
     claim = subparsers.add_parser("claim", help="coordinate temporary node ownership through Git refs")
     claim_subparsers = claim.add_subparsers(dest="claim_command", required=True)
     for operation in ("acquire", "renew", "release"):
         command = claim_subparsers.add_parser(operation)
-        command.add_argument("node_id")
+        command.add_argument("node_id", nargs="+", help="claim target(s); several change all-or-nothing")
         _add_claim_board_arguments(command)
         if operation in {"acquire", "renew"}:
             command.add_argument("--ttl", type=int, default=CLAIM_TTL_S)
@@ -429,6 +461,10 @@ def _project(args: argparse.Namespace) -> int:
 
 
 def _work(args: argparse.Namespace) -> int:
+    if args.work_command == "assumptions":
+        return _work_assumptions(args)
+    if args.work_command == "impact":
+        return _work_impact(args)
     # Only loading the roadmap can fail on the project's paths; printing the
     # result stays outside, so an output error is not reported as one.
     try:
@@ -455,12 +491,18 @@ def _work(args: argparse.Namespace) -> int:
         if args.json:
             print(frontier.to_json())
             return 0
+        if frontier.open_statements:
+            print("Open statements: allowed (a statement may land with a sorry proof)")
         if not frontier.items:
             print("No ready formalization work.")
             return 0
         for item in frontier.items:
             durable = f" [{item.article_id}]" if item.article_id else ""
             print(_human_text(f"{item.phase}: {item.node_id}{durable} - {item.title}"))
+            if item.assumes:
+                print(_human_text("  assumes: " + ", ".join(item.assumes)))
+            if item.revision:
+                print("  revision: start from `autoform work impact`")
         return 0
 
     if args.json:
@@ -485,6 +527,12 @@ def _work(args: argparse.Namespace) -> int:
     print(_human_text(f"{item.title} ({item.node_id})"))
     print(f"State: {item.state}")
     print(f"Phase: {phase}")
+    if item.revision:
+        print("Revision: the statement was retracted; start from `autoform work impact`")
+    if item.open_statements:
+        print("Open statements: allowed")
+        if item.assumes:
+            print(_human_text("Assumes: " + ", ".join(item.assumes)))
     print(_human_text(f"Claim target: {item.claim_target}"))
     if item.blockers:
         print(_human_text("Blocked by: " + ", ".join(item.blockers)))
@@ -501,6 +549,75 @@ def _work(args: argparse.Namespace) -> int:
     return 0
 
 
+def _work_assumptions(args: argparse.Namespace) -> int:
+    # As in `_work`, only loading the roadmap is reported as a path error.
+    try:
+        contract = assumption_contract(args.target)
+        # The text report also names conditional articles without `lean:`,
+        # which the contract leaves out because CI has nothing to check there.
+        runtime = None if args.json else load_runtime_graph(args.target)
+    except (GraphValidationError, RuntimeProjectionError) as error:
+        for issue in error.issues:
+            print(f"error: {_human_text(issue)}", file=sys.stderr)
+        return 2
+    except (OSError, RuntimeError, ValueError):
+        print("error: project or blueprint path cannot be read", file=sys.stderr)
+        return 2
+
+    if args.json:
+        print(contract.to_json())
+        return 0
+    print(_human_text(f"Open statements: {'allowed' if contract.open_statements else 'forbidden'}"))
+    listed = {article.id: article for article in contract.articles}
+    for node in sorted(runtime.nodes, key=lambda candidate: candidate.id):
+        article = listed.get(node.id)
+        assumes = f"assumes {', '.join(node.status.assumes)}"
+        if article is not None and article.open:
+            # An open statement is not conditional: its own proof is missing.
+            line = f"open: {article.id} ({', '.join(article.declarations)})"
+            print(_human_text(f"{line} {assumes}" if article.assumes else line))
+        elif node.status.state == "conditional":
+            print(_human_text(f"conditional: {node.id} {assumes}"))
+        elif article is not None and article.assumes:
+            # Not proved, so nothing is conditional yet; its proof would rest on these.
+            print(_human_text(f"unproved: {article.id} {assumes}"))
+    return 0
+
+
+def _work_impact(args: argparse.Namespace) -> int:
+    try:
+        report = revision_impact(
+            args.target,
+            args.selector,
+            lean_root=args.lean_root,
+            declarations=args.declarations,
+            timeout=args.timeout,
+        )
+    except (GraphValidationError, RuntimeProjectionError) as error:
+        for issue in error.issues:
+            print(f"error: {_human_text(issue)}", file=sys.stderr)
+        return 2
+    except (WorkError, ImpactError) as error:
+        print(f"error: {_human_text(error)}", file=sys.stderr)
+        return 2
+    except SkeletonError as error:
+        # Probe failures carry Lean's multi-line output; escape it line by line.
+        for issue in error.issues:
+            for index, line in enumerate(str(issue).splitlines() or [""]):
+                print(("error: " if index == 0 else "") + _human_text(line), file=sys.stderr)
+        return 2
+    except (OSError, RuntimeError, ValueError):
+        print("error: project, blueprint, or Lean root path cannot be read", file=sys.stderr)
+        return 2
+
+    if args.json:
+        print(report.to_json())
+        return 0
+    for line in format_impact(report):
+        print(_human_text(line))
+    return 0
+
+
 def _print_project_inspection(result) -> None:
     if result.project_root is not None:
         print(f"Project root: {_human_text(result.project_root)}")
@@ -544,18 +661,46 @@ def _claim(args: argparse.Namespace) -> int:
             print(f"removed {board.cleanup()} expired claim(s)")
             return 0
 
-        key = author_claim_key(args.node_id)
+        past_tense = {"acquire": "acquired", "renew": "renewed", "release": "released"}
+        if len(args.node_id) > 1:
+            # Several targets change in one atomic push, so a failure holds none of them.
+            targets: dict[str, str] = {}
+            for node_id in args.node_id:
+                key = author_claim_key(node_id)
+                if key in targets:
+                    print(f"error: duplicate claim target: {node_id}", file=sys.stderr)
+                    return 2
+                targets[key] = node_id
+            if operation == "acquire":
+                result = board.acquire_many(list(targets), ttl=args.ttl, note=args.note)
+            elif operation == "renew":
+                result = board.renew_many(list(targets), ttl=args.ttl)
+            else:
+                result = board.release_many(list(targets))
+            if result:
+                for key, node_id in targets.items():
+                    print(f"{past_tense[operation]} {node_id} ({key})")
+                return 0
+            blocking = ", ".join(targets.get(key, key) for key in result.blocking)
+            reason = f"{result.reason}: {blocking}" if blocking else result.reason
+            print(
+                f"error: could not {operation} {', '.join(args.node_id)}; "
+                f"no claim was {past_tense[operation]}: {reason}"
+            )
+            return 1
+
+        node_id = args.node_id[0]
+        key = author_claim_key(node_id)
         if operation == "acquire":
             succeeded = board.acquire(key, ttl=args.ttl, note=args.note)
         elif operation == "renew":
             succeeded = board.renew(key, ttl=args.ttl)
         else:
             succeeded = board.release(key)
         if succeeded:
-            past_tense = {"acquire": "acquired", "renew": "renewed", "release": "released"}
-            print(f"{past_tense[operation]} {args.node_id} ({key})")
+            print(f"{past_tense[operation]} {node_id} ({key})")
             return 0
-        print(f"error: could not {operation} {args.node_id}; ownership is held or unverifiable")
+        print(f"error: could not {operation} {node_id}; ownership is held or unverifiable")
         return 1
     except (ClaimTransportError, ValueError) as exc:
         print(f"error: {exc}")

autoform_cli/audit.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import json
+import re
 import statistics
 from bisect import bisect_right
 from dataclasses import asdict, dataclass
@@ -16,7 +17,7 @@
 from . import status
 from .coverage import CoverageSummary, load_coverage
 from .graph import Graph, GraphValidationError, Node, load_graph
-from .lean import SourceIndex, declaration_names, index_project
+from .lean import _DECLARATION, Declaration, SourceIndex, _without_lean_comments, declaration_names, index_project
 from .markdown import FENCE as _FENCE
 from .markdown import frontmatter_end as _frontmatter_end
 from .markdown import HEADING as _HEADING
@@ -51,6 +52,17 @@
     "theorem": frozenset({"lemma", "theorem"}),
 }
 
+#: One ``@[...]`` attribute list; a string literal inside it may hold brackets.
+_ATTRIBUTE_LIST = r'@\[(?:[^\]"]|"(?:[^"\\]|\\.)*")*\]'
+#: The attribute lists and modifiers a declaration keyword closes, starting at
+#: a line start, so contiguous attribute lines above the keyword's line count.
+_DECLARATION_HEADER = re.compile(
+    rf"(?:\A|\n)[ \t]*((?:{_ATTRIBUTE_LIST}\s*)*)"
+    r"(?:(?:private|protected|noncomputable|partial|unsafe|scoped|local)\s+)*\Z"
+)
+_STRING_LITERAL = re.compile(r'"(?:[^"\\]|\\.)*"')
+_DEPRECATED_ATTRIBUTE = re.compile(r"(?:\[|,)\s*(?:(?:scoped|local)\s+)?deprecated\b")
+
 
 @dataclass(frozen=True, order=True, slots=True)
 class AuditFinding:
@@ -370,6 +382,7 @@ def _lean_findings(graph: Graph, lean_root: str | Path) -> list[AuditFinding]:
     index = index_project(root)
     spans = _source_spans(index)
     sizes: dict[str, int] = {}
+    sources: dict[Path, list[str] | None] = {}
     for node_id in sorted(graph.nodes):
         node = graph.nodes[node_id]
         article_path = _relative_path(node.path, graph.blueprint_dir)
@@ -398,6 +411,16 @@ def _lean_findings(graph: Graph, lean_root: str | Path) -> list[AuditFinding]:
             else:
                 resolved.append(declaration)
 
+        for declaration in resolved:
+            if _declared_deprecated(declaration, index.root, sources):
+                findings.append(
+                    AuditFinding(
+                        article_path,
+                        "lean-target-deprecated",
+                        f"lean target {declaration.name} is deprecated; point lean: at its replacement",
+                    )
+                )
+
         expected = _DECLARATION_KEYWORDS.get((node.declaration or "").casefold())
         if expected and resolved and not any(declaration.keyword in expected for declaration in resolved):
             actual = ", ".join(sorted({declaration.keyword for declaration in resolved}))
@@ -416,6 +439,40 @@ def _lean_findings(graph: Graph, lean_root: str | Path) -> list[AuditFinding]:
     return findings
 
 
+def _declared_deprecated(declaration: Declaration, root: Path, sources: dict[Path, list[str] | None]) -> bool:
+    """Whether the source declaration carries the ``deprecated`` attribute lexically.
+
+    Only ``@[...]`` lists before the keyword count: on the declaration's line,
+    or on the contiguous attribute lines directly above it. Comments are
+    blanked first, so a commented-out attribute does not count.
+    """
+
+    if declaration.path not in sources:
+        try:
+            text = (root / declaration.path).read_text(encoding="utf-8")
+        except (OSError, UnicodeError):
+            sources[declaration.path] = None
+        else:
+            sources[declaration.path] = _without_lean_comments(text).splitlines()
+    lines = sources[declaration.path]
+    if lines is None or not 0 < declaration.line <= len(lines):
+        return False
+    line = lines[declaration.line - 1]
+    keyword = _DECLARATION.match(line)
+    if keyword is None:
+        return False
+    # The attribute lines cannot reach above the previous declaration, so the
+    # search never rescans the whole file.
+    first = declaration.line - 1
+    while first and not _DECLARATION.match(lines[first - 1]):
+        first -= 1
+    header = _DECLARATION_HEADER.search("\n".join([*lines[first : declaration.line - 1], line[: keyword.start(1)]]))
+    return header is not None and any(
+        _DEPRECATED_ATTRIBUTE.search(_STRING_LITERAL.sub('""', attributes))
+        for attributes in re.findall(_ATTRIBUTE_LIST, header.group(1))
+    )
+
+
 def _source_spans(index: SourceIndex) -> dict[str, int]:
     """Measure each declaration's source span, up to the next declaration.