"""``muse bundle`` — pack and unpack commits for single-file transport. A bundle is a self-contained msgpack binary file carrying commits, snapshots, and objects. It is the porcelain equivalent of ``muse plumbing pack-objects`` / ``unpack-objects``, with friendlier names and the key value-add of auto-updating local branch refs after ``unbundle``. Use bundles to transfer a repository slice between machines without a network connection — copy the file over SSH, USB, or email. Bundle format: binary msgpack encoding of the ``PackBundle`` TypedDict. Objects are stored as raw bytes (no base64 overhead). Subcommands:: muse bundle create [...] [--have ...] [--json] muse bundle unbundle [--no-update-refs] [--json] muse bundle verify [-q] [--json] muse bundle list-heads [--json] Security model:: Symlinks inside ``.muse/refs/heads/`` are silently skipped during branch enumeration. A crafted symlink could otherwise expose external path content. Ref files are read at most 65 bytes; oversized files are treated as invalid refs rather than read entirely into memory. Bundle file parsing is limited by ``MAX_PACK_MSGPACK_BYTES``; both the file size and the msgpack payload are checked before any deserialization occurs. JSON output schemas:: bundle create --json: {"file": str, "commits": int, "objects": int, "size_bytes": int, "branches": [str, ...]} bundle unbundle --json: {"commits_written": int, "snapshots_written": int, "objects_written": int, "objects_skipped": int, "refs_updated": [str, ...]} bundle verify --json: {"objects_checked": int, "snapshots_checked": int, "all_ok": bool, "failures": [str, ...]} bundle list-heads --json: {"": "", ...} Exit codes:: 0 — success 1 — bundle not found, integrity failure, bad arguments 3 — I/O error """ from __future__ import annotations import argparse import hashlib import json import logging import os import pathlib import sys import tempfile from collections import deque from typing import TypedDict, TypeGuard import msgpack from muse.core.errors import ExitCode from muse.core.object_store import has_object, write_object from muse.core.pack import PackBundle, apply_pack, build_pack from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( MAX_PACK_MSGPACK_BYTES, BranchHeads, CommitDict, CommitRecord, MsgpackValue, SnapshotDict, SnapshotRecord, get_head_commit_id, read_commit, read_current_branch, resolve_commit_ref, safe_unpackb, write_branch_ref, write_commit, write_snapshot, ) from muse.core.validation import sanitize_display, validate_branch_name logger = logging.getLogger(__name__) # Maximum bytes read from a branch ref file. A valid SHA-256 hex string is # 64 chars + optional newline = 65 bytes. Anything larger is treated as # corrupt. _MAX_REF_BYTES = 65 # Maximum number of distinct failures collected during ``muse bundle verify``. # Caps memory and output size when a crafted bundle has many corrupted objects. _MAX_VERIFY_FAILURES = 100 # --------------------------------------------------------------------------- # JSON wire-format TypedDicts # --------------------------------------------------------------------------- class _BundleCreateJson(TypedDict): """JSON output for ``muse bundle create --json``.""" file: str commits: int objects: int size_bytes: int branches: list[str] class _BundleUnbundleJson(TypedDict): """JSON output for ``muse bundle unbundle --json``.""" commits_written: int snapshots_written: int objects_written: int objects_skipped: int refs_updated: list[str] class _BundleVerifyJson(TypedDict): """JSON output for ``muse bundle verify --json``.""" objects_checked: int snapshots_checked: int all_ok: bool failures: list[str] # --------------------------------------------------------------------------- # TypeGuards for wire-boundary narrowing # --------------------------------------------------------------------------- def _is_commit_dict(v: MsgpackValue) -> TypeGuard[CommitDict]: """TypeGuard for narrowing MsgpackValue → CommitDict at the wire boundary.""" return isinstance(v, dict) def _is_snapshot_dict(v: MsgpackValue) -> TypeGuard[SnapshotDict]: """TypeGuard for narrowing MsgpackValue → SnapshotDict at the wire boundary.""" return isinstance(v, dict) # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _resolve_refs( root: pathlib.Path, repo_id: str, branch: str, refs: list[str], ) -> list[str]: """Resolve a list of ref strings to commit IDs. Expands ``HEAD``.""" ids: list[str] = [] for ref in refs: if ref.upper() == "HEAD": cid = get_head_commit_id(root, branch) if cid: ids.append(cid) else: rec = resolve_commit_ref(root, repo_id, branch, ref) if rec: ids.append(rec.commit_id) else: print(f"❌ Ref '{sanitize_display(ref)}' not found.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) return ids def _load_bundle(file_path: pathlib.Path) -> PackBundle: """Read a msgpack bundle file and return a :class:`~muse.core.pack.PackBundle`. Security: The file size is checked before reading. The raw bytes are then passed through :func:`~muse.core.store.safe_unpackb` which enforces an upper bound on the deserialized payload as well. This double-check prevents OOM from crafted size-bomb bundles. Exceptions are narrowed to :class:`(OSError, ValueError, msgpack.UnpackException)` — the precise types raised by the I/O and parse path — so genuine programming errors are not silently swallowed. """ try: size = file_path.stat().st_size if size > MAX_PACK_MSGPACK_BYTES: raise OSError( f"Bundle file is {size:,} bytes — exceeds the " f"{MAX_PACK_MSGPACK_BYTES // (1024 * 1024)} MiB safety cap." ) raw_bytes = file_path.read_bytes() parsed = safe_unpackb( raw_bytes, context="bundle file", max_bytes=MAX_PACK_MSGPACK_BYTES, allow_binary=True, ) except FileNotFoundError: print( f"❌ Bundle file not found: {sanitize_display(str(file_path))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) except (OSError, ValueError, msgpack.UnpackException) as exc: print(f"❌ Bundle is not valid msgpack: {sanitize_display(str(exc))}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not isinstance(parsed, dict): print("❌ Bundle has unexpected structure.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from muse.core.pack import ObjectPayload bundle: PackBundle = {} raw_commits = parsed.get("commits") if isinstance(raw_commits, list): bundle["commits"] = [r for r in raw_commits if _is_commit_dict(r)] raw_snapshots = parsed.get("snapshots") if isinstance(raw_snapshots, list): bundle["snapshots"] = [r for r in raw_snapshots if _is_snapshot_dict(r)] if "objects" in parsed and isinstance(parsed["objects"], list): objects: list[ObjectPayload] = [] for item in parsed["objects"]: if isinstance(item, dict): oid = item.get("object_id") content = item.get("content") if isinstance(oid, str) and isinstance(content, (bytes, bytearray)): objects.append(ObjectPayload(object_id=oid, content=bytes(content))) bundle["objects"] = objects if "branch_heads" in parsed and isinstance(parsed["branch_heads"], dict): bundle["branch_heads"] = { k: v for k, v in parsed["branch_heads"].items() if isinstance(k, str) and isinstance(v, str) } return bundle def _iter_branches(root: pathlib.Path) -> list[tuple[str, str]]: """Return ``[(branch_name, commit_id)]`` for all branch ref files. Security: Symlinks inside ``.muse/refs/heads/`` are silently skipped. Ref files are capped at ``_MAX_REF_BYTES`` (65 bytes) before decoding — oversized content is decoded but will fail hex validation downstream. """ heads_dir = root / ".muse" / "refs" / "heads" if not heads_dir.exists(): return [] result: list[tuple[str, str]] = [] for ref_file in sorted(heads_dir.rglob("*")): if ref_file.is_symlink(): logger.warning("⚠️ bundle: skipping symlink ref: %s", ref_file) continue if not ref_file.is_file(): continue raw_bytes = ref_file.read_bytes()[:_MAX_REF_BYTES] cid = raw_bytes.decode("utf-8", errors="replace").strip() if cid: branch_name = str(ref_file.relative_to(heads_dir).as_posix()) result.append((branch_name, cid)) return result def _reachable_from(root: pathlib.Path, tip_ids: list[str]) -> set[str]: """BFS-walk from *tip_ids* and return all reachable commit IDs.""" seen: set[str] = set() q: deque[str] = deque(tip_ids) while q: cid = q.popleft() if cid in seen: continue seen.add(cid) c = read_commit(root, cid) if c: if c.parent_commit_id: q.append(c.parent_commit_id) if c.parent2_commit_id: q.append(c.parent2_commit_id) return seen # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the bundle subcommand.""" parser = subparsers.add_parser( "bundle", help="Pack and unpack commits into a single portable bundle file.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) subs = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND") subs.required = True # create create_p = subs.add_parser( "create", help="Create a bundle file containing commits reachable from .", description=( "Pack commits, snapshots, and objects reachable from into a\n" "single self-contained msgpack binary file. Transfer it over SSH,\n" "USB, or email and apply it with ``muse bundle unbundle``.\n\n" "Use --have to prune commits the receiver already has, producing a\n" "smaller incremental bundle.\n\n" "Agent quickstart\n" "----------------\n" " muse bundle create repo.bundle --json\n" " muse bundle create out.bundle feat/audio --json\n" " muse bundle create out.bundle HEAD --have --json\n\n" "JSON output schema\n" "------------------\n" ' {"file": "", "commits": , "objects": ,\n' ' "size_bytes": , "branches": ["", ...]}\n\n' "Exit codes\n" "----------\n" " 0 — bundle created successfully\n" " 1 — unknown ref, no commits to bundle, or bad --have ref\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) create_p.add_argument("file", help="Output bundle file path.") create_p.add_argument( "refs", nargs="*", default=None, help="Refs to include (default: HEAD).", ) create_p.add_argument( "--have", "-H", nargs="*", default=None, dest="have", help="Commits the receiver already has (exclude from bundle).", ) create_p.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit a machine-readable JSON summary.", ) create_p.set_defaults(func=run_create) # list-heads list_heads_p = subs.add_parser( "list-heads", help="List the branch heads recorded in a bundle file.", description=( "Print the branch → commit_id map embedded in a bundle file.\n" "Does NOT require a Muse repository — runs against any bundle\n" "file in isolation.\n\n" "Agent quickstart\n" "----------------\n" " muse bundle list-heads repo.bundle --json\n" " muse bundle list-heads repo.bundle --json | jq 'keys'\n\n" "JSON output schema\n" "------------------\n" ' {"": "", ...}\n\n' "Exit codes\n" "----------\n" " 0 — always (empty object when no heads are recorded)\n" " 1 — bundle file not found or not valid msgpack\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) list_heads_p.add_argument("file", help="Bundle file to inspect.") list_heads_p.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit a machine-readable JSON map of branch → commit_id.", ) list_heads_p.set_defaults(func=run_list_heads) # unbundle unbundle_p = subs.add_parser( "unbundle", help="Apply a bundle to the local store and optionally advance branch refs.", description=( "Write all commits, snapshots, and objects from a bundle file into\n" "the local repository, then advance local branch refs to match the\n" "bundle's recorded heads. All writes are idempotent — objects\n" "already present are counted as 'skipped', not errors.\n\n" "Use --no-update-refs to import objects without moving any branches\n" "(useful for inspection or partial imports).\n\n" "Agent quickstart\n" "----------------\n" " muse bundle unbundle repo.bundle --json\n" " muse bundle unbundle repo.bundle --no-update-refs --json\n\n" "JSON output schema\n" "------------------\n" ' {"commits_written": , "snapshots_written": ,\n' ' "objects_written": , "objects_skipped": ,\n' ' "refs_updated": ["", ...]}\n\n' "Exit codes\n" "----------\n" " 0 — bundle applied (even if all objects were already present)\n" " 1 — bundle file not found or not valid msgpack\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) unbundle_p.add_argument("file", help="Bundle file to apply.") unbundle_p.add_argument( "--no-update-refs", action="store_false", dest="update_refs", help="Do not update local branch refs from the bundle's branch_heads.", ) unbundle_p.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit a machine-readable JSON summary.", ) unbundle_p.set_defaults(func=run_unbundle, update_refs=True) # verify verify_p = subs.add_parser( "verify", help="Verify the integrity of a bundle file.", description=( "Check every object's SHA-256 against its declared object_id\n" "(detects corruption), and confirm every snapshot's objects are\n" "present in the bundle (detects truncation). Does NOT require a\n" "Muse repository — runs against any bundle file in isolation.\n\n" "Use --quiet for scripting: no output, just the exit code.\n\n" "Agent quickstart\n" "----------------\n" " muse bundle verify repo.bundle --json\n" " muse bundle verify repo.bundle -q && muse bundle unbundle repo.bundle\n\n" "JSON output schema\n" "------------------\n" ' {"objects_checked": , "snapshots_checked": ,\n' ' "all_ok": true|false, "failures": ["", ...]}\n\n' "Exit codes\n" "----------\n" " 0 — bundle is clean\n" " 1 — integrity failures, or bundle file not found / not valid\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) verify_p.add_argument("file", help="Bundle file to verify.") verify_p.add_argument( "--quiet", "-q", action="store_true", dest="quiet", help="No output — exit 0 if clean, 1 on failure.", ) verify_p.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit a machine-readable JSON report.", ) verify_p.set_defaults(func=run_verify) # --------------------------------------------------------------------------- # Subcommand handlers # --------------------------------------------------------------------------- def run_create(args: argparse.Namespace) -> None: """Create a bundle file containing commits reachable from . A bundle is a self-contained **msgpack binary** file carrying commits, snapshots, and objects. Copy it over SSH, USB, or email, then apply it on the receiving end with ``muse bundle unbundle``. ``--have`` prunes commits the receiver already has, reducing bundle size. Pass the tip commit ID(s) of the receiver's matching branch. The output file is written atomically — a temporary file in the same directory is renamed over the target, so a kill signal during write never leaves a partial bundle at the output path. JSON schema (``--json``):: { "file": "", "commits": , "objects": , "size_bytes": , "branches": ["", ...] } Exit codes: 0 — bundle created successfully 1 — unknown ref, no commits to bundle, or bad --have ref 2 — not inside a Muse repository Examples:: muse bundle create repo.bundle # HEAD → bundle muse bundle create out.bundle feat/audio # specific branch muse bundle create out.bundle HEAD --have old-sha muse bundle create repo.bundle --json """ file: str = args.file refs: list[str] | None = args.refs have: list[str] | None = args.have json_out: bool = args.json_out root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) want_refs: list[str] = refs or ["HEAD"] commit_ids = _resolve_refs(root, repo_id, branch, want_refs) if not commit_ids: print("❌ No commits to bundle.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) have_ids: list[str] = have or [] bundle = build_pack(root, commit_ids, have=have_ids) # Pre-compute the reachable set once (O(commits)) so the branch-head # filter below is O(branches) rather than O(branches × commits). reachable: set[str] = _reachable_from(root, commit_ids) heads: BranchHeads = {} for br_name, cid in _iter_branches(root): if cid in commit_ids or cid in reachable: heads[br_name] = cid if heads: bundle["branch_heads"] = heads out_path = pathlib.Path(file) packed = msgpack.packb(bundle, use_bin_type=True) # Atomic write: temp file in same directory → os.replace(). # Ensures a kill signal never leaves a partial bundle at out_path. tmp_dir = out_path.parent if out_path.parent != pathlib.Path("") else pathlib.Path(".") tmp_fd, tmp_str = tempfile.mkstemp(dir=tmp_dir, prefix=".bundle-tmp-") tmp_path_obj = pathlib.Path(tmp_str) try: with os.fdopen(tmp_fd, "wb") as fh: fh.write(packed) os.replace(tmp_path_obj, out_path) except Exception: tmp_path_obj.unlink(missing_ok=True) raise n_commits = len(bundle.get("commits", [])) n_objects = len(bundle.get("objects", [])) size_bytes = out_path.stat().st_size if json_out: payload: _BundleCreateJson = { "file": str(out_path), "commits": n_commits, "objects": n_objects, "size_bytes": size_bytes, "branches": sorted(heads.keys()), } print(json.dumps(payload)) else: size_kb = size_bytes / 1024 print( f"✅ Bundle: {sanitize_display(str(out_path))} " f"({n_commits} commits, {n_objects} objects, {size_kb:.1f} KiB)" ) def run_unbundle(args: argparse.Namespace) -> None: """Apply a bundle to the local store and optionally advance branch refs. This is the key porcelain value-add over ``muse plumbing unpack-objects``: after unpacking, branch refs are updated from ``branch_heads`` in the bundle so the local repo reflects the sender's branch state. Branch names from the bundle are validated with :func:`~muse.core.validation.validate_branch_name` and commit IDs must be exactly 64 lowercase hex characters — invalid entries are skipped with a warning rather than causing the entire operation to fail. JSON schema (``--json``):: { "commits_written": , "snapshots_written": , "objects_written": , "objects_skipped": , "refs_updated": ["", ...] } Exit codes: 0 — bundle applied (idempotent: already-present objects are skipped) 1 — bundle file not found or not valid msgpack 2 — not inside a Muse repository Examples:: muse bundle unbundle repo.bundle muse bundle unbundle repo.bundle --no-update-refs muse bundle unbundle repo.bundle --json """ file: str = args.file update_refs: bool = args.update_refs json_out: bool = args.json_out root = require_repo() bundle = _load_bundle(pathlib.Path(file)) result = apply_pack(root, bundle) updated: list[str] = [] if update_refs: branch_heads: BranchHeads = bundle.get("branch_heads") or {} for br, cid in branch_heads.items(): try: validate_branch_name(br) except ValueError: logger.warning("⚠️ bundle: skipping invalid branch name %r", br) continue if len(cid) != 64 or not all(c in "0123456789abcdef" for c in cid): logger.warning("⚠️ bundle: skipping invalid commit ID for %r", br) continue write_branch_ref(root, br, cid) updated.append(br) if json_out: unbundle_payload: _BundleUnbundleJson = { "commits_written": result["commits_written"], "snapshots_written": result["snapshots_written"], "objects_written": result["objects_written"], "objects_skipped": result["objects_skipped"], "refs_updated": sorted(updated), } print(json.dumps(unbundle_payload)) else: print( f"Unpacked {result['commits_written']} commit(s), " f"{result['snapshots_written']} snapshot(s), " f"{result['objects_written']} object(s) " f"({result['objects_skipped']} skipped)" ) if updated: print( f"Updated refs: {', '.join(sanitize_display(b) for b in updated)}" ) print("✅ Bundle applied.") def run_verify(args: argparse.Namespace) -> None: """Verify the integrity of a bundle file. Checks that every object's SHA-256 matches its declared ``object_id`` (hash mismatch → corruption). Also checks that every snapshot's objects are present in the bundle (detects truncation). Does NOT require a Muse repository — runs against any bundle file in isolation. This makes it safe to verify a bundle before transferring it to a target repo. At most ``_MAX_VERIFY_FAILURES`` (100) distinct failures are recorded. If a bundle has more, a trailing "... and N more" entry is appended so the list stays bounded regardless of bundle size. JSON schema (``--json``):: { "objects_checked": , "snapshots_checked": , "all_ok": , "failures": ["", ...] } Exit codes: 0 — bundle is clean 1 — integrity failures found, or bundle file not found / not valid Examples:: muse bundle verify repo.bundle muse bundle verify repo.bundle --quiet && echo "clean" muse bundle verify repo.bundle --json | jq '.failures' """ file: str = args.file quiet: bool = args.quiet json_out: bool = args.json_out bundle = _load_bundle(pathlib.Path(file)) failures: list[str] = [] objects_checked = 0 snapshots_checked = 0 _total_failures = 0 # tracks raw count before the cap kicks in def _add_failure(msg: str) -> None: nonlocal _total_failures _total_failures += 1 if len(failures) < _MAX_VERIFY_FAILURES: failures.append(msg) # Build set of object IDs in the bundle and verify each hash. bundle_obj_ids: set[str] = set() for obj in bundle.get("objects", []): obj_id = obj["object_id"] raw = obj["content"] if not obj_id or not raw: _add_failure("objects list: entry has empty object_id or content") continue actual = hashlib.sha256(raw).hexdigest() if actual != obj_id: _add_failure(f"object {obj_id[:12]}: hash mismatch (corruption)") else: bundle_obj_ids.add(obj_id) objects_checked += 1 # Check snapshots reference objects present in the bundle. for snap_dict in bundle.get("snapshots", []): snap_id = snap_dict.get("snapshot_id", "") manifest = snap_dict.get("manifest", {}) snapshots_checked += 1 for rel_path, obj_id in manifest.items(): if obj_id not in bundle_obj_ids: _add_failure( f"snapshot {snap_id[:12]}: " f"missing object {obj_id[:12]} for {rel_path}" ) overflow = _total_failures - len(failures) if overflow > 0: failures.append(f"... and {overflow} more failure(s) (cap: {_MAX_VERIFY_FAILURES})") all_ok = _total_failures == 0 if quiet: raise SystemExit(0 if all_ok else ExitCode.USER_ERROR) if json_out: verify_payload: _BundleVerifyJson = { "objects_checked": objects_checked, "snapshots_checked": snapshots_checked, "all_ok": all_ok, "failures": [sanitize_display(f) for f in failures], } print(json.dumps(verify_payload)) else: print(f"Objects checked: {objects_checked}") print(f"Snapshots checked: {snapshots_checked}") if all_ok: print("✅ Bundle is clean.") else: print(f"❌ {_total_failures} failure(s):") for f in failures: print(f" {sanitize_display(str(f))}") raise SystemExit(0 if all_ok else ExitCode.USER_ERROR) def run_list_heads(args: argparse.Namespace) -> None: """List the branch heads recorded in a bundle file. JSON output is a plain ``{branch: commit_id}`` map — suitable for piping directly into ``jq`` or agent decision logic. Does NOT require a Muse repository — runs against any bundle file in isolation. Both branch names and commit IDs from the bundle are sanitized before output to prevent ANSI injection from crafted bundles. Exit codes: 0 — always (empty object when no heads are recorded) 1 — bundle file not found or not valid msgpack Examples:: muse bundle list-heads repo.bundle muse bundle list-heads repo.bundle --json muse bundle list-heads repo.bundle --json | jq 'keys' """ file: str = args.file json_out: bool = args.json_out bundle = _load_bundle(pathlib.Path(file)) heads: BranchHeads = bundle.get("branch_heads") or {} if json_out: safe_heads = { sanitize_display(k): sanitize_display(v) for k, v in heads.items() } print(json.dumps(safe_heads)) else: if not heads: print("No branch heads in bundle.") return for branch, cid in sorted(heads.items()): print(f"{sanitize_display(cid[:12])} {sanitize_display(branch)}")