"""muse plumbing cat-object — read a stored object from the object store. Reads the raw bytes of a content-addressed object and writes them to stdout. Useful for inspecting stored blobs, verifying round-trips, or piping raw content to other tools. Single-object mode (default) ----------------------------- With ``--format raw`` (default): bytes streamed directly to stdout at 64 KiB at a time — no heap spike, no size ceiling. With ``--format info`` / ``--json``: JSON metadata about the object (no content emitted). {"object_id": "", "size_bytes": 1234, "present": true} Batch mode ---------- ``--batch`` reads object IDs from stdin (one per line) and for each emits the batch-protocol header followed by the raw content:: blob \n \n For missing or invalid OIDs the output is:: missing\n ``--batch-check`` is the header-only variant — same protocol but no content bytes are emitted. Useful for bulk presence checks without reading blobs. Batch mode is analogous to ``git cat-file --batch`` / ``--batch-check`` and is intended for agent pipelines and migration tools that need to stream many objects efficiently from a single long-running process. Plumbing contract ----------------- - Exit 0: found — bytes written to stdout or metadata printed. - Exit 1: not found in the store, or invalid object-id format. - Exit 3: I/O error reading from the store. - Batch mode always exits 0 (missing objects are reported inline, not as errors). Agent use --------- Agents that only need metadata (size, presence check) should prefer ``--json`` over ``--format raw`` to avoid pulling large blobs into their context:: muse cat-object --json For bulk reads in agent pipelines, use batch mode:: printf '%s\\n' | muse cat-object --batch printf '%s\\n' | muse cat-object --batch-check """ from __future__ import annotations import argparse import json import logging import pathlib import sys from muse.core.errors import ExitCode from muse.core.object_store import has_object, object_path from muse.core.repo import require_repo from muse.core.validation import sanitize_display, validate_object_id logger = logging.getLogger(__name__) _FORMAT_CHOICES = ("raw", "info") _CHUNK = 65536 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the cat-object subcommand.""" parser = subparsers.add_parser( "cat-object", help="Emit raw bytes of a stored object to stdout.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "object_id", nargs="?", default=None, help=( "SHA-256 object ID to read (64 hex chars). " "Required in single-object mode; omit when using --batch or --batch-check." ), ) batch_group = parser.add_mutually_exclusive_group() batch_group.add_argument( "--batch", action="store_true", dest="batch", help=( "Batch mode: read object IDs from stdin (one per line) and emit " "' blob \\n\\n' for each. " "Missing or invalid OIDs emit ' missing\\n'." ), ) batch_group.add_argument( "--batch-check", action="store_true", dest="batch_check", help=( "Batch-check mode: like --batch but emits only the header line " "' blob \\n' — no content bytes. " "Efficient for bulk presence checks." ), ) parser.add_argument( "--format", "-f", dest="fmt", default="raw", metavar="FORMAT", help="Output format for single-object mode: raw (bytes to stdout) or info (JSON metadata). (default: raw)", ) parser.add_argument( "--json", action="store_const", const="info", dest="fmt", help="Emit JSON metadata instead of raw bytes (alias for --format info).", ) parser.set_defaults(func=run) def _run_batch(root: "pathlib.Path", check_only: bool) -> None: """Process OIDs from stdin in the git cat-file --batch protocol.""" out = sys.stdout.buffer for raw_line in sys.stdin: oid = raw_line.strip() if not oid: continue # Validate format — invalid OIDs are reported as missing, not errors. try: validate_object_id(oid) except ValueError: out.write(f"{oid} missing\n".encode()) out.flush() continue if not has_object(root, oid): out.write(f"{oid} missing\n".encode()) out.flush() continue obj = object_path(root, oid) size = obj.stat().st_size out.write(f"{oid} blob {size}\n".encode()) if not check_only: with obj.open("rb") as fh: for chunk in iter(lambda: fh.read(_CHUNK), b""): out.write(chunk) out.write(b"\n") out.flush() def run(args: argparse.Namespace) -> None: """Read a stored object from the content-addressed object store. Analogous to ``git cat-file``. Single-object mode: with ``--format raw`` (default) the raw bytes are streamed to stdout at 64 KiB at a time — suitable for piping or redirection with no heap spike and no size ceiling. With ``--format info`` (or ``--json``) a JSON summary is printed without emitting the contents. Batch mode (``--batch`` / ``--batch-check``): reads OIDs from stdin and emits the batch protocol for each — one long-running process replaces N subprocess spawns. Analogous to ``git cat-file --batch``. """ batch: bool = args.batch batch_check: bool = args.batch_check fmt: str = args.fmt object_id: str | None = args.object_id # ── Batch mode ──────────────────────────────────────────────────────────── if batch or batch_check: root = require_repo() _run_batch(root, check_only=batch_check) return # ── Single-object mode ──────────────────────────────────────────────────── if object_id is None: print( "❌ object_id is required in single-object mode " "(or use --batch / --batch-check for stdin processing).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if fmt not in _FORMAT_CHOICES: print( f"❌ Unknown format {fmt!r}. Valid choices: {', '.join(_FORMAT_CHOICES)}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) try: validate_object_id(object_id) except ValueError as exc: print(f"❌ Invalid object ID: {sanitize_display(str(exc))}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() if not has_object(root, object_id): if fmt == "info": print(json.dumps({"object_id": object_id, "present": False, "size_bytes": 0})) else: print(f"❌ Object not found: {object_id}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) obj = object_path(root, object_id) if fmt == "info": size = obj.stat().st_size print(json.dumps({"object_id": object_id, "present": True, "size_bytes": size})) return # Raw: two-pass streaming — verify integrity before writing any output. try: import hashlib as _hashlib h = _hashlib.sha256() with obj.open("rb") as fh: for chunk in iter(lambda: fh.read(_CHUNK), b""): h.update(chunk) actual = h.hexdigest() if actual != object_id: print( f"❌ Object {object_id[:8]} failed SHA-256 integrity check — " "store may be corrupt. Run `muse plumbing verify-pack` to audit.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) with obj.open("rb") as fh: for chunk in iter(lambda: fh.read(_CHUNK), b""): sys.stdout.buffer.write(chunk) except SystemExit: raise except OSError as exc: print( f"❌ Failed to read object: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR)