gabriel / muse public
mpack.py python
1,442 lines 56.1 KB Hotspot
Raw
sha256:99fba7e18ebcba8196d02fdc1796a75641959140b208caa4c64657de5fce3302 fix: downgrade verbose per-snapshot logs to debug in mpack.py Sonnet 4.6 34 days ago
1 """Muse MPack format — mpack of commits, snapshots, and blobs for wire transfer.
2
3 An :class:`MPack` is the unit of exchange between the Muse CLI and a remote
4 (e.g. MuseHub). It carries everything needed to reconstruct a slice of commit
5 history locally:
6
7 - :class:`CommitDict` records (full metadata + agent provenance)
8 - :class:`SnapshotDict` records (file manifests)
9 - :class:`BlobPayload` entries (raw blob bytes)
10 - ``summary`` (:class:`MPackSummary`) — advisory counts for agent routing
11
12 :func:`build_mpack` collects all data reachable from a set of commit IDs and
13 populates the summary field.
14 :func:`apply_mpack` writes a mpack into a local ``.muse/`` directory.
15
16 MPack wire encoding
17 ---------------------
18 An MPack is encoded in the MPack binary wire format (``b"MUSE"`` magic,
19 section table, SHA-256 footer) and transmitted with
20 ``Content-Type: application/x-muse-pack``.
21
22 Agent contract
23 --------------
24
25 - ``exit_code`` 0: all data applied successfully.
26 - ``exit_code`` 1: validation error (malformed object ID, path traversal).
27 - ``exit_code`` 3: I/O error reading from local store.
28 - ``duration_ms``: wall-clock milliseconds for build or apply.
29 """
30
31 import collections
32 import datetime
33 import hashlib as _hashlib
34 import logging
35 import os
36 import pathlib
37 import struct as _struct
38 from typing import TypedDict
39
40 from muse.core.graph import iter_ancestors
41 from muse.core.object_availability import ObjectState, load_promisor_remotes, object_state
42 from muse.core.object_store import has_object, read_object
43 from muse.core.pack_store import write_pack
44 from muse.core.ids import hash_snapshot
45 from muse.core.validation import (
46 MAX_OBJECT_WRITE_BYTES,
47 MAX_PACK_OBJECTS,
48 validate_object_id,
49 validate_workspace_path,
50 )
51 from muse.core.types import BranchHeads, blob_id, short_id
52 from muse.core.commits import (
53 CommitDict,
54 CommitRecord,
55 MissingParentError,
56 read_commit,
57 write_commit,
58 )
59 from muse.core.snapshots import (
60 SnapshotDict,
61 SnapshotRecord,
62 read_snapshot,
63 write_snapshot,
64 )
65 from muse.core.tags import (
66 TagDict,
67 TagRecord,
68 get_all_tags,
69 get_tags_for_commit,
70 write_tag,
71 )
72
73 logger = logging.getLogger(__name__)
74
75 # ---------------------------------------------------------------------------
76 # Type aliases — avoid bare dict[str, X] at boundaries
77 # ---------------------------------------------------------------------------
78
79 _Manifest = dict[str, str] # path → object_id
80 _JsonValue = str | int | float | bool | None
81 _MetaDict = dict[str, _JsonValue] # loose metadata (dynamic keys, JSON-serialisable)
82 _SnapshotResolvedMap = dict[str, tuple[_Manifest, list[str]]] # sid → (manifest, dirs)
83
84 # ---------------------------------------------------------------------------
85 # Wire-format TypedDicts
86 # ---------------------------------------------------------------------------
87
88 class _BlobPayloadBase(TypedDict):
89 """Required fields for every blob payload in an MPack."""
90
91 object_id: str
92 content: bytes
93
94 class BlobPayload(_BlobPayloadBase, total=False):
95 """A single content-addressed blob with encoding metadata for mpack transfer.
96
97 Required fields (always present):
98 object_id: Content-addressed SHA-256 identifier (``sha256:<hex>``).
99 content: Raw or encoded bytes — see *encoding*.
100
101 Optional fields (omit for ``"raw"`` with no base):
102 path: Repository path of this blob; used by the server to look
103 up delta base candidates for the next push.
104 encoding: ``"raw"`` (default) | ``"zlib"`` | ``"delta+zlib"``.
105 base_id: Base blob ID for ``"delta+zlib"`` encoding.
106 sz: Uncompressed byte count of the target blob. Required when
107 ``encoding`` is ``"delta+zlib"`` so the server can pre-allocate
108 before decompression. Ignored for ``"raw"`` payloads.
109 """
110
111 path: str
112 encoding: str
113 base_id: str
114 sz: int
115
116 class WireTag(TypedDict):
117 """A tag record serialised for wire transfer inside an :class:`MPack`."""
118
119 tag_id: str
120 repo_id: str
121 commit_id: str
122 tag: str
123 created_at: str
124
125 class MPackMeta(TypedDict, total=False):
126 """Self-describing metadata embedded in every :class:`MPack`.
127
128 Agents read this to understand the mpack's scope without inspecting
129 commits or objects.
130
131 Fields:
132 mode: ``"full"`` — all referenced objects must be in the mpack
133 or the local store. ``"incremental"`` — some objects are
134 expected to exist at the receiver's base (declared in
135 ``base_commits``); they are not included in this mpack.
136 base_commits: Commit IDs passed as ``--have`` when the mpack was
137 built. Empty for full bundles.
138 created_at: ISO 8601 UTC timestamp of when the mpack was assembled.
139 """
140
141 mode: str # "full" | "incremental"
142 base_commits: list[str] # sha256:-prefixed commit IDs
143 created_at: str # ISO 8601 — e.g. "2026-01-01T00:00:00Z"
144
145 class MPackSummary(TypedDict, total=False):
146 """Advisory summary embedded in every :class:`MPack`.
147
148 Agents read this to make routing/accept/reject decisions before
149 touching commits or objects. All fields are advisory — receivers
150 must not rely on them for correctness, only for optimisation.
151
152 Fields:
153 commits_count: Number of commits in this mpack.
154 blobs_count: Number of unique blobs in this mpack.
155 blobs_bytes: Total uncompressed blob bytes.
156 branches: Branch name → tip commit_id at build time.
157 agent_ids: All agent_id values from commits in this mpack.
158 """
159
160 commits_count: int
161 blobs_count: int
162 blobs_bytes: int
163 branches: BranchHeads # branch_name → commit_id
164 agent_ids: list[str] # distinct agent_ids in commits
165
166 class SnapshotDeltaDict(TypedDict, total=False):
167 """Wire representation of a snapshot as a delta from its parent snapshot.
168
169 Guiding principle: content-addressing is a proof, not a label.
170 ``snapshot_id = sha256(sorted path-NUL-oid pairs)``. A receiver who
171 holds ``snapshot_id`` and the delta can reconstruct the full manifest
172 and verify it by hashing — no external store needed.
173
174 Fields:
175 snapshot_id: sha256 of the *full* manifest (the proof).
176 parent_snapshot_id: snapshot_id of the parent, or ``None`` for root.
177 delta_upsert: Paths added or changed relative to parent.
178 delta_remove: Paths removed relative to parent.
179
180 Reconstruction::
181
182 manifest = dict(resolved[parent_snapshot_id]) # or {} if None
183 manifest.update(delta_upsert)
184 for path in delta_remove:
185 del manifest[path]
186 assert hash_snapshot(manifest) == snapshot_id # the math IS the proof
187 """
188
189 snapshot_id: str
190 parent_snapshot_id: str | None
191 delta_upsert: dict[str, str] # path → object_id
192 delta_remove: list[str] # paths removed
193
194
195 class MPack(TypedDict, total=False):
196 """The unit of exchange between the Muse CLI and a remote.
197
198 All fields are optional so that partial bundles (fetch-only, objects-only)
199 are valid wire messages. Callers check for presence before consuming.
200
201 The ``summary`` field carries advisory metadata for agent routing —
202 agents can make decisions from it without deserialising commits or objects.
203
204 The ``meta`` field declares the mpack's scope (full vs incremental) and
205 base commits, allowing receivers to verify it correctly without out-of-band
206 knowledge of how it was built.
207
208 Snapshots are stored as :class:`SnapshotDeltaDict` entries in commit-graph
209 order (oldest first). The first entry has ``parent_snapshot_id=None`` and
210 ``delta_upsert`` equal to the full manifest. Every subsequent entry carries
211 only the paths that changed. Receivers reconstruct full manifests by
212 applying the delta chain and verify correctness by hashing the result.
213 """
214
215 commits: list[CommitDict]
216 snapshots: list[SnapshotDeltaDict]
217 blobs: list[BlobPayload]
218 #: Tags attached to any commit included in this mpack.
219 tags: list[WireTag]
220 #: Advisory summary — populated by :func:`build_mpack`.
221 summary: MPackSummary
222 #: Self-describing metadata — always written by :func:`build_mpack`.
223 meta: MPackMeta
224
225 class RemoteInfo(TypedDict, total=False):
226 """Repository metadata returned by ``GET {url}/refs``."""
227
228 repo_id: str # always present
229 domain: str # always present
230 #: Maps branch name → commit ID for every branch on the remote.
231 branch_heads: BranchHeads # always present
232 default_branch: str # always present
233
234 class PushResult(TypedDict):
235 """Server response after a push attempt."""
236
237 ok: bool
238 message: str
239 #: Updated branch heads on the remote after the push (if successful).
240 branch_heads: BranchHeads
241
242 class FetchRequest(TypedDict, total=False):
243 """Body of ``POST {url}/fetch`` — negotiates which commits to transfer.
244
245 ``want`` lists commit IDs the client wants to receive.
246 ``have`` lists commit IDs already present locally, allowing the server
247 to send only the commits the client lacks (delta negotiation).
248 """
249
250 want: list[str]
251 have: list[str]
252
253 class ApplyResult(TypedDict):
254 """Counts returned by :func:`apply_mpack` describing what was written.
255
256 ``blobs_skipped`` counts blobs already present in the store (not
257 rewritten, idempotent). All other counts reflect *new* writes only.
258 ``tags_written`` counts tag records written from the mpack's ``tags``
259 section (0 for bundles created without tag data).
260 ``failed_blobs`` blob IDs that failed integrity or write checks.
261 ``skipped_snapshots`` snapshot IDs skipped because a referenced blob failed.
262 """
263
264 commits_written: int
265 snapshots_written: int
266 blobs_written: int
267 blobs_skipped: int
268 tags_written: int
269 failed_blobs: list[str]
270 skipped_snapshots: list[str]
271
272 # ---------------------------------------------------------------------------
273 # Pack building
274 # ---------------------------------------------------------------------------
275
276 class _WalkResult(TypedDict):
277 """Cached result of a BFS commit-graph walk.
278
279 Produced once by :func:`walk_commits` and consumed by both
280 :func:`collect_blob_ids_from_walk` (object ID collection) and
281 :func:`build_mpack_from_walk` (load blobs for transmission).
282
283 Sharing this avoids two identical BFS traversals per push: the first to
284 gather object IDs for client-side deduplication, and the second to load
285 blobs and assemble the pack mpack.
286
287 ``missing_snapshots`` is populated by :func:`walk_commits` with the
288 snapshot_ids of any reachable commit whose snapshot file is absent from
289 the local store. Callers should surface this to the user before pushing —
290 a pack that contains a commit but not its snapshot creates a dangling
291 reference on the remote.
292 """
293
294 commits: list[CommitRecord]
295 snapshot_ids: set[str]
296 all_blob_ids: list[str] # sorted, deduplicated — blobs_to_send = manifest_blobs - have_blobs
297 have_blobs: set[str] # blob IDs reachable from any have-commit's snapshot
298 manifest_blobs: set[str] # blob IDs referenced by new commits' manifests (old and new alike)
299 oid_to_path: dict[str, str] # blob_id → repository path (from snapshot manifests)
300 missing_snapshots: set[str] # snapshot_ids present in commits but absent on disk
301 snapshot_deltas: list[SnapshotDeltaDict] # pre-computed, reuse in build_mpack_from_walk
302
303 def walk_commits(
304 repo_root: pathlib.Path,
305 commit_ids: list[str],
306 *,
307 have: list[str] | None = None,
308 ) -> _WalkResult:
309 """BFS-walk the commit graph from *commit_ids*, stopping at *have*.
310
311 Returns a :class:`_WalkResult` that can be passed to both
312 :func:`collect_blob_ids_from_walk` and :func:`build_mpack_from_walk`
313 to avoid repeating the traversal.
314
315 This is the **single source of truth** for what goes into a push mpack.
316 Callers that need both the object ID list and the full pack should call
317 this once and pass the result to both downstream functions.
318
319 Uses ``prune=lambda cid: cid in have_set`` so the walk terminates the
320 moment it reaches a commit the server already has — no ancestor subgraph
321 is expanded beyond the boundary.
322 """
323 have_set: set[str] = set(have or [])
324 commits_to_send: list[CommitRecord] = list(
325 iter_ancestors(repo_root, commit_ids, prune=lambda cid: cid in have_set)
326 )
327
328 # Collect blobs already on the remote (have-commits' snapshots).
329 # Subtracting these gives us only genuinely new blobs to send.
330 have_blobs: set[str] = set()
331 for cid in have_set:
332 have_commit = read_commit(repo_root, cid)
333 if have_commit is not None:
334 have_snap = read_snapshot(repo_root, have_commit.snapshot_id)
335 if have_snap is not None:
336 have_blobs.update(have_snap.manifest.values())
337
338 snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_send}
339 commits_oldest_first = list(reversed(commits_to_send))
340
341 # One pass: compute deltas (one read_snapshot per commit) then derive
342 # object IDs from delta_upsert — no separate manifest scan needed.
343 missing_snapshots: set[str] = set()
344 try:
345 snapshot_deltas = _build_snapshot_deltas(repo_root, commits_oldest_first)
346 except ValueError as exc:
347 # Missing snapshot — extract sid and record it.
348 missing_snapshots = {
349 sid for sid in snapshot_ids
350 if read_snapshot(repo_root, sid) is None
351 }
352 snapshot_deltas = []
353
354 manifest_blobs: set[str] = set(collect_blob_ids_from_deltas(snapshot_deltas))
355
356 # Build oid→path from delta_upsert entries (path → oid in each delta).
357 oid_to_path: dict[str, str] = {}
358 for delta in snapshot_deltas:
359 for path, oid in (delta.get("delta_upsert") or {}).items():
360 oid_to_path[oid] = path
361
362 blobs_to_send: set[str] = manifest_blobs - have_blobs
363
364 if missing_snapshots:
365 for sid in sorted(missing_snapshots):
366 logger.warning(
367 "⚠️ walk_commits: snapshot %s is missing from the local store — "
368 "the commit(s) referencing it will be excluded from the pack. "
369 "Run `muse verify` to audit store integrity.",
370 sid,
371 )
372
373 return _WalkResult(
374 commits=commits_to_send,
375 snapshot_ids=snapshot_ids,
376 all_blob_ids=sorted(blobs_to_send),
377 have_blobs=have_blobs,
378 manifest_blobs=manifest_blobs,
379 oid_to_path=oid_to_path,
380 missing_snapshots=missing_snapshots,
381 snapshot_deltas=snapshot_deltas,
382 )
383
384 def stream_blob_chunks(
385 repo_root: pathlib.Path,
386 blob_ids: list[str],
387 chunk_size: int,
388 ) -> "collections.abc.Iterator[list[BlobPayload]]":
389 """Yield blobs in chunks of *chunk_size* as they are read from disk.
390
391 This is the hot path for ``muse push`` — reads one chunk at a time to
392 avoid loading all blobs into RAM at once (peak RAM = one chunk, not the
393 full set). The caller can start the first upload while the second chunk
394 is still being assembled — reducing both peak memory and time-to-first-upload.
395
396 Missing blobs are logged and skipped, consistent with :func:`build_mpack`.
397 """
398 import collections.abc # local to avoid circular at module level
399
400 chunk: list[BlobPayload] = []
401 for oid in blob_ids:
402 raw = read_object(repo_root, oid)
403 if raw is None:
404 logger.warning("⚠️ stream_blob_chunks: blob %s absent — skipping", oid)
405 continue
406 chunk.append(BlobPayload(object_id=oid, content=raw))
407 if len(chunk) >= chunk_size:
408 yield chunk
409 chunk = []
410 if chunk:
411 yield chunk
412
413 def collect_blob_ids_from_walk(walk: _WalkResult) -> list[str]:
414 """Return the sorted blob ID list from a pre-computed :func:`walk_commits` result.
415
416 Zero disk I/O — the walk already read all snapshots.
417 """
418 return walk["all_blob_ids"]
419
420
421 def collect_blob_ids_from_deltas(deltas: list[SnapshotDeltaDict]) -> list[str]:
422 """Return all unique object IDs from a pre-computed delta list.
423
424 Zero additional disk I/O — extracts oids from delta_upsert.values() only.
425 The first delta encodes the full base manifest; subsequent deltas encode
426 only changed files. Their union is identical to the union of all full
427 manifests (proof: every oid ever introduced appears in exactly one
428 delta_upsert entry).
429
430 Use this instead of collect_blob_ids on the mpack path — the deltas
431 are already computed by _build_snapshot_deltas (one read per snapshot),
432 so this is a pure in-memory operation.
433 """
434 seen: set[str] = set()
435 for delta in deltas:
436 seen.update(delta.get("delta_upsert", {}).values())
437 return sorted(seen)
438
439
440 def _build_snapshot_deltas(
441 repo_root: pathlib.Path,
442 commits_oldest_first: list[CommitRecord],
443 ) -> list[SnapshotDeltaDict]:
444 """Compute delta-encoded snapshots from a commit chain, oldest first.
445
446 Each entry carries only the paths that changed relative to the previous
447 snapshot in the chain. The first entry (no parent in this mpack) uses
448 ``parent_snapshot_id=None`` and encodes the full manifest as ``delta_upsert``.
449
450 Correctness invariant (content-addressing as proof)::
451
452 manifest = apply_delta(prev_manifest, entry)
453 assert hash_snapshot(manifest) == entry["snapshot_id"]
454
455 This replaces per-snapshot full-manifest storage with O(changed_files)
456 deltas — a 10–100× reduction for typical commit chains.
457 """
458 deltas: list[SnapshotDeltaDict] = []
459 seen_sids: set[str] = set()
460 prev_manifest: dict[str, str] = {}
461 prev_sid: str | None = None
462
463 for commit in commits_oldest_first:
464 sid = commit.snapshot_id
465 if sid in seen_sids:
466 continue
467 seen_sids.add(sid)
468
469 snap = read_snapshot(repo_root, sid)
470 if snap is None:
471 raise ValueError(
472 f"Push aborted: snapshot {sid} is missing from the local store "
473 f"but is required by a commit being sent. "
474 f"Run 'muse verify' to audit store integrity."
475 )
476 manifest = snap.manifest
477
478 delta_upsert = {k: v for k, v in manifest.items() if prev_manifest.get(k) != v}
479 delta_remove = [k for k in prev_manifest if k not in manifest]
480
481 # Only include directories in the wire entry if the stored snapshot_id
482 # was computed WITH them. Some snapshots were written before directories
483 # were part of hash_snapshot; their snap.directories field is populated
484 # but the ID was hashed with None. Sending those dirs would cause
485 # hash mismatch on the client side.
486 dirs_for_wire = snap.directories or []
487 if dirs_for_wire and hash_snapshot(manifest, dirs_for_wire) != sid:
488 dirs_for_wire = []
489
490 logger.debug(
491 "[build_snapshot_deltas] sid=%s parent=%s manifest=%d dirs_stored=%d dirs_wire=%d upsert=%d remove=%d",
492 sid[:20], (prev_sid or "none")[:20],
493 len(manifest), len(snap.directories or []), len(dirs_for_wire),
494 len(delta_upsert), len(delta_remove),
495 )
496
497 deltas.append(SnapshotDeltaDict(
498 snapshot_id=sid,
499 parent_snapshot_id=prev_sid,
500 delta_upsert=delta_upsert,
501 delta_remove=delta_remove,
502 directories=dirs_for_wire,
503 ))
504 prev_manifest = manifest
505 prev_sid = sid
506
507 return deltas
508
509
510 def _apply_snapshot_deltas(
511 raw_snapshots: list[SnapshotDeltaDict],
512 ) -> _SnapshotResolvedMap:
513 """Reconstruct full manifests from a delta chain.
514
515 Applies each delta in order and verifies the result by hashing.
516 The hash check IS the integrity proof — no external validation needed.
517
518 Two snapshot formats are accepted:
519 - Delta format (build_mpack): {snapshot_id, parent_snapshot_id,
520 delta_upsert, delta_remove}
521 - Full-manifest format: {snapshot_id, manifest, directories, ...}
522
523 Corrupt or hash-mismatched entries are logged and skipped; they do not
524 block independent valid entries with parent_snapshot_id=None. Dependent
525 entries whose parent was skipped are also skipped (base = {} → hash
526 mismatch → skip).
527
528 Returns {snapshot_id: (full_manifest, directories)} for every valid entry.
529 """
530 resolved: _SnapshotResolvedMap = {}
531 for snap in raw_snapshots:
532 sid = snap.get("snapshot_id", "")
533 if not sid:
534 continue
535
536 # Two formats arrive here:
537 # 1. Delta format (build_mpack): {snapshot_id, parent_snapshot_id,
538 # delta_upsert, delta_remove, directories} — reconstruct from parent + diff.
539 # 2. Full-manifest format (server sends cached manifest directly):
540 # {snapshot_id, manifest, directories, ...} — use manifest directly.
541 directories: list[str] = []
542 manifest_raw = snap.get("manifest")
543 if isinstance(manifest_raw, dict):
544 base = {k: v for k, v in manifest_raw.items()
545 if isinstance(k, str) and isinstance(v, str)}
546 dirs_raw = snap.get("directories")
547 if isinstance(dirs_raw, list):
548 directories = [d for d in dirs_raw if isinstance(d, str)]
549 logger.debug(
550 "[_apply_snapshot_deltas] snap=%s format=full manifest=%d dirs=%d",
551 sid[:20], len(base), len(directories),
552 )
553 else:
554 parent_sid = snap.get("parent_snapshot_id")
555 delta_upsert: dict[str, str] = snap.get("delta_upsert") or {}
556 delta_remove: list[str] = snap.get("delta_remove") or []
557 # BUG FIX: directories must be read in the delta branch too — they
558 # are included in the hash and were previously silently dropped,
559 # causing hash mismatch for every snapshot with non-empty directories.
560 dirs_raw = snap.get("directories")
561 if isinstance(dirs_raw, list):
562 directories = [d for d in dirs_raw if isinstance(d, str)]
563 parent_entry = resolved.get(parent_sid) if parent_sid else None
564 base = dict(parent_entry[0]) if parent_entry else {}
565 base.update(delta_upsert)
566 for path in delta_remove:
567 base.pop(path, None)
568 logger.debug(
569 "[_apply_snapshot_deltas] snap=%s format=delta parent=%s "
570 "upsert=%d remove=%d dirs=%d parent_resolved=%s",
571 sid[:20], (parent_sid or "")[:20],
572 len(delta_upsert), len(delta_remove), len(directories),
573 parent_entry is not None,
574 )
575
576 # Content-addressing IS the proof — hash the result.
577 try:
578 got = hash_snapshot(base, directories or None)
579 except ValueError as _hash_exc:
580 logger.warning(
581 "⚠️ apply_mpack: snapshot %s has invalid object IDs in delta — skipped: %s",
582 sid[:20], _hash_exc,
583 )
584 continue
585 got_nodirs = hash_snapshot(base, None) if directories else got
586 logger.debug(
587 "[apply_snapshot_deltas] sid=%s parent=%s manifest=%d dirs=%d "
588 "got=%s match=%s got_nodirs=%s nodirs_match=%s",
589 sid[:20], (snap.get("parent_snapshot_id") or "none")[:20],
590 len(base), len(directories), got[:20], got == sid,
591 got_nodirs[:20], got_nodirs == sid,
592 )
593 if got != sid:
594 logger.warning(
595 "⚠️ apply_mpack: snapshot %s hash mismatch "
596 "(reconstructed=%s) dirs=%s — skipped",
597 sid[:20], got[:20], directories,
598 )
599 continue
600 resolved[sid] = (base, directories)
601 return resolved
602
603
604 def build_mpack_from_walk(
605 repo_root: pathlib.Path,
606 walk: _WalkResult,
607 *,
608 only_blobs: set[str] | None = None,
609 repo_id: str = "",
610 compress: bool = False,
611 ) -> MPack:
612 """Assemble an :class:`MPack` from a pre-computed :func:`walk_commits` result.
613
614 Avoids the second BFS traversal that :func:`build_mpack` would otherwise
615 perform. Only reads blob bytes for blobs in *only_blobs* (or all blobs
616 when *only_blobs* is ``None``).
617
618 When *compress* is ``True``, each blob is zstd-compressed (level 3).
619 Falls back to ``raw`` when zstd makes the blob larger (rare for binary data).
620 Uses the zstandard C extension — one call per blob, no Python loop.
621
622 Returns:
623 An :class:`MPack` ready for serialisation and transfer.
624 """
625 missing_snapshots: set[str] = walk.get("missing_snapshots") or set()
626 all_blob_ids: set[str] = set(walk["all_blob_ids"])
627
628 # Hard failure on any missing snapshot — silently skipping would push
629 # commits without their snapshots, creating dangling references on the
630 # remote that can never be healed without rewriting history.
631 if missing_snapshots:
632 sample = sorted(missing_snapshots)[:3]
633 sample_str = ", ".join(sample)
634 raise ValueError(
635 f"Push aborted: {len(missing_snapshots)} snapshot(s) are missing from "
636 f"the local store but are required by commits being sent "
637 f"({sample_str}{'…' if len(missing_snapshots) > 3 else ''}). "
638 f"Run 'muse verify' to audit store integrity."
639 )
640
641 commits_to_send = list(walk["commits"])
642 snapshot_deltas = walk["snapshot_deltas"]
643
644 candidate_blob_ids = (
645 all_blob_ids & only_blobs if only_blobs is not None else all_blob_ids
646 )
647
648 blob_payloads: list[BlobPayload] = []
649
650 if not compress:
651 for oid in sorted(candidate_blob_ids):
652 raw = read_object(repo_root, oid)
653 if raw is None:
654 logger.warning("⚠️ build_mpack_from_walk: blob %s absent — skipping", oid)
655 continue
656 blob_payloads.append(BlobPayload(object_id=oid, content=raw))
657 else:
658 import zstandard as _zstd
659 cctx = _zstd.ZstdCompressor(level=3)
660 for oid in sorted(candidate_blob_ids):
661 raw = read_object(repo_root, oid)
662 if raw is None:
663 logger.warning("⚠️ build_mpack_from_walk: blob %s absent — skipping", oid)
664 continue
665 compressed = cctx.compress(raw)
666 if len(compressed) < len(raw):
667 blob_payloads.append(BlobPayload(object_id=oid, content=compressed, encoding="zstd"))
668 else:
669 blob_payloads.append(BlobPayload(object_id=oid, content=raw))
670
671 sent_commit_ids = [c.commit_id for c in commits_to_send]
672 wire_tags = _tags_for_commits(repo_root, sent_commit_ids, repo_id) if repo_id else []
673
674 total_bytes = sum(len(o.get("content") or b"") for o in blob_payloads)
675 agent_ids = sorted({
676 c.to_dict().get("agent_id", "") for c in commits_to_send
677 if c.to_dict().get("agent_id")
678 })
679 summary = MPackSummary(
680 commits_count=len(commits_to_send),
681 blobs_count=len(blob_payloads),
682 blobs_bytes=total_bytes,
683 branches={},
684 agent_ids=agent_ids,
685 )
686 mpack: MPack = {
687 "commits": [c.to_dict() for c in commits_to_send],
688 "snapshots": snapshot_deltas,
689 "blobs": blob_payloads,
690 "summary": summary,
691 "meta": MPackMeta(
692 mode="full",
693 base_commits=[],
694 created_at=datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
695 ),
696 }
697 if wire_tags:
698 mpack["tags"] = wire_tags
699
700 logger.info(
701 "✅ Built MPack (from walk): %d commits, %d snapshots, %d blobs, %d tags",
702 len(commits_to_send),
703 len(snapshot_deltas),
704 len(blob_payloads),
705 len(wire_tags),
706 )
707 return mpack
708
709 def _tags_for_commits(
710 repo_root: pathlib.Path, commit_ids: list[str], repo_id: str
711 ) -> list[WireTag]:
712 """Return all tags attached to *commit_ids* as serialisable :class:`WireTag` dicts."""
713 seen_tag_ids: set[str] = set()
714 wire_tags: list[WireTag] = []
715 for cid in commit_ids:
716 for tag in get_tags_for_commit(repo_root, repo_id, cid):
717 if tag.tag_id not in seen_tag_ids:
718 seen_tag_ids.add(tag.tag_id)
719 wire_tags.append(WireTag(
720 tag_id=tag.tag_id,
721 repo_id=tag.repo_id,
722 commit_id=tag.commit_id,
723 tag=tag.tag,
724 created_at=tag.created_at.isoformat(),
725 ))
726 return wire_tags
727
728 def build_mpack(
729 repo_root: pathlib.Path,
730 commit_ids: list[str],
731 *,
732 have: list[str] | None = None,
733 only_blobs: set[str] | None = None,
734 repo_id: str = "",
735 ) -> MPack:
736 """Assemble an :class:`MPack` from *commit_ids*, excluding commits in *have*.
737
738 Performs a BFS walk of the commit graph from every ID in *commit_ids*,
739 stopping at any commit already in *have*. Collects all snapshot manifests
740 and blobs reachable from the selected commits.
741
742 Missing blobs or snapshots are logged and skipped — the caller decides
743 whether that constitutes an error.
744
745 Args:
746 repo_root: Root of the Muse repository.
747 commit_ids: Tip commit IDs to include (e.g. current branch HEAD).
748 have: Commit IDs already known to the receiver. The BFS stops
749 at these, reducing mpack size. Pass ``None`` or ``[]``
750 to send the full history.
751 only_blobs: When set, only include blobs whose IDs are in this set.
752 Pass the set of blobs missing from the remote so the
753 client only uploads what the remote actually needs.
754 repo_id: Repository content ID used to look up tags. When omitted,
755 tags are not included in the mpack.
756
757 Returns:
758 An :class:`MPack` ready for serialisation and transfer.
759 """
760 walk = walk_commits(repo_root, commit_ids, have=have)
761 if walk["missing_snapshots"]:
762 sample = sorted(walk["missing_snapshots"])[:3]
763 sample_str = ", ".join(sample)
764 raise ValueError(
765 f"Push aborted: {len(walk['missing_snapshots'])} snapshot(s) are missing from "
766 f"the local store but are required by commits being sent "
767 f"({sample_str}{'…' if len(walk['missing_snapshots']) > 3 else ''}). "
768 f"Run 'muse verify' to audit store integrity."
769 )
770
771 commits_to_send: list[CommitRecord] = list(walk["commits"])
772 snapshot_deltas = walk["snapshot_deltas"]
773 all_blob_ids: set[str] = set(walk["all_blob_ids"])
774
775 # When only_blobs is provided skip any blob the remote already has —
776 # only transmit the missing delta.
777 candidate_blob_ids = (
778 all_blob_ids & only_blobs if only_blobs is not None else all_blob_ids
779 )
780
781 promisor_remotes = load_promisor_remotes(repo_root)
782 blob_payloads: list[BlobPayload] = []
783 for oid in sorted(candidate_blob_ids):
784 raw = read_object(repo_root, oid)
785 if raw is None:
786 state = object_state(repo_root, oid, promisor_remotes)
787 if state == ObjectState.PROMISED:
788 logger.debug("build_mpack: blob %s is PROMISED — skipping", short_id(oid))
789 continue
790 raise ValueError(
791 f"Pack aborted: blob {oid} is missing from the local store "
792 f"and no promisor remote is configured. "
793 f"Run 'muse verify' to audit store integrity."
794 )
795 blob_payloads.append(BlobPayload(object_id=oid, content=raw))
796
797 sent_commit_ids = [c.commit_id for c in commits_to_send]
798 wire_tags = _tags_for_commits(repo_root, sent_commit_ids, repo_id) if repo_id else []
799
800 total_bytes = sum(len(b["content"]) for b in blob_payloads)
801 agent_ids = sorted({
802 c.to_dict().get("agent_id", "") for c in commits_to_send
803 if c.to_dict().get("agent_id")
804 })
805 summary = MPackSummary(
806 commits_count=len(commits_to_send),
807 blobs_count=len(blob_payloads),
808 blobs_bytes=total_bytes,
809 branches={},
810 agent_ids=agent_ids,
811 )
812 _have_list = list(have or [])
813 mpack: MPack = {
814 "commits": [c.to_dict() for c in commits_to_send],
815 "snapshots": snapshot_deltas,
816 "blobs": blob_payloads,
817 "summary": summary,
818 "meta": MPackMeta(
819 mode="incremental" if _have_list else "full",
820 base_commits=_have_list,
821 created_at=datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
822 ),
823 }
824 if wire_tags:
825 mpack["tags"] = wire_tags
826
827 logger.info(
828 "✅ Built MPack: %d commits, %d snapshots, %d blobs, %d tags",
829 len(commits_to_send),
830 len(snapshot_deltas),
831 len(blob_payloads),
832 len(wire_tags),
833 )
834 return mpack
835
836 # ---------------------------------------------------------------------------
837 # Object ID collection — for pre-push deduplication negotiation
838 # ---------------------------------------------------------------------------
839
840 def collect_blob_ids(
841 repo_root: pathlib.Path,
842 commit_ids: list[str],
843 *,
844 have: list[str] | None = None,
845 ) -> list[str]:
846 """Return all blob IDs reachable from *commit_ids*, excluding *have*.
847
848 Identical BFS walk to :func:`build_mpack` but without reading object bytes.
849 Used by ``muse push`` for client-side deduplication — the result is compared
850 against the remote's known objects, and :func:`build_mpack` is called with
851 ``only_blobs`` set to the missing subset. This avoids loading any blob
852 content until we know it is actually needed.
853
854 Uses ``prune=lambda cid: cid in have_set`` so the walk terminates the
855 moment it hits a server-known commit, without expanding its ancestor
856 subgraph.
857
858 Args:
859 repo_root: Root of the Muse repository.
860 commit_ids: Tip commit IDs to examine.
861 have: Commit IDs already known to the receiver (BFS stops here).
862
863 Returns:
864 Sorted list of object IDs reachable from the delta.
865 """
866 have_set: set[str] = set(have or [])
867 commits_to_examine: list[CommitRecord] = list(
868 iter_ancestors(repo_root, commit_ids, prune=lambda cid: cid in have_set)
869 )
870
871 # Collect blobs already on the remote (have-commits' snapshots).
872 have_blobs: set[str] = set()
873 for cid in have_set:
874 have_commit = read_commit(repo_root, cid)
875 if have_commit is not None:
876 have_snap = read_snapshot(repo_root, have_commit.snapshot_id)
877 if have_snap is not None:
878 have_blobs.update(have_snap.manifest.values())
879
880 snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_examine}
881 all_blob_ids: set[str] = set()
882 for sid in snapshot_ids:
883 snap = read_snapshot(repo_root, sid)
884 if snap is not None:
885 all_blob_ids.update(snap.manifest.values())
886
887 return sorted(all_blob_ids - have_blobs)
888
889
890 def compute_snapshot_delta(
891 base: _Manifest,
892 new: _Manifest,
893 ) -> tuple[_Manifest, list[str]]:
894 """Compute the delta between two snapshot manifests.
895
896 Returns (added_or_modified, removed):
897 - added_or_modified: paths whose object_id changed or are new in *new*
898 - removed: paths present in *base* but absent from *new*
899 """
900 added = {p: h for p, h in new.items() if base.get(p) != h}
901 removed = [p for p in base if p not in new]
902 return added, removed
903
904
905 def apply_snapshot_delta(
906 base: _Manifest,
907 added: _Manifest,
908 removed: list[str],
909 ) -> _Manifest:
910 """Reconstruct a full manifest by applying a delta to a base manifest.
911
912 Inverse of compute_snapshot_delta:
913 apply_snapshot_delta(base, *compute_snapshot_delta(base, new)) == new
914 """
915 manifest = dict(base)
916 manifest.update(added)
917 for path in removed:
918 manifest.pop(path, None)
919 return manifest
920
921
922 # ---------------------------------------------------------------------------
923 # Presign helpers
924 # ---------------------------------------------------------------------------
925
926 class _PresignPayload(TypedDict):
927 mpack_key: str
928 size_bytes: int
929
930
931 class _UnpackPayload(TypedDict, total=False):
932 mpack_key: str
933 branch: str
934 head: str
935 commits_count: int
936 blobs_count: int
937 force: bool
938
939
940 def build_presign_payload(mpack_bytes: bytes) -> _PresignPayload:
941 """Return the request body for POST /push/mpack-presign.
942
943 The server uses mpack_key to name the MinIO object and to verify integrity
944 after the client PUTs the bytes directly to MinIO.
945 """
946 return {"mpack_key": blob_id(mpack_bytes), "size_bytes": len(mpack_bytes)}
947
948
949 def build_unpack_payload(
950 mpack_key: str,
951 *,
952 branch: str = "main",
953 head: str = "",
954 commits_count: int = 0,
955 blobs_count: int = 0,
956 force: bool = False,
957 ) -> _UnpackPayload:
958 """Return the request body for POST /push/unpack-mpack (Step 3)."""
959 return {
960 "mpack_key": mpack_key,
961 "branch": branch,
962 "head": head,
963 "commits_count": int(commits_count),
964 "blobs_count": int(blobs_count),
965 "force": force,
966 }
967
968
969 # ---------------------------------------------------------------------------
970 # Wire MPack encode / decode (Phase 3)
971 # ---------------------------------------------------------------------------
972 #
973 # Wire format:
974 # [4B] magic: b"MUSE"
975 # [1B] version: 1
976 # [1B] section_count: N
977 # [N*17B] section table: each entry is (1B type, 8B offset LE, 8B length LE)
978 # [...] section data (concatenated, no padding)
979 # [32B] SHA-256 of every byte above (footer, not included in its own hash)
980 #
981 # Section types:
982 # 1 = BLOBS — raw _build_pack() bytes (byte-identical to Phase 1 .mpack)
983 # 2 = COMMITS — [8B count] + N × [8B record_len + JSON bytes]
984 # 3 = SNAPSHOTS — same length-prefixed JSON layout as COMMITS
985 # 4 = TAGS — same layout
986 # 5 = META — [8B json_len + JSON bytes] key-value pairs (repo_id, branch, head_commit_id)
987
988 _WIRE_VERSION = 1
989 _WIRE_SEC_BLOBS = 1
990 _WIRE_SEC_COMMITS = 2
991 _WIRE_SEC_SNAPSHOTS = 3
992 _WIRE_SEC_TAGS = 4
993 _WIRE_SEC_META = 5
994
995
996 def _encode_records(records: list[dict]) -> bytes:
997 """Encode a list of dicts as [8B count] + N × [8B len + JSON bytes]."""
998 import json as _json
999 parts = [_struct.pack("<Q", len(records))]
1000 for rec in records:
1001 enc = _json.dumps(rec, separators=(",", ":")).encode()
1002 parts.append(_struct.pack("<Q", len(enc)))
1003 parts.append(enc)
1004 return b"".join(parts)
1005
1006
1007 def _decode_records(data: bytes) -> list[dict]:
1008 """Decode [8B count] + N × [8B len + JSON bytes] into a list of dicts."""
1009 if len(data) < 8:
1010 return []
1011 import json as _json
1012 count = _struct.unpack_from("<Q", data, 0)[0]
1013 cursor = 8
1014 records: list[dict] = []
1015 for _ in range(count):
1016 if cursor + 8 > len(data):
1017 break
1018 rec_len = _struct.unpack_from("<Q", data, cursor)[0]
1019 cursor += 8
1020 rec = _json.loads(data[cursor: cursor + rec_len])
1021 cursor += rec_len
1022 records.append(rec)
1023 return records
1024
1025
1026 def _encode_meta(meta: _MetaDict) -> bytes:
1027 """Encode a META dict as [8B count] + N × [8B key_len + key + 8B val_len + val]."""
1028 import json as _json
1029 enc = _json.dumps(meta, separators=(",", ":")).encode()
1030 return _struct.pack("<Q", len(enc)) + enc
1031
1032
1033 def _decode_meta(data: bytes) -> _MetaDict:
1034 """Decode a META section back to a dict."""
1035 if len(data) < 8:
1036 return {}
1037 import json as _json
1038 val_len = _struct.unpack_from("<Q", data, 0)[0]
1039 return _json.loads(data[8: 8 + val_len])
1040
1041
1042 def build_wire_mpack(mpack: MPack, *, meta: _MetaDict | None = None) -> bytes:
1043 """Encode an :class:`MPack` as a wire MPack binary bundle.
1044
1045 The OBJECTS section bytes are byte-identical to what :func:`write_pack`
1046 writes to disk in Phase 1, so the client can extract the section and
1047 write it directly without any per-object decode step.
1048
1049 Returns:
1050 Raw bytes of the wire bundle, starting with ``b"MUSE"`` and ending
1051 with a 32-byte SHA-256 footer.
1052 """
1053 from muse.core.pack_store import _build_pack as _ps_build_pack
1054
1055 blobs = mpack.get("blobs") or []
1056 commits = mpack.get("commits") or []
1057 snapshots = mpack.get("snapshots") or []
1058 tags = mpack.get("tags") or []
1059
1060 def _content(blob: BlobPayload) -> bytes:
1061 raw = blob["content"]
1062 if blob.get("encoding") == "zstd":
1063 import zstandard as _zstd
1064 return _zstd.ZstdDecompressor().decompress(raw)
1065 return raw
1066
1067 obj_pairs = [(o["object_id"], _content(o)) for o in blobs]
1068 blobs_bytes = _ps_build_pack(obj_pairs) if obj_pairs else b""
1069 commits_bytes = _encode_records(commits)
1070 snapshots_bytes = _encode_records(snapshots)
1071 tags_bytes = _encode_records(tags)
1072 meta_bytes = _encode_meta(meta) if meta else b""
1073
1074 sections = [
1075 (_WIRE_SEC_BLOBS, blobs_bytes),
1076 (_WIRE_SEC_COMMITS, commits_bytes),
1077 (_WIRE_SEC_SNAPSHOTS, snapshots_bytes),
1078 (_WIRE_SEC_TAGS, tags_bytes),
1079 (_WIRE_SEC_META, meta_bytes),
1080 ]
1081 section_count = len(sections)
1082 # Header: 4B magic + 1B version + 1B section_count = 6B
1083 # Table: section_count × (1B type + 8B offset + 8B length) = section_count × 17B
1084 header_size = 6 + section_count * 17
1085
1086 offset = header_size
1087 table_entries: list[tuple[int, int, int]] = []
1088 for sec_type, sec_data in sections:
1089 table_entries.append((sec_type, offset, len(sec_data)))
1090 offset += len(sec_data)
1091
1092 h = _hashlib.sha256()
1093 parts: list[bytes] = []
1094
1095 def _emit(chunk: bytes) -> None:
1096 h.update(chunk)
1097 parts.append(chunk)
1098
1099 _emit(b"MUSE")
1100 _emit(_struct.pack("<BB", _WIRE_VERSION, section_count))
1101 for sec_type, sec_offset, sec_length in table_entries:
1102 _emit(_struct.pack("<BQQ", sec_type, sec_offset, sec_length))
1103 for _, sec_data in sections:
1104 _emit(sec_data)
1105 parts.append(h.digest()) # footer — not fed back into h
1106 return b"".join(parts)
1107
1108
1109 def parse_wire_mpack(data: bytes) -> MPack:
1110 """Parse a wire MPack binary bundle back into an :class:`MPack` dict.
1111
1112 Verifies the SHA-256 footer before parsing any section.
1113
1114 Raises:
1115 ValueError: Bad magic bytes or unknown version.
1116 OSError: Footer integrity check failed.
1117 """
1118 from muse.core.pack_store import _parse_pack_bytes as _ps_parse
1119
1120 if len(data) < 38: # 4+1+1 header + 1×17 section entry + 32 footer (minimum)
1121 raise ValueError("Wire MPack too short")
1122 if data[:4] != b"MUSE":
1123 raise ValueError(f"Wire MPack bad magic: {data[:4]!r}")
1124 version = data[4]
1125 if version != _WIRE_VERSION:
1126 raise ValueError(f"Wire MPack unknown version: {version}")
1127
1128 body = data[:-32]
1129 stored = data[-32:]
1130 if _hashlib.sha256(body).digest() != stored:
1131 raise OSError("Wire MPack failed SHA-256 integrity check")
1132
1133 section_count = data[5]
1134 cursor = 6
1135 sections: dict[int, bytes] = {}
1136 for _ in range(section_count):
1137 sec_type = data[cursor]
1138 sec_offset, sec_length = _struct.unpack_from("<QQ", data, cursor + 1)
1139 cursor += 17
1140 sections[sec_type] = data[sec_offset: sec_offset + sec_length]
1141
1142 obj_pairs = _ps_parse(sections.get(_WIRE_SEC_BLOBS, b""))
1143 blobs: list[BlobPayload] = [
1144 BlobPayload(object_id=oid, content=content)
1145 for oid, content in obj_pairs
1146 ]
1147 commits = _decode_records(sections.get(_WIRE_SEC_COMMITS, b""))
1148 snapshots = _decode_records(sections.get(_WIRE_SEC_SNAPSHOTS, b""))
1149 tags = _decode_records(sections.get(_WIRE_SEC_TAGS, b""))
1150 meta = _decode_meta(sections.get(_WIRE_SEC_META, b""))
1151
1152 result = MPack(
1153 blobs=blobs,
1154 commits=commits,
1155 snapshots=snapshots,
1156 tags=tags,
1157 )
1158 if meta:
1159 result["meta"] = meta
1160 return result
1161
1162
1163 # ---------------------------------------------------------------------------
1164 # Pack applying
1165 # ---------------------------------------------------------------------------
1166
1167 def apply_mpack(
1168 repo_root: pathlib.Path,
1169 mpack: MPack,
1170 *,
1171 shallow_commits: "collections.abc.Container[str] | None" = None,
1172 ) -> ApplyResult:
1173 """Write the contents of *mpack* into a local ``.muse/`` directory.
1174
1175 Writes in dependency order: objects first (blobs), then snapshots (which
1176 reference object IDs), then commits (which reference snapshot IDs). All
1177 writes are idempotent — already-present items are silently skipped.
1178
1179 Args:
1180 repo_root: Root of the Muse repository to write into.
1181 mpack: :class:`MPack` received from the remote.
1182 shallow_commits: Optional set of boundary commit IDs (shallow clone).
1183 These commits are written even if their parents are
1184 absent from the local store.
1185
1186 Returns:
1187 :class:`ApplyResult` with counts of newly written and skipped items.
1188 """
1189 import sys as _sys
1190 import time as _time
1191 blobs_written = 0
1192 blobs_skipped = 0
1193 snapshots_written = 0
1194 commits_written = 0
1195 tags_written = 0
1196
1197 raw_blobs = mpack.get("blobs") or []
1198 raw_snapshots = mpack.get("snapshots") or []
1199 raw_commits = mpack.get("commits") or []
1200 _t_apply_start = _time.monotonic()
1201
1202 # Pack-bomb guard: cap the total number of items accepted per call.
1203 # A legitimate push carries at most tens of thousands of blobs per chunk;
1204 # a pack claiming millions is an adversarial input.
1205 total_items = len(raw_blobs) + len(raw_snapshots) + len(raw_commits)
1206 if total_items > MAX_PACK_OBJECTS:
1207 raise ValueError(
1208 f"Pack rejected: {total_items:,} total items exceeds the "
1209 f"{MAX_PACK_OBJECTS:,} item limit per apply_mpack call. "
1210 "Split the pack into smaller chunks."
1211 )
1212
1213 # Deduplicate blob IDs before the write loop. A malicious or buggy
1214 # sender may repeat the same blob ID N times, forcing N sha256 hashes
1215 # before the "already exists" short-circuit returns False.
1216 seen_blob_ids: set[str] = set()
1217 # Track blob IDs that failed validation so dependent snapshots and commits
1218 # can be skipped — prevents dangling reference chains in the store.
1219 failed_blob_ids: set[str] = set()
1220 # Blobs that pass all checks and are new to this store — written as a pack.
1221 pack_blobs: list[tuple[str, bytes]] = []
1222
1223 for obj in raw_blobs:
1224 oid = obj.get("object_id", "")
1225 raw = obj.get("content", b"")
1226 if not oid or not isinstance(raw, bytes):
1227 logger.warning("⚠️ apply_mpack: blob entry missing fields — skipped")
1228 continue
1229 if oid in seen_blob_ids:
1230 logger.debug("⚠️ apply_mpack: duplicate blob_id %s — skipped", short_id(oid))
1231 blobs_skipped += 1
1232 continue
1233 seen_blob_ids.add(oid)
1234 if len(raw) > MAX_OBJECT_WRITE_BYTES:
1235 logger.warning(
1236 "⚠️ apply_mpack: blob %s is %d bytes, exceeding %d MiB limit — skipped",
1237 oid, len(raw), MAX_OBJECT_WRITE_BYTES // (1024 * 1024),
1238 )
1239 failed_blob_ids.add(oid)
1240 continue
1241 try:
1242 validate_object_id(oid)
1243 actual = blob_id(raw)
1244 if actual != oid:
1245 raise ValueError(
1246 f"Content integrity failure: expected {oid} got {actual}"
1247 )
1248 except ValueError as exc:
1249 logger.warning("⚠️ apply_mpack: malformed blob entry — skipped: %s", exc)
1250 failed_blob_ids.add(oid)
1251 continue
1252 if has_object(repo_root, oid):
1253 blobs_skipped += 1
1254 continue
1255 pack_blobs.append((oid, raw))
1256 blobs_written += 1
1257
1258 # Write all new blobs as a single MPack file + index — O(1) file writes
1259 # regardless of blob count. Raises OSError on disk failure, which
1260 # propagates before any snapshot or commit is written.
1261 write_pack(repo_root, pack_blobs)
1262
1263 # Reconstruct full manifests from the delta chain, then write snapshots.
1264 # The hash IS the proof: hash_snapshot(reconstructed) == snapshot_id.
1265 # _apply_snapshot_deltas logs and skips corrupt entries — never raises.
1266 resolved_manifests = _apply_snapshot_deltas(raw_snapshots)
1267
1268 # Track snapshot IDs skipped due to referencing failed objects so dependent
1269 # commits can be skipped — prevents dangling commit → snapshot → (missing object).
1270 skipped_snapshot_ids: set[str] = set()
1271 # Track snapshot IDs that were successfully written in this mpack.
1272 # Used below to refuse commits whose snapshots were not sent (snaps=0 bug guard).
1273 written_snapshot_ids: set[str] = set()
1274
1275 for snap_entry in raw_snapshots:
1276 sid = snap_entry.get("snapshot_id", "")
1277 if not sid or sid not in resolved_manifests:
1278 continue
1279 try:
1280 manifest, directories = resolved_manifests[sid]
1281 # Guard against zip-slip: manifest keys are stored as-is and later
1282 # used to construct checkout paths. A malicious mpack could inject
1283 # keys like "../../etc/cron.d/malicious". We validate all keys here —
1284 # before writing — so no traversal path ever enters the store.
1285 for key in manifest:
1286 validate_workspace_path(key)
1287 # Manifest values are object IDs — validate they are safe hex strings.
1288 for oid in manifest.values():
1289 validate_object_id(oid)
1290 # Refuse to write a snapshot whose objects were not fully written —
1291 # that would create a dangling snapshot → object reference.
1292 missing = failed_blob_ids.intersection(manifest.values())
1293 if missing:
1294 logger.warning(
1295 "⚠️ apply_mpack: snapshot %s skipped — references %d object(s) "
1296 "that failed to write",
1297 short_id(sid), len(missing),
1298 )
1299 skipped_snapshot_ids.add(sid)
1300 continue
1301 snap = SnapshotRecord(snapshot_id=sid, manifest=manifest, directories=directories)
1302 is_new = read_snapshot(repo_root, snap.snapshot_id) is None
1303 write_snapshot(repo_root, snap, sync=False)
1304 written_snapshot_ids.add(sid)
1305 if is_new:
1306 snapshots_written += 1
1307 except (KeyError, ValueError) as exc:
1308 logger.warning("⚠️ apply_mpack: malformed snapshot — skipped: %s", exc)
1309
1310 # Parse all commits first so per-commit validation errors are counted
1311 # before any writes begin.
1312 parsed_commits: list[CommitRecord] = []
1313 for commit_dict in raw_commits:
1314 try:
1315 commit = CommitRecord.from_dict(commit_dict)
1316 if not commit.commit_id:
1317 logger.warning("⚠️ apply_mpack: commit missing commit_id — skipped")
1318 continue
1319 if not commit.snapshot_id:
1320 logger.warning(
1321 "⚠️ apply_mpack: commit %s missing snapshot_id — skipped",
1322 commit.commit_id,
1323 )
1324 continue
1325 # Refuse to write a commit whose snapshot was skipped due to failed
1326 # objects — that would create a dangling commit → snapshot reference.
1327 if commit.snapshot_id in skipped_snapshot_ids:
1328 logger.warning(
1329 "⚠️ apply_mpack: commit %s skipped — its snapshot %s was not written "
1330 "(referenced object(s) failed)",
1331 short_id(commit.commit_id), short_id(commit.snapshot_id),
1332 )
1333 continue
1334 # Refuse to write a commit whose snapshot was not included in this
1335 # mpack AND is not already in the local store. Writing such a commit
1336 # corrupts the store: the commit is then present in `have` on the
1337 # next pull, the server returns nothing new, and pull aborts forever
1338 # with "snapshot missing". The correct behaviour is to skip the commit
1339 # so the next pull re-requests it together with its snapshot.
1340 if (
1341 commit.snapshot_id not in written_snapshot_ids
1342 and read_snapshot(repo_root, commit.snapshot_id) is None
1343 ):
1344 logger.warning(
1345 "⚠️ apply_mpack: commit %s skipped — snapshot %s was not in this "
1346 "mpack and is not in the local store (server sent snaps=0). "
1347 "The next pull will re-request this commit with its snapshot.",
1348 short_id(commit.commit_id), short_id(commit.snapshot_id),
1349 )
1350 continue
1351 parsed_commits.append(commit)
1352 except (KeyError, ValueError, TypeError) as exc:
1353 logger.warning("⚠️ apply_mpack: malformed commit — skipped: %s", exc)
1354
1355 # Write commits in dependency order: retry until stable.
1356 # Bundles may arrive with commits in BFS order (newest-first). Phase 2's
1357 # parent-existence guard rejects a commit whose parent hasn't been written
1358 # yet. Keep cycling through MissingParentError commits until every parent
1359 # in the mpack has been written, or until a full pass produces no progress
1360 # (the missing parent isn't in this mpack — log and skip).
1361 _shallow: "collections.abc.Container[str]" = shallow_commits or ()
1362 pending = parsed_commits
1363 while pending:
1364 deferred: list[CommitRecord] = []
1365 for commit in pending:
1366 # Shallow boundary commits have their parent check bypassed so they
1367 # can be written even when their parents are not in the local store.
1368 is_shallow = commit.commit_id in _shallow
1369 try:
1370 is_new = not has_object(repo_root, commit.commit_id)
1371 write_commit(repo_root, commit, skip_parent_check=is_shallow, sync=False)
1372 if is_new:
1373 commits_written += 1
1374 except MissingParentError:
1375 deferred.append(commit)
1376 except OSError as exc:
1377 logger.critical(
1378 "❌ apply_mpack: store integrity violation for commit — skipped: %s", exc
1379 )
1380 except (ValueError, TypeError) as exc:
1381 logger.warning("⚠️ apply_mpack: malformed commit — skipped: %s", exc)
1382 if len(deferred) == len(pending):
1383 # No progress in this pass — the missing parents are not in the mpack.
1384 for commit in deferred:
1385 logger.warning(
1386 "⚠️ apply_mpack: commit %s skipped — parent not in mpack or local store",
1387 commit.commit_id,
1388 )
1389 break
1390 pending = deferred
1391
1392 # Bulk-fsync: all snapshot and commit files were written with sync=False
1393 # (no per-file fsync) for throughput. A single directory fsync here makes
1394 # all preceding renames durable in one barrier — the same strategy git uses
1395 # for pack writes. The ref is not updated until after apply_mpack returns,
1396 # so a crash before this fsync leaves the store in a consistent (if
1397 # incomplete) state that is safe to re-apply.
1398 if snapshots_written > 0 or commits_written > 0:
1399 from muse.core.paths import muse_dir as _muse_dir
1400 _dot = _muse_dir(repo_root)
1401 for _dir in (_dot / "snapshots", _dot / "commits"):
1402 if _dir.exists():
1403 try:
1404 _fd = os.open(str(_dir), os.O_RDONLY)
1405 try:
1406 os.fsync(_fd)
1407 finally:
1408 os.close(_fd)
1409 except OSError:
1410 pass # best-effort — some filesystems (tmpfs) reject dir fsync
1411
1412 for wire_tag in mpack.get("tags") or []:
1413 try:
1414 tag_record = TagRecord.from_dict(TagDict(
1415 tag_id=wire_tag["tag_id"],
1416 repo_id=wire_tag["repo_id"],
1417 commit_id=wire_tag["commit_id"],
1418 tag=wire_tag["tag"],
1419 created_at=wire_tag["created_at"],
1420 ))
1421 write_tag(repo_root, tag_record)
1422 tags_written += 1
1423 except (KeyError, ValueError) as exc:
1424 logger.warning("⚠️ apply_mpack: malformed tag — skipped: %s", exc)
1425
1426 logger.info(
1427 "✅ Applied pack: %d new blobs, %d new snapshots, %d new commits, %d tags (%d blobs skipped)",
1428 blobs_written,
1429 snapshots_written,
1430 commits_written,
1431 tags_written,
1432 blobs_skipped,
1433 )
1434 return ApplyResult(
1435 commits_written=commits_written,
1436 snapshots_written=snapshots_written,
1437 blobs_written=blobs_written,
1438 blobs_skipped=blobs_skipped,
1439 tags_written=tags_written,
1440 failed_blobs=list(failed_blob_ids),
1441 skipped_snapshots=list(skipped_snapshot_ids),
1442 )
File History 15 commits
sha256:99fba7e18ebcba8196d02fdc1796a75641959140b208caa4c64657de5fce3302 fix: downgrade verbose per-snapshot logs to debug in mpack.py Sonnet 4.6 34 days ago
sha256:c177242a2c8988746ba00466fc642f667f1fafaf86cb082ff40926ceb98d09c9 fix: read directories in delta branch of _apply_snapshot_deltas Sonnet 4.6 34 days ago
sha256:889892bcd79c5b77cfb314c90ff6caaeb9f6575dc0cab81b85259d5bbb235971 type fix Human patch 42 days ago
sha256:4d09a52c06fbc389006963ad1e5ca6ee48c3cb72799f1a322561035b263db67d merge conflict resolve Human patch 42 days ago
sha256:b5ec4e4a3a73cae0cd08224f32090f2a4836afa0a804cb3231e70c42a3e89295 fix adapter for agent config Human patch 42 days ago
sha256:8e790785ceed39352c123388b8680542965afb7b2305a48827a9fbf4f3ae75fb fix: build_wire_mpack decompresses zstd-encoded blobs befor… Sonnet 4.6 patch 42 days ago
sha256:2c59968e5fd34f1740180d630338fddfb8c465b71e150a0965f11dbdcba5dec7 fix: apply_mpack refuses commits when their snapshot is abs… Sonnet 4.6 patch 42 days ago
sha256:31316ea2e76b244dea3dd67bd0e608d56279ab9fdd2d4ecacbf41060d91f323f merge dev → main: rc11, urllib migration, object store fixe… Sonnet 4.6 minor ⚠ 44 days ago
sha256:6a70ad4088ca3c07a6b1d85b257a735aa0927db2fc017ed046b982fe95805450 perf: remove debug print statements from mpack hot paths; f… Sonnet 4.6 patch 44 days ago
sha256:35855b7bbce81b93612c655e23587bdebbb5fc09856ff5cbf5cd5b195d0547d4 merge: dev → main (rc11, urllib migration, object store inv… Sonnet 4.6 patch 47 days ago
sha256:633dfa2940e97bf1a3d04996c772027a57d70d103f1693c96da04969613dba6c fix: urllib migration regressions — force flag, job_id, Con… Sonnet 4.6 minor ⚠ 47 days ago
sha256:8ca6bcf5ebccd1b36d8ec716358b2b634d55e75db19c1d5d339082664c773006 fix: include directories in snapshot deltas sent in push mpack Sonnet 4.6 48 days ago
sha256:85a465e59c29055c46f6a2a9efd5e2036f66857c638240b528e63b64b5d1d5af fix: topo-sort snapshot delta chain in apply_mpack; decompr… Sonnet 4.6 49 days ago
sha256:dbebcbb60d67ec6ee0f09918db3266576c7ca66e93381a69b67ba40fe74174f3 fix: decompress zstd blobs in build_wire_mpack — pack secti… Sonnet 4.6 patch 49 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 50 days ago