Prepare lossless all-space review bundles and reconcile contents references

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b

scripts/prepare-bubble-bundle.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""Prepare private offline all-space review trees; never import or contact Bubble."""
+
+import argparse
+import hashlib
+import json
+import os
+import sys
+from collections import Counter
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from coulomb_social.apps.spaces.bubble_bundle import prepare_bundle  # noqa: E402
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("capture", type=Path)
+    parser.add_argument("--inventory-spaces", type=Path)
+    parser.add_argument("--output", type=Path, required=True, help="New directory outside checkout")
+    args = parser.parse_args()
+    output_created = False
+    try:
+        output = args.output.expanduser().resolve()
+        repo = Path(__file__).resolve().parents[1]
+        capture = args.capture.expanduser().resolve()
+        if output.is_relative_to(repo) or output.is_relative_to(capture) or (
+            capture.is_relative_to(output)
+        ):
+            raise ValueError("Unsafe output location")
+        inputs = {name: (capture / (name + ".json")).read_bytes()
+                  for name in ("spaces", "articles")}
+        if args.inventory_spaces:
+            if args.inventory_spaces.resolve().is_relative_to(output):
+                raise ValueError("Output overlaps retained inventory")
+            inputs["retained_spaces"] = args.inventory_spaces.read_bytes()
+        report, files = prepare_bundle(
+            json.loads(inputs["spaces"]), json.loads(inputs["articles"]),
+            inventory_spaces=json.loads(inputs["retained_spaces"])
+            if "retained_spaces" in inputs else None,
+        )
+        report["source_sha256"] = {name: hashlib.sha256(data).hexdigest()
+                                   for name, data in sorted(inputs.items())}
+        # Validate/compile entirely before creating output. Exclusive publication
+        # refuses overwrite; completion receipt is written last if interrupted.
+        output.mkdir(mode=0o700)
+        output_created = True
+        for name, data in sorted({**files, **{
+            "source-capture/" + name + ".json": data for name, data in inputs.items()
+        }}.items()):
+            dest = output / name
+            dest.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            fd = os.open(dest, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            with os.fdopen(fd, "wb") as stream:
+                stream.write(data)
+        report["status"] = "completed_offline_preparation"
+        fd = os.open(output / "reconciliation.json", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(fd, "w") as stream:
+            json.dump(report, stream, indent=2, sort_keys=True)
+            stream.write("\n")
+        print(json.dumps({
+            "spaces": len(report["spaces"]),
+            "draft_trees": sum(bool(entry["tree"]) for entry in report["spaces"]),
+            "exact_body_sidecars": sum(a["body_file"] is not None for a in report["articles"]),
+            "blocker_counts": dict(sorted(Counter(
+                blocker for s in report["spaces"] for blocker in s["blockers"]
+            ).items())), "production_imported": False,
+        }))
+    except (ValueError, OSError, TypeError, KeyError):
+        print("Preparation refused; inspect private input/output locally. " + (
+            "Output may be incomplete; no completion receipt means it is unusable."
+            if output_created else "No output was published."
+        ), file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())