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