pack_objects.py
python
| 1 | """muse plumbing pack-objects — build a PackBundle JSON and write to stdout. |
| 2 | |
| 3 | Collects a set of commits (and all referenced snapshots and objects) into a |
| 4 | single JSON PackBundle suitable for transport to a remote. Analogous to |
| 5 | ``git pack-objects`` but uses JSON + base64 rather than a binary packfile |
| 6 | format — optimised for agent pipelines and HTTP transport. |
| 7 | |
| 8 | Usage:: |
| 9 | |
| 10 | muse plumbing pack-objects <want_id>... [--have <id>...] |
| 11 | |
| 12 | The ``--have`` IDs are commits the receiver already has. Objects reachable |
| 13 | exclusively from ``--have`` ancestors are pruned from the bundle. |
| 14 | |
| 15 | Output: a PackBundle JSON object written to stdout (pipe to a file or HTTP |
| 16 | request body). |
| 17 | |
| 18 | Plumbing contract |
| 19 | ----------------- |
| 20 | |
| 21 | - Exit 0: pack written to stdout. |
| 22 | - Exit 1: a wanted commit not found. |
| 23 | """ |
| 24 | |
| 25 | from __future__ import annotations |
| 26 | |
| 27 | import json |
| 28 | import logging |
| 29 | import pathlib |
| 30 | import sys |
| 31 | from typing import Annotated |
| 32 | |
| 33 | import typer |
| 34 | |
| 35 | from muse.core.errors import ExitCode |
| 36 | from muse.core.pack import build_pack |
| 37 | from muse.core.repo import require_repo |
| 38 | from muse.core.store import get_head_commit_id |
| 39 | |
| 40 | logger = logging.getLogger(__name__) |
| 41 | |
| 42 | app = typer.Typer() |
| 43 | |
| 44 | |
| 45 | def _current_branch(root: pathlib.Path) -> str: |
| 46 | head = (root / ".muse" / "HEAD").read_text().strip() |
| 47 | return head.removeprefix("refs/heads/").strip() |
| 48 | |
| 49 | |
| 50 | @app.callback(invoke_without_command=True) |
| 51 | def pack_objects( |
| 52 | ctx: typer.Context, |
| 53 | want: list[str] = typer.Argument( |
| 54 | ..., |
| 55 | help="Commit IDs to pack. May be full IDs or 'HEAD'.", |
| 56 | ), |
| 57 | have: list[str] = typer.Option( |
| 58 | [], |
| 59 | "--have", |
| 60 | help="Commits the receiver already has (pruned from pack).", |
| 61 | ), |
| 62 | ) -> None: |
| 63 | """Build a PackBundle JSON from wanted commits and write to stdout. |
| 64 | |
| 65 | Traverses the commit graph from each ``want`` ID, collecting all |
| 66 | commits, snapshots, and objects not already reachable from ``--have`` |
| 67 | ancestors. The resulting JSON bundle can be piped directly to |
| 68 | ``muse plumbing unpack-objects`` on the receiving side, or sent via |
| 69 | HTTP to a MuseHub endpoint. |
| 70 | """ |
| 71 | root = require_repo() |
| 72 | |
| 73 | resolved_wants: list[str] = [] |
| 74 | for w in want: |
| 75 | if w.upper() == "HEAD": |
| 76 | branch = _current_branch(root) |
| 77 | cid = get_head_commit_id(root, branch) |
| 78 | if cid is None: |
| 79 | typer.echo( |
| 80 | json.dumps({"error": "HEAD has no commits"}), err=True |
| 81 | ) |
| 82 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 83 | resolved_wants.append(cid) |
| 84 | else: |
| 85 | resolved_wants.append(w) |
| 86 | |
| 87 | bundle = build_pack(root, commit_ids=resolved_wants, have=have or None) |
| 88 | sys.stdout.write(json.dumps(bundle)) |