gabriel / muse public
backup.py python
271 lines 10.4 KB
Raw
sha256:a57e9ca1e385a1e7a0e3e28094bc35799950a3ed7b2421712d6fb7f0e523a4a0 feat(dev-safety): Phase 5 of #185 — belt-and-suspenders aut… Sonnet 5 patch 3 days ago
1 #!/usr/bin/env python3
2 """Belt-and-suspenders backups for canonical muse/musehub repos (#185 Phase 5).
3
4 Two independent mechanisms, on purpose — not either/or:
5
6 - **Suspenders** (fast, frequent, local): APFS copy-on-write snapshots of
7 just ``.muse/`` (not the working tree). Near-instant, cheap enough to take
8 before any risky operation.
9 - **Belt** (verified, portable, survives an APFS-level disaster): a
10 `muse bundle` archive of every branch, immediately verified after
11 creation.
12
13 Restore refuses to overwrite a canonical repo that currently passes
14 `muse verify` unless ``force=True`` — don't clobber a good copy by accident.
15 """
16 from __future__ import annotations
17
18 import json
19 import shutil
20 import subprocess
21 import sys
22 from datetime import datetime, timezone
23 from pathlib import Path
24
25 DEFAULT_CANONICAL_ROOTS = {
26 "muse": Path.home() / "ecosystem" / "muse",
27 "musehub": Path.home() / "ecosystem" / "musehub",
28 }
29 DEFAULT_BACKUP_BASE = Path.home() / "dev" / "backups"
30 DEFAULT_SNAPSHOT_KEEP = 20
31 DEFAULT_BUNDLE_KEEP = 20
32
33
34 class CanonicalHealthyError(RuntimeError):
35 """Raised when a restore is refused because canonical already passes `muse verify`."""
36
37
38 def _resolve_canonical(repo: str, canonical_root: Path | None) -> Path:
39 if canonical_root is not None:
40 return canonical_root
41 if repo not in DEFAULT_CANONICAL_ROOTS:
42 raise ValueError(f"Unknown repo {repo!r} — no default canonical root registered for it.")
43 return DEFAULT_CANONICAL_ROOTS[repo]
44
45
46 def _timestamp() -> str:
47 return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%f")
48
49
50 def _canonical_is_healthy(canonical_root: Path) -> bool:
51 proc = subprocess.run(
52 ["muse", "verify", "--json"], cwd=canonical_root, capture_output=True, text=True,
53 )
54 if proc.returncode not in (0, 1):
55 raise RuntimeError(f"`muse verify` failed unexpectedly: {proc.stderr.strip()}")
56 try:
57 return bool(json.loads(proc.stdout).get("all_ok", False))
58 except json.JSONDecodeError:
59 return False
60
61
62 def _guard_restore(canonical_root: Path, force: bool) -> None:
63 if force:
64 return
65 if _canonical_is_healthy(canonical_root):
66 raise CanonicalHealthyError(
67 f"{canonical_root} currently passes `muse verify` — refusing to restore over "
68 f"a healthy repo. Pass force=True if you're certain."
69 )
70
71
72 def _prune(dir_path: Path, *, keep: int, is_relevant) -> None:
73 entries = sorted((p for p in dir_path.iterdir() if is_relevant(p)), key=lambda p: p.name)
74 for stale in entries[:-keep] if keep > 0 else entries:
75 if stale.is_dir():
76 shutil.rmtree(stale)
77 else:
78 stale.unlink()
79
80
81 # ── Snapshots ("suspenders") ──────────────────────────────────────────────
82
83 def _snapshot_dir(repo: str, backup_base: Path) -> Path:
84 return backup_base / f"{repo}-snapshots"
85
86
87 def create_snapshot(
88 repo: str, *, canonical_root: Path | None = None, backup_base: Path | None = None,
89 keep: int = DEFAULT_SNAPSHOT_KEEP,
90 ) -> Path:
91 """APFS copy-on-write snapshot of canonical's ``.muse/`` only."""
92 root = _resolve_canonical(repo, canonical_root)
93 base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE
94 snap_root = _snapshot_dir(repo, base)
95 snap_root.mkdir(parents=True, exist_ok=True)
96
97 dest = snap_root / _timestamp()
98 if dest.exists():
99 raise RuntimeError(f"Snapshot destination already exists (clock collision?): {dest}")
100 dest.mkdir(parents=True)
101
102 subprocess.run(
103 ["cp", "-c", "-R", str(root / ".muse"), str(dest / ".muse")],
104 check=True, capture_output=True, text=True,
105 )
106
107 _prune(snap_root, keep=keep, is_relevant=lambda p: p.is_dir())
108 return dest
109
110
111 def list_snapshots(repo: str, *, backup_base: Path | None = None) -> list[Path]:
112 base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE
113 snap_root = _snapshot_dir(repo, base)
114 if not snap_root.exists():
115 return []
116 return sorted((p for p in snap_root.iterdir() if p.is_dir()), key=lambda p: p.name)
117
118
119 def restore_from_snapshot(
120 repo: str, snapshot_name: str, *, canonical_root: Path | None = None,
121 backup_base: Path | None = None, force: bool = False,
122 ) -> None:
123 root = _resolve_canonical(repo, canonical_root)
124 base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE
125 snapshot_path = _snapshot_dir(repo, base) / snapshot_name / ".muse"
126 if not snapshot_path.exists():
127 raise FileNotFoundError(f"No such snapshot: {snapshot_path}")
128
129 _guard_restore(root, force)
130
131 live_muse = root / ".muse"
132 if live_muse.exists():
133 shutil.rmtree(live_muse)
134 subprocess.run(
135 ["cp", "-c", "-R", str(snapshot_path), str(live_muse)],
136 check=True, capture_output=True, text=True,
137 )
138
139
140 # ── Bundles ("belt") ──────────────────────────────────────────────────────
141
142 def _bundle_dir(repo: str, backup_base: Path) -> Path:
143 return backup_base / f"{repo}-bundles"
144
145
146 def create_bundle_backup(
147 repo: str, *, canonical_root: Path | None = None, backup_base: Path | None = None,
148 keep: int = DEFAULT_BUNDLE_KEEP,
149 ) -> Path:
150 """`muse bundle create` over every local branch, immediately verified."""
151 root = _resolve_canonical(repo, canonical_root)
152 base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE
153 bundle_root = _bundle_dir(repo, base)
154 bundle_root.mkdir(parents=True, exist_ok=True)
155
156 branches_proc = subprocess.run(
157 ["muse", "branch", "--json"], cwd=root, capture_output=True, text=True, check=True,
158 )
159 branch_names = [b["name"] for b in json.loads(branches_proc.stdout)]
160 if not branch_names:
161 raise RuntimeError(f"No branches found in {root} — nothing to bundle.")
162
163 dest = bundle_root / f"{_timestamp()}.muse"
164 subprocess.run(
165 ["muse", "bundle", "create", str(dest), *branch_names, "--json"],
166 cwd=root, check=True, capture_output=True, text=True,
167 )
168
169 verify_proc = subprocess.run(
170 ["muse", "bundle", "verify", str(dest), "--json"],
171 cwd=root, capture_output=True, text=True,
172 )
173 if not json.loads(verify_proc.stdout).get("all_ok", False):
174 dest.unlink(missing_ok=True)
175 raise RuntimeError(f"Freshly created bundle failed verification: {verify_proc.stdout}")
176
177 _prune(bundle_root, keep=keep, is_relevant=lambda p: p.is_file() and p.suffix == ".muse")
178 return dest
179
180
181 def list_bundles(repo: str, *, backup_base: Path | None = None) -> list[Path]:
182 base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE
183 bundle_root = _bundle_dir(repo, base)
184 if not bundle_root.exists():
185 return []
186 return sorted(
187 (p for p in bundle_root.iterdir() if p.is_file() and p.suffix == ".muse"),
188 key=lambda p: p.name,
189 )
190
191
192 def restore_from_bundle(
193 repo: str, bundle_name: str, *, canonical_root: Path | None = None,
194 backup_base: Path | None = None, force: bool = False,
195 ) -> None:
196 root = _resolve_canonical(repo, canonical_root)
197 base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE
198 bundle_path = _bundle_dir(repo, base) / bundle_name
199 if not bundle_path.exists():
200 raise FileNotFoundError(f"No such bundle: {bundle_path}")
201
202 _guard_restore(root, force)
203
204 # `muse bundle unbundle` treats an object as "already present" purely by
205 # path existing on disk — it never re-validates content, so unbundling
206 # on top of an existing (possibly corrupted) store can silently leave
207 # corruption in place. A restore must fully replace, the same way
208 # restore_from_snapshot does, not overlay.
209 live_muse = root / ".muse"
210 if live_muse.exists():
211 shutil.rmtree(live_muse)
212 subprocess.run(["muse", "init", "--json"], cwd=root, check=True, capture_output=True, text=True)
213
214 subprocess.run(
215 ["muse", "bundle", "unbundle", str(bundle_path), "--verify", "--json"],
216 cwd=root, check=True, capture_output=True, text=True,
217 )
218
219
220 def main(argv: list[str] | None = None) -> int:
221 import argparse
222
223 parser = argparse.ArgumentParser(description=__doc__)
224 subparsers = parser.add_subparsers(dest="action", required=True)
225
226 for name, fn in (("snapshot", create_snapshot), ("bundle", create_bundle_backup)):
227 p = subparsers.add_parser(name, help=f"Create a {name} backup.")
228 p.add_argument("repo")
229 p.add_argument("--keep", type=int, default=DEFAULT_SNAPSHOT_KEEP if name == "snapshot" else DEFAULT_BUNDLE_KEEP)
230 p.add_argument("--backup-base", default=None, help=argparse.SUPPRESS)
231 p.add_argument("--json", action="store_true")
232 p.set_defaults(_create=fn)
233
234 restore_p = subparsers.add_parser("restore", help="Restore canonical from a snapshot or bundle.")
235 restore_p.add_argument("repo")
236 restore_p.add_argument("--from-snapshot", default=None)
237 restore_p.add_argument("--from-bundle", default=None)
238 restore_p.add_argument("--force", action="store_true")
239 restore_p.add_argument("--backup-base", default=None, help=argparse.SUPPRESS)
240 restore_p.add_argument("--json", action="store_true")
241
242 args = parser.parse_args(argv)
243 backup_base = Path(args.backup_base) if args.backup_base else None
244
245 try:
246 if args.action in ("snapshot", "bundle"):
247 path = args._create(args.repo, backup_base=backup_base, keep=args.keep)
248 print(json.dumps({"repo": args.repo, "path": str(path)}) if args.json
249 else f"✅ Created {args.action} backup: {path}")
250 return 0
251
252 if args.action == "restore":
253 if bool(args.from_snapshot) == bool(args.from_bundle):
254 print("❌ Exactly one of --from-snapshot or --from-bundle is required.", file=sys.stderr)
255 return 1
256 if args.from_snapshot:
257 restore_from_snapshot(args.repo, args.from_snapshot, backup_base=backup_base, force=args.force)
258 else:
259 restore_from_bundle(args.repo, args.from_bundle, backup_base=backup_base, force=args.force)
260 print(json.dumps({"repo": args.repo, "status": "restored"}) if args.json
261 else f"✅ Restored {args.repo}.")
262 return 0
263 except (ValueError, FileNotFoundError, RuntimeError, CanonicalHealthyError) as e:
264 print(f"❌ {e}", file=sys.stderr)
265 return 1
266
267 return 1
268
269
270 if __name__ == "__main__":
271 sys.exit(main())
File History 1 commit
sha256:a57e9ca1e385a1e7a0e3e28094bc35799950a3ed7b2421712d6fb7f0e523a4a0 feat(dev-safety): Phase 5 of #185 — belt-and-suspenders aut… Sonnet 5 patch 3 days ago