sandbox.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 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()) |