"""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 ``<repo_root>/.muse/objects/`` using a sharded directory
layout that mirrors Git's loose-object format::

    .muse/objects/<prefix>/<remainder>

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).

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.
"""

from __future__ import annotations

import fcntl
import functools
import hashlib
import logging
import os
import pathlib
import shutil
import sys
import tempfile
import time

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__)

_OBJECTS_DIR = "objects"
_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

# 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.
    Used after ``shutil.copy2`` which does not expose an fd.
    """
    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/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 = repo_root / ".muse" / _OBJECTS_DIR
    if not store.exists():
        return 0
    now = time.time()
    removed = 0
    for shard in store.iterdir():
        # Skip symlinked shard directories — never delete files inside an
        # attacker-controlled location that was swapped in via symlink.
        if not shard.is_dir() or shard.is_symlink():
            continue
        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()
                removed += 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 removed


def objects_dir(repo_root: pathlib.Path) -> pathlib.Path:
    """Return the path to the local object store root directory.

    The store lives at ``<repo_root>/.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 repo_root / ".muse" / _OBJECTS_DIR


def object_path(
    repo_root: pathlib.Path,
    object_id: str,
    prefix_len: int | 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::

        .muse/objects/<prefix>/<remainder>

    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 Muse repository.
        object_id:  SHA-256 hex digest of the object's content (64 chars).
        prefix_len: Shard prefix length override (2 or 4). ``None`` reads
                    from config.

    Returns:
        Absolute path to the object file (may not yet exist).

    Raises:
        ValueError: If *object_id* is not exactly 64 lowercase hex characters.
    """
    validate_object_id(object_id)
    plen = prefix_len if prefix_len is not None else _shard_prefix_len(repo_root)
    return objects_dir(repo_root) / object_id[:plen] / object_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 migration fallback.

    When the configured prefix length is 4 but the object was written with a
    2-char prefix (legacy), this function returns the 2-char path so callers
    transparently read existing objects after a sharding upgrade.

    Args:
        repo_root: Root of the Muse repository.
        object_id: SHA-256 hex digest (64 chars).

    Returns:
        The path that exists on disk, or the configured-prefix path if neither
        exists (callers can treat a non-existent return as "object absent").
    """
    plen = _shard_prefix_len(repo_root)
    primary = object_path(repo_root, object_id, prefix_len=plen)
    if primary.exists():
        return primary
    # Fallback: if primary uses 4-char prefix, also check 2-char (migration).
    if plen != _DEFAULT_SHARD_PREFIX_LEN:
        fallback = object_path(repo_root, object_id, prefix_len=_DEFAULT_SHARD_PREFIX_LEN)
        if fallback.exists():
            return fallback
    return primary


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 the configured shard path first; if not found and the configured
    prefix is 4, also checks the legacy 2-char path for migration compatibility.

    Args:
        repo_root: Root of the Muse repository.
        object_id: SHA-256 hex digest to check.
    """
    return _object_path_with_fallback(repo_root, object_id).exists()


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: SHA-256 hex digest that identifies this object (64 chars).
        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 64-char hex string, 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[:8]}… is {len(content):,} bytes, exceeding the "
            f"{MAX_OBJECT_WRITE_BYTES // (1024 * 1024)} MiB per-object write limit."
        )

    actual = hashlib.sha256(content).hexdigest()
    if actual != object_id:
        raise ValueError(
            f"Content integrity failure: expected object {object_id[:8]}…, "
            f"got {actual[:8]}…"
        )

    dest = object_path(repo_root, object_id)
    # Also check legacy 2-char path when configured prefix differs.
    if _object_path_with_fallback(repo_root, object_id).exists():
        logger.debug("⚠️ Object %s already in store — skipped", object_id[:8])
        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}/)")

    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(content)
            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)", object_id[:8], 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 ``shutil.copy2`` delegates to the OS copy
    mechanism, 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: SHA-256 hex digest of *src*'s content (64 chars).
        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 2-char path when configured prefix differs.
    if _object_path_with_fallback(repo_root, object_id).exists():
        logger.debug("⚠️ Object %s already in store — skipped", object_id[:8])
        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.
    h = hashlib.sha256()
    with src.open("rb") as fh:
        for chunk in iter(lambda: fh.read(65536), b""):
            h.update(chunk)
    actual = h.hexdigest()
    if actual != object_id:
        raise ValueError(
            f"Content integrity failure for {src}: expected {object_id[:8]}…, "
            f"got {actual[:8]}…"
        )

    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}/)")

    fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".obj-tmp-")
    tmp = pathlib.Path(tmp_str)
    try:
        os.close(fd)
        shutil.copy2(src, tmp)
        # Re-open tmp for an fd so we can fchmod and fsync via descriptor rather
        # than path.  Path-based chmod after close is vulnerable to
        # cleanup_stale_object_temps unlinking the temp between the close and
        # the chmod call.
        with tmp.open("rb") as fh:
            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)", object_id[:8], 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: SHA-256 hex digest 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 64-char hex string.
        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():
        logger.debug("⚠️ Object %s not found in local store", object_id[:8])
        return None
    size = dest.stat().st_size
    if size > MAX_FILE_BYTES:
        raise OSError(
            f"Object {object_id[:8]} 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 = h.hexdigest()
    if actual != object_id:
        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[:8],
            object_id[:16],
            actual[:16],
        )
        raise OSError(
            f"Object {object_id[:8]}… failed SHA-256 integrity check. "
            f"Expected digest starting {object_id[:16]}, got {actual[:16]}. "
            "The object store may be corrupt. Run `muse plumbing verify-pack` "
            "to audit the full store."
        )
    return b"".join(chunks)


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: SHA-256 hex digest of the desired object (64 chars).
        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 64-char hex string.
        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 == "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855":
            write_object(repo_root, object_id, b"")
            src = _object_path_with_fallback(repo_root, object_id)
            if not src.exists():
                return False
        else:
            logger.debug(
                "⚠️ Object %s not found in local store — cannot restore", object_id[:8]
            )
            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[:8]} is {size:,} bytes, exceeding the "
            f"{MAX_FILE_BYTES // (1024 * 1024)} MiB read limit."
        )
    h = hashlib.sha256()
    with src.open("rb") as fh:
        for chunk in iter(lambda: fh.read(65536), b""):
            h.update(chunk)
    actual = 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[:8], object_id[:16], actual[:16],
        )
        raise OSError(
            f"Object {object_id[:8]}… failed SHA-256 integrity check during restore. "
            f"Expected digest starting {object_id[:16]}, got {actual[:16]}. "
            "Run `muse plumbing verify-pack` to audit the store."
        )

    # Idempotent check-before-write: if dest already contains object_id's
    # 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; hashing is only
    # done when sizes match, amortising the cost over the common mismatch case.
    if dest.exists():
        try:
            if dest.stat().st_size == size:
                h_dest = hashlib.sha256()
                with dest.open("rb") as fh:
                    for chunk in iter(lambda: fh.read(65536), b""):
                        h_dest.update(chunk)
                if h_dest.hexdigest() == object_id:
                    logger.debug(
                        "✅ Object %s already at dest — skipping restore (inode preserved)",
                        object_id[:8],
                    )
                    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:
        os.close(fd)
        shutil.copy2(src, tmp)
        # Stored objects are 0o444 (immutable); shutil.copy2 would propagate that
        # mode to the restored working-tree file, making it read-only and
        # uneditable.  Working-tree files must be writable by their owner.
        os.chmod(tmp, 0o644)
        # shutil.copy2 copies the source mtime, which for object-store files is
        # the time the object was originally committed — potentially days or weeks
        # in the past.  After os.replace the destination would carry that old
        # mtime, causing editors (Cursor, VS Code, Vim) to see "new mtime <
        # cached mtime" and decide the file was not modified, keeping the stale
        # buffer open.  Resetting to "now" guarantees the destination always has
        # a fresh timestamp, so file-system watchers reliably trigger a buffer
        # reload for any content change.
        os.utime(tmp, None)
        _fsync_path(tmp)
        os.replace(tmp, dest)
    except Exception:
        tmp.unlink(missing_ok=True)
        raise
    logger.debug("✅ Restored object %s → %s", object_id[:8], dest)
    return True