"""Muse MPack format — mpack of commits, snapshots, and blobs for wire transfer. An :class:`MPack` is the unit of exchange between the Muse CLI and a remote (e.g. MuseHub). It carries everything needed to reconstruct a slice of commit history locally: - :class:`CommitDict` records (full metadata + agent provenance) - :class:`SnapshotDict` records (file manifests) - :class:`BlobPayload` entries (raw blob bytes) - ``summary`` (:class:`MPackSummary`) — advisory counts for agent routing :func:`build_mpack` collects all data reachable from a set of commit IDs and populates the summary field. :func:`apply_mpack` writes a mpack into a local ``.muse/`` directory. MPack wire encoding --------------------- An MPack is encoded in the MPack binary wire format (``b"MUSE"`` magic, section table, SHA-256 footer) and transmitted with ``Content-Type: application/x-muse-pack``. Agent contract -------------- - ``exit_code`` 0: all data applied successfully. - ``exit_code`` 1: validation error (malformed object ID, path traversal). - ``exit_code`` 3: I/O error reading from local store. - ``duration_ms``: wall-clock milliseconds for build or apply. """ import collections import datetime import hashlib as _hashlib import logging import os import pathlib import struct as _struct from typing import TypedDict from muse.core.graph import iter_ancestors from muse.core.object_availability import ObjectState, load_promisor_remotes, object_state from muse.core.object_store import has_object, read_object from muse.core.pack_store import write_pack from muse.core.ids import hash_snapshot from muse.core.validation import ( MAX_OBJECT_WRITE_BYTES, MAX_PACK_OBJECTS, validate_object_id, validate_workspace_path, ) from muse.core.types import BranchHeads, blob_id, short_id from muse.core.commits import ( CommitDict, CommitRecord, MissingParentError, read_commit, write_commit, ) from muse.core.snapshots import ( SnapshotDict, SnapshotRecord, read_snapshot, write_snapshot, ) from muse.core.tags import ( TagDict, TagRecord, get_all_tags, get_tags_for_commit, write_tag, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Type aliases — avoid bare dict[str, X] at boundaries # --------------------------------------------------------------------------- _Manifest = dict[str, str] # path → object_id _JsonValue = str | int | float | bool | None _MetaDict = dict[str, _JsonValue] # loose metadata (dynamic keys, JSON-serialisable) _SnapshotResolvedMap = dict[str, tuple[_Manifest, list[str]]] # sid → (manifest, dirs) # --------------------------------------------------------------------------- # Wire-format TypedDicts # --------------------------------------------------------------------------- class _BlobPayloadBase(TypedDict): """Required fields for every blob payload in an MPack.""" object_id: str content: bytes class BlobPayload(_BlobPayloadBase, total=False): """A single content-addressed blob with encoding metadata for mpack transfer. Required fields (always present): object_id: Content-addressed SHA-256 identifier (``sha256:``). content: Raw or encoded bytes — see *encoding*. Optional fields (omit for ``"raw"`` with no base): path: Repository path of this blob; used by the server to look up delta base candidates for the next push. encoding: ``"raw"`` (default) | ``"zlib"`` | ``"delta+zlib"``. base_id: Base blob ID for ``"delta+zlib"`` encoding. sz: Uncompressed byte count of the target blob. Required when ``encoding`` is ``"delta+zlib"`` so the server can pre-allocate before decompression. Ignored for ``"raw"`` payloads. """ path: str encoding: str base_id: str sz: int class WireTag(TypedDict): """A tag record serialised for wire transfer inside an :class:`MPack`.""" tag_id: str repo_id: str commit_id: str tag: str created_at: str class MPackMeta(TypedDict, total=False): """Self-describing metadata embedded in every :class:`MPack`. Agents read this to understand the mpack's scope without inspecting commits or objects. Fields: mode: ``"full"`` — all referenced objects must be in the mpack or the local store. ``"incremental"`` — some objects are expected to exist at the receiver's base (declared in ``base_commits``); they are not included in this mpack. base_commits: Commit IDs passed as ``--have`` when the mpack was built. Empty for full bundles. created_at: ISO 8601 UTC timestamp of when the mpack was assembled. """ mode: str # "full" | "incremental" base_commits: list[str] # sha256:-prefixed commit IDs created_at: str # ISO 8601 — e.g. "2026-01-01T00:00:00Z" class MPackSummary(TypedDict, total=False): """Advisory summary embedded in every :class:`MPack`. Agents read this to make routing/accept/reject decisions before touching commits or objects. All fields are advisory — receivers must not rely on them for correctness, only for optimisation. Fields: commits_count: Number of commits in this mpack. blobs_count: Number of unique blobs in this mpack. blobs_bytes: Total uncompressed blob bytes. branches: Branch name → tip commit_id at build time. agent_ids: All agent_id values from commits in this mpack. """ commits_count: int blobs_count: int blobs_bytes: int branches: BranchHeads # branch_name → commit_id agent_ids: list[str] # distinct agent_ids in commits class SnapshotDeltaDict(TypedDict, total=False): """Wire representation of a snapshot as a delta from its parent snapshot. Guiding principle: content-addressing is a proof, not a label. ``snapshot_id = sha256(sorted path-NUL-oid pairs)``. A receiver who holds ``snapshot_id`` and the delta can reconstruct the full manifest and verify it by hashing — no external store needed. Fields: snapshot_id: sha256 of the *full* manifest (the proof). parent_snapshot_id: snapshot_id of the parent, or ``None`` for root. delta_upsert: Paths added or changed relative to parent. delta_remove: Paths removed relative to parent. Reconstruction:: manifest = dict(resolved[parent_snapshot_id]) # or {} if None manifest.update(delta_upsert) for path in delta_remove: del manifest[path] assert hash_snapshot(manifest) == snapshot_id # the math IS the proof """ snapshot_id: str parent_snapshot_id: str | None delta_upsert: dict[str, str] # path → object_id delta_remove: list[str] # paths removed class MPack(TypedDict, total=False): """The unit of exchange between the Muse CLI and a remote. All fields are optional so that partial bundles (fetch-only, objects-only) are valid wire messages. Callers check for presence before consuming. The ``summary`` field carries advisory metadata for agent routing — agents can make decisions from it without deserialising commits or objects. The ``meta`` field declares the mpack's scope (full vs incremental) and base commits, allowing receivers to verify it correctly without out-of-band knowledge of how it was built. Snapshots are stored as :class:`SnapshotDeltaDict` entries in commit-graph order (oldest first). The first entry has ``parent_snapshot_id=None`` and ``delta_upsert`` equal to the full manifest. Every subsequent entry carries only the paths that changed. Receivers reconstruct full manifests by applying the delta chain and verify correctness by hashing the result. """ commits: list[CommitDict] snapshots: list[SnapshotDeltaDict] blobs: list[BlobPayload] #: Tags attached to any commit included in this mpack. tags: list[WireTag] #: Advisory summary — populated by :func:`build_mpack`. summary: MPackSummary #: Self-describing metadata — always written by :func:`build_mpack`. meta: MPackMeta class RemoteInfo(TypedDict, total=False): """Repository metadata returned by ``GET {url}/refs``.""" repo_id: str # always present domain: str # always present #: Maps branch name → commit ID for every branch on the remote. branch_heads: BranchHeads # always present default_branch: str # always present class PushResult(TypedDict): """Server response after a push attempt.""" ok: bool message: str #: Updated branch heads on the remote after the push (if successful). branch_heads: BranchHeads class FetchRequest(TypedDict, total=False): """Body of ``POST {url}/fetch`` — negotiates which commits to transfer. ``want`` lists commit IDs the client wants to receive. ``have`` lists commit IDs already present locally, allowing the server to send only the commits the client lacks (delta negotiation). """ want: list[str] have: list[str] class ApplyResult(TypedDict): """Counts returned by :func:`apply_mpack` describing what was written. ``blobs_skipped`` counts blobs already present in the store (not rewritten, idempotent). All other counts reflect *new* writes only. ``tags_written`` counts tag records written from the mpack's ``tags`` section (0 for bundles created without tag data). ``failed_blobs`` blob IDs that failed integrity or write checks. ``skipped_snapshots`` snapshot IDs skipped because a referenced blob failed. """ commits_written: int snapshots_written: int blobs_written: int blobs_skipped: int tags_written: int failed_blobs: list[str] skipped_snapshots: list[str] # --------------------------------------------------------------------------- # Pack building # --------------------------------------------------------------------------- class _WalkResult(TypedDict): """Cached result of a BFS commit-graph walk. Produced once by :func:`walk_commits` and consumed by both :func:`collect_blob_ids_from_walk` (object ID collection) and :func:`build_mpack_from_walk` (load blobs for transmission). Sharing this avoids two identical BFS traversals per push: the first to gather object IDs for client-side deduplication, and the second to load blobs and assemble the pack mpack. ``missing_snapshots`` is populated by :func:`walk_commits` with the snapshot_ids of any reachable commit whose snapshot file is absent from the local store. Callers should surface this to the user before pushing — a pack that contains a commit but not its snapshot creates a dangling reference on the remote. """ commits: list[CommitRecord] snapshot_ids: set[str] all_blob_ids: list[str] # sorted, deduplicated — blobs_to_send = manifest_blobs - have_blobs have_blobs: set[str] # blob IDs reachable from any have-commit's snapshot manifest_blobs: set[str] # blob IDs referenced by new commits' manifests (old and new alike) oid_to_path: dict[str, str] # blob_id → repository path (from snapshot manifests) missing_snapshots: set[str] # snapshot_ids present in commits but absent on disk snapshot_deltas: list[SnapshotDeltaDict] # pre-computed, reuse in build_mpack_from_walk def walk_commits( repo_root: pathlib.Path, commit_ids: list[str], *, have: list[str] | None = None, ) -> _WalkResult: """BFS-walk the commit graph from *commit_ids*, stopping at *have*. Returns a :class:`_WalkResult` that can be passed to both :func:`collect_blob_ids_from_walk` and :func:`build_mpack_from_walk` to avoid repeating the traversal. This is the **single source of truth** for what goes into a push mpack. Callers that need both the object ID list and the full pack should call this once and pass the result to both downstream functions. Uses ``prune=lambda cid: cid in have_set`` so the walk terminates the moment it reaches a commit the server already has — no ancestor subgraph is expanded beyond the boundary. """ have_set: set[str] = set(have or []) commits_to_send: list[CommitRecord] = list( iter_ancestors(repo_root, commit_ids, prune=lambda cid: cid in have_set) ) # Collect blobs already on the remote (have-commits' snapshots). # Subtracting these gives us only genuinely new blobs to send. have_blobs: set[str] = set() for cid in have_set: have_commit = read_commit(repo_root, cid) if have_commit is not None: have_snap = read_snapshot(repo_root, have_commit.snapshot_id) if have_snap is not None: have_blobs.update(have_snap.manifest.values()) snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_send} commits_oldest_first = list(reversed(commits_to_send)) # One pass: compute deltas (one read_snapshot per commit) then derive # object IDs from delta_upsert — no separate manifest scan needed. missing_snapshots: set[str] = set() try: snapshot_deltas = _build_snapshot_deltas(repo_root, commits_oldest_first) except ValueError as exc: # Missing snapshot — extract sid and record it. missing_snapshots = { sid for sid in snapshot_ids if read_snapshot(repo_root, sid) is None } snapshot_deltas = [] manifest_blobs: set[str] = set(collect_blob_ids_from_deltas(snapshot_deltas)) # Build oid→path from delta_upsert entries (path → oid in each delta). oid_to_path: dict[str, str] = {} for delta in snapshot_deltas: for path, oid in (delta.get("delta_upsert") or {}).items(): oid_to_path[oid] = path blobs_to_send: set[str] = manifest_blobs - have_blobs if missing_snapshots: for sid in sorted(missing_snapshots): logger.warning( "⚠️ walk_commits: snapshot %s is missing from the local store — " "the commit(s) referencing it will be excluded from the pack. " "Run `muse verify` to audit store integrity.", sid, ) return _WalkResult( commits=commits_to_send, snapshot_ids=snapshot_ids, all_blob_ids=sorted(blobs_to_send), have_blobs=have_blobs, manifest_blobs=manifest_blobs, oid_to_path=oid_to_path, missing_snapshots=missing_snapshots, snapshot_deltas=snapshot_deltas, ) def stream_blob_chunks( repo_root: pathlib.Path, blob_ids: list[str], chunk_size: int, ) -> "collections.abc.Iterator[list[BlobPayload]]": """Yield blobs in chunks of *chunk_size* as they are read from disk. This is the hot path for ``muse push`` — reads one chunk at a time to avoid loading all blobs into RAM at once (peak RAM = one chunk, not the full set). The caller can start the first upload while the second chunk is still being assembled — reducing both peak memory and time-to-first-upload. Missing blobs are logged and skipped, consistent with :func:`build_mpack`. """ import collections.abc # local to avoid circular at module level chunk: list[BlobPayload] = [] for oid in blob_ids: raw = read_object(repo_root, oid) if raw is None: logger.warning("⚠️ stream_blob_chunks: blob %s absent — skipping", oid) continue chunk.append(BlobPayload(object_id=oid, content=raw)) if len(chunk) >= chunk_size: yield chunk chunk = [] if chunk: yield chunk def collect_blob_ids_from_walk(walk: _WalkResult) -> list[str]: """Return the sorted blob ID list from a pre-computed :func:`walk_commits` result. Zero disk I/O — the walk already read all snapshots. """ return walk["all_blob_ids"] def collect_blob_ids_from_deltas(deltas: list[SnapshotDeltaDict]) -> list[str]: """Return all unique object IDs from a pre-computed delta list. Zero additional disk I/O — extracts oids from delta_upsert.values() only. The first delta encodes the full base manifest; subsequent deltas encode only changed files. Their union is identical to the union of all full manifests (proof: every oid ever introduced appears in exactly one delta_upsert entry). Use this instead of collect_blob_ids on the mpack path — the deltas are already computed by _build_snapshot_deltas (one read per snapshot), so this is a pure in-memory operation. """ seen: set[str] = set() for delta in deltas: seen.update(delta.get("delta_upsert", {}).values()) return sorted(seen) def _build_snapshot_deltas( repo_root: pathlib.Path, commits_oldest_first: list[CommitRecord], ) -> list[SnapshotDeltaDict]: """Compute delta-encoded snapshots from a commit chain, oldest first. Each entry carries only the paths that changed relative to the previous snapshot in the chain. The first entry (no parent in this mpack) uses ``parent_snapshot_id=None`` and encodes the full manifest as ``delta_upsert``. Correctness invariant (content-addressing as proof):: manifest = apply_delta(prev_manifest, entry) assert hash_snapshot(manifest) == entry["snapshot_id"] This replaces per-snapshot full-manifest storage with O(changed_files) deltas — a 10–100× reduction for typical commit chains. """ deltas: list[SnapshotDeltaDict] = [] seen_sids: set[str] = set() prev_manifest: dict[str, str] = {} prev_sid: str | None = None for commit in commits_oldest_first: sid = commit.snapshot_id if sid in seen_sids: continue seen_sids.add(sid) snap = read_snapshot(repo_root, sid) if snap is None: raise ValueError( f"Push aborted: snapshot {sid} is missing from the local store " f"but is required by a commit being sent. " f"Run 'muse verify' to audit store integrity." ) manifest = snap.manifest delta_upsert = {k: v for k, v in manifest.items() if prev_manifest.get(k) != v} delta_remove = [k for k in prev_manifest if k not in manifest] # Only include directories in the wire entry if the stored snapshot_id # was computed WITH them. Some snapshots were written before directories # were part of hash_snapshot; their snap.directories field is populated # but the ID was hashed with None. Sending those dirs would cause # hash mismatch on the client side. dirs_for_wire = snap.directories or [] if dirs_for_wire and hash_snapshot(manifest, dirs_for_wire) != sid: dirs_for_wire = [] logger.debug( "[build_snapshot_deltas] sid=%s parent=%s manifest=%d dirs_stored=%d dirs_wire=%d upsert=%d remove=%d", sid[:20], (prev_sid or "none")[:20], len(manifest), len(snap.directories or []), len(dirs_for_wire), len(delta_upsert), len(delta_remove), ) deltas.append(SnapshotDeltaDict( snapshot_id=sid, parent_snapshot_id=prev_sid, delta_upsert=delta_upsert, delta_remove=delta_remove, directories=dirs_for_wire, )) prev_manifest = manifest prev_sid = sid return deltas def _apply_snapshot_deltas( raw_snapshots: list[SnapshotDeltaDict], ) -> _SnapshotResolvedMap: """Reconstruct full manifests from a delta chain. Applies each delta in order and verifies the result by hashing. The hash check IS the integrity proof — no external validation needed. Two snapshot formats are accepted: - Delta format (build_mpack): {snapshot_id, parent_snapshot_id, delta_upsert, delta_remove} - Full-manifest format: {snapshot_id, manifest, directories, ...} Corrupt or hash-mismatched entries are logged and skipped; they do not block independent valid entries with parent_snapshot_id=None. Dependent entries whose parent was skipped are also skipped (base = {} → hash mismatch → skip). Returns {snapshot_id: (full_manifest, directories)} for every valid entry. """ resolved: _SnapshotResolvedMap = {} for snap in raw_snapshots: sid = snap.get("snapshot_id", "") if not sid: continue # Two formats arrive here: # 1. Delta format (build_mpack): {snapshot_id, parent_snapshot_id, # delta_upsert, delta_remove, directories} — reconstruct from parent + diff. # 2. Full-manifest format (server sends cached manifest directly): # {snapshot_id, manifest, directories, ...} — use manifest directly. directories: list[str] = [] manifest_raw = snap.get("manifest") if isinstance(manifest_raw, dict): base = {k: v for k, v in manifest_raw.items() if isinstance(k, str) and isinstance(v, str)} dirs_raw = snap.get("directories") if isinstance(dirs_raw, list): directories = [d for d in dirs_raw if isinstance(d, str)] logger.debug( "[_apply_snapshot_deltas] snap=%s format=full manifest=%d dirs=%d", sid[:20], len(base), len(directories), ) else: parent_sid = snap.get("parent_snapshot_id") delta_upsert: dict[str, str] = snap.get("delta_upsert") or {} delta_remove: list[str] = snap.get("delta_remove") or [] # BUG FIX: directories must be read in the delta branch too — they # are included in the hash and were previously silently dropped, # causing hash mismatch for every snapshot with non-empty directories. dirs_raw = snap.get("directories") if isinstance(dirs_raw, list): directories = [d for d in dirs_raw if isinstance(d, str)] parent_entry = resolved.get(parent_sid) if parent_sid else None base = dict(parent_entry[0]) if parent_entry else {} base.update(delta_upsert) for path in delta_remove: base.pop(path, None) logger.debug( "[_apply_snapshot_deltas] snap=%s format=delta parent=%s " "upsert=%d remove=%d dirs=%d parent_resolved=%s", sid[:20], (parent_sid or "")[:20], len(delta_upsert), len(delta_remove), len(directories), parent_entry is not None, ) # Content-addressing IS the proof — hash the result. try: got = hash_snapshot(base, directories or None) except ValueError as _hash_exc: logger.warning( "⚠️ apply_mpack: snapshot %s has invalid object IDs in delta — skipped: %s", sid[:20], _hash_exc, ) continue got_nodirs = hash_snapshot(base, None) if directories else got logger.debug( "[apply_snapshot_deltas] sid=%s parent=%s manifest=%d dirs=%d " "got=%s match=%s got_nodirs=%s nodirs_match=%s", sid[:20], (snap.get("parent_snapshot_id") or "none")[:20], len(base), len(directories), got[:20], got == sid, got_nodirs[:20], got_nodirs == sid, ) if got != sid: logger.warning( "⚠️ apply_mpack: snapshot %s hash mismatch " "(reconstructed=%s) dirs=%s — skipped", sid[:20], got[:20], directories, ) continue resolved[sid] = (base, directories) return resolved def build_mpack_from_walk( repo_root: pathlib.Path, walk: _WalkResult, *, only_blobs: set[str] | None = None, repo_id: str = "", compress: bool = False, ) -> MPack: """Assemble an :class:`MPack` from a pre-computed :func:`walk_commits` result. Avoids the second BFS traversal that :func:`build_mpack` would otherwise perform. Only reads blob bytes for blobs in *only_blobs* (or all blobs when *only_blobs* is ``None``). When *compress* is ``True``, each blob is zstd-compressed (level 3). Falls back to ``raw`` when zstd makes the blob larger (rare for binary data). Uses the zstandard C extension — one call per blob, no Python loop. Returns: An :class:`MPack` ready for serialisation and transfer. """ missing_snapshots: set[str] = walk.get("missing_snapshots") or set() all_blob_ids: set[str] = set(walk["all_blob_ids"]) # Hard failure on any missing snapshot — silently skipping would push # commits without their snapshots, creating dangling references on the # remote that can never be healed without rewriting history. if missing_snapshots: sample = sorted(missing_snapshots)[:3] sample_str = ", ".join(sample) raise ValueError( f"Push aborted: {len(missing_snapshots)} snapshot(s) are missing from " f"the local store but are required by commits being sent " f"({sample_str}{'…' if len(missing_snapshots) > 3 else ''}). " f"Run 'muse verify' to audit store integrity." ) commits_to_send = list(walk["commits"]) snapshot_deltas = walk["snapshot_deltas"] candidate_blob_ids = ( all_blob_ids & only_blobs if only_blobs is not None else all_blob_ids ) blob_payloads: list[BlobPayload] = [] if not compress: for oid in sorted(candidate_blob_ids): raw = read_object(repo_root, oid) if raw is None: logger.warning("⚠️ build_mpack_from_walk: blob %s absent — skipping", oid) continue blob_payloads.append(BlobPayload(object_id=oid, content=raw)) else: import zstandard as _zstd cctx = _zstd.ZstdCompressor(level=3) for oid in sorted(candidate_blob_ids): raw = read_object(repo_root, oid) if raw is None: logger.warning("⚠️ build_mpack_from_walk: blob %s absent — skipping", oid) continue compressed = cctx.compress(raw) if len(compressed) < len(raw): blob_payloads.append(BlobPayload(object_id=oid, content=compressed, encoding="zstd")) else: blob_payloads.append(BlobPayload(object_id=oid, content=raw)) sent_commit_ids = [c.commit_id for c in commits_to_send] wire_tags = _tags_for_commits(repo_root, sent_commit_ids, repo_id) if repo_id else [] total_bytes = sum(len(o.get("content") or b"") for o in blob_payloads) agent_ids = sorted({ c.to_dict().get("agent_id", "") for c in commits_to_send if c.to_dict().get("agent_id") }) summary = MPackSummary( commits_count=len(commits_to_send), blobs_count=len(blob_payloads), blobs_bytes=total_bytes, branches={}, agent_ids=agent_ids, ) mpack: MPack = { "commits": [c.to_dict() for c in commits_to_send], "snapshots": snapshot_deltas, "blobs": blob_payloads, "summary": summary, "meta": MPackMeta( mode="full", base_commits=[], created_at=datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), ), } if wire_tags: mpack["tags"] = wire_tags logger.info( "✅ Built MPack (from walk): %d commits, %d snapshots, %d blobs, %d tags", len(commits_to_send), len(snapshot_deltas), len(blob_payloads), len(wire_tags), ) return mpack def _tags_for_commits( repo_root: pathlib.Path, commit_ids: list[str], repo_id: str ) -> list[WireTag]: """Return all tags attached to *commit_ids* as serialisable :class:`WireTag` dicts.""" seen_tag_ids: set[str] = set() wire_tags: list[WireTag] = [] for cid in commit_ids: for tag in get_tags_for_commit(repo_root, repo_id, cid): if tag.tag_id not in seen_tag_ids: seen_tag_ids.add(tag.tag_id) wire_tags.append(WireTag( tag_id=tag.tag_id, repo_id=tag.repo_id, commit_id=tag.commit_id, tag=tag.tag, created_at=tag.created_at.isoformat(), )) return wire_tags def build_mpack( repo_root: pathlib.Path, commit_ids: list[str], *, have: list[str] | None = None, only_blobs: set[str] | None = None, repo_id: str = "", ) -> MPack: """Assemble an :class:`MPack` from *commit_ids*, excluding commits in *have*. Performs a BFS walk of the commit graph from every ID in *commit_ids*, stopping at any commit already in *have*. Collects all snapshot manifests and blobs reachable from the selected commits. Missing blobs or snapshots are logged and skipped — the caller decides whether that constitutes an error. Args: repo_root: Root of the Muse repository. commit_ids: Tip commit IDs to include (e.g. current branch HEAD). have: Commit IDs already known to the receiver. The BFS stops at these, reducing mpack size. Pass ``None`` or ``[]`` to send the full history. only_blobs: When set, only include blobs whose IDs are in this set. Pass the set of blobs missing from the remote so the client only uploads what the remote actually needs. repo_id: Repository content ID used to look up tags. When omitted, tags are not included in the mpack. Returns: An :class:`MPack` ready for serialisation and transfer. """ walk = walk_commits(repo_root, commit_ids, have=have) if walk["missing_snapshots"]: sample = sorted(walk["missing_snapshots"])[:3] sample_str = ", ".join(sample) raise ValueError( f"Push aborted: {len(walk['missing_snapshots'])} snapshot(s) are missing from " f"the local store but are required by commits being sent " f"({sample_str}{'…' if len(walk['missing_snapshots']) > 3 else ''}). " f"Run 'muse verify' to audit store integrity." ) commits_to_send: list[CommitRecord] = list(walk["commits"]) snapshot_deltas = walk["snapshot_deltas"] all_blob_ids: set[str] = set(walk["all_blob_ids"]) # When only_blobs is provided skip any blob the remote already has — # only transmit the missing delta. candidate_blob_ids = ( all_blob_ids & only_blobs if only_blobs is not None else all_blob_ids ) promisor_remotes = load_promisor_remotes(repo_root) blob_payloads: list[BlobPayload] = [] for oid in sorted(candidate_blob_ids): raw = read_object(repo_root, oid) if raw is None: state = object_state(repo_root, oid, promisor_remotes) if state == ObjectState.PROMISED: logger.debug("build_mpack: blob %s is PROMISED — skipping", short_id(oid)) continue raise ValueError( f"Pack aborted: blob {oid} is missing from the local store " f"and no promisor remote is configured. " f"Run 'muse verify' to audit store integrity." ) blob_payloads.append(BlobPayload(object_id=oid, content=raw)) sent_commit_ids = [c.commit_id for c in commits_to_send] wire_tags = _tags_for_commits(repo_root, sent_commit_ids, repo_id) if repo_id else [] total_bytes = sum(len(b["content"]) for b in blob_payloads) agent_ids = sorted({ c.to_dict().get("agent_id", "") for c in commits_to_send if c.to_dict().get("agent_id") }) summary = MPackSummary( commits_count=len(commits_to_send), blobs_count=len(blob_payloads), blobs_bytes=total_bytes, branches={}, agent_ids=agent_ids, ) _have_list = list(have or []) mpack: MPack = { "commits": [c.to_dict() for c in commits_to_send], "snapshots": snapshot_deltas, "blobs": blob_payloads, "summary": summary, "meta": MPackMeta( mode="incremental" if _have_list else "full", base_commits=_have_list, created_at=datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), ), } if wire_tags: mpack["tags"] = wire_tags logger.info( "✅ Built MPack: %d commits, %d snapshots, %d blobs, %d tags", len(commits_to_send), len(snapshot_deltas), len(blob_payloads), len(wire_tags), ) return mpack # --------------------------------------------------------------------------- # Object ID collection — for pre-push deduplication negotiation # --------------------------------------------------------------------------- def collect_blob_ids( repo_root: pathlib.Path, commit_ids: list[str], *, have: list[str] | None = None, ) -> list[str]: """Return all blob IDs reachable from *commit_ids*, excluding *have*. Identical BFS walk to :func:`build_mpack` but without reading object bytes. Used by ``muse push`` for client-side deduplication — the result is compared against the remote's known objects, and :func:`build_mpack` is called with ``only_blobs`` set to the missing subset. This avoids loading any blob content until we know it is actually needed. Uses ``prune=lambda cid: cid in have_set`` so the walk terminates the moment it hits a server-known commit, without expanding its ancestor subgraph. Args: repo_root: Root of the Muse repository. commit_ids: Tip commit IDs to examine. have: Commit IDs already known to the receiver (BFS stops here). Returns: Sorted list of object IDs reachable from the delta. """ have_set: set[str] = set(have or []) commits_to_examine: list[CommitRecord] = list( iter_ancestors(repo_root, commit_ids, prune=lambda cid: cid in have_set) ) # Collect blobs already on the remote (have-commits' snapshots). have_blobs: set[str] = set() for cid in have_set: have_commit = read_commit(repo_root, cid) if have_commit is not None: have_snap = read_snapshot(repo_root, have_commit.snapshot_id) if have_snap is not None: have_blobs.update(have_snap.manifest.values()) snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_examine} all_blob_ids: set[str] = set() for sid in snapshot_ids: snap = read_snapshot(repo_root, sid) if snap is not None: all_blob_ids.update(snap.manifest.values()) return sorted(all_blob_ids - have_blobs) def compute_snapshot_delta( base: _Manifest, new: _Manifest, ) -> tuple[_Manifest, list[str]]: """Compute the delta between two snapshot manifests. Returns (added_or_modified, removed): - added_or_modified: paths whose object_id changed or are new in *new* - removed: paths present in *base* but absent from *new* """ added = {p: h for p, h in new.items() if base.get(p) != h} removed = [p for p in base if p not in new] return added, removed def apply_snapshot_delta( base: _Manifest, added: _Manifest, removed: list[str], ) -> _Manifest: """Reconstruct a full manifest by applying a delta to a base manifest. Inverse of compute_snapshot_delta: apply_snapshot_delta(base, *compute_snapshot_delta(base, new)) == new """ manifest = dict(base) manifest.update(added) for path in removed: manifest.pop(path, None) return manifest # --------------------------------------------------------------------------- # Presign helpers # --------------------------------------------------------------------------- class _PresignPayload(TypedDict): mpack_key: str size_bytes: int class _UnpackPayload(TypedDict, total=False): mpack_key: str branch: str head: str commits_count: int blobs_count: int force: bool def build_presign_payload(mpack_bytes: bytes) -> _PresignPayload: """Return the request body for POST /push/mpack-presign. The server uses mpack_key to name the MinIO object and to verify integrity after the client PUTs the bytes directly to MinIO. """ return {"mpack_key": blob_id(mpack_bytes), "size_bytes": len(mpack_bytes)} def build_unpack_payload( mpack_key: str, *, branch: str = "main", head: str = "", commits_count: int = 0, blobs_count: int = 0, force: bool = False, ) -> _UnpackPayload: """Return the request body for POST /push/unpack-mpack (Step 3).""" return { "mpack_key": mpack_key, "branch": branch, "head": head, "commits_count": int(commits_count), "blobs_count": int(blobs_count), "force": force, } # --------------------------------------------------------------------------- # Wire MPack encode / decode (Phase 3) # --------------------------------------------------------------------------- # # Wire format: # [4B] magic: b"MUSE" # [1B] version: 1 # [1B] section_count: N # [N*17B] section table: each entry is (1B type, 8B offset LE, 8B length LE) # [...] section data (concatenated, no padding) # [32B] SHA-256 of every byte above (footer, not included in its own hash) # # Section types: # 1 = BLOBS — raw _build_pack() bytes (byte-identical to Phase 1 .mpack) # 2 = COMMITS — [8B count] + N × [8B record_len + JSON bytes] # 3 = SNAPSHOTS — same length-prefixed JSON layout as COMMITS # 4 = TAGS — same layout # 5 = META — [8B json_len + JSON bytes] key-value pairs (repo_id, branch, head_commit_id) _WIRE_VERSION = 1 _WIRE_SEC_BLOBS = 1 _WIRE_SEC_COMMITS = 2 _WIRE_SEC_SNAPSHOTS = 3 _WIRE_SEC_TAGS = 4 _WIRE_SEC_META = 5 def _encode_records(records: list[dict]) -> bytes: """Encode a list of dicts as [8B count] + N × [8B len + JSON bytes].""" import json as _json parts = [_struct.pack(" list[dict]: """Decode [8B count] + N × [8B len + JSON bytes] into a list of dicts.""" if len(data) < 8: return [] import json as _json count = _struct.unpack_from(" len(data): break rec_len = _struct.unpack_from(" bytes: """Encode a META dict as [8B count] + N × [8B key_len + key + 8B val_len + val].""" import json as _json enc = _json.dumps(meta, separators=(",", ":")).encode() return _struct.pack(" _MetaDict: """Decode a META section back to a dict.""" if len(data) < 8: return {} import json as _json val_len = _struct.unpack_from(" bytes: """Encode an :class:`MPack` as a wire MPack binary bundle. The OBJECTS section bytes are byte-identical to what :func:`write_pack` writes to disk in Phase 1, so the client can extract the section and write it directly without any per-object decode step. Returns: Raw bytes of the wire bundle, starting with ``b"MUSE"`` and ending with a 32-byte SHA-256 footer. """ from muse.core.pack_store import _build_pack as _ps_build_pack blobs = mpack.get("blobs") or [] commits = mpack.get("commits") or [] snapshots = mpack.get("snapshots") or [] tags = mpack.get("tags") or [] def _content(blob: BlobPayload) -> bytes: raw = blob["content"] if blob.get("encoding") == "zstd": import zstandard as _zstd return _zstd.ZstdDecompressor().decompress(raw) return raw obj_pairs = [(o["object_id"], _content(o)) for o in blobs] blobs_bytes = _ps_build_pack(obj_pairs) if obj_pairs else b"" commits_bytes = _encode_records(commits) snapshots_bytes = _encode_records(snapshots) tags_bytes = _encode_records(tags) meta_bytes = _encode_meta(meta) if meta else b"" sections = [ (_WIRE_SEC_BLOBS, blobs_bytes), (_WIRE_SEC_COMMITS, commits_bytes), (_WIRE_SEC_SNAPSHOTS, snapshots_bytes), (_WIRE_SEC_TAGS, tags_bytes), (_WIRE_SEC_META, meta_bytes), ] section_count = len(sections) # Header: 4B magic + 1B version + 1B section_count = 6B # Table: section_count × (1B type + 8B offset + 8B length) = section_count × 17B header_size = 6 + section_count * 17 offset = header_size table_entries: list[tuple[int, int, int]] = [] for sec_type, sec_data in sections: table_entries.append((sec_type, offset, len(sec_data))) offset += len(sec_data) h = _hashlib.sha256() parts: list[bytes] = [] def _emit(chunk: bytes) -> None: h.update(chunk) parts.append(chunk) _emit(b"MUSE") _emit(_struct.pack(" MPack: """Parse a wire MPack binary bundle back into an :class:`MPack` dict. Verifies the SHA-256 footer before parsing any section. Raises: ValueError: Bad magic bytes or unknown version. OSError: Footer integrity check failed. """ from muse.core.pack_store import _parse_pack_bytes as _ps_parse if len(data) < 38: # 4+1+1 header + 1×17 section entry + 32 footer (minimum) raise ValueError("Wire MPack too short") if data[:4] != b"MUSE": raise ValueError(f"Wire MPack bad magic: {data[:4]!r}") version = data[4] if version != _WIRE_VERSION: raise ValueError(f"Wire MPack unknown version: {version}") body = data[:-32] stored = data[-32:] if _hashlib.sha256(body).digest() != stored: raise OSError("Wire MPack failed SHA-256 integrity check") section_count = data[5] cursor = 6 sections: dict[int, bytes] = {} for _ in range(section_count): sec_type = data[cursor] sec_offset, sec_length = _struct.unpack_from(" ApplyResult: """Write the contents of *mpack* into a local ``.muse/`` directory. Writes in dependency order: objects first (blobs), then snapshots (which reference object IDs), then commits (which reference snapshot IDs). All writes are idempotent — already-present items are silently skipped. Args: repo_root: Root of the Muse repository to write into. mpack: :class:`MPack` received from the remote. shallow_commits: Optional set of boundary commit IDs (shallow clone). These commits are written even if their parents are absent from the local store. Returns: :class:`ApplyResult` with counts of newly written and skipped items. """ import sys as _sys import time as _time blobs_written = 0 blobs_skipped = 0 snapshots_written = 0 commits_written = 0 tags_written = 0 raw_blobs = mpack.get("blobs") or [] raw_snapshots = mpack.get("snapshots") or [] raw_commits = mpack.get("commits") or [] _t_apply_start = _time.monotonic() # Pack-bomb guard: cap the total number of items accepted per call. # A legitimate push carries at most tens of thousands of blobs per chunk; # a pack claiming millions is an adversarial input. total_items = len(raw_blobs) + len(raw_snapshots) + len(raw_commits) if total_items > MAX_PACK_OBJECTS: raise ValueError( f"Pack rejected: {total_items:,} total items exceeds the " f"{MAX_PACK_OBJECTS:,} item limit per apply_mpack call. " "Split the pack into smaller chunks." ) # Deduplicate blob IDs before the write loop. A malicious or buggy # sender may repeat the same blob ID N times, forcing N sha256 hashes # before the "already exists" short-circuit returns False. seen_blob_ids: set[str] = set() # Track blob IDs that failed validation so dependent snapshots and commits # can be skipped — prevents dangling reference chains in the store. failed_blob_ids: set[str] = set() # Blobs that pass all checks and are new to this store — written as a pack. pack_blobs: list[tuple[str, bytes]] = [] for obj in raw_blobs: oid = obj.get("object_id", "") raw = obj.get("content", b"") if not oid or not isinstance(raw, bytes): logger.warning("⚠️ apply_mpack: blob entry missing fields — skipped") continue if oid in seen_blob_ids: logger.debug("⚠️ apply_mpack: duplicate blob_id %s — skipped", short_id(oid)) blobs_skipped += 1 continue seen_blob_ids.add(oid) if len(raw) > MAX_OBJECT_WRITE_BYTES: logger.warning( "⚠️ apply_mpack: blob %s is %d bytes, exceeding %d MiB limit — skipped", oid, len(raw), MAX_OBJECT_WRITE_BYTES // (1024 * 1024), ) failed_blob_ids.add(oid) continue try: validate_object_id(oid) actual = blob_id(raw) if actual != oid: raise ValueError( f"Content integrity failure: expected {oid} got {actual}" ) except ValueError as exc: logger.warning("⚠️ apply_mpack: malformed blob entry — skipped: %s", exc) failed_blob_ids.add(oid) continue if has_object(repo_root, oid): blobs_skipped += 1 continue pack_blobs.append((oid, raw)) blobs_written += 1 # Write all new blobs as a single MPack file + index — O(1) file writes # regardless of blob count. Raises OSError on disk failure, which # propagates before any snapshot or commit is written. write_pack(repo_root, pack_blobs) # Reconstruct full manifests from the delta chain, then write snapshots. # The hash IS the proof: hash_snapshot(reconstructed) == snapshot_id. # _apply_snapshot_deltas logs and skips corrupt entries — never raises. resolved_manifests = _apply_snapshot_deltas(raw_snapshots) # Track snapshot IDs skipped due to referencing failed objects so dependent # commits can be skipped — prevents dangling commit → snapshot → (missing object). skipped_snapshot_ids: set[str] = set() # Track snapshot IDs that were successfully written in this mpack. # Used below to refuse commits whose snapshots were not sent (snaps=0 bug guard). written_snapshot_ids: set[str] = set() for snap_entry in raw_snapshots: sid = snap_entry.get("snapshot_id", "") if not sid or sid not in resolved_manifests: continue try: manifest, directories = resolved_manifests[sid] # Guard against zip-slip: manifest keys are stored as-is and later # used to construct checkout paths. A malicious mpack could inject # keys like "../../etc/cron.d/malicious". We validate all keys here — # before writing — so no traversal path ever enters the store. for key in manifest: validate_workspace_path(key) # Manifest values are object IDs — validate they are safe hex strings. for oid in manifest.values(): validate_object_id(oid) # Refuse to write a snapshot whose objects were not fully written — # that would create a dangling snapshot → object reference. missing = failed_blob_ids.intersection(manifest.values()) if missing: logger.warning( "⚠️ apply_mpack: snapshot %s skipped — references %d object(s) " "that failed to write", short_id(sid), len(missing), ) skipped_snapshot_ids.add(sid) continue snap = SnapshotRecord(snapshot_id=sid, manifest=manifest, directories=directories) is_new = read_snapshot(repo_root, snap.snapshot_id) is None write_snapshot(repo_root, snap, sync=False) written_snapshot_ids.add(sid) if is_new: snapshots_written += 1 except (KeyError, ValueError) as exc: logger.warning("⚠️ apply_mpack: malformed snapshot — skipped: %s", exc) # Parse all commits first so per-commit validation errors are counted # before any writes begin. parsed_commits: list[CommitRecord] = [] for commit_dict in raw_commits: try: commit = CommitRecord.from_dict(commit_dict) if not commit.commit_id: logger.warning("⚠️ apply_mpack: commit missing commit_id — skipped") continue if not commit.snapshot_id: logger.warning( "⚠️ apply_mpack: commit %s missing snapshot_id — skipped", commit.commit_id, ) continue # Refuse to write a commit whose snapshot was skipped due to failed # objects — that would create a dangling commit → snapshot reference. if commit.snapshot_id in skipped_snapshot_ids: logger.warning( "⚠️ apply_mpack: commit %s skipped — its snapshot %s was not written " "(referenced object(s) failed)", short_id(commit.commit_id), short_id(commit.snapshot_id), ) continue # Refuse to write a commit whose snapshot was not included in this # mpack AND is not already in the local store. Writing such a commit # corrupts the store: the commit is then present in `have` on the # next pull, the server returns nothing new, and pull aborts forever # with "snapshot missing". The correct behaviour is to skip the commit # so the next pull re-requests it together with its snapshot. if ( commit.snapshot_id not in written_snapshot_ids and read_snapshot(repo_root, commit.snapshot_id) is None ): logger.warning( "⚠️ apply_mpack: commit %s skipped — snapshot %s was not in this " "mpack and is not in the local store (server sent snaps=0). " "The next pull will re-request this commit with its snapshot.", short_id(commit.commit_id), short_id(commit.snapshot_id), ) continue parsed_commits.append(commit) except (KeyError, ValueError, TypeError) as exc: logger.warning("⚠️ apply_mpack: malformed commit — skipped: %s", exc) # Write commits in dependency order: retry until stable. # Bundles may arrive with commits in BFS order (newest-first). Phase 2's # parent-existence guard rejects a commit whose parent hasn't been written # yet. Keep cycling through MissingParentError commits until every parent # in the mpack has been written, or until a full pass produces no progress # (the missing parent isn't in this mpack — log and skip). _shallow: "collections.abc.Container[str]" = shallow_commits or () pending = parsed_commits while pending: deferred: list[CommitRecord] = [] for commit in pending: # Shallow boundary commits have their parent check bypassed so they # can be written even when their parents are not in the local store. is_shallow = commit.commit_id in _shallow try: is_new = not has_object(repo_root, commit.commit_id) write_commit(repo_root, commit, skip_parent_check=is_shallow, sync=False) if is_new: commits_written += 1 except MissingParentError: deferred.append(commit) except OSError as exc: logger.critical( "❌ apply_mpack: store integrity violation for commit — skipped: %s", exc ) except (ValueError, TypeError) as exc: logger.warning("⚠️ apply_mpack: malformed commit — skipped: %s", exc) if len(deferred) == len(pending): # No progress in this pass — the missing parents are not in the mpack. for commit in deferred: logger.warning( "⚠️ apply_mpack: commit %s skipped — parent not in mpack or local store", commit.commit_id, ) break pending = deferred # Bulk-fsync: all snapshot and commit files were written with sync=False # (no per-file fsync) for throughput. A single directory fsync here makes # all preceding renames durable in one barrier — the same strategy git uses # for pack writes. The ref is not updated until after apply_mpack returns, # so a crash before this fsync leaves the store in a consistent (if # incomplete) state that is safe to re-apply. if snapshots_written > 0 or commits_written > 0: from muse.core.paths import muse_dir as _muse_dir _dot = _muse_dir(repo_root) for _dir in (_dot / "snapshots", _dot / "commits"): if _dir.exists(): try: _fd = os.open(str(_dir), os.O_RDONLY) try: os.fsync(_fd) finally: os.close(_fd) except OSError: pass # best-effort — some filesystems (tmpfs) reject dir fsync for wire_tag in mpack.get("tags") or []: try: tag_record = TagRecord.from_dict(TagDict( tag_id=wire_tag["tag_id"], repo_id=wire_tag["repo_id"], commit_id=wire_tag["commit_id"], tag=wire_tag["tag"], created_at=wire_tag["created_at"], )) write_tag(repo_root, tag_record) tags_written += 1 except (KeyError, ValueError) as exc: logger.warning("⚠️ apply_mpack: malformed tag — skipped: %s", exc) logger.info( "✅ Applied pack: %d new blobs, %d new snapshots, %d new commits, %d tags (%d blobs skipped)", blobs_written, snapshots_written, commits_written, tags_written, blobs_skipped, ) return ApplyResult( commits_written=commits_written, snapshots_written=snapshots_written, blobs_written=blobs_written, blobs_skipped=blobs_skipped, tags_written=tags_written, failed_blobs=list(failed_blob_ids), skipped_snapshots=list(skipped_snapshot_ids), )