gabriel / muse public
read_snapshot.py python
185 lines 5.7 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 days ago
1 """muse read-snapshot — emit full snapshot metadata as JSON.
2
3 Reads a snapshot record by its SHA-256 ID and emits the complete JSON
4 representation including the file manifest.
5
6 Output::
7
8 {
9 "snapshot_id": "<sha256>",
10 "created_at": "2026-03-18T12:00:00+00:00",
11 "file_count": 3,
12 "manifest": {
13 "tracks/drums.mid": "<sha256>",
14 "tracks/bass.mid": "<sha256>",
15 "tracks/piano.mid": "<sha256>"
16 }
17 }
18
19 Output contract
20 ---------------
21
22 - Exit 0: snapshot found and printed.
23 - Exit 1: snapshot not found or invalid snapshot ID format.
24
25 Agent use
26 ---------
27
28 Skip the manifest when you only need metadata::
29
30 muse read-snapshot <id> --no-manifest
31 # → {"snapshot_id": "...", "created_at": "...", "file_count": 3}
32
33 Filter to a path prefix to avoid pulling the full manifest::
34
35 muse read-snapshot <id> --path-prefix src/
36 """
37
38 import argparse
39 import json
40 import logging
41 import sys
42 from typing import TypedDict
43
44 from muse.core.envelope import EnvelopeJson, make_envelope
45 from muse.core.errors import ExitCode
46 from muse.core.repo import require_repo
47 from muse.core.store import read_snapshot
48 from muse.core.validation import validate_object_id
49 from muse.core.types import Manifest
50 from muse.core.timing import start_timer
51
52 logger = logging.getLogger(__name__)
53
54 class _SnapshotOutput(EnvelopeJson, total=False):
55 """JSON output schema for ``muse read-snapshot``.
56
57 Always present: envelope fields + ``snapshot_id``, ``created_at``,
58 ``file_count``.
59
60 Optional: ``manifest`` (omitted with ``--no-manifest``).
61
62 ``file_count`` reflects the number of entries actually returned — i.e.
63 the filtered count when ``--path-prefix`` is used.
64 """
65 snapshot_id: str
66 created_at: str
67 file_count: int
68 manifest: Manifest
69
70 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
71 """Register the read-snapshot subcommand."""
72 parser = subparsers.add_parser(
73 "read-snapshot",
74 help="Emit full snapshot metadata and manifest as JSON.",
75 description=__doc__,
76 formatter_class=argparse.RawDescriptionHelpFormatter,
77 )
78 parser.add_argument(
79 "snapshot_id",
80 help="SHA-256 snapshot ID (64 hex chars).",
81 )
82 parser.add_argument(
83 "--no-manifest",
84 action="store_true",
85 dest="no_manifest",
86 help=(
87 "Omit the file manifest from JSON output. "
88 "Returns only snapshot_id, created_at, and file_count — "
89 "ideal for agents doing metadata-only queries."
90 ),
91 )
92 parser.add_argument(
93 "--path-prefix", "-p",
94 default=None,
95 dest="path_prefix",
96 metavar="PREFIX",
97 help="Filter manifest to paths starting with PREFIX.",
98 )
99 parser.add_argument(
100 "--json", "-j",
101 action="store_true",
102 dest="json_out",
103 help="Emit machine-readable JSON.",
104 )
105 parser.set_defaults(func=run)
106
107 def run(args: argparse.Namespace) -> None:
108 """Emit full snapshot metadata as JSON (default) or a compact text summary.
109
110 A snapshot holds the complete file manifest (path → object_id mapping)
111 for a point in time. Every commit references exactly one snapshot.
112 Use ``muse ls-files --commit <id>`` if you want to look up a snapshot from
113 a commit ID rather than the snapshot ID directly.
114
115 Use ``--no-manifest`` for lightweight metadata queries (returns only
116 ``snapshot_id``, ``created_at``, ``file_count``, ``elapsed()``,
117 ``exit_code``).
118
119 Use ``--path-prefix`` to scope the manifest to a subtree. ``file_count``
120 in the response reflects the filtered count, not the total snapshot size.
121
122 ``--no-manifest`` and ``--path-prefix`` may be combined: useful for
123 counting files under a path without downloading any object IDs::
124
125 muse read-snapshot <id> --no-manifest --path-prefix src/
126 # → {"snapshot_id": "...", "created_at": "...", "file_count": 5,
127 # "duration_ms": 1.2, "exit_code": 0}
128
129 Text format (``--format text``)::
130
131 <sha256:12-hex> <file_count> files <created_at>
132 """
133 elapsed = start_timer()
134 json_out: bool = args.json_out
135 snapshot_id: str = args.snapshot_id
136 no_manifest: bool = args.no_manifest
137 path_prefix: str | None = args.path_prefix
138
139 if no_manifest and not json_out:
140 print(
141 json.dumps({"error": "--no-manifest is only valid with --json"}),
142 file=sys.stderr,
143 )
144 raise SystemExit(ExitCode.USER_ERROR)
145
146 if path_prefix is not None and not json_out:
147 print(
148 json.dumps({"error": "--path-prefix is only valid with --json"}),
149 file=sys.stderr,
150 )
151 raise SystemExit(ExitCode.USER_ERROR)
152
153 try:
154 validate_object_id(snapshot_id)
155 except ValueError as exc:
156 print(json.dumps({"error": f"Invalid snapshot ID: {exc}"}), file=sys.stderr)
157 raise SystemExit(ExitCode.USER_ERROR)
158
159 root = require_repo()
160
161 record = read_snapshot(root, snapshot_id)
162 if record is None:
163 print(json.dumps({"error": f"Snapshot not found: {snapshot_id}"}), file=sys.stderr)
164 raise SystemExit(ExitCode.USER_ERROR)
165
166 if not json_out:
167 print(
168 f"{record.snapshot_id} {len(record.manifest)} files "
169 f"{record.created_at.isoformat()}"
170 )
171 return
172
173 manifest = record.manifest
174 if path_prefix is not None:
175 manifest = {p: oid for p, oid in manifest.items() if p.startswith(path_prefix)}
176
177 output = _SnapshotOutput(
178 **make_envelope(elapsed),
179 snapshot_id=record.snapshot_id,
180 created_at=record.created_at.isoformat(),
181 file_count=len(manifest),
182 )
183 if not no_manifest:
184 output["manifest"] = manifest
185 print(json.dumps(output))
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago