Stream large skeleton packets and reports, and name each skeleton phase on stderr

CHANGELOG.md

@@ -74,6 +74,14 @@ date when the release is tagged.
 - **Check and render.** One reused converter reads every page, and each linked
   path is resolved once per run, so `check` uses about a fifth less CPU on an
   1,806-article vault; neither command's output changes (#415).
+- **Large skeletons.** Packets, the report, and the check that guards their
+  publication are hashed and written in pieces, so writing a large report no
+  longer holds every packet in memory, and shared closure material is no longer
+  parsed and escaped again for every root; output bytes and hashes are unchanged.
+  `skeleton` names each phase on stderr as it starts (`skeleton: probing <n>
+  modules...`, then `assembling` once no probe runs, then `writing`). Agent
+  Review and Formalize say what a user's pause of skeleton evidence during proof
+  development still requires (#497).
 - **Builds.** `project new` turns on Lake's artifact cache with
   `enableArtifactCache` and `restoreAllArtifacts`, so checkouts and worktrees
   on one toolchain reuse each other's builds; Setup offers the same two lines

autoform_cli/README.md

@@ -587,6 +587,10 @@ check, the helper's build, and each probe have their own 600-second budget;
 run at a time. These limits are resource controls, not a security or
 authenticity boundary.
 
+Each phase is named on stderr as it starts: `skeleton: probing <n> modules...`
+when Lean probes start, `skeleton: assembling...` once none runs (the build must
+still not change), and `skeleton: writing...` before `--output` or `--packets`.
+
 `--packets DIR` writes one comment-stripped packet per skeleton, named after
 its article and declaration, with a manifest of articles and hashes; a blind
 auditor gets the hash-named packets of `review prepare` instead. The

autoform_cli/__main__.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import argparse
+import contextlib
 import hashlib
 import json
 import os
@@ -1754,22 +1755,37 @@ def _skeleton(args: argparse.Namespace) -> int:
             print("error: --output must be disjoint from packet and passage directories", file=sys.stderr)
             return 2
     stream = sys.stderr if args.json else sys.stdout
+
+    def progress(phase: str) -> None:
+        # Information only. Without fd 2, print(file=None) would write to stdout,
+        # and a line left in a broken stderr's buffer would fail the exit-time flush.
+        if sys.stderr is None:
+            return
+        try:
+            print(f"skeleton: {phase}...", file=sys.stderr, flush=True)
+        except OSError:
+            with contextlib.suppress(OSError, ValueError):
+                os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stderr.fileno())
+
     try:
         report = extract_skeletons(
             args.blueprint_dir,
             lean_root=args.lean_root,
             timeout=args.timeout,
             node_ids=tuple(args.nodes) if args.nodes else None,
+            progress=progress,
         )
         if args.packets is not None and not report.clean:
             print("error: refusing to publish review packets from an incomplete skeleton report", file=sys.stderr)
         elif args.packets is not None:
+            progress("writing")
             written = write_packets(report, args.packets, passages=args.passages, report_path=args.output)
             print(f"{args.packets}: {len(written)} blind packet(s) written", file=stream)
             if args.passages is not None:
                 cited = sum(1 for node in report.nodes if node.passage is not None)
                 print(f"{args.passages}: {cited} source passage(s) written", file=stream)
         if args.output is not None and (args.packets is None or not report.clean):
+            progress("writing")
             write_skeleton_report(report, args.output)
     except SkeletonError as exc:
         for issue in exc.issues:
@@ -1784,7 +1800,9 @@ def _skeleton(args: argparse.Namespace) -> int:
         for issue in report.unresolved:
             print(f"error: {issue.message}", file=stream)
     elif args.json:
-        print(report.to_json())
+        # In pieces, as `--output` writes it: a large report need not be held twice.
+        sys.stdout.writelines(report._json_pieces())
+        sys.stdout.write("\n")
     else:
         print(format_report(report, lean_root=args.lean_root), end="")
     return 0 if report.clean else 1

autoform_cli/_managed_output.py

@@ -27,6 +27,7 @@
 import sys
 import tempfile
 import warnings
+from collections.abc import Iterable
 from pathlib import Path
 
 from ._tree_snapshot import _normalized_name
@@ -60,7 +61,12 @@ def output_identity(path: Path) -> tuple[int, int, str] | None:
             if child.is_dir():
                 digest.update(b"directory\0")
             elif child.is_file():
-                digest.update(b"file\0" + child.read_bytes() + b"\0")
+                # In blocks: one packet of a large closure can be gigabytes.
+                digest.update(b"file\0")
+                with child.open("rb") as stream:
+                    while block := stream.read(1 << 20):
+                        digest.update(block)
+                digest.update(b"\0")
             else:
                 raise SkeletonError([f"refusing special file in packet output: {child}"])
     except OSError as exc:
@@ -101,9 +107,13 @@ def validate_managed_output(path: Path, *, kind: str, schema: str | None = None)
     return identity
 
 
-def stage_output(destination: Path, content: str | None = None, *, stages: list[Path] | None = None) -> Path:
+def stage_output(
+    destination: Path, content: str | Iterable[str] | None = None, *, stages: list[Path] | None = None
+) -> Path:
     """Create a same-directory stage, a directory or a file of ``content``, with the old mode.
 
+    ``content`` may come in pieces, which are written as they come.
+
     With ``stages``, the stage is registered there before it is created, so
     whoever owns ``stages`` removes it however this call or a later step ends.
     Without, a stage this call cannot finish is removed before the error leaves.
@@ -130,7 +140,7 @@ def stage_output(destination: Path, content: str | None = None, *, stages: list[
     try:
         if stream is not None:
             with stream:
-                stream.write(content)
+                stream.writelines((content,) if isinstance(content, str) else content)
                 stream.flush()
                 os.fsync(stream.fileno())
         if mode is not None: