snapshot_cmd.py
python
sha256:502e812e12319e30302d07d165425ceeb21687ad30e8af3a439b1f6de36c64d6
fix: update stale tag tests to use label for free-form anno…
Sonnet 4.6
19 days ago
| 1 | """``muse snapshot`` — explicit snapshot management. |
| 2 | |
| 3 | A snapshot is Muse's fundamental unit of state: a content-addressed, |
| 4 | immutable record mapping workspace-relative paths to their SHA-256 object IDs. |
| 5 | Every commit points to exactly one snapshot. |
| 6 | |
| 7 | ``muse snapshot`` makes snapshots a first-class operation — you can capture, |
| 8 | list, inspect, and export them independently of the commit workflow. This is |
| 9 | especially useful for agents that want to checkpoint mid-work without creating |
| 10 | a formal commit. |
| 11 | |
| 12 | Subcommands:: |
| 13 | |
| 14 | muse snapshot create [-m <note>] [--json] |
| 15 | Capture current working-tree state. Notes are persisted in the record. |
| 16 | |
| 17 | muse snapshot list [--limit N] [--json] |
| 18 | List all stored snapshots, newest first. |
| 19 | |
| 20 | muse snapshot read <id> [--text] |
| 21 | Print a snapshot manifest (default: JSON; --text for human output). |
| 22 | |
| 23 | muse snapshot export <id> [-f tar.gz|zip] [-o file] [--json] |
| 24 | Export tracked files to a portable archive. --json emits a JSON |
| 25 | summary of the export alongside creating the archive. |
| 26 | |
| 27 | Security model:: |
| 28 | |
| 29 | Symlinks inside ``.muse/objects/sha256/`` are silently skipped during listing |
| 30 | and prefix-scanning. A crafted symlink pointing outside the repository |
| 31 | boundary could otherwise expose external file contents. |
| 32 | |
| 33 | The zip-slip guard (``_safe_arcname``) rejects any manifest entry whose |
| 34 | normalised path escapes the archive root. Both ``..`` segments and |
| 35 | absolute paths are rejected before any file is opened. |
| 36 | |
| 37 | JSON output schemas:: |
| 38 | |
| 39 | snapshot create --json: |
| 40 | {"repo_id": str, "snapshot_id": str, "file_count": int, |
| 41 | "note": str, "created_at": str, |
| 42 | "duration_ms": float, "exit_code": int} |
| 43 | |
| 44 | snapshot list --json: |
| 45 | {"snapshots": [{"snapshot_id": str, "file_count": int, |
| 46 | "note": str, "created_at": str}, ...], |
| 47 | "duration_ms": float, "exit_code": int} |
| 48 | |
| 49 | snapshot read --json: |
| 50 | {"snapshot_id": str, "created_at": str, "file_count": int, |
| 51 | "note": str, "manifest": {path: object_id, ...}, |
| 52 | "duration_ms": float, "exit_code": int} |
| 53 | |
| 54 | snapshot export --json: |
| 55 | {"snapshot_id": str, "output": str, "format": str, |
| 56 | "file_count": int, "size_bytes": int, |
| 57 | "duration_ms": float, "exit_code": int} |
| 58 | |
| 59 | Exit codes:: |
| 60 | |
| 61 | 0 — success |
| 62 | 1 — snapshot not found, bad arguments |
| 63 | 3 — I/O error |
| 64 | """ |
| 65 | |
| 66 | import argparse |
| 67 | import json |
| 68 | import logging |
| 69 | import pathlib |
| 70 | import tarfile |
| 71 | import zipfile |
| 72 | import sys |
| 73 | from typing import TypedDict |
| 74 | |
| 75 | from muse.core.envelope import EnvelopeJson, make_envelope |
| 76 | from muse.core.errors import ExitCode |
| 77 | from muse.core.object_store import object_path, read_object, write_object_from_path |
| 78 | from muse.core.repo import read_repo_id, require_repo |
| 79 | from muse.core.ids import hash_snapshot |
| 80 | from muse.core.snapshot import directories_from_manifest |
| 81 | from muse.core.types import Manifest, long_id, short_id, split_id |
| 82 | from muse.core.snapshots import ( |
| 83 | SnapshotRecord, |
| 84 | read_snapshot, |
| 85 | write_snapshot, |
| 86 | ) |
| 87 | from muse.core.validation import clamp_int, sanitize_display |
| 88 | from muse.core.timing import start_timer |
| 89 | from muse.plugins.registry import resolve_plugin |
| 90 | |
| 91 | logger = logging.getLogger(__name__) |
| 92 | |
| 93 | _HEX_CHARS = frozenset("0123456789abcdef") |
| 94 | |
| 95 | # --------------------------------------------------------------------------- |
| 96 | # JSON wire-format TypedDicts |
| 97 | # --------------------------------------------------------------------------- |
| 98 | |
| 99 | class _SnapshotCreateJson(EnvelopeJson): |
| 100 | """JSON output for ``muse snapshot create --json``.""" |
| 101 | |
| 102 | repo_id: str |
| 103 | snapshot_id: str |
| 104 | file_count: int |
| 105 | note: str |
| 106 | created_at: str |
| 107 | |
| 108 | class _SnapshotListItemJson(TypedDict): |
| 109 | """One entry in the ``muse snapshot list --json`` array. |
| 110 | |
| 111 | Fields |
| 112 | ------ |
| 113 | snapshot_id |
| 114 | 64-character hex content-hash of the snapshot manifest. |
| 115 | file_count |
| 116 | Number of files in the snapshot. |
| 117 | note |
| 118 | The note label stored at create time (empty string if none). |
| 119 | created_at |
| 120 | ISO-8601 timestamp of when the snapshot was created. |
| 121 | """ |
| 122 | |
| 123 | snapshot_id: str |
| 124 | file_count: int |
| 125 | note: str |
| 126 | created_at: str |
| 127 | |
| 128 | class _SnapshotListJson(EnvelopeJson): |
| 129 | """Top-level JSON output for ``muse snapshot list --json``.""" |
| 130 | |
| 131 | snapshots: list[_SnapshotListItemJson] |
| 132 | |
| 133 | class _SnapshotReadJson(EnvelopeJson): |
| 134 | """JSON output for ``muse snapshot read --json``.""" |
| 135 | |
| 136 | snapshot_id: str |
| 137 | created_at: str |
| 138 | file_count: int |
| 139 | note: str |
| 140 | manifest: Manifest |
| 141 | |
| 142 | class _SnapshotExportJson(EnvelopeJson): |
| 143 | """JSON output for ``muse snapshot export --json``.""" |
| 144 | |
| 145 | snapshot_id: str |
| 146 | output: str |
| 147 | format: str |
| 148 | file_count: int |
| 149 | size_bytes: int |
| 150 | |
| 151 | # --------------------------------------------------------------------------- |
| 152 | # Internal helpers |
| 153 | # --------------------------------------------------------------------------- |
| 154 | |
| 155 | def _safe_arcname(prefix: str, rel_path: str) -> str | None: |
| 156 | """Build an archive entry name that cannot escape the archive root (zip-slip guard). |
| 157 | |
| 158 | Returns ``None`` when *rel_path* resolves outside the intended prefix, in |
| 159 | which case the caller should skip that entry. ``prefix`` must not contain |
| 160 | ``..`` segments; the caller is responsible for validating it. |
| 161 | """ |
| 162 | clean_prefix = prefix.rstrip("/").strip() if prefix else "" |
| 163 | if ".." in clean_prefix.split("/"): |
| 164 | return None |
| 165 | |
| 166 | resolved = pathlib.PurePosixPath(rel_path) |
| 167 | if resolved.is_absolute(): |
| 168 | return None |
| 169 | parts = resolved.parts |
| 170 | if ".." in parts: |
| 171 | return None |
| 172 | safe_rel = str(resolved) |
| 173 | |
| 174 | return f"{clean_prefix}/{safe_rel}" if clean_prefix else safe_rel |
| 175 | |
| 176 | def _validate_snapshot_id_prefix(snapshot_id: str) -> str: |
| 177 | """Return a glob-safe prefix from *snapshot_id* (hex chars only, max 64).""" |
| 178 | return "".join(c for c in snapshot_id[:64] if c in _HEX_CHARS) |
| 179 | |
| 180 | def _is_bare_hex(s: str) -> bool: |
| 181 | """Return True when *s* is bare hex with no ``sha256:`` prefix. |
| 182 | |
| 183 | The sha256: prefix is a type tag — it identifies the algorithm, not just |
| 184 | labels the hash. Bare hex is ambiguous: if blake3: IDs are ever added, |
| 185 | a bare hex string cannot be mapped to the correct algorithm. |
| 186 | |
| 187 | This predicate guards every CLI boundary that accepts a snapshot ID. |
| 188 | """ |
| 189 | return bool(s) and not s.startswith("sha256:") and all(c in _HEX_CHARS for c in s) |
| 190 | |
| 191 | def _reject_if_bare_hex(snapshot_id: str) -> None: |
| 192 | """Exit with USER_ERROR and a helpful message if *snapshot_id* is bare hex. |
| 193 | |
| 194 | Called at every CLI boundary that accepts a snapshot ID, before any |
| 195 | resolution is attempted. Keeps resolution functions clean — they receive |
| 196 | only validated input. |
| 197 | """ |
| 198 | if _is_bare_hex(snapshot_id): |
| 199 | safe = sanitize_display(snapshot_id) |
| 200 | print( |
| 201 | f"❌ Bare hex IDs are not accepted — use 'sha256:{safe}' instead.\n" |
| 202 | f" Even a short prefix works: 'sha256:{safe[:12]}'", |
| 203 | file=sys.stderr, |
| 204 | ) |
| 205 | raise SystemExit(ExitCode.USER_ERROR) |
| 206 | |
| 207 | def _list_all_snapshots(root: pathlib.Path) -> list[SnapshotRecord]: |
| 208 | """Return all stored snapshots sorted newest-first.""" |
| 209 | from muse.core.object_store import iter_stored_objects, read_muse_object |
| 210 | import json as _json |
| 211 | results: list[SnapshotRecord] = [] |
| 212 | for object_id, _ in iter_stored_objects(root): |
| 213 | result = read_muse_object(root, object_id) |
| 214 | if result is None or result[0] != "snapshot": |
| 215 | continue |
| 216 | try: |
| 217 | results.append(SnapshotRecord.from_dict(_json.loads(result[1]))) |
| 218 | except Exception: |
| 219 | continue |
| 220 | return sorted(results, key=lambda s: s.created_at, reverse=True) |
| 221 | |
| 222 | def _resolve_snapshot(root: pathlib.Path, snapshot_id: str) -> SnapshotRecord | None: |
| 223 | """Resolve *snapshot_id* (full or prefix) to a :class:`SnapshotRecord`.""" |
| 224 | try: |
| 225 | snap = read_snapshot(root, snapshot_id) |
| 226 | except ValueError: |
| 227 | snap = None |
| 228 | if snap is not None: |
| 229 | return snap |
| 230 | from muse.core.object_store import iter_stored_objects, read_muse_object |
| 231 | import json as _json |
| 232 | bare_id = long_id(snapshot_id, strip=True) |
| 233 | safe_prefix = _validate_snapshot_id_prefix(bare_id) |
| 234 | for object_id, _ in iter_stored_objects(root): |
| 235 | _, hex_id = split_id(object_id) |
| 236 | if not hex_id.startswith(safe_prefix): |
| 237 | continue |
| 238 | result = read_muse_object(root, object_id) |
| 239 | if result is None or result[0] != "snapshot": |
| 240 | continue |
| 241 | try: |
| 242 | snap = SnapshotRecord.from_dict(_json.loads(result[1])) |
| 243 | return snap |
| 244 | except Exception: |
| 245 | continue |
| 246 | return None |
| 247 | |
| 248 | def _build_tar( |
| 249 | root: pathlib.Path, |
| 250 | manifest: Manifest, |
| 251 | output_path: pathlib.Path, |
| 252 | prefix: str, |
| 253 | ) -> int: |
| 254 | """Write a tar.gz archive from *manifest*; return the number of files written.""" |
| 255 | import io |
| 256 | count = 0 |
| 257 | with tarfile.open(output_path, "w:gz") as tar: |
| 258 | for rel_path, object_id in sorted(manifest.items()): |
| 259 | arcname = _safe_arcname(prefix, rel_path) |
| 260 | if arcname is None: |
| 261 | logger.warning("⚠️ Skipping unsafe path in manifest: %s", rel_path) |
| 262 | continue |
| 263 | content = read_object(root, object_id) |
| 264 | if content is None: |
| 265 | logger.warning( |
| 266 | "⚠️ Missing object %s for %s — skipping", object_id, rel_path |
| 267 | ) |
| 268 | continue |
| 269 | info = tarfile.TarInfo(name=arcname) |
| 270 | info.size = len(content) |
| 271 | tar.addfile(info, io.BytesIO(content)) |
| 272 | count += 1 |
| 273 | return count |
| 274 | |
| 275 | def _build_zip( |
| 276 | root: pathlib.Path, |
| 277 | manifest: Manifest, |
| 278 | output_path: pathlib.Path, |
| 279 | prefix: str, |
| 280 | ) -> int: |
| 281 | """Write a zip archive from *manifest*; return the number of files written.""" |
| 282 | count = 0 |
| 283 | with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: |
| 284 | for rel_path, object_id in sorted(manifest.items()): |
| 285 | arcname = _safe_arcname(prefix, rel_path) |
| 286 | if arcname is None: |
| 287 | logger.warning("⚠️ Skipping unsafe path in manifest: %s", rel_path) |
| 288 | continue |
| 289 | content = read_object(root, object_id) |
| 290 | if content is None: |
| 291 | logger.warning( |
| 292 | "⚠️ Missing object %s for %s — skipping", object_id, rel_path |
| 293 | ) |
| 294 | continue |
| 295 | zf.writestr(arcname, content) |
| 296 | count += 1 |
| 297 | return count |
| 298 | |
| 299 | # --------------------------------------------------------------------------- |
| 300 | # Registration |
| 301 | # --------------------------------------------------------------------------- |
| 302 | |
| 303 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 304 | """Register the snapshot subcommand.""" |
| 305 | parser = subparsers.add_parser( |
| 306 | "snapshot", |
| 307 | help="Explicit snapshot management.", |
| 308 | description=__doc__, |
| 309 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 310 | ) |
| 311 | subs = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND") |
| 312 | subs.required = True |
| 313 | |
| 314 | # --- create --- |
| 315 | create_p = subs.add_parser( |
| 316 | "create", |
| 317 | help="Capture the current working tree as a snapshot without committing.", |
| 318 | description=( |
| 319 | "Hash every tracked file, store content in the object store, and\n" |
| 320 | "write a SnapshotRecord to ``.muse/objects/sha256/``. No commit is\n" |
| 321 | "created — the snapshot is a standalone checkpoint usable for\n" |
| 322 | "inspection, export, or rollback independently of the commit graph.\n\n" |
| 323 | "The optional ``-m`` / ``--note`` label is persisted inside the\n" |
| 324 | "snapshot record and appears in ``muse snapshot list`` and\n" |
| 325 | "``muse snapshot read``.\n\n" |
| 326 | "Agent quickstart\n" |
| 327 | "----------------\n" |
| 328 | " muse snapshot create --json\n" |
| 329 | " muse snapshot create -m 'before refactor' --json\n" |
| 330 | " SNAP=$(muse snapshot create --json | jq -r .snapshot_id)\n\n" |
| 331 | "JSON output schema\n" |
| 332 | "------------------\n" |
| 333 | ' {"repo_id": "<str>", "snapshot_id": "<hex64>",\n' |
| 334 | ' "file_count": <int>, "note": "<str>",\n' |
| 335 | ' "created_at": "<iso8601>", "duration_ms": <float>, "exit_code": 0}\n\n' |
| 336 | "Exit codes\n" |
| 337 | "----------\n" |
| 338 | " 0 — snapshot created\n" |
| 339 | " 2 — not inside a Muse repository\n" |
| 340 | " 3 — I/O error writing to the object store\n" |
| 341 | ), |
| 342 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 343 | ) |
| 344 | create_p.add_argument( |
| 345 | "-m", "--note", default="", metavar="NOTE", |
| 346 | help="Optional note stored with the snapshot.", |
| 347 | ) |
| 348 | create_p.add_argument( |
| 349 | "--json", "-j", action="store_true", dest="json_out", |
| 350 | help="Emit a machine-readable JSON summary.", |
| 351 | ) |
| 352 | create_p.set_defaults(func=run_snapshot_create) |
| 353 | |
| 354 | # --- export --- |
| 355 | export_p = subs.add_parser( |
| 356 | "export", |
| 357 | help="Export a snapshot as a portable archive.", |
| 358 | description=( |
| 359 | "Write tracked snapshot files to a tar.gz or zip archive. No\n" |
| 360 | "``.muse/`` metadata is included. A zip-slip guard rejects any\n" |
| 361 | "manifest entry whose path escapes the archive root (absolute paths\n" |
| 362 | "and ``..`` segments are silently skipped with a warning).\n\n" |
| 363 | "If ``--output`` is omitted the archive is written to\n" |
| 364 | "``<id12>.<format>`` in the current directory. Use ``--prefix``\n" |
| 365 | "to nest all files under a directory inside the archive.\n\n" |
| 366 | "Agent quickstart\n" |
| 367 | "----------------\n" |
| 368 | " muse snapshot export <id> --json\n" |
| 369 | " muse snapshot export <id> -f zip -o release.zip --json\n" |
| 370 | " muse snapshot export <id> --prefix myproject/ --json\n\n" |
| 371 | "JSON output schema\n" |
| 372 | "------------------\n" |
| 373 | ' {"snapshot_id": "<hex64>", "output": "<path>",\n' |
| 374 | ' "format": "tar.gz"|"zip", "file_count": <int>,\n' |
| 375 | ' "size_bytes": <int>, "duration_ms": <float>, "exit_code": 0}\n\n' |
| 376 | "Exit codes\n" |
| 377 | "----------\n" |
| 378 | " 0 — archive written successfully\n" |
| 379 | " 1 — snapshot ID not found\n" |
| 380 | " 2 — not inside a Muse repository\n" |
| 381 | " 3 — I/O error writing the archive\n" |
| 382 | ), |
| 383 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 384 | ) |
| 385 | export_p.add_argument("snapshot_id", metavar="ID", help="Snapshot ID (full or prefix).") |
| 386 | export_p.add_argument( |
| 387 | "--format", dest="fmt", default="tar.gz", choices=["tar.gz", "zip"], |
| 388 | help="Archive format: tar.gz (default) or zip.", |
| 389 | ) |
| 390 | export_p.add_argument( |
| 391 | "--output", "-o", default=None, metavar="FILE", help="Output file path." |
| 392 | ) |
| 393 | export_p.add_argument( |
| 394 | "--prefix", default="", metavar="PREFIX", |
| 395 | help="Path prefix inside the archive.", |
| 396 | ) |
| 397 | export_p.add_argument( |
| 398 | "--json", "-j", action="store_true", dest="json_out", |
| 399 | help="Emit a JSON summary (output path, file count, size) after creating the archive.", |
| 400 | ) |
| 401 | export_p.set_defaults(func=run_snapshot_export) |
| 402 | |
| 403 | # --- list --- |
| 404 | list_p = subs.add_parser( |
| 405 | "list", |
| 406 | help="List all stored snapshots, newest first.", |
| 407 | description=( |
| 408 | "Scan ``.muse/objects/sha256/`` and print all snapshot records sorted\n" |
| 409 | "newest-first. Symlinks in the object store are silently\n" |
| 410 | "skipped. Use ``--limit`` / ``-n`` to cap results (default 20).\n\n" |
| 411 | "Agent quickstart\n" |
| 412 | "----------------\n" |
| 413 | " muse snapshot list --json\n" |
| 414 | " muse snapshot list -n 5 --json\n" |
| 415 | " muse snapshot list -n 100 --json\n\n" |
| 416 | "JSON output schema\n" |
| 417 | "------------------\n" |
| 418 | " Object with top-level duration_ms/exit_code envelope:\n" |
| 419 | ' {"snapshots": [{"snapshot_id": "<hex64>", "file_count": <int>,\n' |
| 420 | ' "note": "<str>", "created_at": "<iso8601>"}, ...],\n' |
| 421 | ' "duration_ms": <float>, "exit_code": 0}\n\n' |
| 422 | "Exit codes\n" |
| 423 | "----------\n" |
| 424 | " 0 — list returned (may be empty)\n" |
| 425 | " 1 — --limit value out of range\n" |
| 426 | " 2 — not inside a Muse repository\n" |
| 427 | ), |
| 428 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 429 | ) |
| 430 | list_p.add_argument( |
| 431 | "--limit", type=int, default=20, metavar="N", |
| 432 | help="Maximum snapshots to show (default: 20, max: 100000).", |
| 433 | ) |
| 434 | list_p.add_argument( |
| 435 | "--json", "-j", action="store_true", dest="json_out", |
| 436 | help="Emit a machine-readable JSON array.", |
| 437 | ) |
| 438 | list_p.set_defaults(func=run_snapshot_list) |
| 439 | |
| 440 | # --- read --- |
| 441 | read_p = subs.add_parser( |
| 442 | "read", |
| 443 | help="Read the full manifest of a snapshot.", |
| 444 | description=( |
| 445 | "Display every file in a snapshot. Output is JSON by default —\n" |
| 446 | "the most useful format for agents. Pass ``--text`` for a\n" |
| 447 | "human-readable path listing.\n\n" |
| 448 | "Accepts a full 64-character snapshot ID or any unambiguous\n" |
| 449 | "prefix. The prefix scan skips symlinks in ``.muse/objects/sha256/``\n" |
| 450 | "(security guard against crafted symlinks).\n\n" |
| 451 | "Agent quickstart\n" |
| 452 | "----------------\n" |
| 453 | " muse snapshot read <id>\n" |
| 454 | " muse snapshot read <id12prefix>\n" |
| 455 | " SNAP=$(muse snapshot create --json | jq -r .snapshot_id)\n" |
| 456 | " muse snapshot read \"$SNAP\"\n\n" |
| 457 | "JSON output schema (default — no flag needed)\n" |
| 458 | "----------------------------------------------\n" |
| 459 | ' {"snapshot_id": "<hex64>", "created_at": "<iso8601>",\n' |
| 460 | ' "file_count": <int>, "note": "<str>",\n' |
| 461 | ' "manifest": {"<path>": "<object_id>", ...},\n' |
| 462 | ' "duration_ms": <float>, "exit_code": 0}\n\n' |
| 463 | "Exit codes\n" |
| 464 | "----------\n" |
| 465 | " 0 — snapshot found and printed\n" |
| 466 | " 1 — snapshot ID not found\n" |
| 467 | " 2 — not inside a Muse repository\n" |
| 468 | ), |
| 469 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 470 | ) |
| 471 | read_p.add_argument("snapshot_id", metavar="ID", help="Snapshot ID (full or prefix).") |
| 472 | read_p.add_argument( |
| 473 | "--json", "-j", action="store_true", dest="json_out", |
| 474 | help="Emit machine-readable JSON.", |
| 475 | ) |
| 476 | read_p.set_defaults(func=run_snapshot_read) |
| 477 | |
| 478 | # --------------------------------------------------------------------------- |
| 479 | # Subcommand handlers |
| 480 | # --------------------------------------------------------------------------- |
| 481 | |
| 482 | def run_snapshot_create(args: argparse.Namespace) -> None: |
| 483 | """Capture the current working tree as a snapshot without committing. |
| 484 | |
| 485 | Hashes every tracked file, stores their content in the object store, and |
| 486 | writes a :class:`SnapshotRecord` to ``.muse/objects/sha256/``. No commit is |
| 487 | created — the snapshot is a standalone checkpoint that can be listed, |
| 488 | inspected, and exported independently. |
| 489 | |
| 490 | Agent quickstart:: |
| 491 | |
| 492 | muse snapshot create --json |
| 493 | muse snapshot create -m "before refactor" --json |
| 494 | muse snapshot create -m "checkpoint" --json |
| 495 | |
| 496 | JSON fields:: |
| 497 | |
| 498 | repo_id str Repository identifier |
| 499 | snapshot_id str Content-addressed snapshot ID (sha256: prefix) |
| 500 | file_count int Number of files captured |
| 501 | note str The --note value (empty string if not supplied) |
| 502 | created_at str ISO 8601 creation timestamp |
| 503 | |
| 504 | Exit codes:: |
| 505 | |
| 506 | 0 Success. |
| 507 | 2 Not inside a Muse repository. |
| 508 | 3 I/O error writing to the object store. |
| 509 | """ |
| 510 | note: str = args.note |
| 511 | json_out: bool = args.json_out |
| 512 | |
| 513 | elapsed = start_timer() |
| 514 | |
| 515 | root = require_repo() |
| 516 | plugin = resolve_plugin(root) |
| 517 | |
| 518 | snap_result = plugin.snapshot(root) |
| 519 | manifest: Manifest = snap_result["files"] |
| 520 | snap_dirs = list(snap_result.get("directories") or []) |
| 521 | |
| 522 | for rel_path, object_id in manifest.items(): |
| 523 | src = root / rel_path |
| 524 | if src.exists(): |
| 525 | try: |
| 526 | write_object_from_path(root, object_id, src) |
| 527 | except (ValueError, OSError) as exc: |
| 528 | logger.warning("⚠️ Could not store %s: %s", rel_path, exc) |
| 529 | |
| 530 | if not snap_dirs: |
| 531 | snap_dirs = directories_from_manifest(manifest) |
| 532 | snapshot_id = hash_snapshot(manifest, snap_dirs) |
| 533 | record = SnapshotRecord(snapshot_id=snapshot_id, manifest=manifest, directories=snap_dirs, note=note) |
| 534 | write_snapshot(root, record) |
| 535 | |
| 536 | if json_out: |
| 537 | repo_id = read_repo_id(root) or "" |
| 538 | payload = _SnapshotCreateJson( |
| 539 | **make_envelope(elapsed), |
| 540 | repo_id=repo_id, |
| 541 | snapshot_id=snapshot_id, |
| 542 | file_count=len(manifest), |
| 543 | note=note, |
| 544 | created_at=record.created_at.isoformat(), |
| 545 | ) |
| 546 | print(json.dumps(payload)) |
| 547 | else: |
| 548 | print(f"Snapshot {snapshot_id} ({len(manifest)} file(s))") |
| 549 | if note: |
| 550 | print(f"Note: {sanitize_display(note)}") |
| 551 | |
| 552 | def run_snapshot_list(args: argparse.Namespace) -> None: |
| 553 | """List all stored snapshots, newest first. |
| 554 | |
| 555 | Scans ``.muse/objects/sha256/`` and returns snapshot records sorted by creation |
| 556 | timestamp descending. Symlinks inside the object store are |
| 557 | silently skipped (security guard against crafted symlinks). |
| 558 | |
| 559 | Agent quickstart:: |
| 560 | |
| 561 | muse snapshot list --json |
| 562 | muse snapshot list -n 5 --json |
| 563 | muse snapshot list -n 100 --json |
| 564 | |
| 565 | JSON fields:: |
| 566 | |
| 567 | snapshots list Snapshot entries (may be empty) |
| 568 | snapshots[].snapshot_id str Content-addressed snapshot ID |
| 569 | snapshots[].file_count int Files in the snapshot |
| 570 | snapshots[].note str Label stored at create time |
| 571 | snapshots[].created_at str ISO 8601 creation timestamp |
| 572 | |
| 573 | Exit codes:: |
| 574 | |
| 575 | 0 Success (empty list is valid). |
| 576 | 1 --limit value out of range. |
| 577 | 2 Not inside a Muse repository. |
| 578 | """ |
| 579 | try: |
| 580 | limit: int = clamp_int(args.limit, 1, 100000, "limit") |
| 581 | except ValueError as exc: |
| 582 | print(f"❌ {exc}", file=sys.stderr) |
| 583 | raise SystemExit(ExitCode.USER_ERROR) |
| 584 | json_out: bool = args.json_out |
| 585 | |
| 586 | elapsed = start_timer() |
| 587 | |
| 588 | root = require_repo() |
| 589 | snapshots = _list_all_snapshots(root) |
| 590 | |
| 591 | if limit: |
| 592 | snapshots = snapshots[:limit] |
| 593 | |
| 594 | if not snapshots: |
| 595 | if json_out: |
| 596 | print(json.dumps(_SnapshotListJson( |
| 597 | **make_envelope(elapsed), |
| 598 | snapshots=[], |
| 599 | ))) |
| 600 | else: |
| 601 | print("No snapshots found.") |
| 602 | return |
| 603 | |
| 604 | if json_out: |
| 605 | items: list[_SnapshotListItemJson] = [ |
| 606 | _SnapshotListItemJson( |
| 607 | snapshot_id=s.snapshot_id, |
| 608 | file_count=len(s.manifest), |
| 609 | note=s.note, |
| 610 | created_at=s.created_at.isoformat(), |
| 611 | ) |
| 612 | for s in snapshots |
| 613 | ] |
| 614 | print(json.dumps(_SnapshotListJson( |
| 615 | **make_envelope(elapsed), |
| 616 | snapshots=items, |
| 617 | ))) |
| 618 | else: |
| 619 | for s in snapshots: |
| 620 | when = s.created_at.strftime("%Y-%m-%d %H:%M:%S UTC") |
| 621 | note_suffix = f" {sanitize_display(s.note)}" if s.note else "" |
| 622 | print(f"{s.snapshot_id} {when} {len(s.manifest)} file(s){note_suffix}") |
| 623 | |
| 624 | def run_snapshot_read(args: argparse.Namespace) -> None: |
| 625 | """Read and print the full manifest of a snapshot. |
| 626 | |
| 627 | Defaults to JSON output — the most useful form for agents. Pass |
| 628 | ``--text`` for a human-readable path listing. Accepts a full ID or |
| 629 | any unambiguous sha256: prefix. |
| 630 | |
| 631 | Agent quickstart:: |
| 632 | |
| 633 | muse snapshot read sha256:abc123 --json |
| 634 | muse snapshot read sha256:abc123 --text |
| 635 | |
| 636 | JSON fields:: |
| 637 | |
| 638 | snapshot_id str Content-addressed snapshot ID |
| 639 | created_at str ISO 8601 creation timestamp |
| 640 | file_count int Number of files in the manifest |
| 641 | note str Label stored at create time |
| 642 | manifest dict path → object_id for all tracked files |
| 643 | |
| 644 | Exit codes:: |
| 645 | |
| 646 | 0 Success. |
| 647 | 1 Snapshot ID not found. |
| 648 | 2 Not inside a Muse repository. |
| 649 | """ |
| 650 | snapshot_id: str = args.snapshot_id |
| 651 | json_out: bool = args.json_out |
| 652 | |
| 653 | elapsed = start_timer() |
| 654 | |
| 655 | _reject_if_bare_hex(snapshot_id) |
| 656 | root = require_repo() |
| 657 | snap = _resolve_snapshot(root, snapshot_id) |
| 658 | |
| 659 | if snap is None: |
| 660 | print( |
| 661 | f"❌ Snapshot '{sanitize_display(snapshot_id)}' not found.", |
| 662 | file=sys.stderr, |
| 663 | ) |
| 664 | raise SystemExit(ExitCode.USER_ERROR) |
| 665 | |
| 666 | if json_out: |
| 667 | payload = _SnapshotReadJson( |
| 668 | **make_envelope(elapsed), |
| 669 | snapshot_id=snap.snapshot_id, |
| 670 | created_at=snap.created_at.isoformat(), |
| 671 | file_count=len(snap.manifest), |
| 672 | note=snap.note, |
| 673 | manifest=dict(sorted(snap.manifest.items())), |
| 674 | ) |
| 675 | print(json.dumps(payload)) |
| 676 | else: |
| 677 | print(f"snapshot_id: {snap.snapshot_id}") |
| 678 | print(f"created_at: {snap.created_at.isoformat()}") |
| 679 | if snap.note: |
| 680 | print(f"note: {sanitize_display(snap.note)}") |
| 681 | print(f"files ({len(snap.manifest)}):") |
| 682 | for rel_path, obj_id in sorted(snap.manifest.items()): |
| 683 | print(f" {obj_id} {sanitize_display(rel_path)}") |
| 684 | |
| 685 | def run_snapshot_export(args: argparse.Namespace) -> None: |
| 686 | """Export a snapshot as a portable tar.gz or zip archive. |
| 687 | |
| 688 | The archive contains only tracked files — no ``.muse/`` metadata. A |
| 689 | zip-slip guard rejects manifest entries whose paths escape the archive root. |
| 690 | Missing objects are skipped with a warning; the archive is still created. |
| 691 | |
| 692 | Agent quickstart:: |
| 693 | |
| 694 | muse snapshot export sha256:abc123 --json |
| 695 | muse snapshot export sha256:abc123 -f zip -o release.zip --json |
| 696 | muse snapshot export sha256:abc123 --prefix myproject/ --json |
| 697 | |
| 698 | JSON fields:: |
| 699 | |
| 700 | snapshot_id str Full content-addressed snapshot ID |
| 701 | output str Path of the archive file written |
| 702 | format str "tar.gz" or "zip" |
| 703 | file_count int Files written (unsafe/missing entries excluded) |
| 704 | size_bytes int Archive file size in bytes |
| 705 | |
| 706 | Exit codes:: |
| 707 | |
| 708 | 0 Success. |
| 709 | 1 Snapshot ID not found. |
| 710 | 2 Not inside a Muse repository. |
| 711 | 3 I/O error writing the archive. |
| 712 | """ |
| 713 | snapshot_id: str = args.snapshot_id |
| 714 | fmt: str = args.fmt |
| 715 | output: str | None = args.output |
| 716 | prefix: str = args.prefix |
| 717 | json_out: bool = args.json_out |
| 718 | |
| 719 | elapsed = start_timer() |
| 720 | |
| 721 | _reject_if_bare_hex(snapshot_id) |
| 722 | root = require_repo() |
| 723 | snap = _resolve_snapshot(root, snapshot_id) |
| 724 | |
| 725 | if snap is None: |
| 726 | print( |
| 727 | f"❌ Snapshot '{sanitize_display(snapshot_id)}' not found.", |
| 728 | file=sys.stderr, |
| 729 | ) |
| 730 | raise SystemExit(ExitCode.USER_ERROR) |
| 731 | |
| 732 | _, snap_hex = split_id(snap.snapshot_id) |
| 733 | short = short_id(snap.snapshot_id) |
| 734 | out_name = output or f"{snap_hex}.{fmt}" |
| 735 | out_path = pathlib.Path(out_name) |
| 736 | |
| 737 | if fmt == "tar.gz": |
| 738 | count = _build_tar(root, snap.manifest, out_path, prefix) |
| 739 | else: |
| 740 | count = _build_zip(root, snap.manifest, out_path, prefix) |
| 741 | |
| 742 | size_bytes = out_path.stat().st_size if out_path.exists() else 0 |
| 743 | |
| 744 | if json_out: |
| 745 | print(json.dumps(_SnapshotExportJson( |
| 746 | **make_envelope(elapsed), |
| 747 | snapshot_id=snap.snapshot_id, |
| 748 | output=str(out_path), |
| 749 | format=fmt, |
| 750 | file_count=count, |
| 751 | size_bytes=size_bytes, |
| 752 | ))) |
| 753 | else: |
| 754 | size_kb = size_bytes / 1024 |
| 755 | print( |
| 756 | f"✅ Archive: {sanitize_display(str(out_path))} ({count} file(s), {size_kb:.1f} KiB)\n" |
| 757 | f" Snapshot: {short}" |
| 758 | ) |
File History
1 commit
sha256:502e812e12319e30302d07d165425ceeb21687ad30e8af3a439b1f6de36c64d6
fix: update stale tag tests to use label for free-form anno…
Sonnet 4.6
19 days ago