aaronrene / gabriel-muse public
unpack_objects.py python
160 lines 5.2 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago
1 """muse plumbing unpack-objects — read a PackBundle from stdin and write to store.
2
3 Reads a PackBundle msgpack binary from stdin and idempotently writes its
4 commits, snapshots, and objects into the local ``.muse/`` store. Analogous
5 to ``git unpack-objects``.
6
7 Usage::
8
9 cat pack.muse | muse plumbing unpack-objects
10 muse plumbing pack-objects HEAD | muse plumbing unpack-objects
11
12 Output::
13
14 {
15 "commits_written": 12,
16 "snapshots_written": 12,
17 "objects_written": 47,
18 "objects_skipped": 3
19 }
20
21 Plumbing contract
22 -----------------
23
24 - Exit 0: objects unpacked (idempotent — already-present objects are skipped).
25 - Exit 1: invalid msgpack from stdin.
26 - Exit 3: write failure.
27 """
28
29 from __future__ import annotations
30
31 import argparse
32 import json
33 import logging
34 import sys
35 from typing import TypeGuard
36
37 from muse.core.errors import ExitCode
38 from muse.core.pack import PackBundle, apply_pack
39 from muse.core.repo import require_repo
40 from muse.core.store import MAX_PACK_MSGPACK_BYTES, CommitDict, MsgpackValue, SnapshotDict, safe_unpackb
41 from muse.core._types import BranchHeads
42
43 logger = logging.getLogger(__name__)
44
45 _FORMAT_CHOICES = ("json", "text")
46
47
48 def _is_commit_dict(v: MsgpackValue) -> TypeGuard[CommitDict]:
49 """TypeGuard for narrowing MsgpackValue → CommitDict at the wire boundary."""
50 return isinstance(v, dict)
51
52
53 def _is_snapshot_dict(v: MsgpackValue) -> TypeGuard[SnapshotDict]:
54 """TypeGuard for narrowing MsgpackValue → SnapshotDict at the wire boundary."""
55 return isinstance(v, dict)
56
57
58 def _as_branch_heads(v: MsgpackValue) -> BranchHeads:
59 """Extract branch_heads from a wire payload, narrowing safely."""
60 if not isinstance(v, dict):
61 return {}
62 return {str(k): str(val) for k, val in v.items() if isinstance(val, str)}
63
64
65 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
66 """Register the unpack-objects subcommand."""
67 parser = subparsers.add_parser(
68 "unpack-objects",
69 help="Read a PackBundle JSON from stdin, write to the local store.",
70 description=__doc__,
71 )
72 parser.add_argument(
73 "--format", "-f",
74 dest="fmt",
75 default="json",
76 metavar="FORMAT",
77 help="Output format: json (default) or text (human-readable summary).",
78 )
79 parser.add_argument(
80 "--json", action="store_const", const="json", dest="fmt",
81 help="Shorthand for --format json."
82 )
83 parser.set_defaults(func=run)
84
85
86 def run(args: argparse.Namespace) -> None:
87 """Read a PackBundle JSON from stdin and write to the local store.
88
89 Idempotent: if a commit, snapshot, or object already exists in the store
90 it is silently skipped. Partial packs (interrupted transfers) are safe
91 to re-apply. The exit code is 0 as long as the store is consistent at
92 the end of the operation.
93 """
94 fmt: str = args.fmt
95
96 if fmt not in _FORMAT_CHOICES:
97 print(
98 json.dumps({"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"}),
99 file=sys.stderr,
100 )
101 raise SystemExit(ExitCode.USER_ERROR)
102
103 root = require_repo()
104
105 raw_bytes = sys.stdin.buffer.read()
106 try:
107 raw_dict = safe_unpackb(
108 raw_bytes,
109 context="stdin",
110 max_bytes=MAX_PACK_MSGPACK_BYTES,
111 allow_binary=True,
112 )
113 except Exception as exc:
114 print(json.dumps({"error": f"Invalid msgpack from stdin: {exc}"}), file=sys.stderr)
115 raise SystemExit(ExitCode.USER_ERROR)
116
117 if not isinstance(raw_dict, dict):
118 print(
119 json.dumps({"error": "Expected a msgpack map at the top level."}),
120 file=sys.stderr,
121 )
122 raise SystemExit(ExitCode.USER_ERROR)
123
124 from muse.core.pack import ObjectPayload
125
126 raw_objects: list[ObjectPayload] = []
127 raw_objects_v = raw_dict.get("objects")
128 for item in (raw_objects_v if isinstance(raw_objects_v, list) else []):
129 if isinstance(item, dict):
130 oid = item.get("object_id")
131 content = item.get("content")
132 if isinstance(oid, str) and isinstance(content, (bytes, bytearray)):
133 raw_objects.append(ObjectPayload(object_id=oid, content=bytes(content)))
134
135 raw_commits = raw_dict.get("commits")
136 raw_snapshots = raw_dict.get("snapshots")
137 bundle = PackBundle(
138 commits=[r for r in raw_commits if _is_commit_dict(r)] if isinstance(raw_commits, list) else [],
139 snapshots=[r for r in raw_snapshots if _is_snapshot_dict(r)] if isinstance(raw_snapshots, list) else [],
140 objects=raw_objects,
141 branch_heads=_as_branch_heads(raw_dict.get("branch_heads")),
142 )
143
144 result = apply_pack(root, bundle)
145
146 if fmt == "text":
147 print(
148 f"Wrote {result['commits_written']} commits, "
149 f"{result['snapshots_written']} snapshots, "
150 f"{result['objects_written']} objects "
151 f"({result['objects_skipped']} skipped)."
152 )
153 return
154
155 print(json.dumps({
156 "commits_written": result["commits_written"],
157 "snapshots_written": result["snapshots_written"],
158 "objects_written": result["objects_written"],
159 "objects_skipped": result["objects_skipped"],
160 }))
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago