Harden original-date migration and verify Coulomb recovery admission

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

scripts/plan-bubble-transfer.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""Plan all captured spaces newest first, offline; private output, no import."""
+
+import argparse
+import hashlib
+import json
+import os
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from coulomb_social.apps.spaces.bubble_transfer import transfer_plan  # noqa: E402
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("capture", type=Path, help="Private spaces.json/articles.json directory")
+    parser.add_argument("--output", type=Path, required=True, help="New private JSON file")
+    parser.add_argument("--inventory-spaces", type=Path,
+                        help="Retained spaces.json; keep absent spaces queued as not refreshed")
+    args = parser.parse_args()
+    try:
+        sources = {name: (args.capture / (name + ".json")).read_bytes()
+                   for name in ("spaces", "articles")}
+        inventory = None
+        if args.inventory_spaces:
+            sources["retained_spaces"] = args.inventory_spaces.read_bytes()
+            inventory = json.loads(sources["retained_spaces"])
+        plan = transfer_plan(json.loads(sources["spaces"]), json.loads(sources["articles"]),
+                             inventory_spaces=inventory)
+        plan["source_sha256"] = {name: hashlib.sha256(data).hexdigest()
+                                 for name, data in sources.items()}
+        # Never print private IDs, names, notes or source errors to the terminal.
+        fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(fd, "w") as stream:
+            json.dump(plan, stream, indent=2)
+            stream.write("\n")
+        print(json.dumps({
+            "spaces": len(plan["spaces"]),
+            "captured_graph_complete": sum(s["captured_graph_complete"] for s in plan["spaces"]),
+            "unassigned_articles": len(plan["unassigned_article_ids"]),
+            "inventory_only_spaces": len(plan["inventory_only_space_ids"]),
+            "live_source_completeness_verified": False, "production_imported": False,
+        }))
+    except (ValueError, OSError, TypeError):
+        print("Transfer plan refused; inspect the private capture/output locally", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())