gabriel / muse public
sandbox.py python
185 lines 6.5 KB
Raw
sha256:408631a8dc866047a2cff7440933b461af5abd4b55e755599765e433abceb88b feat(dev-safety): Phase 6 of #185 — wire together, document… Sonnet 5 minor ⚠ breaking 3 days ago
1 #!/usr/bin/env python3
2 """Disposable sandbox tooling for muse's own dev/testing (musehub staging #185 Phase 4).
3
4 Phase 3's guard rail (muse/cli/dev_guard.py) refuses mutating muse-dev
5 commands against canonical repos. This module is where that dev work is
6 supposed to actually happen instead: an APFS copy-on-write clone
7 (`cp -c -R`) of a canonical repo, timestamped and rotated, so testing
8 muse's own engine changes never risks the canonical object store — a full
9 clone costs almost nothing on APFS until the copy diverges from the
10 original.
11
12 Usage::
13
14 python3 sandbox.py refresh muse --json
15 python3 sandbox.py run muse -- status --json
16 """
17 from __future__ import annotations
18
19 import argparse
20 import json
21 import shutil
22 import subprocess
23 import sys
24 from datetime import datetime, timezone
25 from pathlib import Path
26
27 DEFAULT_CANONICAL_ROOTS = {
28 "muse": Path.home() / "ecosystem" / "muse",
29 "musehub": Path.home() / "ecosystem" / "musehub",
30 }
31 DEFAULT_SANDBOX_BASE = Path.home() / "dev" / "sandboxes"
32 DEFAULT_KEEP = 5
33
34
35 class SandboxNotFoundError(RuntimeError):
36 pass
37
38
39 def _sandbox_dir(repo: str, sandbox_base: Path) -> Path:
40 return sandbox_base / f"{repo}-sandbox"
41
42
43 def refresh_sandbox(
44 repo: str,
45 *,
46 canonical_root: Path | None = None,
47 sandbox_base: Path | None = None,
48 backup_base: Path | None = None,
49 keep: int = DEFAULT_KEEP,
50 ) -> Path:
51 """Clone ``repo``'s canonical checkout into a new timestamped sandbox.
52
53 Always takes a snapshot-store.sh backup of canonical first (#185 Phase 6)
54 — cheap insurance, belt-and-suspenders even though the sandbox clone
55 itself never touches canonical.
56
57 Updates the ``current`` symlink to point at the new clone and prunes
58 older timestamped clones beyond ``keep``. Returns the new sandbox path.
59 """
60 base = sandbox_base if sandbox_base is not None else DEFAULT_SANDBOX_BASE
61 if canonical_root is None:
62 if repo not in DEFAULT_CANONICAL_ROOTS:
63 raise ValueError(f"Unknown repo {repo!r} — no default canonical root registered for it.")
64 canonical_root = DEFAULT_CANONICAL_ROOTS[repo]
65 if not canonical_root.exists():
66 raise FileNotFoundError(f"Canonical repo root does not exist: {canonical_root}")
67
68 from backup import create_snapshot
69 create_snapshot(repo, canonical_root=canonical_root, backup_base=backup_base)
70
71 sandbox_dir = _sandbox_dir(repo, base)
72 sandbox_dir.mkdir(parents=True, exist_ok=True)
73
74 timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%f")
75 dest = sandbox_dir / timestamp
76 if dest.exists():
77 raise RuntimeError(f"Sandbox destination already exists (clock collision?): {dest}")
78
79 subprocess.run(
80 ["cp", "-c", "-R", str(canonical_root), str(dest)],
81 check=True, capture_output=True, text=True,
82 )
83
84 current_link = sandbox_dir / "current"
85 if current_link.exists() or current_link.is_symlink():
86 current_link.unlink()
87 current_link.symlink_to(timestamp)
88
89 _prune_old_sandboxes(sandbox_dir, keep=keep)
90
91 return dest
92
93
94 def _prune_old_sandboxes(sandbox_dir: Path, *, keep: int) -> None:
95 timestamped_dirs = sorted(
96 (p for p in sandbox_dir.iterdir() if p.is_dir() and not p.is_symlink()),
97 key=lambda p: p.name,
98 )
99 for stale in timestamped_dirs[:-keep] if keep > 0 else timestamped_dirs:
100 shutil.rmtree(stale)
101
102
103 def resolve_current_sandbox(repo: str, *, sandbox_base: Path | None = None) -> Path:
104 base = sandbox_base if sandbox_base is not None else DEFAULT_SANDBOX_BASE
105 current_link = _sandbox_dir(repo, base) / "current"
106 if not current_link.exists():
107 raise SandboxNotFoundError(
108 f"No sandbox found for {repo!r} — run `sandbox-refresh.sh {repo}` first."
109 )
110 return current_link.resolve()
111
112
113 def _cmd_refresh(args: argparse.Namespace) -> int:
114 try:
115 path = refresh_sandbox(
116 args.repo,
117 sandbox_base=Path(args.sandbox_base) if args.sandbox_base else None,
118 backup_base=Path(args.backup_base) if args.backup_base else None,
119 keep=args.keep,
120 )
121 except (ValueError, FileNotFoundError, RuntimeError) as e:
122 print(f"❌ {e}", file=sys.stderr)
123 return 1
124 if args.json:
125 print(json.dumps({"repo": args.repo, "sandbox_path": str(path)}))
126 else:
127 print(f"✅ Refreshed sandbox for {args.repo}: {path}")
128 return 0
129
130
131 def _cmd_run(args: argparse.Namespace) -> int:
132 try:
133 sandbox_path = resolve_current_sandbox(
134 args.repo,
135 sandbox_base=Path(args.sandbox_base) if args.sandbox_base else None,
136 )
137 except SandboxNotFoundError as e:
138 print(f"❌ {e}", file=sys.stderr)
139 return 1
140
141 muse_dev = shutil.which("muse-dev")
142 if muse_dev is None:
143 print("❌ `muse-dev` not found on PATH.", file=sys.stderr)
144 return 1
145
146 proc = subprocess.run([muse_dev, "-C", str(sandbox_path), *args.muse_dev_args])
147 return proc.returncode
148
149
150 def main(argv: list[str] | None = None) -> int:
151 raw_argv = list(sys.argv[1:] if argv is None else argv)
152
153 # `run`'s trailing muse-dev args are split out manually on the first
154 # literal `--`, rather than via argparse.REMAINDER: REMAINDER greedily
155 # swallows *everything* from the first token after the leading positional
156 # (including our own --sandbox-base), regardless of where `--` appears.
157 muse_dev_args: list[str] = []
158 if raw_argv[:1] == ["run"] and "--" in raw_argv:
159 sep = raw_argv.index("--")
160 muse_dev_args = raw_argv[sep + 1:]
161 raw_argv = raw_argv[:sep]
162
163 parser = argparse.ArgumentParser(description=__doc__)
164 subparsers = parser.add_subparsers(dest="action", required=True)
165
166 refresh_p = subparsers.add_parser("refresh", help="Create a fresh sandbox clone of a canonical repo.")
167 refresh_p.add_argument("repo", help="Repo name, e.g. muse or musehub.")
168 refresh_p.add_argument("--sandbox-base", default=None, help=argparse.SUPPRESS)
169 refresh_p.add_argument("--backup-base", default=None, help=argparse.SUPPRESS)
170 refresh_p.add_argument("--keep", type=int, default=DEFAULT_KEEP)
171 refresh_p.add_argument("--json", action="store_true")
172 refresh_p.set_defaults(func=_cmd_refresh)
173
174 run_p = subparsers.add_parser("run", help="Run a muse-dev command against the current sandbox.")
175 run_p.add_argument("repo")
176 run_p.add_argument("--sandbox-base", default=None, help=argparse.SUPPRESS)
177 run_p.set_defaults(func=_cmd_run)
178
179 args = parser.parse_args(raw_argv)
180 args.muse_dev_args = muse_dev_args
181 return args.func(args)
182
183
184 if __name__ == "__main__":
185 sys.exit(main())
File History 1 commit
sha256:408631a8dc866047a2cff7440933b461af5abd4b55e755599765e433abceb88b feat(dev-safety): Phase 6 of #185 — wire together, document… Sonnet 5 minor ⚠ 3 days ago