"""muse hash-object — compute the SHA-256 object ID of a file. Computes the content-addressed object ID (SHA-256 hex digest) of a file or stdin stream. With ``--write`` the object is also stored in ``.muse/objects/`` so it can be referenced by future snapshots and commits. Output (JSON, default):: {"object_id": "", "stored": false} Output (--format text):: Output contract --------------- - Exit 0: hash computed successfully. - Exit 1: file not found, path is a directory, or unknown --format value. - Exit 3: I/O error writing to the store, or integrity check failed. Agent use --------- Read from stdin to avoid temp-file overhead:: echo -n "hello" | muse hash-object --stdin cat large_file.bin | muse hash-object --stdin --write """ import argparse import json import logging import pathlib import sys from collections.abc import Callable from typing import TypedDict from muse.core.types import blob_id from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.object_store import write_object, write_object_from_path from muse.core.repo import find_repo_root from muse.core.snapshot import hash_file from muse.core.validation import sanitize_display from muse.core.timing import start_timer logger = logging.getLogger(__name__) _FORMAT_CHOICES = ("json", "text") class _HashObjectJson(EnvelopeJson): """JSON output for ``muse hash-object --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ object_id Full content-addressed object ID: ``sha256:<64 hex chars>``. Deterministic — identical bytes always produce the same ID. stored True when the object was written to ``.muse/objects/`` (i.e. ``--write`` was passed and the object was not already present). size_bytes Size of the hashed input in bytes. """ object_id: str stored: bool size_bytes: int def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the hash-object subcommand.""" parser = subparsers.add_parser( "hash-object", help="SHA-256 a file; optionally store it in the object store.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "path", type=pathlib.Path, nargs="?", default=None, help="File to hash. Omit when using --stdin.", ) parser.add_argument( "--stdin", action="store_true", help="Read content from stdin instead of a file path.", ) parser.add_argument( "--write", "-w", action="store_true", help="Store the object in .muse/objects/.", ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON.", ) parser.set_defaults(func=run) def _hash_bytes(data: bytes) -> str: """Return the canonical ``sha256:`` object ID for *data*.""" return blob_id(data) def run(args: argparse.Namespace) -> None: """Compute the SHA-256 object ID of a file or stdin stream. The object ID is deterministic — identical bytes always produce the same ID. Use ``--write`` to store the object so it can be referenced by future ``muse commit-tree`` calls; use ``--stdin`` to hash content piped from another command without writing a temporary file. Agent quickstart ---------------- :: muse hash-object file.py --format json muse hash-object file.py --write --format json echo "content" | muse hash-object --stdin --format json muse hash-object file.py --stdin --write --format json JSON fields ----------- object_id Full ``sha256:`` object ID. stored ``true`` if the object was written to the store (``--write``). size_bytes Byte length of the content hashed. Exit codes ---------- 0 Success. 1 Invalid arguments or conflicting flags. 3 I/O error reading file or writing to the object store. """ elapsed = start_timer() json_out: bool = args.json_out path: pathlib.Path | None = args.path use_stdin: bool = args.stdin write: bool = args.write if use_stdin and path is not None: print("❌ Cannot use both a path argument and --stdin.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not use_stdin and path is None: print("❌ Provide a file path or use --stdin.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── stdin mode ─────────────────────────────────────────────────────────── if use_stdin: data = sys.stdin.buffer.read() object_id = _hash_bytes(data) size_bytes = len(data) stored = False if write: root = find_repo_root(pathlib.Path.cwd()) if root is None: print( "❌ Not inside a Muse repository. Cannot write object.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) try: stored = write_object(root, object_id, data) except ValueError as exc: print( f"❌ Integrity check failed: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) except OSError as exc: print( f"❌ Failed to write object: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) _emit(json_out, object_id, stored, size_bytes, elapsed) return # ── file mode ──────────────────────────────────────────────────────────── assert path is not None # Null bytes are not valid in POSIX paths; pathlib raises ValueError on # Python ≥3.12 but the explicit check is clearer and version-independent. try: _path_str = str(path) except Exception: _path_str = repr(path) if "\x00" in _path_str: print("❌ Path contains a null byte.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not path.exists(): print( f"❌ Path does not exist: {sanitize_display(str(path))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if path.is_dir(): print( f"❌ Path is a directory, not a file: {sanitize_display(str(path))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Reject special files (block/char devices, named pipes, sockets). # These would either block forever (FIFO), return device-specific data # (block/char), or fail unpredictably (socket). Only regular files and # symlinks to regular files are accepted — consistent with git hash-object. if not path.is_file(): print( f"❌ Path is not a regular file (device, pipe, or socket): " f"{sanitize_display(str(path))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) object_id = hash_file(path) size_bytes = path.stat().st_size stored = False if write: root = find_repo_root(pathlib.Path.cwd()) if root is None: print("❌ Not inside a Muse repository. Cannot write object.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) try: # write_object_from_path streams at 64 KiB chunks — heap-safe for # arbitrarily large blobs. Re-verifies hash before writing to catch # any TOCTOU race between hash_file() above and the store write. stored = write_object_from_path(root, object_id, path) except ValueError as exc: print( f"❌ Integrity check failed (file may have changed during write): " f"{sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) except OSError as exc: print( f"❌ Failed to write object: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) _emit(json_out, object_id, stored, size_bytes, elapsed) def _emit(json_out: bool, object_id: str, stored: bool, size_bytes: int, elapsed: Callable[[], float]) -> None: """Print the result in the requested format.""" if not json_out: print(object_id) else: print(json.dumps(_HashObjectJson( **make_envelope(elapsed), object_id=object_id, stored=stored, size_bytes=size_bytes, )))