Plan accepted-space date corrections and bind routine recovery to scoped APIs

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

scripts/plan-source-date-backfill.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+"""Plan an existing space's original-date correction from private read-only metadata."""
+
+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.date_backfill import plan_date_backfill  # noqa: E402
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("capture", type=Path, help="Private spaces.json/articles.json capture")
+    parser.add_argument("--observation", type=Path, required=True)
+    parser.add_argument("--selection", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    try:
+        output = args.output.expanduser().resolve()
+        repo = Path(__file__).resolve().parents[1]
+        if output.is_relative_to(repo):
+            raise ValueError("Output must be private outside checkout")
+        paths = {"spaces": args.capture / "spaces.json", "articles": args.capture / "articles.json",
+                 "observation": args.observation, "selection": args.selection}
+        if any(path.resolve() == output for path in paths.values()):
+            raise ValueError("Output overlaps input")
+        raw = {name: path.read_bytes() for name, path in paths.items()}
+        inputs = {name: json.loads(value) for name, value in raw.items()}
+        plan = plan_date_backfill(inputs["observation"], inputs["spaces"], inputs["articles"],
+                                  inputs["selection"])
+        plan["source_sha256"] = {name: hashlib.sha256(value).hexdigest()
+                                 for name, value in sorted(raw.items())}
+        fd = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(fd, "w") as stream:
+            json.dump(plan, stream, indent=2, sort_keys=True)
+            stream.write("\n")
+        print(json.dumps({
+            "proposed_changes": len(plan["proposed_changes"]),
+            "page_status_counts": dict(sorted(Counter(p["status"] for p in plan["pages"]).items())),
+            "blockers": len(plan["blockers"]), "ready_to_apply": False,
+            "production_writes": False,
+        }))
+    except (ValueError, OSError, TypeError, KeyError, AttributeError, OverflowError):
+        print("Date plan refused; inspect private inputs/output locally.", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())