"""Canonical content-addressed object store for the Muse VCS. All Muse commands that read or write blobs — ``muse commit``, ``muse read-tree``, ``muse reset`` — go through this module exclusively. No command may implement its own path logic or copy its own blobs. Layout ------ Objects are stored under ``/.muse/objects/`` using a sharded directory layout with an algorithm subdirectory:: .muse/objects/sha256// The algorithm directory (``sha256/``) makes the on-disk layout self-describing. The prefix length is configurable via ``[limits] shard_prefix_length`` in ``.muse/config.toml``: * **2** (default, 256 shards) — matches Git's layout; safe up to ~25 million objects before any shard exceeds 100 000 entries on ext4. * **4** (65 536 shards) — use for repos exceeding 10 million objects to keep per-shard counts below 1 000 at Linux-kernel scale (17 million objects). Legacy layout (no algo dir) is transparently supported via fallback — repos written before this layout change are read without any migration. When switching from 2-char to 4-char sharding, existing objects are still found via a fallback lookup at the 2-char path so no migration is required. File permissions ---------------- Every object file is written with mode ``0o444`` (read-only for all users). Content-addressing makes objects immutable by definition; the read-only mode expresses that constraint at the OS level, preventing accidental truncation by any tool that opens the file for writing. Startup cleanup --------------- A call to :func:`cleanup_stale_object_temps` removes any ``.obj-tmp-*`` files left behind by a previous SIGKILL. :func:`objects_dir` calls this automatically on the first access so callers never need to invoke it directly. This module is the single source of truth for all local object I/O. The store is append-only: writing the same object twice is always a no-op. """ import fcntl import functools import hashlib import logging import os import pathlib import sys import tempfile import time from collections.abc import Iterator from muse.core.types import DEFAULT_HASH_ALGO, blob_id, hash_file, long_id, short_id, split_id from muse.core.paths import objects_dir as _objects_dir from muse.core.validation import ( MAX_FILE_BYTES, MAX_OBJECT_WRITE_BYTES, assert_not_symlink, assert_write_inside_repo, validate_object_id, ) logger = logging.getLogger(__name__) _DEFAULT_SHARD_PREFIX_LEN: int = 2 _VALID_SHARD_PREFIX_LENS: frozenset[int] = frozenset({2, 4}) # Mode applied to every newly-written object: read-only for owner, group, other. # Content-addressed objects are immutable; this enforces that at the OS level. _OBJECT_MODE: int = 0o444 # Hex character set used by iter_stored_objects to reject stray files. _HEX_CHARS: frozenset[str] = frozenset("0123456789abcdef") # Minimum age (seconds) before cleanup_stale_object_temps will delete a temp # file. This prevents a concurrent process that just started up from deleting # temp files that belong to in-progress writes from another process. A write # that cannot complete within 60 seconds is effectively stalled and its temp # file is genuinely stale. _CLEANUP_MIN_AGE_SECS: float = 60.0 def _fsync_fd(fd: int) -> None: """Flush *fd* to durable storage, using the fastest safe syscall available. Flushes OS page-cache writes to durable storage before the caller performs ``os.replace``. Without it, a power loss between the kernel accepting the rename syscall and flushing the page cache can leave a zero-byte or partially-written file at the destination even though ``os.replace`` returned successfully (observed on APFS, ext4, btrfs). On macOS, ``os.fsync()`` maps to ``fcntl(F_FULLFSYNC)`` which forces a full hardware flush — ~4–5 ms per call on APFS. ``F_BARRIERFSYNC`` (85) provides the same crash-safety guarantee (write barrier to APFS journal) at ~0.1 ms per call. We prefer it on macOS and fall through to ``os.fsync()`` only if the kernel rejects it. On other platforms ``os.fsync()`` is used directly; on virtual filesystems (tmpfs, ramfs) where fsync is a no-op it is silently skipped. """ try: if sys.platform == "darwin": # F_BARRIERFSYNC (85) ensures APFS crash safety without a full # hardware flush — ~50× faster than F_FULLFSYNC on APFS SSDs. try: fcntl.fcntl(fd, 85) # F_BARRIERFSYNC return except OSError: pass # kernel too old or FS doesn't support it — fall through os.fsync(fd) except OSError: pass # best-effort — unsupported on some virtual filesystems def _fsync_path(path: pathlib.Path) -> None: """``fsync`` a file by path when no file descriptor is available. Opens the file read-only solely to obtain an fd, syncs, and closes. """ try: with path.open("rb") as fh: os.fsync(fh.fileno()) except OSError: pass # best-effort — unsupported on some virtual filesystems @functools.lru_cache(maxsize=16) def _cached_shard_prefix_len(repo_root_str: str) -> int: """LRU-cached shard prefix length reader. The shard prefix length is effectively static for the lifetime of a process: it is set once in ``.muse/config.toml`` and almost never changed. Caching eliminates redundant TOML file reads on every call to :func:`object_path` and :func:`_object_path_with_fallback` — both called at least twice per :func:`write_object` invocation. Cache capacity of 16 covers all realistic multi-repo scenarios in a single agent process. Values that fail validation fall through to the default and are never cached with an invalid value. """ from muse.cli.config import get_limit raw = get_limit("shard_prefix_length", pathlib.Path(repo_root_str)) return raw if raw in _VALID_SHARD_PREFIX_LENS else _DEFAULT_SHARD_PREFIX_LEN def _shard_prefix_len(repo_root: pathlib.Path) -> int: """Return the configured shard prefix length (2 or 4) for *repo_root*. Reads ``[limits] shard_prefix_length`` from ``.muse/config.toml``. Defaults to ``2`` when absent. Values outside ``{2, 4}`` are ignored and the default is used. Results are cached per repo root for the lifetime of the process — see :func:`_cached_shard_prefix_len`. """ return _cached_shard_prefix_len(str(repo_root)) # --------------------------------------------------------------------------- # Shard directory validation cache # --------------------------------------------------------------------------- # Set of shard directories that have been created AND path-traversal-validated # at least once this process run. Key: str(shard_dir). # # What is amortised (happens only on the FIRST write per shard): # • assert_write_inside_repo — expensive resolve() chain (~6 lstat calls) # • mkdir(parents=True) — stat checks on each parent component # # What is NOT amortised (happens on EVERY write): # • assert_not_symlink(shard_dir) — single lstat, essential TOCTOU guard # # Safety rationale: # assert_write_inside_repo guards against a symlink placed at shard_dir # BEFORE the first write. assert_not_symlink (below, in write_object) guards # against a concurrent swap AFTER the first write — because it fires just # before mkstemp, the race window between the check and the actual write is # nanoseconds, not milliseconds. Both guards must be in place; only the # expensive path-resolution check is amortised. _created_object_shards: set[str] = set() def _ensure_object_shard(repo_root: pathlib.Path, shard_dir: pathlib.Path) -> None: """Create and path-traversal-validate *shard_dir*, amortised per process run. On the first call for a given *shard_dir* this process run: 1. Validates *shard_dir* resolves inside *repo_root* (path-traversal guard, prevents a pre-placed symlink from redirecting writes outside the repo). 2. Creates the directory with ``mkdir -p``. 3. Records the shard as created so future calls skip steps 1 and 2. Subsequent calls for the same *shard_dir* return immediately — the ``resolve()`` chain and parent ``stat`` calls are not repeated. **The caller is still required to call** ``assert_not_symlink(shard_dir)`` after this function returns. That per-write check catches concurrent symlink-swap attacks (attacker removes the real shard dir and replaces it with a symlink between the path-traversal validation and the ``mkstemp`` call). Args: repo_root: Repository root (parent of ``.muse/``). shard_dir: The shard subdirectory (e.g. ``.muse/objects/sha256/ab/``). """ shard_str = str(shard_dir) if shard_str in _created_object_shards: return # First write to this shard this process run: path-traversal check + mkdir. assert_write_inside_repo(repo_root, shard_dir) shard_dir.mkdir(parents=True, exist_ok=True) _created_object_shards.add(shard_str) def cleanup_stale_object_temps(repo_root: pathlib.Path) -> int: """Remove stale ``.obj-tmp-*`` files left by a previous SIGKILL. A SIGKILL between :func:`write_object`'s ``mkstemp`` call and the ``os.replace`` rename leaves a partial object temp file in the shard directory. These files are safe to delete because the object was never committed to the store (the rename never happened). **Age gate**: only files older than :data:`_CLEANUP_MIN_AGE_SECS` seconds are deleted. This prevents a newly-started process from deleting temp files that belong to in-progress writes from another concurrently-running process — a genuine risk when multiple agents start simultaneously and one calls ``require_repo()`` while another is mid-write. A write that cannot complete within 60 seconds is treated as genuinely stalled. This function is called automatically by :func:`objects_dir` on first access and by :func:`muse.core.repo.require_repo` at startup. Args: repo_root: Root of the Muse repository. Returns: Number of stale temp files removed. """ store = _objects_dir(repo_root) if not store.exists(): return 0 now = time.time() removed = 0 def _clean_shard(shard: pathlib.Path) -> int: count = 0 for f in shard.iterdir(): if not (f.name.startswith(".obj-tmp-") or f.name.startswith(".restore-tmp-")): continue try: age = now - f.stat().st_mtime except OSError: continue # file disappeared between iterdir and stat — skip if age < _CLEANUP_MIN_AGE_SECS: logger.debug( "⏳ Skipping recent temp %s (age %.1fs < %ss — may be in-progress write)", f.name, age, _CLEANUP_MIN_AGE_SECS, ) continue try: f.unlink() count += 1 logger.warning( "⚠️ Removed stale object temp %s (left by prior crash)", f ) except OSError as exc: logger.warning("⚠️ Could not remove stale temp %s: %s", f, exc) return count algo_dir = store / DEFAULT_HASH_ALGO if not algo_dir.exists(): return removed for shard in algo_dir.iterdir(): if not shard.is_dir() or shard.is_symlink(): continue removed += _clean_shard(shard) return removed def objects_dir(repo_root: pathlib.Path) -> pathlib.Path: """Return the path to the local object store root directory. The store lives at ``/.muse/objects/``. Shard subdirectories are created lazily by :func:`write_object` and :func:`write_object_from_path`. Stale ``.obj-tmp-*`` files from a prior SIGKILL are swept by :func:`muse.core.repo.require_repo` at every command startup via :func:`cleanup_stale_object_temps`. Callers do not need to invoke the cleanup function directly. Args: repo_root: Root of the Muse repository (the directory containing ``.muse/``). Returns: Absolute path to the objects directory (may not yet exist). """ return _objects_dir(repo_root) def objects_algo_dir(repo_root: pathlib.Path) -> pathlib.Path: """Return the algorithm-prefixed subdirectory inside the object store. All objects are stored under ``//``. Tests and tools that need to create or inspect shard directories should use this function rather than hardcoding ``"sha256"``. Args: repo_root: Root of the Muse repository. Returns: Absolute path to the algorithm subdirectory (e.g. ``.muse/objects/sha256/``). """ return objects_dir(repo_root) / DEFAULT_HASH_ALGO def object_path( repo_root: pathlib.Path, object_id: str, prefix_len: int | None = None, *, objects_base: pathlib.Path | None = None, ) -> pathlib.Path: """Return the canonical on-disk path for a single object. Objects are sharded by the first *prefix_len* hex characters of their SHA-256 digest, nested under the algorithm subdirectory:: /sha256// The default *objects_base* is ``/.muse/objects/`` (client layout). Pass ``objects_base`` explicitly to use a different store root — for example, the server-side bare repo layout uses ``/objects/`` (no ``.muse/`` wrapper) and passes that path here directly. The sharding logic is identical regardless of which base is used. When *prefix_len* is ``None`` (the default) the value is read from ``[limits] shard_prefix_length`` in ``.muse/config.toml`` (default ``2``). Args: repo_root: Root of the repository (used to locate config and the default *objects_base*). object_id: Prefixed SHA-256 object ID (``sha256:<64hex>``). prefix_len: Shard prefix length override (2 or 4). ``None`` reads from config. objects_base: Override for the objects directory. Defaults to ``/.muse/objects/`` (client layout). Returns: Absolute path to the object file (may not yet exist). Raises: ValueError: If *object_id* is not a valid prefixed SHA-256 object ID. """ validate_object_id(object_id) _, hex_id = split_id(object_id) plen = prefix_len if prefix_len is not None else _shard_prefix_len(repo_root) base = objects_base if objects_base is not None else objects_dir(repo_root) return base / DEFAULT_HASH_ALGO / hex_id[:plen] / hex_id[plen:] def _object_path_with_fallback( repo_root: pathlib.Path, object_id: str ) -> pathlib.Path: """Return the path where *object_id* actually lives, with shard-prefix fallback. Checks in preference order: 1. Canonical layout with configured shard prefix — ``objects/sha256//`` 2. If the configured prefix is 4-char, also checks the 2-char shard prefix (migration path when upgrading from 2-char to 4-char sharding). Args: repo_root: Root of the Muse repository. object_id: Prefixed SHA-256 object ID (``sha256:<64hex>``). Returns: The path that exists on disk, or the primary path if the object is absent (callers treat a non-existent return as "absent"). """ validate_object_id(object_id) _, hex_id = split_id(object_id) plen = _shard_prefix_len(repo_root) store = objects_dir(repo_root) primary = store / DEFAULT_HASH_ALGO / hex_id[:plen] / hex_id[plen:] if primary.exists(): return primary # Shard-prefix migration: if configured for 4-char, also check 2-char. if plen != _DEFAULT_SHARD_PREFIX_LEN: alt = store / DEFAULT_HASH_ALGO / hex_id[:_DEFAULT_SHARD_PREFIX_LEN] / hex_id[_DEFAULT_SHARD_PREFIX_LEN:] if alt.exists(): return alt return primary def iter_stored_objects( repo_root: pathlib.Path, *, skip_symlinks: bool = True, ) -> Iterator[tuple[str, pathlib.Path]]: """Yield ``(prefixed_object_id, path)`` for every valid object in the store. This is the single canonical walker for the object store. It walks only ``objects/sha256//`` — the current layout after the algo-prefix migration. Only real files (not symlinks, not directories) whose shard + filename form a valid 64-char lowercase hex string are yielded. Stray files (``.DS_Store``, editor temporaries, etc.) are silently skipped. Args: repo_root: Root of the Muse repository. skip_symlinks: When ``True`` (default), symlinked shard dirs and symlinked object files are skipped. Set ``False`` only in tests that need to verify the guard. Yields: ``(object_id, path)`` where *object_id* is the canonical ``sha256:<64hex>`` string and *path* is the absolute on-disk path. """ store = objects_dir(repo_root) if not store.exists(): return seen: set[str] = set() def _is_hex(s: str) -> bool: return bool(s) and all(c in _HEX_CHARS for c in s) def _walk_shard(shard_dir: pathlib.Path) -> Iterator[tuple[str, pathlib.Path]]: prefix = shard_dir.name if not _is_hex(prefix): return for obj_file in shard_dir.iterdir(): if skip_symlinks and obj_file.is_symlink(): continue if not obj_file.is_file(): continue rest = obj_file.name if not _is_hex(rest): continue hex_id = prefix + rest if len(hex_id) != 64: continue oid = long_id(hex_id) if oid not in seen: seen.add(oid) yield oid, obj_file algo_dir = store / DEFAULT_HASH_ALGO if not algo_dir.exists() or not algo_dir.is_dir(): return if skip_symlinks and algo_dir.is_symlink(): return for shard_dir in algo_dir.iterdir(): if skip_symlinks and shard_dir.is_symlink(): continue if not shard_dir.is_dir(): continue yield from _walk_shard(shard_dir) def has_object(repo_root: pathlib.Path, object_id: str) -> bool: """Return ``True`` if *object_id* is present in the local store. Cheaper than :func:`read_object` when the caller only needs to check existence (e.g. to pre-flight a hard reset before touching the working tree). Checks loose objects first (canonical layout + legacy fallback), then falls through to the MPack local pack store. Args: repo_root: Root of the Muse repository. object_id: Prefixed SHA-256 object ID to check. """ if _object_path_with_fallback(repo_root, object_id).exists(): return True from muse.core.pack_store import has_object_in_packs return has_object_in_packs(repo_root, object_id) def write_object(repo_root: pathlib.Path, object_id: str, content: bytes) -> bool: """Write *content* to the local object store under *object_id*. If the object already exists (same ID = same content, content-addressed) the write is skipped and ``False`` is returned. Returns ``True`` when a new object was written. The shard directory is created on first write. Subsequent writes for the same ``object_id`` are no-ops — they never overwrite existing content. The content hash is verified against *object_id* before writing to prevent corrupt or malicious blobs from entering the store. Writes are atomic: content is written to a temp file then renamed, so a crash mid-write never leaves a partial object. Args: repo_root: Root of the Muse repository. object_id: Prefixed SHA-256 object ID that identifies this object. content: Raw bytes to persist. Returns: ``True`` if the object was newly written, ``False`` if it already existed (idempotent). Raises: ValueError: If *object_id* is not a valid prefixed object ID, or if the hash of *content* does not match *object_id*. """ validate_object_id(object_id) # Size cap: reject blobs that exceed the per-object write ceiling before # doing any I/O. This mirrors the read-time MAX_FILE_BYTES ceiling in # read_object so the same limit applies at both boundaries. if len(content) > MAX_OBJECT_WRITE_BYTES: raise ValueError( f"Object {object_id} is {len(content):,} bytes, exceeding the " f"{MAX_OBJECT_WRITE_BYTES // (1024 * 1024)} MiB per-object write limit." ) from muse.core.ids import hash_blob as _hash_blob actual = _hash_blob(content) if actual != object_id: raise ValueError( f"Content integrity failure: expected object {object_id} " f"got {actual}" ) dest = object_path(repo_root, object_id) # Also check legacy paths when configured prefix differs. if _object_path_with_fallback(repo_root, object_id).exists(): logger.debug("⚠️ Object %s already in store — skipped", short_id(object_id)) return False # Amortised path-traversal check + mkdir. The first write per shard calls # assert_write_inside_repo (expensive resolve chain) and mkdir; subsequent # writes to the same shard skip both. _ensure_object_shard(repo_root, dest.parent) # Per-write symlink guard: assert the shard is STILL a real directory. # This catches a concurrent attacker who removes the real shard and # replaces it with a symlink after our first-write validation. The check # fires just before mkstemp so the race window is nanoseconds wide. assert_not_symlink(dest.parent, label=f"object shard directory ({dest.parent.name}/)") # Store the muse-idiomatic format: " \0". # The hash of this full string equals object_id (verified above via hash_blob). blob_header = f"blob {len(content)}\0".encode() stored_bytes = blob_header + content fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".obj-tmp-") tmp = pathlib.Path(tmp_str) try: with os.fdopen(fd, "wb") as fh: fh.write(stored_bytes) fh.flush() # Set read-only via fd *before* closing — path-based os.chmod after # fh.close() is vulnerable to cleanup_stale_object_temps unlinking # the temp file between the close and the chmod call. os.fchmod # uses the open fd (immune to concurrent unlink of the path). os.fchmod(fh.fileno(), _OBJECT_MODE) _fsync_fd(fh.fileno()) os.replace(tmp, dest) except Exception: tmp.unlink(missing_ok=True) raise logger.debug("✅ Stored object %s (%d bytes)", short_id(object_id), len(content)) return True def write_object_from_path( repo_root: pathlib.Path, object_id: str, src: pathlib.Path, ) -> bool: """Copy *src* into the object store without loading it into memory. Preferred over :func:`write_object` for large blobs (dense MIDI renders, audio previews) because ``pathlib.Path.copy`` delegates to the OS copy mechanism (sendfile / copy_file_range), keeping the interpreter heap clean. Idempotent: if the object already exists it is never overwritten. The source file's hash is verified against *object_id* before writing. Writes are atomic (temp file + rename). Args: repo_root: Root of the Muse repository. object_id: Prefixed SHA-256 object ID of *src*'s content. src: Absolute path of the source file to store. Returns: ``True`` if the object was newly written, ``False`` if it already existed (idempotent). Raises: ValueError: If *object_id* is invalid or the file's hash does not match. """ validate_object_id(object_id) dest = object_path(repo_root, object_id) # Also check legacy paths when configured prefix differs. if _object_path_with_fallback(repo_root, object_id).exists(): logger.debug("⚠️ Object %s already in store — skipped", short_id(object_id)) return False # Guard BEFORE any filesystem modification. assert_write_inside_repo(repo_root, dest) # Size cap: stat the source file before reading it. This avoids loading # an oversized blob into the hash loop and then discovering it's too large. src_size = src.stat().st_size if src_size > MAX_OBJECT_WRITE_BYTES: raise ValueError( f"Source file {src.name!r} is {src_size:,} bytes, exceeding the " f"{MAX_OBJECT_WRITE_BYTES // (1024 * 1024)} MiB per-object write limit." ) # Verify hash before writing. actual = hash_file(src) if actual != object_id: raise ValueError( f"Content integrity failure for {src}: expected {object_id} " f"got {actual}" ) dest.parent.mkdir(parents=True, exist_ok=True) # Second guard after mkdir: catch symlink swap between the resolve check # and mkdir completing. assert_not_symlink(dest.parent, label=f"object shard directory ({dest.parent.name}/)") # Write the muse blob format: "blob \0". # Matches write_object's on-disk layout exactly — the two functions must # produce structurally identical object files. blob_header = f"blob {src_size}\0".encode() fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".obj-tmp-") tmp = pathlib.Path(tmp_str) try: with os.fdopen(fd, "wb") as fh: fh.write(blob_header) with src.open("rb") as src_fh: for chunk in iter(lambda: src_fh.read(65536), b""): fh.write(chunk) os.fchmod(fh.fileno(), _OBJECT_MODE) _fsync_fd(fh.fileno()) os.replace(tmp, dest) except Exception: tmp.unlink(missing_ok=True) raise logger.debug("✅ Stored object %s (%s)", short_id(object_id), src.name) return True def read_object(repo_root: pathlib.Path, object_id: str) -> bytes | None: """Read and return the raw bytes for *object_id* from the local store. Every read re-verifies the SHA-256 digest of the returned bytes against *object_id*. If the file has been silently corrupted (disk error, cosmic ray, filesystem bug, or deliberate tamper) this function raises ``OSError`` rather than returning bad data. Silent corruption is the worst possible failure mode for a version-control system. Returns ``None`` when the object is not present in the store so callers can produce a user-facing error rather than raising ``FileNotFoundError``. Args: repo_root: Root of the Muse repository. object_id: Prefixed SHA-256 object ID of the desired object. Returns: Raw bytes, or ``None`` when the object is absent from the store. Raises: ValueError: If *object_id* is not a valid prefixed object ID. OSError: If the object file exceeds MAX_FILE_BYTES or fails the SHA-256 integrity check. """ validate_object_id(object_id) dest = _object_path_with_fallback(repo_root, object_id) if not dest.exists(): from muse.core.pack_store import read_object_from_packs return read_object_from_packs(repo_root, object_id) size = dest.stat().st_size if size > MAX_FILE_BYTES: raise OSError( f"Object {object_id} is {size} bytes, exceeding the " f"{MAX_FILE_BYTES // (1024 * 1024)} MiB read limit." ) # Stream-hash while reading so memory usage stays constant for large blobs. h = hashlib.sha256() chunks: list[bytes] = [] with dest.open("rb") as fh: for chunk in iter(lambda: fh.read(65536), b""): h.update(chunk) chunks.append(chunk) actual = long_id(h.hexdigest()) if actual != object_id: # Recovery: bare objects written without the typed header have a hash of # their raw content rather than hash("blob \0" + content). # If blob_id(raw) matches the expected object_id the content is intact. data_raw = b"".join(chunks) from muse.core.ids import hash_blob as _hash_blob if _hash_blob(data_raw) == object_id: logger.warning( "⚠️ Object %s has no typed header (bare format). " "Content intact. Run `muse code migrate` to normalise.", short_id(object_id), ) return data_raw logger.critical( "❌ Object %s failed integrity check — store may be corrupt " "(expected %s, got %s). Do not trust any data from this object.", object_id, object_id, actual, ) raise OSError( f"Object {object_id} failed SHA-256 integrity check. " f"Expected {object_id}, got {actual}. " "The object store may be corrupt. Run `muse verify-pack` " "to audit the full store." ) data = b"".join(chunks) # Strip muse object header ("blob \0", "commit \0", etc.) # if present. The header is valid when the type token and declared size # both match — this avoids misidentifying raw blobs that happen to contain # an early null byte. null_idx = data.find(b"\0") if null_idx > 0: header = data[:null_idx].decode(errors="replace") parts = header.split(" ", 1) if len(parts) == 2 and parts[1].isdigit(): payload = data[null_idx + 1:] if len(payload) == int(parts[1]): return payload return data # --------------------------------------------------------------------------- # Muse object store — idiomatic format # # On-disk layout mirrors the approach used by Git: # " \0" # # The full string (header + payload) is hashed to produce the object ID, # so the type is part of the object's identity — two objects with identical # payload bytes but different types will never share an ID. # --------------------------------------------------------------------------- def write_muse_object( repo_root: pathlib.Path, type_str: str, payload: bytes, ) -> str: """Write a Muse object to the unified store. Prepends ``" \\0"`` to *payload*, hashes the result to produce the object ID, and writes to ``objects//<2>/<62>``. Idempotent — returns the object ID without writing if already present. Args: repo_root: Repository root. type_str: Object type — ``"blob"``, ``"snapshot"``, or ``"commit"``. payload: Raw bytes to store. Returns: The prefixed SHA-256 object ID. """ header = f"{type_str} {len(payload)}\0".encode() full = header + payload object_id = long_id(hashlib.sha256(full).hexdigest()) path = object_path(repo_root, object_id) if path.exists(): return object_id path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(full) return object_id def read_muse_object( repo_root: pathlib.Path, object_id: str, ) -> tuple[str, bytes] | None: """Read a Muse object from the unified store. Returns ``(type_str, payload)`` or ``None`` if the object is absent or malformed (callers treat malformed the same as missing). Args: repo_root: Repository root. object_id: Prefixed SHA-256 object ID. """ path = object_path(repo_root, object_id) if not path.exists(): return None data = path.read_bytes() try: null_idx = data.index(b"\0") type_str = data[:null_idx].decode().split(" ", 1)[0] return type_str, data[null_idx + 1:] except (ValueError, UnicodeDecodeError): return None def restore_object( repo_root: pathlib.Path, object_id: str, dest: pathlib.Path, ) -> bool: """Copy an object from the store to *dest*, preserving the destination inode. When *dest* already contains the correct content this function returns immediately **without any write**. Skipping the write keeps the inode number and mtime unchanged, which is critical for editors (Cursor, VS Code, Vim, …) that watch open buffers via inode-based filesystem events. A spurious ``os.replace`` (rename syscall) replaces the inode even when the bytes are identical; the editor was watching the old inode and goes blind, leaving a permanently stale buffer that shows the previous file content until the user closes and reopens the tab. When *dest* does not yet exist or contains different content, the write is done atomically: content is staged to a sibling temp file and renamed into place via :func:`os.replace` so a partially-written destination is never visible to other processes. Creates parent directories of *dest* if they do not exist. The caller is responsible for ensuring *dest* is within a safe base directory (use :func:`muse.core.validation.contain_path` before calling). Args: repo_root: Root of the Muse repository. object_id: Prefixed SHA-256 object ID of the desired object. dest: Absolute path to write the restored file. Returns: ``True`` on success, ``False`` if the object is not in the store. Raises: ValueError: If *object_id* is not a valid prefixed object ID. OSError: If the atomic write fails after the object was found. """ validate_object_id(object_id) src = _object_path_with_fallback(repo_root, object_id) if not src.exists(): # The empty-content object (SHA-256("") = e3b0c44…) is deterministic — # its content is always b"". Repos pushed before the server-side # empty-object fix may be missing it from the store. Synthesize it # inline so apply_manifest does not fail, and persist it into the store # for consistency so future calls find it normally. if object_id == blob_id(b""): write_object(repo_root, object_id, b"") src = _object_path_with_fallback(repo_root, object_id) if not src.exists(): return False else: # Object may be in a pack file (Phase 2+). Extract it to the loose # store so the path-based copy path below can proceed normally. from muse.core.pack_store import read_object_from_packs packed_bytes = read_object_from_packs(repo_root, object_id) if packed_bytes is None: logger.debug( "⚠️ Object %s not found in local store or packs — cannot restore", short_id(object_id), ) return False write_object(repo_root, object_id, packed_bytes) src = _object_path_with_fallback(repo_root, object_id) if not src.exists(): return False # Re-verify the source object's hash before restoring it to the working # tree. The in-store file is 0o444 (immutable) but disk errors or cosmic # rays can corrupt it silently after it was written. Detecting corruption # here prevents restoring bad bytes to the working tree without error. size = src.stat().st_size if size > MAX_FILE_BYTES: raise OSError( f"Object {object_id} is {size:,} bytes, exceeding the " f"{MAX_FILE_BYTES // (1024 * 1024)} MiB read limit." ) # Hash the raw object file bytes to verify integrity. Object files store # the full muse blob format ("blob \0"), so the raw bytes # ARE the hash input — do NOT call hash_file() here because that function # now computes the blob-prefix hash for working-tree files (raw content). _h = hashlib.sha256() with src.open("rb") as _fh: for _chunk in iter(lambda: _fh.read(65536), b""): _h.update(_chunk) actual = long_id(_h.hexdigest()) if actual != object_id: logger.critical( "❌ Object %s failed integrity check during restore" "(expected %s, got %s). Refusing to restore corrupt data.", object_id, object_id, actual, ) raise OSError( f"Object {object_id} failed SHA-256 integrity check during restore. " f"Expected {object_id}, got {actual}. " "Run `muse verify-pack` to audit the store." ) # Read the raw object file and strip the muse header to get the actual # content bytes ("blob \0" → ""). raw = src.read_bytes() null_idx = raw.find(b"\0") if null_idx > 0: header = raw[:null_idx].decode(errors="replace") parts = header.split(" ", 1) if len(parts) == 2 and parts[1].isdigit(): payload = raw[null_idx + 1:] if len(payload) == int(parts[1]): content = payload else: content = raw else: content = raw else: content = raw # Idempotent check-before-write: if dest already contains the same bytes, # skip the rename entirely. Preserving the inode keeps any editor # watching the file alive — an atomic rename (os.replace) always produces a # new inode, silently blinding inode-based watchers even for identical # content. Size is checked first as a cheap pre-filter. if dest.exists(): try: if dest.stat().st_size == len(content) and dest.read_bytes() == content: logger.debug( "✅ Object %s already at dest — skipping restore (inode preserved)", short_id(object_id), ) return True except OSError: pass # dest unreadable or raced away — fall through to the write path dest.parent.mkdir(parents=True, exist_ok=True) fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".restore-tmp-") tmp = pathlib.Path(tmp_str) try: with os.fdopen(fd, "wb") as fh: fh.write(content) os.chmod(tmp, 0o644) _fsync_path(tmp) os.replace(tmp, dest) except Exception: tmp.unlink(missing_ok=True) raise logger.debug("✅ Restored object %s → %s", short_id(object_id), dest) return True