gabriel / muse public
pack.py python
863 lines 32.9 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
1 """Muse MPack format — bundle of commits, snapshots, and blobs for wire transfer.
2
3 An :class:`MPackBundle` 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:`ObjectPayload` entries (raw blob bytes)
10 - ``branch_heads`` mapping (branch name → commit ID, reflecting remote state)
11 - ``summary`` (:class:`MPackSummary`) — advisory counts for agent routing
12
13 :func:`build_mpack` collects all data reachable from a set of commit IDs and
14 populates the summary field.
15 :func:`apply_mpack` writes a bundle into a local ``.muse/`` directory.
16
17 MPack wire encoding
18 ---------------------
19 Object bytes are transmitted as msgpack frames over the streaming MWP endpoint
20 (``Content-Type: application/x-muse-mpack``). See :mod:`muse.core.mpack` for
21 the frame TypedDicts and stream writer/reader.
22
23 Agent contract
24 --------------
25
26 - ``exit_code`` 0: all data applied successfully.
27 - ``exit_code`` 1: validation error (malformed object ID, path traversal).
28 - ``exit_code`` 3: I/O error reading from local store.
29 - ``duration_ms``: wall-clock milliseconds for build or apply.
30 """
31
32 from __future__ import annotations
33
34 import collections
35 import datetime
36 import logging
37 import pathlib
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 read_object, write_object
43 from muse.core.validation import (
44 MAX_OBJECT_WRITE_BYTES,
45 MAX_PACK_OBJECTS,
46 validate_object_id,
47 validate_workspace_path,
48 )
49 from muse.core._types import BranchHeads, short_id
50 from muse.core.store import (
51 CommitDict,
52 CommitRecord,
53 SnapshotDict,
54 SnapshotRecord,
55 TagDict,
56 TagRecord,
57 get_all_tags,
58 get_tags_for_commit,
59 read_commit,
60 read_snapshot,
61 write_commit,
62 write_snapshot,
63 write_tag,
64 )
65
66 logger = logging.getLogger(__name__)
67
68
69 # ---------------------------------------------------------------------------
70 # Wire-format TypedDicts
71 # ---------------------------------------------------------------------------
72
73
74 class _ObjectPayloadBase(TypedDict):
75 """Required fields for every object payload transmitted over MWP."""
76
77 object_id: str
78 content: bytes
79
80
81 class ObjectPayload(_ObjectPayloadBase, total=False):
82 """A single content-addressed blob with encoding metadata for MWP transport.
83
84 Required fields (always present):
85 object_id: Content-addressed SHA-256 identifier (``sha256:<hex>``).
86 content: Raw or encoded bytes — see *encoding*.
87
88 Optional fields (omit for ``"raw"`` with no base):
89 path: Repository path of this object; used by the server to look
90 up delta base candidates for the next push.
91 encoding: ``"raw"`` (default) | ``"zlib"`` | ``"delta+zlib"``.
92 base_id: Base object ID for ``"delta+zlib"`` encoding.
93 sz: Uncompressed byte count of the target object. Required when
94 ``encoding`` is ``"delta+zlib"`` so the server can pre-allocate
95 before decompression. Ignored for ``"raw"`` payloads.
96 """
97
98 path: str
99 encoding: str
100 base_id: str
101 sz: int
102
103
104 class WireTag(TypedDict):
105 """A tag record serialised for wire transfer inside an :class:`MPackBundle`."""
106
107 tag_id: str
108 repo_id: str
109 commit_id: str
110 tag: str
111 created_at: str
112
113
114 class BundleMeta(TypedDict, total=False):
115 """Self-describing metadata embedded in every :class:`MPackBundle`.
116
117 Agents read this to understand the bundle's scope without inspecting
118 commits or objects.
119
120 Fields:
121 mode: ``"full"`` — all referenced objects must be in the bundle
122 or the local store. ``"incremental"`` — some objects are
123 expected to exist at the receiver's base (declared in
124 ``base_commits``); they are not included in this bundle.
125 base_commits: Commit IDs passed as ``--have`` when the bundle was
126 built. Empty for full bundles.
127 created_at: ISO 8601 UTC timestamp of when the bundle was assembled.
128 """
129
130 mode: str # "full" | "incremental"
131 base_commits: list[str] # sha256:-prefixed commit IDs
132 created_at: str # ISO 8601 — e.g. "2026-01-01T00:00:00Z"
133
134
135 class MPackSummary(TypedDict, total=False):
136 """Advisory summary embedded in every :class:`MPackBundle`.
137
138 Agents read this to make routing/accept/reject decisions before
139 touching commits or objects. All fields are advisory — receivers
140 must not rely on them for correctness, only for optimisation.
141
142 Fields:
143 commits_count: Number of commits in this bundle.
144 objects_count: Number of unique objects (blobs) in this bundle.
145 objects_bytes: Total uncompressed object bytes.
146 branches: Branch name → tip commit_id at build time.
147 agent_ids: All agent_id values from commits in this bundle.
148 """
149
150 commits_count: int
151 objects_count: int
152 objects_bytes: int
153 branches: BranchHeads # branch_name → commit_id
154 agent_ids: list[str] # distinct agent_ids in commits
155
156
157 class MPackBundle(TypedDict, total=False):
158 """The unit of exchange between the Muse CLI and a remote.
159
160 All fields are optional so that partial bundles (fetch-only, objects-only)
161 are valid wire messages. Callers check for presence before consuming.
162
163 The ``summary`` field carries advisory metadata for agent routing —
164 agents can make decisions from it without deserialising commits or objects.
165
166 The ``meta`` field declares the bundle's scope (full vs incremental) and
167 base commits, allowing receivers to verify it correctly without out-of-band
168 knowledge of how it was built.
169 """
170
171 commits: list[CommitDict]
172 snapshots: list[SnapshotDict]
173 objects: list[ObjectPayload]
174 #: Tags attached to any commit included in this bundle.
175 tags: list[WireTag]
176 #: Remote branch heads at the time the bundle was produced.
177 branch_heads: BranchHeads
178 #: Advisory summary — populated by :func:`build_mpack`.
179 summary: MPackSummary
180 #: Self-describing metadata — always written by :func:`build_mpack`.
181 meta: BundleMeta
182
183
184 class RemoteInfo(TypedDict, total=False):
185 """Repository metadata returned by ``GET {url}/refs``."""
186
187 repo_id: str # always present
188 domain: str # always present
189 #: Maps branch name → commit ID for every branch on the remote.
190 branch_heads: BranchHeads # always present
191 default_branch: str # always present
192 #: Optional Worker URL advertised by the server. When present, the CLI
193 #: routes POST /push/object-pack calls here instead of the primary URL.
194 pack_origin: str
195
196
197 class PushResult(TypedDict):
198 """Server response after a push attempt."""
199
200 ok: bool
201 message: str
202 #: Updated branch heads on the remote after the push (if successful).
203 branch_heads: BranchHeads
204
205
206 class FetchRequest(TypedDict, total=False):
207 """Body of ``POST {url}/fetch`` — negotiates which commits to transfer.
208
209 ``want`` lists commit IDs the client wants to receive.
210 ``have`` lists commit IDs already present locally, allowing the server
211 to send only the commits the client lacks (delta negotiation).
212 """
213
214 want: list[str]
215 have: list[str]
216
217
218 class ApplyResult(TypedDict):
219 """Counts returned by :func:`apply_mpack` describing what was written.
220
221 ``objects_skipped`` counts blobs already present in the store (not
222 rewritten, idempotent). All other counts reflect *new* writes only.
223 ``tags_written`` counts tag records written from the bundle's ``tags``
224 section (0 for bundles created without tag data).
225 """
226
227 commits_written: int
228 snapshots_written: int
229 objects_written: int
230 objects_skipped: int
231 tags_written: int
232
233
234 class ObjectsChunkResponse(TypedDict):
235 """Response from ``POST {url}/push/objects`` — one chunk of a chunked push.
236
237 Returned by both :class:`~muse.core.transport.HttpTransport` and
238 :class:`~muse.core.transport.LocalFileTransport` after pre-uploading a
239 batch of content-addressed objects.
240
241 ``stored`` — objects written to storage in this call.
242 ``skipped`` — objects already present on the remote (idempotent no-ops).
243 """
244
245 stored: int
246 skipped: int
247
248
249 # ---------------------------------------------------------------------------
250 # Pack building
251 # ---------------------------------------------------------------------------
252
253
254 class _WalkResult(TypedDict):
255 """Cached result of a BFS commit-graph walk.
256
257 Produced once by :func:`walk_commits` and consumed by both
258 :func:`collect_object_ids` (Phase 1 — send IDs to filter-objects) and
259 :func:`build_mpack` (Phase 2 — load blobs for missing objects).
260
261 Sharing this avoids two identical BFS traversals per push: the first to
262 gather object IDs for ``POST /filter-objects``, and the second to actually
263 load blobs and assemble the pack bundle.
264
265 ``missing_snapshots`` is populated by :func:`walk_commits` with the
266 snapshot_ids of any reachable commit whose snapshot file is absent from
267 the local store. Callers should surface this to the user before pushing —
268 a pack that contains a commit but not its snapshot creates a dangling
269 reference on the remote.
270 """
271
272 commits: list[CommitRecord]
273 snapshot_ids: set[str]
274 all_object_ids: list[str] # sorted, deduplicated
275 oid_to_path: dict[str, str] # object_id → repository path (from snapshot manifests)
276 missing_snapshots: set[str] # snapshot_ids present in commits but absent on disk
277
278
279 def walk_commits(
280 repo_root: pathlib.Path,
281 commit_ids: list[str],
282 *,
283 have: list[str] | None = None,
284 ) -> _WalkResult:
285 """BFS-walk the commit graph from *commit_ids*, stopping at *have*.
286
287 Returns a :class:`_WalkResult` that can be passed to both
288 :func:`collect_object_ids_from_walk` and :func:`build_mpack_from_walk`
289 to avoid repeating the traversal.
290
291 This is the **single source of truth** for what goes into a push bundle.
292 Callers that need both the object ID list (for ``POST /filter-objects``)
293 and the full pack (for uploading blobs) should call this once and pass
294 the result to both downstream functions.
295 """
296 have_set: set[str] = set(have or [])
297 commits_to_send: list[CommitRecord] = list(
298 iter_ancestors(repo_root, commit_ids, exclude=have_set)
299 )
300
301 # Collect objects already on the remote (have-commits' snapshots).
302 # Subtracting these gives us only genuinely new objects to send.
303 have_object_ids: set[str] = set()
304 for cid in have_set:
305 have_commit = read_commit(repo_root, cid)
306 if have_commit is not None:
307 have_snap = read_snapshot(repo_root, have_commit.snapshot_id)
308 if have_snap is not None:
309 have_object_ids.update(have_snap.manifest.values())
310
311 snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_send}
312 missing_snapshots: set[str] = set()
313 all_object_ids: set[str] = set()
314 oid_to_path: dict[str, str] = {}
315 for sid in snapshot_ids:
316 snap = read_snapshot(repo_root, sid)
317 if snap is not None:
318 all_object_ids.update(snap.manifest.values())
319 # Build oid→path from each snapshot manifest (path → oid).
320 # Later snapshots win on collision — any path is fine for delta lookup.
321 for path, oid in snap.manifest.items():
322 oid_to_path[oid] = path
323 else:
324 missing_snapshots.add(sid)
325
326 all_object_ids -= have_object_ids
327
328 if missing_snapshots:
329 for sid in sorted(missing_snapshots):
330 logger.warning(
331 "⚠️ walk_commits: snapshot %s is missing from the local store — "
332 "the commit(s) referencing it will be excluded from the pack. "
333 "Run `muse verify` to audit store integrity.",
334 short_id(sid),
335 )
336
337 return _WalkResult(
338 commits=commits_to_send,
339 snapshot_ids=snapshot_ids,
340 all_object_ids=sorted(all_object_ids),
341 oid_to_path=oid_to_path,
342 missing_snapshots=missing_snapshots,
343 )
344
345
346 def stream_object_chunks(
347 repo_root: pathlib.Path,
348 object_ids: list[str],
349 chunk_size: int,
350 ) -> "collections.abc.Iterator[list[ObjectPayload]]":
351 """Yield blobs in chunks of *chunk_size* as they are read from disk.
352
353 This is the hot path for ``muse push`` after ``POST /filter-objects``.
354 Instead of reading all blobs into RAM before uploading (peak RAM = all
355 missing objects), we read one chunk at a time and yield it immediately.
356 The caller can start the first upload while the second chunk is still
357 being assembled — reducing both peak memory and time-to-first-upload.
358
359 Missing blobs are logged and skipped, consistent with :func:`build_mpack`.
360 """
361 import collections.abc # local to avoid circular at module level
362
363 chunk: list[ObjectPayload] = []
364 for oid in object_ids:
365 raw = read_object(repo_root, oid)
366 if raw is None:
367 logger.warning("⚠️ stream_object_chunks: blob %s absent — skipping", short_id(oid))
368 continue
369 chunk.append(ObjectPayload(object_id=oid, content=raw))
370 if len(chunk) >= chunk_size:
371 yield chunk
372 chunk = []
373 if chunk:
374 yield chunk
375
376
377 def collect_object_ids_from_walk(walk: _WalkResult) -> list[str]:
378 """Return the sorted object ID list from a pre-computed :func:`walk_commits` result.
379
380 Zero disk I/O — the walk already read all snapshots.
381 """
382 return walk["all_object_ids"]
383
384
385 def build_mpack_from_walk(
386 repo_root: pathlib.Path,
387 walk: _WalkResult,
388 *,
389 only_objects: set[str] | None = None,
390 repo_id: str = "",
391 ) -> MPackBundle:
392 """Assemble an :class:`MPackBundle` from a pre-computed :func:`walk_commits` result.
393
394 Avoids the second BFS traversal that :func:`build_mpack` would otherwise
395 perform. Only reads blob bytes for objects in *only_objects* (or all
396 objects when *only_objects* is ``None``).
397
398 This is the hot path for ``muse push`` after ``POST /filter-objects``
399 narrows the upload set — we already have the commit+snapshot lists in
400 memory from the earlier :func:`walk_commits` call.
401
402 Returns:
403 An :class:`MPackBundle` ready for serialisation and transfer.
404 """
405 missing_snapshots: set[str] = walk.get("missing_snapshots") or set()
406 all_object_ids_set: set[str] = set(walk["all_object_ids"])
407
408 # Hard failure on any missing snapshot — silently skipping would push
409 # commits without their snapshots, creating dangling references on the
410 # remote that can never be healed without rewriting history.
411 if missing_snapshots:
412 sample = sorted(missing_snapshots)[:3]
413 sample_str = ", ".join(short_id(s) for s in sample)
414 raise ValueError(
415 f"Push aborted: {len(missing_snapshots)} snapshot(s) are missing from "
416 f"the local store but are required by commits being sent "
417 f"({sample_str}{'…' if len(missing_snapshots) > 3 else ''}). "
418 f"Run 'muse verify' to audit store integrity."
419 )
420
421 snapshot_dicts: list[SnapshotDict] = []
422 for sid in sorted(walk["snapshot_ids"]):
423 snap = read_snapshot(repo_root, sid)
424 if snap is None:
425 # This should not happen — missing_snapshots guard above catches it.
426 raise ValueError(
427 f"Push aborted: snapshot {short_id(sid)} is missing from the local store. "
428 f"Run 'muse verify' to audit store integrity."
429 )
430 snapshot_dicts.append(snap.to_dict())
431
432 commits_to_send = list(walk["commits"])
433
434 candidate_ids = (
435 all_object_ids_set & only_objects if only_objects is not None else all_object_ids_set
436 )
437
438 object_payloads: list[ObjectPayload] = []
439 for oid in sorted(candidate_ids):
440 raw = read_object(repo_root, oid)
441 if raw is None:
442 logger.warning("⚠️ build_mpack_from_walk: blob %s absent — skipping", short_id(oid))
443 continue
444 object_payloads.append(ObjectPayload(object_id=oid, content=raw))
445
446 sent_commit_ids = [c.commit_id for c in commits_to_send]
447 wire_tags = _tags_for_commits(repo_root, sent_commit_ids, repo_id) if repo_id else []
448
449 total_bytes = sum(len(read_object(repo_root, oid) or b"") for oid in sorted(candidate_ids))
450 agent_ids = sorted({
451 c.to_dict().get("agent_id", "") for c in commits_to_send
452 if c.to_dict().get("agent_id")
453 })
454 summary = MPackSummary(
455 commits_count=len(commits_to_send),
456 objects_count=len(object_payloads),
457 objects_bytes=total_bytes,
458 branches={},
459 agent_ids=agent_ids,
460 )
461 bundle: MPackBundle = {
462 "commits": [c.to_dict() for c in commits_to_send],
463 "snapshots": snapshot_dicts,
464 "objects": object_payloads,
465 "summary": summary,
466 "meta": BundleMeta(
467 mode="full",
468 base_commits=[],
469 created_at=datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
470 ),
471 }
472 if wire_tags:
473 bundle["tags"] = wire_tags
474
475 logger.info(
476 "✅ Built MPack (from walk): %d commits, %d snapshots, %d objects, %d tags",
477 len(commits_to_send),
478 len(snapshot_dicts),
479 len(object_payloads),
480 len(wire_tags),
481 )
482 return bundle
483
484
485 def _tags_for_commits(
486 repo_root: pathlib.Path, commit_ids: list[str], repo_id: str
487 ) -> list[WireTag]:
488 """Return all tags attached to *commit_ids* as serialisable :class:`WireTag` dicts."""
489 seen_tag_ids: set[str] = set()
490 wire_tags: list[WireTag] = []
491 for cid in commit_ids:
492 for tag in get_tags_for_commit(repo_root, repo_id, cid):
493 if tag.tag_id not in seen_tag_ids:
494 seen_tag_ids.add(tag.tag_id)
495 wire_tags.append(WireTag(
496 tag_id=tag.tag_id,
497 repo_id=tag.repo_id,
498 commit_id=tag.commit_id,
499 tag=tag.tag,
500 created_at=tag.created_at.isoformat(),
501 ))
502 return wire_tags
503
504
505 def build_mpack(
506 repo_root: pathlib.Path,
507 commit_ids: list[str],
508 *,
509 have: list[str] | None = None,
510 only_objects: set[str] | None = None,
511 repo_id: str = "",
512 ) -> MPackBundle:
513 """Assemble an :class:`MPackBundle` from *commit_ids*, excluding commits in *have*.
514
515 Performs a BFS walk of the commit graph from every ID in *commit_ids*,
516 stopping at any commit already in *have*. Collects all snapshot manifests
517 and object blobs reachable from the selected commits.
518
519 Missing objects or snapshots are logged and skipped — the caller decides
520 whether that constitutes an error.
521
522 Args:
523 repo_root: Root of the Muse repository.
524 commit_ids: Tip commit IDs to include (e.g. current branch HEAD).
525 have: Commit IDs already known to the receiver. The BFS stops
526 at these, reducing bundle size. Pass ``None`` or ``[]``
527 to send the full history.
528 only_objects: When set, only include objects whose IDs are in this set.
529 Used after a ``POST /filter-objects`` negotiation so the
530 client only uploads objects the remote is missing.
531 repo_id: Repository UUID used to look up tags. When omitted, tags
532 are not included in the bundle.
533
534 Returns:
535 An :class:`MPackBundle` ready for serialisation and transfer.
536 """
537 have_set: set[str] = set(have or [])
538
539 # BFS walk from every tip, treating have_set as already-visited.
540 commits_to_send: list[CommitRecord] = []
541 seen: set[str] = set(have_set)
542 queue: collections.deque[str] = collections.deque(
543 cid for cid in commit_ids if cid not in seen
544 )
545
546 while queue:
547 cid = queue.popleft()
548 if cid in seen:
549 continue
550 seen.add(cid)
551 commit = read_commit(repo_root, cid)
552 if commit is None:
553 logger.warning("⚠️ build_mpack: commit %s not found — skipping", short_id(cid))
554 continue
555 commits_to_send.append(commit)
556 if commit.parent_commit_id and commit.parent_commit_id not in seen:
557 queue.append(commit.parent_commit_id)
558 if commit.parent2_commit_id and commit.parent2_commit_id not in seen:
559 queue.append(commit.parent2_commit_id)
560
561 # Unique snapshot IDs referenced by selected commits.
562 snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_send}
563
564
565 snapshot_dicts: list[SnapshotDict] = []
566 all_object_ids: set[str] = set()
567 for sid in sorted(snapshot_ids):
568 snap = read_snapshot(repo_root, sid)
569 if snap is None:
570 # Hard failure — a commit whose snapshot is absent locally cannot
571 # be pushed safely. Silently skipping would create a dangling
572 # commit reference on the remote: the commit lands, the snapshot
573 # never does, and every subsequent pull restores the commit but
574 # not its snapshot. Fail loudly so the user knows the store is
575 # corrupt and can take corrective action before data is lost.
576 raise ValueError(
577 f"Push aborted: snapshot {short_id(sid)} is missing from the local store "
578 f"but is required by a commit being sent. "
579 f"Run 'muse verify' to audit store integrity."
580 )
581 snapshot_dicts.append(snap.to_dict())
582 all_object_ids.update(snap.manifest.values())
583
584 # When only_objects is provided (post filter-objects negotiation) skip
585 # any object the remote already has — only transmit the missing delta.
586 candidate_ids = (
587 all_object_ids & only_objects if only_objects is not None else all_object_ids
588 )
589
590 promisor_remotes = load_promisor_remotes(repo_root)
591 object_payloads: list[ObjectPayload] = []
592 for oid in sorted(candidate_ids):
593 raw = read_object(repo_root, oid)
594 if raw is None:
595 state = object_state(repo_root, oid, promisor_remotes)
596 if state == ObjectState.PROMISED:
597 logger.debug("build_mpack: blob %s is PROMISED — skipping", short_id(oid))
598 continue
599 raise ValueError(
600 f"Pack aborted: object {short_id(oid)}... is missing from the local store "
601 f"and no promisor remote is configured. "
602 f"Run 'muse verify' to audit store integrity."
603 )
604 object_payloads.append(ObjectPayload(object_id=oid, content=raw))
605
606 sent_commit_ids = [c.commit_id for c in commits_to_send]
607 wire_tags = _tags_for_commits(repo_root, sent_commit_ids, repo_id) if repo_id else []
608
609 total_bytes = sum(len(read_object(repo_root, oid) or b"") for oid in sorted(candidate_ids))
610 agent_ids = sorted({
611 c.to_dict().get("agent_id", "") for c in commits_to_send
612 if c.to_dict().get("agent_id")
613 })
614 summary = MPackSummary(
615 commits_count=len(commits_to_send),
616 objects_count=len(object_payloads),
617 objects_bytes=total_bytes,
618 branches={},
619 agent_ids=agent_ids,
620 )
621 _have_list = list(have or [])
622 bundle: MPackBundle = {
623 "commits": [c.to_dict() for c in commits_to_send],
624 "snapshots": snapshot_dicts,
625 "objects": object_payloads,
626 "summary": summary,
627 "meta": BundleMeta(
628 mode="incremental" if _have_list else "full",
629 base_commits=_have_list,
630 created_at=datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
631 ),
632 }
633 if wire_tags:
634 bundle["tags"] = wire_tags
635
636 logger.info(
637 "✅ Built MPack: %d commits, %d snapshots, %d objects, %d tags",
638 len(commits_to_send),
639 len(snapshot_dicts),
640 len(object_payloads),
641 len(wire_tags),
642 )
643 return bundle
644
645
646 # ---------------------------------------------------------------------------
647 # Object ID collection — for pre-push deduplication negotiation
648 # ---------------------------------------------------------------------------
649
650
651 def collect_object_ids(
652 repo_root: pathlib.Path,
653 commit_ids: list[str],
654 *,
655 have: list[str] | None = None,
656 ) -> list[str]:
657 """Return all object IDs reachable from *commit_ids*, excluding *have*.
658
659 Identical BFS walk to :func:`build_mpack` but without reading object bytes.
660 Used by ``muse push`` to call ``POST /filter-objects`` before building the
661 full bundle — the client discovers which objects are missing on the remote
662 and then calls :func:`build_mpack` with ``only_objects`` set to that subset.
663 This avoids loading any blob content until we know it is actually needed.
664
665 Args:
666 repo_root: Root of the Muse repository.
667 commit_ids: Tip commit IDs to examine.
668 have: Commit IDs already known to the receiver (BFS stops here).
669
670 Returns:
671 Sorted list of object IDs reachable from the delta.
672 """
673 have_set: set[str] = set(have or [])
674 commits_to_examine: list[CommitRecord] = []
675 seen: set[str] = set(have_set)
676 queue: collections.deque[str] = collections.deque(
677 cid for cid in commit_ids if cid not in seen
678 )
679 while queue:
680 cid = queue.popleft()
681 if cid in seen:
682 continue
683 seen.add(cid)
684 commit = read_commit(repo_root, cid)
685 if commit is None:
686 continue
687 commits_to_examine.append(commit)
688 if commit.parent_commit_id and commit.parent_commit_id not in seen:
689 queue.append(commit.parent_commit_id)
690 if commit.parent2_commit_id and commit.parent2_commit_id not in seen:
691 queue.append(commit.parent2_commit_id)
692
693 # Collect objects already on the remote (have-commits' snapshots).
694 have_object_ids: set[str] = set()
695 for cid in have_set:
696 have_commit = read_commit(repo_root, cid)
697 if have_commit is not None:
698 have_snap = read_snapshot(repo_root, have_commit.snapshot_id)
699 if have_snap is not None:
700 have_object_ids.update(have_snap.manifest.values())
701
702 snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_examine}
703 all_object_ids: set[str] = set()
704 for sid in snapshot_ids:
705 snap = read_snapshot(repo_root, sid)
706 if snap is not None:
707 all_object_ids.update(snap.manifest.values())
708
709 return sorted(all_object_ids - have_object_ids)
710
711
712 # ---------------------------------------------------------------------------
713 # Pack applying
714 # ---------------------------------------------------------------------------
715
716
717 def apply_mpack(repo_root: pathlib.Path, bundle: MPackBundle) -> ApplyResult:
718 """Write the contents of *bundle* into a local ``.muse/`` directory.
719
720 Writes in dependency order: objects first (blobs), then snapshots (which
721 reference object IDs), then commits (which reference snapshot IDs). All
722 writes are idempotent — already-present items are silently skipped.
723
724 Args:
725 repo_root: Root of the Muse repository to write into.
726 bundle: :class:`MPackBundle` received from the remote.
727
728 Returns:
729 :class:`ApplyResult` with counts of newly written and skipped items.
730 """
731 objects_written = 0
732 objects_skipped = 0
733 snapshots_written = 0
734 commits_written = 0
735 tags_written = 0
736
737 raw_objects = bundle.get("objects") or []
738 raw_snapshots = bundle.get("snapshots") or []
739 raw_commits = bundle.get("commits") or []
740
741 # Pack-bomb guard: cap the total number of items accepted per call.
742 # A legitimate push carries at most tens of thousands of objects per chunk;
743 # a pack claiming millions is an adversarial input.
744 total_items = len(raw_objects) + len(raw_snapshots) + len(raw_commits)
745 if total_items > MAX_PACK_OBJECTS:
746 raise ValueError(
747 f"Pack rejected: {total_items:,} total items exceeds the "
748 f"{MAX_PACK_OBJECTS:,} item limit per apply_mpack call. "
749 "Split the pack into smaller chunks."
750 )
751
752 # Deduplicate object IDs before the write loop. A malicious or buggy
753 # sender may repeat the same object_id N times, forcing N sha256 hashes
754 # before the "already exists" short-circuit returns False.
755 seen_object_ids: set[str] = set()
756
757 for obj in raw_objects:
758 oid = obj.get("object_id", "")
759 raw = obj.get("content", b"")
760 if not oid or not isinstance(raw, bytes):
761 logger.warning("⚠️ apply_mpack: blob entry missing fields — skipped")
762 continue
763 if oid in seen_object_ids:
764 logger.debug("⚠️ apply_mpack: duplicate object_id %s — skipped", short_id(oid))
765 objects_skipped += 1
766 continue
767 seen_object_ids.add(oid)
768 # Per-object size cap: check before calling write_object to avoid
769 # hashing a huge payload that will be rejected anyway.
770 if len(raw) > MAX_OBJECT_WRITE_BYTES:
771 logger.warning(
772 "⚠️ apply_mpack: object %s is %d bytes, exceeding %d MiB limit — skipped",
773 short_id(oid), len(raw), MAX_OBJECT_WRITE_BYTES // (1024 * 1024),
774 )
775 continue
776 try:
777 if write_object(repo_root, oid, raw):
778 objects_written += 1
779 else:
780 objects_skipped += 1
781 except ValueError as exc:
782 # Malicious object IDs (non-hex, path traversal) or content/ID
783 # mismatch — log and skip rather than aborting the entire pack.
784 logger.warning("⚠️ apply_mpack: malformed object entry — skipped: %s", exc)
785
786 for snap_dict in raw_snapshots:
787 try:
788 snap = SnapshotRecord.from_dict(snap_dict)
789 # Guard against zip-slip: manifest keys are stored as-is and later
790 # used to construct checkout paths. A malicious bundle could inject
791 # keys like "../../etc/cron.d/evil". We validate all keys here —
792 # before writing — so no traversal path ever enters the store.
793 for key in snap.manifest:
794 validate_workspace_path(key)
795 # Manifest values are object IDs — validate they are safe hex strings.
796 for oid in snap.manifest.values():
797 validate_object_id(oid)
798 is_new = read_snapshot(repo_root, snap.snapshot_id) is None
799 write_snapshot(repo_root, snap)
800 if is_new:
801 snapshots_written += 1
802 except (KeyError, ValueError) as exc:
803 logger.warning("⚠️ apply_mpack: malformed snapshot — skipped: %s", exc)
804
805 for commit_dict in raw_commits:
806 try:
807 commit = CommitRecord.from_dict(commit_dict)
808 # from_dict is designed for trusted, typed callers and does not
809 # validate essential fields — guard them here before the commit
810 # reaches write_commit (which constructs paths from commit_id).
811 if not commit.commit_id:
812 logger.warning("⚠️ apply_mpack: commit missing commit_id — skipped")
813 continue
814 if not commit.snapshot_id:
815 logger.warning(
816 "⚠️ apply_mpack: commit %s missing snapshot_id — skipped",
817 short_id(commit.commit_id),
818 )
819 continue
820 is_new = read_commit(repo_root, commit.commit_id) is None
821 write_commit(repo_root, commit)
822 if is_new:
823 commits_written += 1
824 except OSError as exc:
825 # write_commit raises OSError("Store integrity violation") when the
826 # existing commit file contains a DIFFERENT commit_id than the one
827 # being written — indicating the local store has been tampered with.
828 # Log CRITICAL and skip rather than crashing the entire apply_mpack.
829 logger.critical(
830 "❌ apply_mpack: store integrity violation for commit — skipped: %s", exc
831 )
832 except (KeyError, ValueError, TypeError) as exc:
833 logger.warning("⚠️ apply_mpack: malformed commit — skipped: %s", exc)
834
835 for wire_tag in bundle.get("tags") or []:
836 try:
837 tag_record = TagRecord.from_dict(TagDict(
838 tag_id=wire_tag["tag_id"],
839 repo_id=wire_tag["repo_id"],
840 commit_id=wire_tag["commit_id"],
841 tag=wire_tag["tag"],
842 created_at=wire_tag["created_at"],
843 ))
844 write_tag(repo_root, tag_record)
845 tags_written += 1
846 except (KeyError, ValueError) as exc:
847 logger.warning("⚠️ apply_mpack: malformed tag — skipped: %s", exc)
848
849 logger.info(
850 "✅ Applied pack: %d new blobs, %d new snapshots, %d new commits, %d tags (%d blobs skipped)",
851 objects_written,
852 snapshots_written,
853 commits_written,
854 tags_written,
855 objects_skipped,
856 )
857 return ApplyResult(
858 commits_written=commits_written,
859 snapshots_written=snapshots_written,
860 objects_written=objects_written,
861 objects_skipped=objects_skipped,
862 tags_written=tags_written,
863 )
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 137 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 140 days ago