"""Muse pack format — bundle of commits, snapshots, and blobs for wire transfer. A :class:`PackBundle` 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) - :class:`SnapshotDict` records (file manifests) - :class:`ObjectPayload` entries (raw blob bytes) - ``branch_heads`` mapping (branch name → commit ID, reflecting remote state) :func:`build_pack` collects all data reachable from a set of commit IDs. :func:`apply_pack` writes a bundle into a local ``.muse/`` directory. MWP wire encoding -------------------- Object bytes are transmitted as raw ``bytes`` in :class:`ObjectPayload`. The :class:`~muse.core.transport.HttpTransport` serialises the pack using ``msgpack`` (``Content-Type: application/x-msgpack``) which handles binary natively — no base64 inflation or encoding overhead. """ from __future__ import annotations import collections import logging import pathlib from typing import TypedDict from muse.core.object_store import read_object, write_object from muse.core.validation import ( MAX_OBJECT_WRITE_BYTES, MAX_PACK_OBJECTS, validate_object_id, validate_workspace_path, ) from muse.core._types import BranchHeads from muse.core.store import ( CommitDict, CommitRecord, SnapshotDict, SnapshotRecord, TagDict, TagRecord, get_all_tags, get_tags_for_commit, read_commit, read_snapshot, write_commit, write_snapshot, write_tag, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Wire-format TypedDicts # --------------------------------------------------------------------------- class _ObjectPayloadBase(TypedDict): """Required fields for every object payload transmitted over MWP.""" object_id: str content: bytes class ObjectPayload(_ObjectPayloadBase, total=False): """A single content-addressed blob with encoding metadata for MWP transport. 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 object; used by the server to look up delta base candidates for the next push. encoding: ``"raw"`` (default) | ``"zlib"`` | ``"delta+zlib"``. base_id: Base object ID for ``"delta+zlib"`` encoding. """ path: str encoding: str base_id: str class WireTag(TypedDict): """A tag record serialised for wire transfer inside a :class:`PackBundle`.""" tag_id: str repo_id: str commit_id: str tag: str created_at: str class PackBundle(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. """ commits: list[CommitDict] snapshots: list[SnapshotDict] objects: list[ObjectPayload] #: Tags attached to any commit included in this bundle. tags: list[WireTag] #: Remote branch heads at the time the bundle was produced. branch_heads: BranchHeads 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 #: Optional Worker URL advertised by the server. When present, the CLI #: routes POST /push/object-pack calls here instead of the primary URL. pack_origin: str 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_pack` describing what was written. ``objects_skipped`` counts blobs already present in the store (not rewritten, idempotent). All other counts reflect *new* writes only. """ commits_written: int snapshots_written: int objects_written: int objects_skipped: int class ObjectsChunkResponse(TypedDict): """Response from ``POST {url}/push/objects`` — one chunk of a chunked push. Returned by both :class:`~muse.core.transport.HttpTransport` and :class:`~muse.core.transport.LocalFileTransport` after pre-uploading a batch of content-addressed objects. ``stored`` — objects written to storage in this call. ``skipped`` — objects already present on the remote (idempotent no-ops). """ stored: int skipped: int # --------------------------------------------------------------------------- # 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_object_ids` (Phase 1 — send IDs to filter-objects) and :func:`build_pack` (Phase 2 — load blobs for missing objects). Sharing this avoids two identical BFS traversals per push: the first to gather object IDs for ``POST /filter-objects``, and the second to actually load blobs and assemble the pack bundle. ``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_object_ids: list[str] # sorted, deduplicated oid_to_path: dict[str, str] # object_id → repository path (from snapshot manifests) missing_snapshots: set[str] # snapshot_ids present in commits but absent on disk 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_object_ids_from_walk` and :func:`build_pack_from_walk` to avoid repeating the traversal. This is the **single source of truth** for what goes into a push bundle. Callers that need both the object ID list (for ``POST /filter-objects``) and the full pack (for uploading blobs) should call this once and pass the result to both downstream functions. """ have_set: set[str] = set(have or []) commits_to_send: list[CommitRecord] = [] seen: set[str] = set(have_set) queue: collections.deque[str] = collections.deque( cid for cid in commit_ids if cid not in seen ) while queue: cid = queue.popleft() if cid in seen: continue seen.add(cid) commit = read_commit(repo_root, cid) if commit is None: logger.warning("⚠️ walk_commits: commit %s not found — skipping", cid[:8]) continue commits_to_send.append(commit) if commit.parent_commit_id and commit.parent_commit_id not in seen: queue.append(commit.parent_commit_id) if commit.parent2_commit_id and commit.parent2_commit_id not in seen: queue.append(commit.parent2_commit_id) snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_send} missing_snapshots: set[str] = set() all_object_ids: set[str] = set() oid_to_path: dict[str, str] = {} for sid in snapshot_ids: snap = read_snapshot(repo_root, sid) if snap is not None: all_object_ids.update(snap.manifest.values()) # Build oid→path from each snapshot manifest (path → oid). # Later snapshots win on collision — any path is fine for delta lookup. for path, oid in snap.manifest.items(): oid_to_path[oid] = path else: missing_snapshots.add(sid) 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[:8], ) return _WalkResult( commits=commits_to_send, snapshot_ids=snapshot_ids, all_object_ids=sorted(all_object_ids), oid_to_path=oid_to_path, missing_snapshots=missing_snapshots, ) def stream_object_chunks( repo_root: pathlib.Path, object_ids: list[str], chunk_size: int, ) -> "collections.abc.Iterator[list[ObjectPayload]]": """Yield blobs in chunks of *chunk_size* as they are read from disk. This is the hot path for ``muse push`` after ``POST /filter-objects``. Instead of reading all blobs into RAM before uploading (peak RAM = all missing objects), we read one chunk at a time and yield it immediately. 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_pack`. """ import collections.abc # local to avoid circular at module level chunk: list[ObjectPayload] = [] for oid in object_ids: raw = read_object(repo_root, oid) if raw is None: logger.warning("⚠️ stream_object_chunks: blob %s absent — skipping", oid[:8]) continue chunk.append(ObjectPayload(object_id=oid, content=raw)) if len(chunk) >= chunk_size: yield chunk chunk = [] if chunk: yield chunk def collect_object_ids_from_walk(walk: _WalkResult) -> list[str]: """Return the sorted object ID list from a pre-computed :func:`walk_commits` result. Zero disk I/O — the walk already read all snapshots. """ return walk["all_object_ids"] def build_pack_from_walk( repo_root: pathlib.Path, walk: _WalkResult, *, only_objects: set[str] | None = None, repo_id: str = "", ) -> PackBundle: """Assemble a :class:`PackBundle` from a pre-computed :func:`walk_commits` result. Avoids the second BFS traversal that :func:`build_pack` would otherwise perform. Only reads blob bytes for objects in *only_objects* (or all objects when *only_objects* is ``None``). This is the hot path for ``muse push`` after ``POST /filter-objects`` narrows the upload set — we already have the commit+snapshot lists in memory from the earlier :func:`walk_commits` call. """ missing_snapshots: set[str] = walk.get("missing_snapshots") or set() all_object_ids_set: set[str] = set(walk["all_object_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(s[:8] for s in 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." ) snapshot_dicts: list[SnapshotDict] = [] for sid in sorted(walk["snapshot_ids"]): snap = read_snapshot(repo_root, sid) if snap is None: # This should not happen — missing_snapshots guard above catches it. raise ValueError( f"Push aborted: snapshot {sid[:8]} is missing from the local store. " f"Run 'muse verify' to audit store integrity." ) snapshot_dicts.append(snap.to_dict()) commits_to_send = list(walk["commits"]) candidate_ids = ( all_object_ids_set & only_objects if only_objects is not None else all_object_ids_set ) object_payloads: list[ObjectPayload] = [] for oid in sorted(candidate_ids): raw = read_object(repo_root, oid) if raw is None: logger.warning("⚠️ build_pack_from_walk: blob %s absent — skipping", oid[:8]) continue object_payloads.append(ObjectPayload(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 [] bundle: PackBundle = { "commits": [c.to_dict() for c in commits_to_send], "snapshots": snapshot_dicts, "objects": object_payloads, } if wire_tags: bundle["tags"] = wire_tags logger.info( "✅ Built pack (from walk): %d commits, %d snapshots, %d objects, %d tags", len(commits_to_send), len(snapshot_dicts), len(object_payloads), len(wire_tags), ) return bundle 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_pack( repo_root: pathlib.Path, commit_ids: list[str], *, have: list[str] | None = None, only_objects: set[str] | None = None, repo_id: str = "", ) -> PackBundle: """Assemble a :class:`PackBundle` 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 object blobs reachable from the selected commits. Missing objects 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 bundle size. Pass ``None`` or ``[]`` to send the full history. only_objects: When set, only include objects whose IDs are in this set. Used after a ``POST /filter-objects`` negotiation so the client only uploads objects the remote is missing. repo_id: Repository UUID used to look up tags. When omitted, tags are not included in the bundle. Returns: A :class:`PackBundle` ready for serialisation and transfer. """ have_set: set[str] = set(have or []) # BFS walk from every tip, treating have_set as already-visited. commits_to_send: list[CommitRecord] = [] seen: set[str] = set(have_set) queue: collections.deque[str] = collections.deque( cid for cid in commit_ids if cid not in seen ) while queue: cid = queue.popleft() if cid in seen: continue seen.add(cid) commit = read_commit(repo_root, cid) if commit is None: logger.warning("⚠️ build_pack: commit %s not found — skipping", cid[:8]) continue commits_to_send.append(commit) if commit.parent_commit_id and commit.parent_commit_id not in seen: queue.append(commit.parent_commit_id) if commit.parent2_commit_id and commit.parent2_commit_id not in seen: queue.append(commit.parent2_commit_id) # Unique snapshot IDs referenced by selected commits. snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_send} snapshot_dicts: list[SnapshotDict] = [] all_object_ids: set[str] = set() for sid in sorted(snapshot_ids): snap = read_snapshot(repo_root, sid) if snap is None: # Hard failure — a commit whose snapshot is absent locally cannot # be pushed safely. Silently skipping would create a dangling # commit reference on the remote: the commit lands, the snapshot # never does, and every subsequent pull restores the commit but # not its snapshot. Fail loudly so the user knows the store is # corrupt and can take corrective action before data is lost. raise ValueError( f"Push aborted: snapshot {sid[:8]} is missing from the local store " f"but is required by a commit being sent. " f"Run 'muse verify' to audit store integrity." ) snapshot_dicts.append(snap.to_dict()) all_object_ids.update(snap.manifest.values()) # When only_objects is provided (post filter-objects negotiation) skip # any object the remote already has — only transmit the missing delta. candidate_ids = ( all_object_ids & only_objects if only_objects is not None else all_object_ids ) object_payloads: list[ObjectPayload] = [] for oid in sorted(candidate_ids): raw = read_object(repo_root, oid) if raw is None: logger.warning("⚠️ build_pack: blob %s absent from store — skipping", oid[:8]) continue object_payloads.append(ObjectPayload(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 [] bundle: PackBundle = { "commits": [c.to_dict() for c in commits_to_send], "snapshots": snapshot_dicts, "objects": object_payloads, } if wire_tags: bundle["tags"] = wire_tags logger.info( "✅ Built pack: %d commits, %d snapshots, %d objects, %d tags", len(commits_to_send), len(snapshot_dicts), len(object_payloads), len(wire_tags), ) return bundle # --------------------------------------------------------------------------- # Object ID collection — for pre-push deduplication negotiation # --------------------------------------------------------------------------- def collect_object_ids( repo_root: pathlib.Path, commit_ids: list[str], *, have: list[str] | None = None, ) -> list[str]: """Return all object IDs reachable from *commit_ids*, excluding *have*. Identical BFS walk to :func:`build_pack` but without reading object bytes. Used by ``muse push`` to call ``POST /filter-objects`` before building the full pack — the client discovers which objects are missing on the remote and then calls :func:`build_pack` with ``only_objects`` set to that subset. This avoids loading any blob content until we know it is actually needed. 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] = [] seen: set[str] = set(have_set) queue: collections.deque[str] = collections.deque( cid for cid in commit_ids if cid not in seen ) while queue: cid = queue.popleft() if cid in seen: continue seen.add(cid) commit = read_commit(repo_root, cid) if commit is None: continue commits_to_examine.append(commit) if commit.parent_commit_id and commit.parent_commit_id not in seen: queue.append(commit.parent_commit_id) if commit.parent2_commit_id and commit.parent2_commit_id not in seen: queue.append(commit.parent2_commit_id) snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_examine} all_object_ids: set[str] = set() for sid in snapshot_ids: snap = read_snapshot(repo_root, sid) if snap is not None: all_object_ids.update(snap.manifest.values()) return sorted(all_object_ids) # --------------------------------------------------------------------------- # Pack applying # --------------------------------------------------------------------------- def apply_pack(repo_root: pathlib.Path, bundle: PackBundle) -> ApplyResult: """Write the contents of *bundle* 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. bundle: :class:`PackBundle` received from the remote. Returns: :class:`ApplyResult` with counts of newly written and skipped items. """ objects_written = 0 objects_skipped = 0 snapshots_written = 0 commits_written = 0 raw_objects = bundle.get("objects") or [] raw_snapshots = bundle.get("snapshots") or [] raw_commits = bundle.get("commits") or [] # Pack-bomb guard: cap the total number of items accepted per call. # A legitimate push carries at most tens of thousands of objects per chunk; # a pack claiming millions is an adversarial input. total_items = len(raw_objects) + 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_pack call. " "Split the pack into smaller chunks." ) # Deduplicate object IDs before the write loop. A malicious or buggy # sender may repeat the same object_id N times, forcing N sha256 hashes # before the "already exists" short-circuit returns False. seen_object_ids: set[str] = set() for obj in raw_objects: oid = obj.get("object_id", "") raw = obj.get("content", b"") if not oid or not isinstance(raw, bytes): logger.warning("⚠️ apply_pack: blob entry missing fields — skipped") continue if oid in seen_object_ids: logger.debug("⚠️ apply_pack: duplicate object_id %s — skipped", oid[:8]) objects_skipped += 1 continue seen_object_ids.add(oid) # Per-object size cap: check before calling write_object to avoid # hashing a huge payload that will be rejected anyway. if len(raw) > MAX_OBJECT_WRITE_BYTES: logger.warning( "⚠️ apply_pack: object %s is %d bytes, exceeding %d MiB limit — skipped", oid[:8], len(raw), MAX_OBJECT_WRITE_BYTES // (1024 * 1024), ) continue try: if write_object(repo_root, oid, raw): objects_written += 1 else: objects_skipped += 1 except ValueError as exc: # Malicious object IDs (non-hex, path traversal) or content/ID # mismatch — log and skip rather than aborting the entire pack. logger.warning("⚠️ apply_pack: malformed object entry — skipped: %s", exc) for snap_dict in raw_snapshots: try: snap = SnapshotRecord.from_dict(snap_dict) # Guard against zip-slip: manifest keys are stored as-is and later # used to construct checkout paths. A malicious bundle could inject # keys like "../../etc/cron.d/evil". We validate all keys here — # before writing — so no traversal path ever enters the store. for key in snap.manifest: validate_workspace_path(key) # Manifest values are object IDs — validate they are safe hex strings. for oid in snap.manifest.values(): validate_object_id(oid) is_new = read_snapshot(repo_root, snap.snapshot_id) is None write_snapshot(repo_root, snap) if is_new: snapshots_written += 1 except (KeyError, ValueError) as exc: logger.warning("⚠️ apply_pack: malformed snapshot — skipped: %s", exc) for commit_dict in raw_commits: try: commit = CommitRecord.from_dict(commit_dict) # from_dict is designed for trusted, typed callers and does not # validate essential fields — guard them here before the commit # reaches write_commit (which constructs paths from commit_id). if not commit.commit_id: logger.warning("⚠️ apply_pack: commit missing commit_id — skipped") continue if not commit.snapshot_id: logger.warning( "⚠️ apply_pack: commit %s missing snapshot_id — skipped", commit.commit_id[:8], ) continue is_new = read_commit(repo_root, commit.commit_id) is None write_commit(repo_root, commit) if is_new: commits_written += 1 except OSError as exc: # write_commit raises OSError("Store integrity violation") when the # existing commit file contains a DIFFERENT commit_id than the one # being written — indicating the local store has been tampered with. # Log CRITICAL and skip rather than crashing the entire apply_pack. logger.critical( "❌ apply_pack: store integrity violation for commit — skipped: %s", exc ) except (KeyError, ValueError, TypeError) as exc: logger.warning("⚠️ apply_pack: malformed commit — skipped: %s", exc) for wire_tag in bundle.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) except (KeyError, ValueError) as exc: logger.warning("⚠️ apply_pack: malformed tag — skipped: %s", exc) logger.info( "✅ Applied pack: %d new blobs, %d new snapshots, %d new commits (%d blobs skipped)", objects_written, snapshots_written, commits_written, objects_skipped, ) return ApplyResult( commits_written=commits_written, snapshots_written=snapshots_written, objects_written=objects_written, objects_skipped=objects_skipped, )