"""File-based commit and snapshot store for the Muse VCS. All commit and snapshot metadata is stored as **msgpack** files under ``.muse/`` — no external database required. msgpack is used instead of JSON because it is 3–6× faster to serialize/deserialize on real repository data and is already a first-class dependency (also used by the MWP wire protocol and ``SymbolCache``). All user-facing output (``muse log --json``, ``muse show``, etc.) is still JSON built from the deserialized in-memory structures. Writes are **durable and atomic**: data is flushed with ``fsync`` to a unique ``mkstemp`` temp file then renamed over the target file, so a crash or power loss mid-write never corrupts store data. Layout ------ .muse/ commits/.msgpack — one msgpack file per commit snapshots/.msgpack — one msgpack file per snapshot manifest tags//.msgpack — tag records releases//.msgpack — release records objects// — content-addressed blobs (via object_store.py) refs/heads/ — branch HEAD pointers (plain text commit IDs) HEAD — "ref: refs/heads/main" | "commit: " repo.json — repository identity (human-editable, JSON) Commit schema (in-memory dict shape, serialized as msgpack) ----------------------------------------------------------- { "commit_id": "", "repo_id": "", "branch": "main", "parent_commit_id": null | "", "parent2_commit_id": null | "", "snapshot_id": "", "message": "Add verse melody", "author": "gabriel", "committed_at": "2026-03-16T12:00:00+00:00", "metadata": {} } Snapshot schema (in-memory dict shape, serialized as msgpack) ------------------------------------------------------------- { "snapshot_id": "", "manifest": {"tracks/drums.mid": "", ...}, "created_at": "2026-03-16T12:00:00+00:00" } All functions are synchronous — file I/O on a local ``.muse/`` directory does not require async. This removes the SQLAlchemy/asyncpg dependency from the CLI entirely. """ from __future__ import annotations import datetime import errno import fcntl import logging import os import pathlib import re import sys import tempfile import uuid import msgpack from dataclasses import dataclass, field from typing import Literal, TypedDict, TypeGuard from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.validation import ( assert_not_symlink, sanitize_glob_prefix, validate_branch_name, validate_ref_id, validate_repo_id, ) from muse.core.semver import ( ApiChangeSummary, ChangelogEntry, FileHotspot, LanguageStat, ReleaseChannel, RefactorEventSummary, SemanticReleaseReport, SemVerTag, SymbolKindCount, _CHANNEL_MAP, _ChannelMap, parse_semver, semver_channel, semver_to_str, ) from muse.domain import SemVerBump, StructuredDelta logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Read safety constants # --------------------------------------------------------------------------- # Maximum size of a msgpack file that _read_msgpack will load into memory. # Commits, snapshots, tags, and releases should never approach this limit in # any realistic repository. It exists to prevent a malicious or corrupted # store file from triggering an OOM before read_bytes() is even called. # Callers (read_commit, read_snapshot, etc.) catch OSError and return None. MAX_MSGPACK_BYTES: int = 64 * 1024 * 1024 # 64 MiB # Upper bound for pack/bundle deserialization. Pack files legitimately # contain many binary blobs (raw file content), so they need a larger cap. # Matches MAX_OBJECT_WRITE_BYTES * 2 — a generous but bounded envelope. MAX_PACK_MSGPACK_BYTES: int = 512 * 1024 * 1024 # 512 MiB # Per-value limits passed to msgpack.unpackb. Even if a file is within the # size cap, a pathological document with a single enormous string, an # astronomically deep nesting, or a billion-entry map can still exhaust RAM. # These limits are deliberately generous for real workloads (snapshot manifests # with 75k+ files, long commit messages) while rejecting absurd inputs. _MSGPACK_MAX_STR_LEN: int = 1_048_576 # 1 MiB per string value _MSGPACK_MAX_BIN_LEN: int = 0 # no binary blobs in commit/snapshot/tag records _MSGPACK_MAX_BIN_LEN_PACK: int = 256 * 1024 * 1024 # 256 MiB per blob in pack/bundle files _MSGPACK_MAX_ARRAY_LEN: int = 1_000_000 # generous for large manifests _MSGPACK_MAX_MAP_LEN: int = 1_000_000 # generous for 75k-file snapshot manifests # --------------------------------------------------------------------------- # Serialization boundary types # --------------------------------------------------------------------------- # ``MsgpackValue`` is the recursive union of every value type that # ``msgpack.unpackb(raw=False)`` can produce. Using this type (instead of # ``object`` or ``Any``) keeps deserialization boundaries fully typed while # preserving the fact that the exact shape of a persisted record is unknown # until the caller narrows it via ``isinstance``. # # Mirrors the Rust ``Value`` enum that will own this boundary after the port: # enum Value { Nil, Bool(bool), Int(i64), Float(f64), Bytes(Vec), # Str(String), Array(Vec), Map(HashMap) } # Type aliases are defined in _types.py to avoid circular imports. # Re-exported here so callers can import from muse.core.store as before. from muse.core._types import ( # noqa: E402 BranchHeads, JsonValue, Manifest, MsgpackDict, MsgpackValue, Metadata, ) # --------------------------------------------------------------------------- # Typed accessor helpers for deserialized msgpack dicts # --------------------------------------------------------------------------- def _str_val( d: MsgpackDict, key: str, default: str = "" ) -> str: """Return ``d[key]`` as ``str``, or *default* if absent or wrong type. Never raises — callers that need strict validation should check the return value or use the runtime guards inside ``from_dict``. """ val = d.get(key, default) return val if isinstance(val, str) else default def _int_val( d: MsgpackDict, key: str, default: int = 0 ) -> int: """Return ``d[key]`` as ``int``, or *default* if absent or wrong type. Excludes ``bool`` (which is a subclass of ``int`` in Python) to prevent accidental coercion. """ val = d.get(key, default) return val if isinstance(val, int) and not isinstance(val, bool) else default def _str_or_none( d: MsgpackDict, key: str ) -> str | None: """Return ``d[key]`` as ``str | None``. Returns ``None`` if the key is absent or the value is not a ``str``. """ val = d.get(key) return val if isinstance(val, str) else None def _str_list( d: MsgpackDict, key: str ) -> list[str]: """Return ``d[key]`` as ``list[str]``, filtering non-string elements.""" val = d.get(key) if not isinstance(val, list): return [] return [x for x in val if isinstance(x, str)] def _str_dict( d: MsgpackDict, key: str ) -> Manifest: """Return ``d[key]`` as ``dict[str, str]``, filtering non-string values.""" val = d.get(key) if not isinstance(val, dict): return {} return {k: v for k, v in val.items() if isinstance(v, str)} def _as_structured_delta( v: MsgpackDict, ) -> TypeGuard[StructuredDelta]: """Narrow a raw msgpack dict to :class:`StructuredDelta`. ``StructuredDelta`` is ``TypedDict(total=False)`` — every field is optional — so any ``dict`` is structurally compatible. This TypeGuard exists solely to satisfy the type checker without ``# type: ignore``. """ return True def _sem_ver_bump_val(d: MsgpackDict) -> "SemVerBump": """Extract and validate a ``sem_ver_bump`` field from a raw msgpack dict. Falls back to ``"none"`` if the value is absent or not a recognised Literal — guards against tampered or forward-versioned records. """ val = _str_val(d, "sem_ver_bump", "none") if val == "major": return "major" if val == "minor": return "minor" if val == "patch": return "patch" return "none" def _parse_semver_tag( d: MsgpackDict, key: str ) -> "SemVerTag": """Extract a nested :class:`SemVerTag` from a raw msgpack mapping. Returns a zeroed ``SemVerTag`` if the key is absent or the value is not a dict — callers can inspect ``major == 0`` to detect the fallback. """ val = d.get(key) if isinstance(val, dict): return SemVerTag( major=_int_val(val, "major"), minor=_int_val(val, "minor"), patch=_int_val(val, "patch"), pre=_str_val(val, "pre"), build=_str_val(val, "build"), ) return SemVerTag(major=0, minor=0, patch=0, pre="", build="") def _parse_changelog_entries( d: MsgpackDict, key: str ) -> "list[ChangelogEntry]": """Extract a ``list[ChangelogEntry]`` from a raw msgpack mapping. Silently skips items that are not dicts — defensive deserialization at the boundary between wire format and in-memory types. """ val = d.get(key) if not isinstance(val, list): return [] entries: list[ChangelogEntry] = [] for item in val: if not isinstance(item, dict): continue entries.append( ChangelogEntry( commit_id=_str_val(item, "commit_id"), message=_str_val(item, "message"), sem_ver_bump=_sem_ver_bump_val(item), breaking_changes=_str_list(item, "breaking_changes"), author=_str_val(item, "author"), committed_at=_str_val(item, "committed_at"), agent_id=_str_val(item, "agent_id"), model_id=_str_val(item, "model_id"), ) ) return entries # --------------------------------------------------------------------------- # msgpack I/O helpers # --------------------------------------------------------------------------- def write_text_atomic( path: pathlib.Path, text: str, encoding: str = "utf-8", ) -> None: """Write *text* to *path* atomically and durably. Uses the same ``mkstemp`` → ``flush`` → ``fsync`` → ``rename`` pattern as :func:`_write_msgpack_atomic` so every VCS state file — HEAD, branch refs, merge state, config — is protected against: * **Torn writes**: a crash mid-write leaves the old file intact (atomic rename). * **Page-cache loss**: a power loss after rename cannot produce an empty file (fsync ensures data is durable before the rename commits it). * **Concurrent writer races**: every caller gets its own mkstemp name so two concurrent writers to the same path do not corrupt each other's temp file. ``fsync`` failure is non-fatal and is silently ignored — some virtual filesystems (tmpfs, certain Docker volumes) do not support it. The atomicity guarantee (torn-write protection) still holds. Raises: ValueError: If *path*'s parent directory is a symlink (symlink-swap attack guard — writing into a symlinked directory would redirect the write outside the repository root). """ path.parent.mkdir(parents=True, exist_ok=True) # Guard: reject symlinked parent directories to prevent symlink-swap # attacks that redirect store writes to attacker-controlled locations. assert_not_symlink(path.parent, label=f"write target parent ({path.parent.name}/)") fd, tmp_str = tempfile.mkstemp(dir=path.parent, prefix=".muse-tmp-") tmp = pathlib.Path(tmp_str) try: with os.fdopen(fd, "w", encoding=encoding) as fh: fh.write(text) fh.flush() try: os.fsync(fh.fileno()) except OSError as exc: if exc.errno not in (errno.EINVAL, None): raise tmp.replace(path) except OSError: tmp.unlink(missing_ok=True) raise def write_branch_ref( repo_root: pathlib.Path, branch: str, commit_id: str, ) -> None: """Atomically update the branch tip pointer in ``.muse/refs/heads/``. This is the **canonical** way to advance a branch ref. All commands that record a new commit on a branch — ``commit``, ``merge``, ``cherry-pick``, ``revert``, ``reset``, ``pull``, ``rebase`` — must call this function rather than writing the ref file directly. Using a bare ``path.write_text()`` is forbidden for ref files because: * It is not atomic — a crash mid-write leaves a zero-length or partial file, orphaning all commits reachable only from this branch. * It is not fsynced — a power loss after the write syscall returns but before the page cache is flushed produces the same corruption. Args: repo_root: Repository root (parent of ``.muse/``). branch: Branch name; validated before use. commit_id: 64-char SHA-256 hex string of the new tip commit. Raises: ValueError: If *branch* or *commit_id* is invalid. """ from muse.core.validation import validate_branch_name validate_branch_name(branch) if not re.fullmatch(r"[0-9a-f]{64}", commit_id): raise ValueError(f"commit_id must be 64 hex chars, got: {commit_id!r}") ref_path = repo_root / ".muse" / "refs" / "heads" / branch write_text_atomic(ref_path, commit_id) # Store parent directories that have already been validated as non-symlinks # this process run. The same amortisation strategy as object_store.py's # _validated_object_shards: commits/, snapshots/, tags/ are checked once. _validated_store_parents: set[str] = set() def _write_msgpack_atomic( path: pathlib.Path, data: "CommitDict | SnapshotDict | TagDict | ReleaseDict", ) -> None: """Serialize *data* to msgpack and atomically replace *path*. Uses a unique ``mkstemp`` temp file → ``fsync`` → ``rename`` pattern so: - A crash mid-write never leaves a corrupt store file (atomic rename). - A power loss after rename returns but before the page cache is flushed cannot produce a zero-length or partially-written file (fsync ensures the data is durable before the rename commits it). Raises: ValueError: If *path*'s parent directory is a symlink (symlink-swap attack guard). """ # Guard: reject symlinked parent directories to prevent symlink-swap # attacks that redirect metadata writes to attacker-controlled locations. # Amortised: after the first write to a given parent directory this process # run the check is skipped — the parent cannot become a symlink without # write access to .muse/, which would represent a more fundamental breach. parent_str = str(path.parent) if parent_str not in _validated_store_parents: assert_not_symlink(path.parent, label=f"write target parent ({path.parent.name}/)") _validated_store_parents.add(parent_str) packed = msgpack.packb(data, use_bin_type=True) fd, tmp_str = tempfile.mkstemp(dir=path.parent, prefix=".muse-tmp-") tmp = pathlib.Path(tmp_str) try: with os.fdopen(fd, "wb") as fh: fh.write(packed) fh.flush() try: if sys.platform == "darwin": # F_BARRIERFSYNC (85) provides the same APFS crash-safety # guarantee as F_FULLFSYNC at ~50× lower latency (~0.1 ms # vs ~5 ms per call). Fall through to os.fsync() if the # kernel rejects it (old macOS or non-APFS filesystem). try: fcntl.fcntl(fh.fileno(), 85) # F_BARRIERFSYNC except OSError: os.fsync(fh.fileno()) else: os.fsync(fh.fileno()) except OSError as exc: if exc.errno != errno.EINVAL: raise tmp.replace(path) except OSError: tmp.unlink(missing_ok=True) raise def safe_unpackb( raw: bytes, *, context: str = "", max_bytes: int = MAX_MSGPACK_BYTES, strict_map_key: bool = True, allow_binary: bool = False, ) -> MsgpackValue: """Deserialize msgpack bytes with strict, configurable safety limits. This is the single canonical entry-point for all in-memory msgpack deserialization in Muse. Every callsite that receives untrusted bytes (stdin, network, cache files) must go through here. Args: raw: The msgpack bytes to unpack. context: Human-readable label for error messages (e.g. "stdin", "server response"). Omit for internal callers. max_bytes: Hard cap on ``len(raw)`` checked *before* any parsing, preventing OOM from size-bomb payloads. Defaults to :data:`MAX_MSGPACK_BYTES` (64 MiB). Use :data:`MAX_PACK_MSGPACK_BYTES` for pack/bundle data. strict_map_key: When ``True`` (default) only ``str`` keys are accepted in msgpack maps, matching Muse's own serialisation convention. Set ``False`` only for legacy formats that may carry integer keys (e.g. the staging index v1). allow_binary: When ``False`` (default) binary blobs are rejected (``max_bin_len=0``), enforcing that commit/snapshot/tag records never embed raw bytes. Set ``True`` for pack/bundle payloads that legitimately carry blob content. Raises: ValueError: If ``len(raw)`` exceeds *max_bytes* — checked before any allocation from ``unpackb``. msgpack.UnpackException: If the bytes are not valid msgpack. msgpack.ExtraData: If there is trailing data after the top-level value. """ n = len(raw) if n > max_bytes: label = f" ({context})" if context else "" raise ValueError( f"Msgpack payload{label} is {n:,} bytes — exceeds the " f"{max_bytes // (1024 * 1024)} MiB safety cap. " "Possible size-bomb or corrupted input." ) result: MsgpackValue = msgpack.unpackb( raw, raw=False, strict_map_key=strict_map_key, max_str_len=_MSGPACK_MAX_STR_LEN, max_bin_len=_MSGPACK_MAX_BIN_LEN_PACK if allow_binary else _MSGPACK_MAX_BIN_LEN, max_array_len=_MSGPACK_MAX_ARRAY_LEN, max_map_len=_MSGPACK_MAX_MAP_LEN, ) return result def read_msgpack_file( path: pathlib.Path, *, max_bytes: int = MAX_MSGPACK_BYTES, strict_map_key: bool = True, allow_binary: bool = False, ) -> MsgpackValue: """Read a msgpack file and deserialize it with strict safety limits. The file size is checked via :func:`os.stat` *before* ``read_bytes()`` is called, so a multi-GiB file never causes an allocation. The deserialized value is then bounded by per-field limits via :func:`safe_unpackb`. This is the canonical helper for all file-based msgpack reads outside of the core commit/snapshot/tag store (which uses the private ``_read_msgpack``). Raises: OSError: If the file size exceeds *max_bytes*. ValueError: Forwarded from :func:`safe_unpackb` if deserialization limits are violated. msgpack.UnpackException: If the file is not valid msgpack. """ size = path.stat().st_size if size > max_bytes: raise OSError( f"Msgpack file {path.name!r} is {size:,} bytes — exceeds the " f"{max_bytes // (1024 * 1024)} MiB safety cap. " "File may be corrupt or tampered." ) return safe_unpackb( path.read_bytes(), context=path.name, max_bytes=max_bytes, strict_map_key=strict_map_key, allow_binary=allow_binary, ) def _read_msgpack(path: pathlib.Path) -> MsgpackValue: """Read and unpack a msgpack file, enforcing size and per-value limits. Raises :exc:`OSError` if the file exceeds :data:`MAX_MSGPACK_BYTES` — the stat check fires *before* ``read_bytes()`` so a 10 GiB file never causes an allocation. Callers that want ``None`` on failure should catch :exc:`Exception` and return ``None``. Per-value limits (``max_str_len``, ``max_bin_len``, etc.) bound memory use during unpacking even when the total file size is within the cap. The size limit is read from :data:`MAX_MSGPACK_BYTES` at call time so that test patches to the module attribute are respected. """ size = path.stat().st_size limit = MAX_MSGPACK_BYTES if size > limit: raise OSError( f"Msgpack file {path.name!r} is {size:,} bytes — exceeds the " f"{limit:,} bytes read limit." ) return safe_unpackb(path.read_bytes(), context=path.name, max_bytes=limit) def _read_msgpack_dict(path: pathlib.Path) -> MsgpackDict: """Read a msgpack file and return the top-level mapping. Raises :exc:`TypeError` if the deserialized value is not a ``dict`` — which indicates a corrupt or tampered store file. Callers that want ``None`` on failure should catch :exc:`Exception` and return ``None``. """ raw = _read_msgpack(path) if not isinstance(raw, dict): raise TypeError( f"Expected dict from {path}, got {type(raw).__name__!r}" ) return raw _COMMITS_DIR = "commits" _SNAPSHOTS_DIR = "snapshots" _TAGS_DIR = "tags" _RELEASES_DIR = "releases" class ReleaseDict(TypedDict): """JSON-serialisable representation of a ReleaseRecord.""" release_id: str repo_id: str tag: str semver: SemVerTag channel: str commit_id: str snapshot_id: str title: str body: str changelog: list[ChangelogEntry] agent_id: str model_id: str is_draft: bool gpg_signature: str created_at: str # --------------------------------------------------------------------------- # HEAD file — typed I/O # --------------------------------------------------------------------------- # # Muse HEAD format # ---------------- # The ``.muse/HEAD`` file is always one of two self-describing forms: # # ref: refs/heads/ — symbolic ref; HEAD points to a branch # commit: — detached HEAD; HEAD points to a commit # # The ``ref:`` prefix is adopted from Git because it is the right design: # a file that can hold two semantically different things should say which # one it holds. The ``commit:`` prefix for detached HEAD is a Muse # extension — Git uses a bare SHA, which is ambiguous (SHA-1? SHA-256?). # Muse makes the hash algorithm implicit in the prefix, leaving the door # open for future algorithm identifiers without changing the parsing rule. # # There is no backward-compatibility layer; every write site uses # ``write_head_branch`` / ``write_head_commit`` and every read site uses # ``read_head`` / ``read_current_branch``. class SymbolicHead(TypedDict): """HEAD points to a named branch.""" kind: Literal["branch"] branch: str class DetachedHead(TypedDict): """HEAD points directly to a commit (detached HEAD state).""" kind: Literal["commit"] commit_id: str HeadState = SymbolicHead | DetachedHead def read_head(repo_root: pathlib.Path) -> HeadState: """Parse ``.muse/HEAD`` and return a typed :data:`HeadState`. Raises :exc:`ValueError` for any content that does not match the two expected forms, and when the HEAD file does not exist (uninitialised or corrupt repository), so callers never receive an ambiguous raw string or an unhandled :exc:`FileNotFoundError`. """ head_path = repo_root / ".muse" / "HEAD" try: raw = head_path.read_text(encoding="utf-8").strip() except FileNotFoundError: raise ValueError( f"Repository HEAD file missing: {head_path}\n" "The repository may be uninitialised. Run 'muse init' to fix it." ) if raw.startswith("ref: refs/heads/"): branch = raw.removeprefix("ref: refs/heads/").strip() validate_branch_name(branch) return SymbolicHead(kind="branch", branch=branch) if raw.startswith("commit: "): commit_id = raw.removeprefix("commit: ").strip() if not re.fullmatch(r"[0-9a-f]{64}", commit_id): raise ValueError(f"Malformed commit ID in HEAD: {commit_id!r}") return DetachedHead(kind="commit", commit_id=commit_id) raise ValueError( f"Malformed HEAD: {raw!r}. " "Expected 'ref: refs/heads/' or 'commit: '." ) def read_current_branch(repo_root: pathlib.Path) -> str: """Return the currently checked-out branch name. Raises :exc:`ValueError` when the repository is in detached HEAD state so callers that cannot operate without a branch get a clear error rather than silently receiving a commit ID as a branch name. """ state = read_head(repo_root) if state["kind"] != "branch": raise ValueError( "Repository is in detached HEAD state. " "Run 'muse checkout ' to return to a branch." ) return state["branch"] def write_head_branch(repo_root: pathlib.Path, branch: str) -> None: """Write a symbolic ref to ``.muse/HEAD`` atomically. Format: ``ref: refs/heads/`` — self-describing; the ``ref:`` prefix unambiguously identifies the entry as a symbolic reference. Uses :func:`write_text_atomic` so a crash or power loss during ``muse checkout`` or ``muse init`` cannot corrupt or zero-out HEAD. """ validate_branch_name(branch) write_text_atomic(repo_root / ".muse" / "HEAD", f"ref: refs/heads/{branch}\n") def write_head_commit(repo_root: pathlib.Path, commit_id: str) -> None: """Write a direct commit reference to ``.muse/HEAD`` atomically (detached HEAD). Format: ``commit: `` — the ``commit:`` prefix is a Muse extension that makes the entry self-describing in all states. Unlike Git (which stores a bare hash), this makes the hash type explicit and leaves room for future algorithm prefixes without parsing heuristics. Uses :func:`write_text_atomic` so a crash or power loss cannot zero-out HEAD. """ if not re.fullmatch(r"[0-9a-f]{64}", commit_id): raise ValueError(f"commit_id must be a 64-char hex string, got: {commit_id!r}") write_text_atomic(repo_root / ".muse" / "HEAD", f"commit: {commit_id}\n") # --------------------------------------------------------------------------- # Wire-format TypedDicts (JSON-serialisable, used by to_dict / from_dict) # --------------------------------------------------------------------------- class CommitDict(TypedDict, total=False): """JSON-serialisable representation of a CommitRecord. ``structured_delta`` is the typed delta produced by the domain plugin's ``diff()`` at commit time. ``None`` on the initial commit (no parent to diff against). ``sem_ver_bump`` and ``breaking_changes`` are semantic versioning metadata. Absent (treated as ``"none"`` / ``[]``) for older records and non-code domains. Agent provenance fields (all optional, default ``""`` for older records): ``agent_id`` Stable identity string for the committing agent or human (e.g. ``"counterpoint-bot"`` or ``"gabriel"``). ``model_id`` Model identifier when the author is an AI agent (e.g. ``"claude-opus-4"``). Empty for human authors. ``toolchain_id`` Toolchain that produced the commit (e.g. ``"cursor-agent-v2"``). ``prompt_hash`` SHA-256 of the instruction/prompt that triggered this commit. Privacy-preserving: the hash identifies the prompt without storing its content. ``signature`` Base64url-encoded Ed25519 signature (no padding, 86 chars) over the provenance payload (SHA-256 of commit_id + authorship fields). Verifiable with :func:`muse.core.provenance.verify_commit_ed25519` using the embedded ``signer_public_key``. ``signer_public_key`` Base64url-encoded raw Ed25519 public key bytes (32 bytes → 43 chars). Embedded in the commit so that verification is fully offline — no hub lookup required. ``signer_key_id`` First 16 hex chars of SHA-256(raw public key bytes). ``format_version`` Schema evolution counter. Each phase of the Muse supercharge plan that extends the commit record bumps this value. Readers use it to know which optional fields are present: - ``1`` — base record (commit_id, snapshot_id, parent, message, author) - ``2`` — adds ``structured_delta`` (Phase 1: Typed Delta Algebra) - ``3`` — adds ``sem_ver_bump``, ``breaking_changes`` (Phase 2: Domain Schema) - ``4`` — adds agent provenance: ``agent_id``, ``model_id``, ``toolchain_id``, ``prompt_hash``, ``signature``, ``signer_key_id`` (Phase 4: Agent Identity) - ``5`` — adds CRDT annotation fields: ``reviewed_by`` (ORSet of reviewer IDs), ``test_runs`` (GCounter of test-run events) - ``6`` — legacy; HMAC-SHA256 provenance signing (removed). Records at this version carry an HMAC signature that can no longer be verified — treat as unsigned. - ``7`` — Ed25519 provenance signing: ``signature`` is a base64url Ed25519 signature over the provenance payload; ``signer_public_key`` contains the signer's raw public key bytes (base64url, 43 chars) for offline verification. Old records without this field default to ``1``. """ commit_id: str repo_id: str branch: str snapshot_id: str message: str committed_at: str parent_commit_id: str | None parent2_commit_id: str | None author: str metadata: Metadata structured_delta: StructuredDelta | None sem_ver_bump: SemVerBump breaking_changes: list[str] agent_id: str model_id: str toolchain_id: str prompt_hash: str signature: str signer_public_key: str signer_key_id: str format_version: int # CRDT-backed annotation fields (format_version >= 5). # ``reviewed_by`` is the logical state of an ORSet: a list of unique # reviewer identifiers. Merging two records takes the union (set join). # ``test_runs`` is a GCounter: monotonically increasing test-run count. # Both fields are absent in older records and default to [] / 0. reviewed_by: list[str] test_runs: int class SnapshotDict(TypedDict): """JSON-serialisable representation of a SnapshotRecord.""" snapshot_id: str manifest: Manifest directories: list[str] created_at: str note: str class TagDict(TypedDict): """JSON-serialisable representation of a TagRecord.""" tag_id: str repo_id: str commit_id: str tag: str created_at: str # --------------------------------------------------------------------------- # Data classes # --------------------------------------------------------------------------- @dataclass class CommitRecord: """An immutable commit record stored as a JSON file under .muse/commits/. ``sem_ver_bump`` and ``breaking_changes`` are populated by the commit command when a code-domain delta is available. They default to ``"none"`` and ``[]`` for older records and non-code domains. Agent provenance fields default to ``""`` so that existing JSON without them deserialises without error. See :class:`CommitDict` for field semantics. """ commit_id: str repo_id: str branch: str snapshot_id: str message: str committed_at: datetime.datetime parent_commit_id: str | None = None parent2_commit_id: str | None = None author: str = "" metadata: Metadata = field(default_factory=dict) structured_delta: StructuredDelta | None = None sem_ver_bump: SemVerBump = "none" breaking_changes: list[str] = field(default_factory=list) agent_id: str = "" model_id: str = "" toolchain_id: str = "" prompt_hash: str = "" signature: str = "" signer_public_key: str = "" signer_key_id: str = "" #: Schema evolution counter — see :class:`CommitDict` for the version table. #: Version 5 adds ``reviewed_by`` / ``test_runs`` (CRDT fields). #: Version 7 switches commit signing from HMAC to Ed25519; adds ``signer_public_key``. format_version: int = 7 reviewed_by: list[str] = field(default_factory=list) test_runs: int = 0 def to_dict(self) -> CommitDict: return CommitDict( commit_id=self.commit_id, repo_id=self.repo_id, branch=self.branch, snapshot_id=self.snapshot_id, message=self.message, committed_at=self.committed_at.isoformat(), parent_commit_id=self.parent_commit_id, parent2_commit_id=self.parent2_commit_id, author=self.author, metadata=dict(self.metadata), structured_delta=self.structured_delta, sem_ver_bump=self.sem_ver_bump, breaking_changes=list(self.breaking_changes), agent_id=self.agent_id, model_id=self.model_id, toolchain_id=self.toolchain_id, prompt_hash=self.prompt_hash, signature=self.signature, signer_public_key=self.signer_public_key, signer_key_id=self.signer_key_id, format_version=self.format_version, reviewed_by=list(self.reviewed_by), test_runs=self.test_runs, ) @classmethod def from_msgpack(cls, d: MsgpackDict) -> "CommitRecord": """Deserialise a :class:`CommitRecord` from a raw msgpack mapping. Uses typed accessor helpers (:func:`_str_val`, :func:`_str_or_none`, etc.) so every field access is type-safe without ``# type: ignore``. Runtime guards on the three ID fields fail loud — a corrupt commit_id would propagate into path construction which is a security boundary. """ committed_at_str = _str_val(d, "committed_at") try: committed_at = datetime.datetime.fromisoformat(committed_at_str) except ValueError as exc: # Do NOT substitute datetime.now(). committed_at feeds directly into # compute_commit_id(), so a fake timestamp would produce a CommitRecord # whose hash never matches the stored commit_id. Worse, the substitution # masks the corruption from write_commit's self-repair check: write_commit # compares stored commit_id (intact) and sees a match, so it silently # skips overwriting the corrupt file. The commit then becomes permanently # unreadable. Raising here lets write_commit's `except Exception` branch # detect the corruption and overwrite with the incoming good record. raise ValueError( f"Commit record has missing or unparseable committed_at " f"({committed_at_str!r}): {exc}" ) from exc # Runtime type guards — fail loud rather than silently carrying # non-string IDs into path construction (a security boundary). commit_id = _str_val(d, "commit_id") if not commit_id: raise TypeError(f"commit_id must be a non-empty str, got {d.get('commit_id')!r}") snapshot_id = _str_val(d, "snapshot_id") if not snapshot_id: raise TypeError(f"snapshot_id must be a non-empty str, got {d.get('snapshot_id')!r}") branch = _str_val(d, "branch") if not branch: raise TypeError(f"branch must be a non-empty str, got {d.get('branch')!r}") raw_delta = d.get("structured_delta") structured_delta: StructuredDelta | None = None if isinstance(raw_delta, dict) and _as_structured_delta(raw_delta): structured_delta = raw_delta return cls( commit_id=commit_id, repo_id=_str_val(d, "repo_id"), branch=branch, snapshot_id=snapshot_id, message=_str_val(d, "message"), committed_at=committed_at, parent_commit_id=_str_or_none(d, "parent_commit_id"), parent2_commit_id=_str_or_none(d, "parent2_commit_id"), author=_str_val(d, "author"), metadata=_str_dict(d, "metadata"), structured_delta=structured_delta, sem_ver_bump=_sem_ver_bump_val(d), breaking_changes=_str_list(d, "breaking_changes"), agent_id=_str_val(d, "agent_id"), model_id=_str_val(d, "model_id"), toolchain_id=_str_val(d, "toolchain_id"), prompt_hash=_str_val(d, "prompt_hash"), signature=_str_val(d, "signature"), signer_public_key=_str_val(d, "signer_public_key"), signer_key_id=_str_val(d, "signer_key_id"), format_version=_int_val(d, "format_version", 1), reviewed_by=_str_list(d, "reviewed_by"), test_runs=_int_val(d, "test_runs"), ) @classmethod def from_dict(cls, d: "CommitDict") -> "CommitRecord": """Construct a :class:`CommitRecord` from a fully typed :class:`CommitDict`. The caller guarantees that *d* is structurally valid — all field types match the TypedDict declaration. Use :meth:`from_msgpack` when deserialising untrusted wire or disk data. """ committed_at_str = d.get("committed_at", "") try: committed_at = datetime.datetime.fromisoformat(committed_at_str) except ValueError as exc: # Do NOT substitute datetime.now(). committed_at feeds directly into # compute_commit_id(), so a fake timestamp would produce a CommitRecord # whose hash never matches the stored commit_id. When this bad record is # written via write_commit / apply_pack, read_commit's hash verification # always fails and the commit is permanently unreadable. raise ValueError( f"Commit record has missing or unparseable committed_at " f"({committed_at_str!r}): {exc}" ) from exc return cls( commit_id=d.get("commit_id", ""), repo_id=d.get("repo_id", ""), branch=d.get("branch", ""), snapshot_id=d.get("snapshot_id", ""), message=d.get("message", ""), committed_at=committed_at, parent_commit_id=d.get("parent_commit_id"), parent2_commit_id=d.get("parent2_commit_id"), author=d.get("author", ""), metadata=dict(d.get("metadata") or {}), structured_delta=d.get("structured_delta"), sem_ver_bump=d.get("sem_ver_bump", "none"), breaking_changes=list(d.get("breaking_changes") or []), agent_id=d.get("agent_id", ""), model_id=d.get("model_id", ""), toolchain_id=d.get("toolchain_id", ""), prompt_hash=d.get("prompt_hash", ""), signature=d.get("signature", ""), signer_public_key=d.get("signer_public_key", ""), signer_key_id=d.get("signer_key_id", ""), format_version=d.get("format_version", 1) or 1, reviewed_by=list(d.get("reviewed_by") or []), test_runs=d.get("test_runs", 0) or 0, ) @dataclass class SnapshotRecord: """An immutable snapshot record stored as a msgpack file under .muse/snapshots/. ``directories`` is the sorted list of workspace-relative POSIX directory paths that were explicitly tracked at snapshot time. It is included in the snapshot ID hash so that a directory rename produces a distinct snapshot even when file contents are unchanged. ``note`` is an optional human-readable label set at capture time. """ snapshot_id: str manifest: Manifest directories: list[str] = field(default_factory=list) created_at: datetime.datetime = field( default_factory=lambda: datetime.datetime.now(datetime.timezone.utc) ) note: str = "" def to_dict(self) -> SnapshotDict: return SnapshotDict( snapshot_id=self.snapshot_id, manifest=self.manifest, directories=list(self.directories), created_at=self.created_at.isoformat(), note=self.note, ) @classmethod def from_msgpack(cls, d: MsgpackDict) -> "SnapshotRecord": """Deserialise a :class:`SnapshotRecord` from a raw msgpack mapping.""" created_at_str = _str_val(d, "created_at") try: created_at = datetime.datetime.fromisoformat(created_at_str) except ValueError as exc: raise ValueError( f"Snapshot record has missing or unparseable created_at " f"({created_at_str!r}): {exc}" ) from exc raw_dirs = d.get("directories") directories = ( [v for v in raw_dirs if isinstance(v, str)] if isinstance(raw_dirs, list) else [] ) return cls( snapshot_id=_str_val(d, "snapshot_id"), manifest=_str_dict(d, "manifest"), directories=directories, created_at=created_at, note=_str_val(d, "note"), ) @classmethod def from_dict(cls, d: "SnapshotDict") -> "SnapshotRecord": """Construct a :class:`SnapshotRecord` from a typed :class:`SnapshotDict`.""" created_at_str = d.get("created_at", "") try: created_at = datetime.datetime.fromisoformat(created_at_str) except ValueError as exc: raise ValueError( f"Snapshot record has missing or unparseable created_at " f"({created_at_str!r}): {exc}" ) from exc return cls( snapshot_id=d.get("snapshot_id", ""), manifest=dict(d.get("manifest") or {}), directories=list(d.get("directories") or []), created_at=created_at, note=d.get("note", ""), ) @dataclass class TagRecord: """A semantic tag attached to a commit.""" tag_id: str repo_id: str commit_id: str tag: str created_at: datetime.datetime = field( default_factory=lambda: datetime.datetime.now(datetime.timezone.utc) ) def to_dict(self) -> TagDict: return TagDict( tag_id=self.tag_id, repo_id=self.repo_id, commit_id=self.commit_id, tag=self.tag, created_at=self.created_at.isoformat(), ) @classmethod def from_msgpack(cls, d: MsgpackDict) -> "TagRecord": """Deserialise a :class:`TagRecord` from a raw msgpack mapping.""" created_at_str = _str_val(d, "created_at") try: created_at = datetime.datetime.fromisoformat(created_at_str) except ValueError: created_at = datetime.datetime.now(datetime.timezone.utc) return cls( tag_id=_str_val(d, "tag_id") or str(uuid.uuid4()), repo_id=_str_val(d, "repo_id"), commit_id=_str_val(d, "commit_id"), tag=_str_val(d, "tag"), created_at=created_at, ) @classmethod def from_dict(cls, d: "TagDict") -> "TagRecord": """Construct a :class:`TagRecord` from a typed :class:`TagDict`.""" created_at_str = d.get("created_at", "") try: created_at = datetime.datetime.fromisoformat(created_at_str) except ValueError: created_at = datetime.datetime.now(datetime.timezone.utc) return cls( tag_id=d.get("tag_id", "") or str(uuid.uuid4()), repo_id=d.get("repo_id", ""), commit_id=d.get("commit_id", ""), tag=d.get("tag", ""), created_at=created_at, ) @dataclass class ReleaseRecord: """A versioned release attached to a commit. A release is richer than a Git tag: - ``semver`` is a parsed struct (major/minor/patch/pre/build) — queryable by version component without string parsing. - ``channel`` replaces the boolean ``is_prerelease`` flag with a named distribution channel: stable | beta | alpha | nightly. - ``changelog`` is auto-generated from the typed ``sem_ver_bump`` and ``breaking_changes`` fields on commits since the previous release, so no conventional-commit parsing is needed. - ``snapshot_id`` makes the release byte-for-byte reproducible from the content-addressed object store forever. - ``agent_id`` / ``model_id`` surface AI provenance from the tip commit. - ``gpg_signature`` signs the release for tamper-evident distribution. """ release_id: str repo_id: str tag: str semver: SemVerTag channel: ReleaseChannel commit_id: str snapshot_id: str title: str body: str changelog: list[ChangelogEntry] agent_id: str = "" model_id: str = "" is_draft: bool = False gpg_signature: str = "" created_at: datetime.datetime = field( default_factory=lambda: datetime.datetime.now(datetime.timezone.utc) ) def to_dict(self) -> ReleaseDict: d = ReleaseDict( release_id=self.release_id, repo_id=self.repo_id, tag=self.tag, semver=self.semver, channel=self.channel, commit_id=self.commit_id, snapshot_id=self.snapshot_id, title=self.title, body=self.body, changelog=list(self.changelog), agent_id=self.agent_id, model_id=self.model_id, is_draft=self.is_draft, gpg_signature=self.gpg_signature, created_at=self.created_at.isoformat(), ) return d @classmethod def from_msgpack(cls, d: MsgpackDict) -> "ReleaseRecord": """Deserialise a :class:`ReleaseRecord` from a raw msgpack mapping.""" created_at_str = _str_val(d, "created_at") try: created_at = datetime.datetime.fromisoformat(created_at_str) except ValueError: created_at = datetime.datetime.now(datetime.timezone.utc) channel = _CHANNEL_MAP.get(_str_val(d, "channel", "stable"), "stable") is_draft_val = d.get("is_draft", False) return cls( release_id=_str_val(d, "release_id"), repo_id=_str_val(d, "repo_id"), tag=_str_val(d, "tag"), semver=_parse_semver_tag(d, "semver"), channel=channel, commit_id=_str_val(d, "commit_id"), snapshot_id=_str_val(d, "snapshot_id"), title=_str_val(d, "title"), body=_str_val(d, "body"), changelog=_parse_changelog_entries(d, "changelog"), agent_id=_str_val(d, "agent_id"), model_id=_str_val(d, "model_id"), is_draft=bool(is_draft_val), gpg_signature=_str_val(d, "gpg_signature"), created_at=created_at, ) @classmethod def from_dict(cls, d: "ReleaseDict") -> "ReleaseRecord": """Construct a :class:`ReleaseRecord` from a typed :class:`ReleaseDict`.""" try: created_at = datetime.datetime.fromisoformat(d["created_at"]) except ValueError: created_at = datetime.datetime.now(datetime.timezone.utc) channel = _CHANNEL_MAP.get(d["channel"], "stable") return cls( release_id=d["release_id"], repo_id=d["repo_id"], tag=d["tag"], semver=d["semver"], channel=channel, commit_id=d["commit_id"], snapshot_id=d["snapshot_id"], title=d["title"], body=d["body"], changelog=d["changelog"], agent_id=d["agent_id"], model_id=d["model_id"], is_draft=d["is_draft"], gpg_signature=d["gpg_signature"], created_at=created_at, ) # --------------------------------------------------------------------------- # Path helpers # --------------------------------------------------------------------------- def _commits_dir(repo_root: pathlib.Path) -> pathlib.Path: return repo_root / ".muse" / _COMMITS_DIR def _snapshots_dir(repo_root: pathlib.Path) -> pathlib.Path: return repo_root / ".muse" / _SNAPSHOTS_DIR def _tags_dir(repo_root: pathlib.Path, repo_id: str) -> pathlib.Path: # Validate repo_id to prevent path traversal via crafted IDs from remote data. # Uses a best-effort guard (no path separators or dot-sequences). if "/" in repo_id or "\\" in repo_id or ".." in repo_id or not repo_id: raise ValueError(f"repo_id {repo_id!r} contains unsafe path components.") return repo_root / ".muse" / _TAGS_DIR / repo_id def _releases_dir(repo_root: pathlib.Path, repo_id: str) -> pathlib.Path: if "/" in repo_id or "\\" in repo_id or ".." in repo_id or not repo_id: raise ValueError(f"repo_id {repo_id!r} contains unsafe path components.") return repo_root / ".muse" / _RELEASES_DIR / repo_id def _commit_path(repo_root: pathlib.Path, commit_id: str) -> pathlib.Path: return _commits_dir(repo_root) / f"{commit_id}.msgpack" def _snapshot_path(repo_root: pathlib.Path, snapshot_id: str) -> pathlib.Path: return _snapshots_dir(repo_root) / f"{snapshot_id}.msgpack" # --------------------------------------------------------------------------- # Commit operations # --------------------------------------------------------------------------- def commit_exists(repo_root: pathlib.Path, commit_id: str) -> bool: """Return ``True`` when the commit file for *commit_id* is present on disk. Uses the canonical ``.msgpack`` path — the same path that :func:`write_commit` writes and :func:`read_commit` reads. Callers that need to filter a list of candidate have-anchors to those the local BFS can actually stop at should use this function rather than building the path themselves. """ return _commit_path(repo_root, commit_id).exists() # Commit directories that have been created this process run. Eliminates # the per-write mkdir(exist_ok=True) stat call after the first commit. _known_commits_dirs: set[str] = set() def write_commit(repo_root: pathlib.Path, commit: CommitRecord) -> None: """Persist a commit record to ``.muse/commits/.msgpack``. Idempotent: if the file already exists and is a valid commit record, the write is skipped. If the file exists but is *corrupt*, a CRITICAL is logged and the file is overwritten with the incoming (good) record — the store prefers a live good record over a corrupt existing one. Raises: OSError: If an existing record's ``commit_id`` field does not match the filename (data-integrity violation — indicates a tampered or severely corrupted store). """ cd = _commits_dir(repo_root) cd_str = str(cd) if cd_str not in _known_commits_dirs: cd.mkdir(parents=True, exist_ok=True) _known_commits_dirs.add(cd_str) # Validate the INCOMING record before touching disk. from_dict callers # that silently substituted now() for a corrupt committed_at produce a # record whose content hash never matches commit_id — if we wrote it, # every subsequent read_commit would return None (permanent data loss). # This is the last line of defence before the atomic write. try: _verify_commit_id(commit, commit.commit_id, pathlib.Path("")) except OSError as exc: raise ValueError( f"Refusing to write commit {commit.commit_id[:8]!r}: " f"incoming record failed hash verification — {exc}" ) from exc path = _commit_path(repo_root, commit.commit_id) if path.exists(): try: existing = CommitRecord.from_msgpack(_read_msgpack_dict(path)) if existing.commit_id != commit.commit_id: # The stored record's commit_id does not match the filename. # This should never happen with correct Muse code; it signals # tampering or catastrophic bit-rot. raise OSError( f"Store integrity violation: commit file {path.name!r} " f"contains commit_id {existing.commit_id[:8]!r}, " f"expected {commit.commit_id[:8]!r}" ) # Content-hash verification: re-derive the commit ID from the core # fields (snapshot_id, message, committed_at, parent IDs) to catch # bit-rot that leaves commit_id intact but corrupts the content. # Wrap OSError as ValueError so the except Exception branch below # triggers repair rather than the except OSError re-raise. try: _verify_commit_id(existing, commit.commit_id, path) except OSError as verify_exc: raise ValueError(str(verify_exc)) from verify_exc # Existing record is valid and hash-verified — normal idempotent skip. logger.debug("⚠️ Commit %s already exists — skipped", commit.commit_id[:8]) return except OSError: raise # propagate hard integrity violations (commit_id field mismatch) except Exception as exc: # Existing file is corrupt (parse failure or hash mismatch) — # overwrite it with the incoming (good) record. logger.critical( "❌ Corrupt commit file %s — overwriting with incoming record: %s", path, exc, ) _write_msgpack_atomic(path, commit.to_dict()) logger.debug("✅ Stored commit %s branch=%r", commit.commit_id[:8], commit.branch) def _verify_commit_id( record: "CommitRecord", expected_id: str, path: pathlib.Path ) -> None: """Re-derive the commit ID from stored fields and assert it matches *expected_id*. A bit flip that keeps msgpack structure intact but corrupts a core commit field (``snapshot_id``, ``message``, ``committed_at``, or any parent ID) will produce a different derived hash, exposing the corruption. **Coverage:** fields in :func:`~muse.core.snapshot.compute_commit_id` — ``parent_commit_id``, ``parent2_commit_id``, ``snapshot_id``, ``message``, and ``committed_at``. Metadata-only fields (``branch``, ``author``, ``repo_id``, ``metadata``) are NOT included in the commit ID by design — those can be updated post-hoc via :func:`overwrite_commit`. Raises: OSError: If the recomputed ID does not match *expected_id*, indicating silent field-level corruption that survived msgpack structural validation. """ parent_ids: list[str] = [] if record.parent_commit_id: parent_ids.append(record.parent_commit_id) if record.parent2_commit_id: parent_ids.append(record.parent2_commit_id) recomputed = compute_commit_id( parent_ids, record.snapshot_id, record.message, record.committed_at.isoformat(), ) if recomputed != expected_id: logger.critical( "❌ Commit %s failed content-hash verification — " "core fields are corrupt (snapshot_id, message, committed_at, or " "parent IDs). Expected %s, recomputed %s. " "Run `muse verify-pack` to audit the full store.", expected_id[:8], expected_id[:16], recomputed[:16], ) raise OSError( f"Commit {expected_id[:8]}… failed content-hash verification. " f"Core fields (snapshot_id, message, committed_at, parent IDs) " f"have been silently corrupted in {path.name}. " "Run `muse verify-pack` to audit the full store." ) def _verify_snapshot_id( record: "SnapshotRecord", expected_id: str, path: pathlib.Path ) -> None: """Re-derive the snapshot ID from the manifest and assert it matches *expected_id*. The snapshot ID is a hash of every ``path → object_id`` pair in the manifest, so any bit flip in any file path or object ID — however subtle — produces a different hash. This catches the class of corruptions that keep msgpack structure valid while silently altering manifest entries. Raises: OSError: If the recomputed ID does not match *expected_id*, indicating silent manifest corruption. """ recomputed = compute_snapshot_id(record.manifest, record.directories) if recomputed != expected_id: logger.critical( "❌ Snapshot %s failed content-hash verification — " "manifest entries are corrupt. Expected %s, recomputed %s. " "Run `muse verify-pack` to audit the full store.", expected_id[:8], expected_id[:16], recomputed[:16], ) raise OSError( f"Snapshot {expected_id[:8]}… failed content-hash verification. " f"One or more manifest entries (file paths or object IDs) have " f"been silently corrupted in {path.name}. " "Run `muse verify-pack` to audit the full store." ) def read_commit(repo_root: pathlib.Path, commit_id: str) -> CommitRecord | None: """Load a commit record by ID, or ``None`` if it does not exist or is corrupt. Every read re-verifies the commit ID by recomputing it from the stored core fields (``snapshot_id``, ``message``, ``committed_at``, parent IDs). This catches bit-flip corruptions that survive msgpack structural validation but silently alter the commit's essential content. Callers that need to distinguish "not found" from "corrupt" should use :func:`read_commit_result` instead. Callers that accept user-supplied or remote-supplied commit IDs should validate the ID with :func:`~muse.core.validation.validate_ref_id` before calling this function. This function itself accepts any string to support internal uses with computed IDs. """ path = _commit_path(repo_root, commit_id) if not path.exists(): return None try: record = CommitRecord.from_msgpack(_read_msgpack_dict(path)) _verify_commit_id(record, commit_id, path) return record except Exception as exc: logger.critical("❌ Corrupt commit file %s: %s", path, exc) return None class CommitReadOk(TypedDict): status: str commit: CommitRecord class CommitReadNotFound(TypedDict): status: str class CommitReadCorrupt(TypedDict): status: str path: str error: str def commit_read_is_ok( r: CommitReadOk | CommitReadNotFound | CommitReadCorrupt, ) -> TypeGuard[CommitReadOk]: """``True`` when *r* is a successful :func:`read_commit_result`.""" return r["status"] == "ok" def commit_read_is_not_found( r: CommitReadOk | CommitReadNotFound | CommitReadCorrupt, ) -> TypeGuard[CommitReadNotFound]: """``True`` when *r* represents a missing commit.""" return r["status"] == "not_found" def commit_read_is_corrupt( r: CommitReadOk | CommitReadNotFound | CommitReadCorrupt, ) -> TypeGuard[CommitReadCorrupt]: """``True`` when *r* represents a corrupt commit file.""" return r["status"] == "corrupt" def read_commit_result( repo_root: pathlib.Path, commit_id: str ) -> CommitReadOk | CommitReadNotFound | CommitReadCorrupt: """Load a commit record with a typed result that distinguishes all outcomes. Unlike :func:`read_commit` (which returns ``CommitRecord | None``), this function enables callers to take different actions depending on whether a commit is absent, corrupt, or healthy. Returns one of: * ``{"status": "ok", "commit": CommitRecord}`` * ``{"status": "not_found"}`` * ``{"status": "corrupt", "path": str, "error": str}`` """ path = _commit_path(repo_root, commit_id) if not path.exists(): return CommitReadNotFound(status="not_found") try: record = CommitRecord.from_msgpack(_read_msgpack_dict(path)) _verify_commit_id(record, commit_id, path) return CommitReadOk(status="ok", commit=record) except Exception as exc: logger.critical("❌ Corrupt commit file %s: %s", path, exc) return CommitReadCorrupt(status="corrupt", path=str(path), error=str(exc)) def overwrite_commit(repo_root: pathlib.Path, commit: CommitRecord) -> None: """Overwrite an existing commit record on disk (e.g. for annotation updates). Unlike :func:`write_commit`, this function always writes the record even if the file already exists. Use only for annotation fields (``reviewed_by``, ``test_runs``) that are semantically additive — never for changing history (commit_id, parent, snapshot, message). Args: repo_root: Repository root. commit: The updated commit record to persist. """ _commits_dir(repo_root).mkdir(parents=True, exist_ok=True) path = _commit_path(repo_root, commit.commit_id) _write_msgpack_atomic(path, commit.to_dict()) logger.debug("✅ Updated annotation on commit %s", commit.commit_id[:8]) def update_commit_metadata( repo_root: pathlib.Path, commit_id: str, key: str, value: str, ) -> bool: """Set a single string key in a commit's metadata dict. Returns ``True`` on success, ``False`` if the commit is not found. """ commit = read_commit(repo_root, commit_id) if commit is None: logger.warning("⚠️ Commit %s not found — cannot update metadata", commit_id[:8]) return False commit.metadata[key] = value path = _commit_path(repo_root, commit_id) _write_msgpack_atomic(path, commit.to_dict()) logger.debug("✅ Set %s=%r on commit %s", key, value, commit_id[:8]) return True def get_head_commit_id(repo_root: pathlib.Path, branch: str) -> str | None: """Return the commit ID at HEAD of *branch*, or ``None`` for an empty branch.""" validate_branch_name(branch) ref_path = repo_root / ".muse" / "refs" / "heads" / branch if not ref_path.exists(): return None raw = ref_path.read_text(encoding="utf-8").strip() return raw if raw else None def get_all_branch_heads(repo_root: pathlib.Path) -> BranchHeads: """Return a mapping of branch name → commit ID for every branch in *repo_root*. Reads all ref files under ``.muse/refs/heads/``. Branches whose ref file is empty or contains an invalid commit ID are silently skipped. Args: repo_root: Repository root directory (contains ``.muse/``). Returns: ``{branch_name: commit_id}`` for every non-empty branch ref. """ heads_dir = repo_root / ".muse" / "refs" / "heads" if not heads_dir.is_dir(): return {} result: BranchHeads = {} for ref_file in heads_dir.iterdir(): if not ref_file.is_file(): continue raw = ref_file.read_text(encoding="utf-8").strip() if raw: result[ref_file.name] = raw return result def get_head_snapshot_id( repo_root: pathlib.Path, repo_id: str, branch: str, ) -> str | None: """Return the snapshot_id at HEAD of *branch*, or ``None``.""" commit_id = get_head_commit_id(repo_root, branch) if commit_id is None: return None commit = read_commit(repo_root, commit_id) if commit is None: return None return commit.snapshot_id def resolve_commit_ref( repo_root: pathlib.Path, repo_id: str, branch: str, ref: str | None, ) -> CommitRecord | None: """Resolve a commit reference to a ``CommitRecord``. *ref* may be: - ``None`` / ``"HEAD"`` — the most recent commit on *branch*. - ``"HEAD~N"`` or ``"~N"`` — walk *N* first-parent steps back. - A full or abbreviated commit SHA — resolved by prefix scan. Performs a safe prefix scan (glob metacharacters stripped from *ref*) so user-supplied references cannot glob the entire commits directory. """ import re as _re if ref is None or ref.upper() == "HEAD": commit_id = get_head_commit_id(repo_root, branch) if commit_id is None: return None return read_commit(repo_root, commit_id) # Handle tilde notation: HEAD~N or ~N _tilde_match = _re.fullmatch(r"(.+?)~(\d+)", ref, _re.IGNORECASE) if _tilde_match: base_ref, steps_str = _tilde_match.group(1), _tilde_match.group(2) steps = int(steps_str) # Resolve the base ref (may itself be HEAD or a SHA) base = resolve_commit_ref(repo_root, repo_id, branch, base_ref if base_ref.upper() != "HEAD" else None) if base is None: return None commit = base for _ in range(steps): if commit.parent_commit_id is None: return None # walked past the initial commit next_commit = read_commit(repo_root, commit.parent_commit_id) if next_commit is None: return None commit = next_commit return commit # Sanitize user-supplied ref before using it in any filesystem operation. safe_ref = sanitize_glob_prefix(ref) # Try exact match — only if it looks like a full 64-char hex ID. try: validate_ref_id(safe_ref) exact: CommitRecord | None = read_commit(repo_root, safe_ref) if exact is not None: return exact except ValueError: pass # Not a full hex ID — fall through to prefix scan. # Prefix scan with sanitized prefix. return _find_commit_by_prefix(repo_root, safe_ref) def _find_commit_by_prefix( repo_root: pathlib.Path, prefix: str ) -> CommitRecord | None: """Find the first commit whose ID starts with *prefix*. Glob metacharacters are stripped from *prefix* before use to prevent callers from turning a targeted lookup into an arbitrary directory scan. """ commits_dir = _commits_dir(repo_root) if not commits_dir.exists(): return None safe_prefix = sanitize_glob_prefix(prefix) for path in commits_dir.glob(f"{safe_prefix}*.msgpack"): try: return CommitRecord.from_msgpack(_read_msgpack_dict(path)) except Exception: continue return None def find_commits_by_prefix( repo_root: pathlib.Path, prefix: str ) -> list[CommitRecord]: """Return all commits whose ID starts with *prefix*.""" commits_dir = _commits_dir(repo_root) if not commits_dir.exists(): return [] safe_prefix = sanitize_glob_prefix(prefix) results: list[CommitRecord] = [] for path in commits_dir.glob(f"{safe_prefix}*.msgpack"): try: results.append(CommitRecord.from_msgpack(_read_msgpack_dict(path))) except Exception: continue return results def _resolve_branch_commit_id(repo_root: pathlib.Path, branch: str) -> str | None: """Resolve *branch* to a commit ID, handling both local and remote tracking refs. Resolution order: 1. Local branch — ``.muse/refs/heads/`` 2. Remote tracking ref — ``.muse/remotes//`` when *branch* contains a ``/`` (e.g. ``"origin/dev"`` → remote ``"origin"``, name ``"dev"``). 3. Returns ``None`` when neither exists. Kept in ``store.py`` (not ``cli/config.py``) so any store-layer caller can resolve remote refs without introducing an upward import dependency. """ local = get_head_commit_id(repo_root, branch) if local is not None: return local if "/" in branch: remote, _, name = branch.partition("/") if name: ref_path = repo_root / ".muse" / "remotes" / remote / name if ref_path.is_file(): raw = ref_path.read_text(encoding="utf-8").strip() return raw if raw else None return None def get_commits_for_branch( repo_root: pathlib.Path, repo_id: str, branch: str, max_count: int = 0, ) -> list[CommitRecord]: """Return commits on *branch*, newest first, by walking the parent chain. *branch* may be a local branch name (``"dev"``) or a remote tracking ref (``"origin/dev"``). Both are resolved via :func:`_resolve_branch_commit_id`. Args: repo_root: Repository root. repo_id: Repository UUID (reserved for future index use). branch: Branch name or remote tracking ref to walk from HEAD. max_count: Stop after this many commits. ``0`` (default) means walk the entire chain. Pass the caller's ``--max-count`` / ``-n`` value here so the walk stops early instead of loading every commit file from disk before the caller slices the result. """ commits: list[CommitRecord] = [] commit_id = _resolve_branch_commit_id(repo_root, branch) seen: set[str] = set() while commit_id and commit_id not in seen: if max_count > 0 and len(commits) >= max_count: break seen.add(commit_id) commit = read_commit(repo_root, commit_id) if commit is None: break commits.append(commit) commit_id = commit.parent_commit_id return commits def get_all_commits(repo_root: pathlib.Path) -> list[CommitRecord]: """Return all commits in the store (order not guaranteed). Corrupt commit files are skipped with a CRITICAL log entry — callers receive a best-effort list without the corrupt records. Use :func:`muse.core.verify.run_verify` for a full integrity audit. """ commits_dir = _commits_dir(repo_root) if not commits_dir.exists(): return [] results: list[CommitRecord] = [] for path in commits_dir.glob("*.msgpack"): try: results.append(CommitRecord.from_msgpack(_read_msgpack_dict(path))) except Exception as exc: logger.critical("❌ Corrupt commit file %s — skipped in listing: %s", path, exc) return results class WalkResult(TypedDict): """Result of a bounded history walk. Returned by :func:`walk_commits_between_result` so callers can distinguish between a complete walk and one that was truncated by the safety cap. """ commits: list[CommitRecord] truncated: bool count: int def walk_commits_between( repo_root: pathlib.Path, to_commit_id: str, from_commit_id: str | None = None, max_commits: int = 10_000, ) -> list[CommitRecord]: """Return commits reachable from *to_commit_id*, stopping before *from_commit_id*. Walks the parent chain from *to_commit_id* backwards. Returns commits in newest-first order (callers can reverse for oldest-first). .. note:: If the walk reaches *max_commits* before the chain is exhausted the result is silently truncated. Use :func:`walk_commits_between_result` when the caller must know whether truncation occurred. Args: repo_root: Repository root. to_commit_id: Inclusive end of the range. from_commit_id: Exclusive start; ``None`` means walk to the initial commit. max_commits: Safety cap — default 10 000. Returns: List of ``CommitRecord`` objects, newest first. """ return walk_commits_between_result( repo_root, to_commit_id, from_commit_id, max_commits )["commits"] def walk_commits_between_result( repo_root: pathlib.Path, to_commit_id: str, from_commit_id: str | None = None, max_commits: int = 10_000, ) -> WalkResult: """Bounded history walk with explicit truncation signalling. Identical to :func:`walk_commits_between` but returns a :class:`WalkResult` TypedDict that includes ``"truncated": True`` when the safety cap was reached before the chain was exhausted. This is the preferred function for any agent or CLI path that must emit ``"truncated"`` in its JSON output. Args: repo_root: Repository root. to_commit_id: Inclusive end of the range. from_commit_id: Exclusive start; ``None`` means walk to the initial commit. max_commits: Safety cap — default 10 000. Returns: ``WalkResult`` with ``commits``, ``truncated``, and ``count``. """ commits: list[CommitRecord] = [] seen: set[str] = set() current_id: str | None = to_commit_id while current_id and current_id not in seen: if len(commits) >= max_commits: return WalkResult(commits=commits, truncated=True, count=len(commits)) seen.add(current_id) if current_id == from_commit_id: break commit = read_commit(repo_root, current_id) if commit is None: break commits.append(commit) current_id = commit.parent_commit_id return WalkResult(commits=commits, truncated=False, count=len(commits)) # --------------------------------------------------------------------------- # Snapshot operations # --------------------------------------------------------------------------- def write_snapshot(repo_root: pathlib.Path, snapshot: SnapshotRecord) -> None: """Persist a snapshot record to ``.muse/snapshots/.msgpack``.""" _snapshots_dir(repo_root).mkdir(parents=True, exist_ok=True) # Validate the INCOMING record before touching disk. A SnapshotRecord # constructed with a mismatched snapshot_id (from a corrupt bundle, a # tampered manifest, or a bad from_dict call) would be written successfully # but read_snapshot would immediately fail hash verification and return None — # the snapshot becomes permanently unreadable. Reject it here. try: _verify_snapshot_id(snapshot, snapshot.snapshot_id, pathlib.Path("")) except OSError as exc: raise ValueError( f"Refusing to write snapshot {snapshot.snapshot_id[:8]!r}: " f"incoming record failed hash verification — {exc}" ) from exc path = _snapshot_path(repo_root, snapshot.snapshot_id) if path.exists(): # Verify the existing file before skipping — a corrupt manifest entry # (wrong object ID, missing file, injected file) would otherwise be # permanent: read_snapshot always calls _verify_snapshot_id and returns # None on mismatch, but write_snapshot would never overwrite it. try: existing = SnapshotRecord.from_msgpack(_read_msgpack_dict(path)) _verify_snapshot_id(existing, snapshot.snapshot_id, path) logger.debug("⚠️ Snapshot %s already exists — skipped", snapshot.snapshot_id[:8]) return except Exception as exc: logger.critical( "❌ Corrupt snapshot file %s — overwriting with incoming record: %s", path, exc, ) _write_msgpack_atomic(path, snapshot.to_dict()) logger.debug("✅ Stored snapshot %s (%d files, %d dirs)", snapshot.snapshot_id[:8], len(snapshot.manifest), len(snapshot.directories)) def read_snapshot(repo_root: pathlib.Path, snapshot_id: str) -> SnapshotRecord | None: """Load a snapshot record by ID, or ``None`` if it does not exist or is corrupt. Every read re-verifies the snapshot ID by recomputing it from the stored manifest. Any bit flip that alters a file path or object ID in the manifest — even without breaking msgpack structure — is caught here. Callers that need to distinguish "not found" from "corrupt" should use :func:`read_snapshot_result` instead. Callers that accept user-supplied or remote-supplied snapshot IDs should validate the ID with :func:`~muse.core.validation.validate_ref_id` before calling this function. This function itself accepts any string to support internal uses with computed IDs. """ path = _snapshot_path(repo_root, snapshot_id) if not path.exists(): return None try: record = SnapshotRecord.from_msgpack(_read_msgpack_dict(path)) _verify_snapshot_id(record, snapshot_id, path) return record except Exception as exc: logger.critical("❌ Corrupt snapshot file %s: %s", path, exc) return None class SnapshotReadOk(TypedDict): status: str snapshot: SnapshotRecord class SnapshotReadNotFound(TypedDict): status: str class SnapshotReadCorrupt(TypedDict): status: str path: str error: str def snapshot_read_is_ok( r: SnapshotReadOk | SnapshotReadNotFound | SnapshotReadCorrupt, ) -> TypeGuard[SnapshotReadOk]: """``True`` when *r* is a successful :func:`read_snapshot_result`.""" return r["status"] == "ok" def snapshot_read_is_corrupt( r: SnapshotReadOk | SnapshotReadNotFound | SnapshotReadCorrupt, ) -> TypeGuard[SnapshotReadCorrupt]: """``True`` when *r* represents a corrupt snapshot file.""" return r["status"] == "corrupt" def read_snapshot_result( repo_root: pathlib.Path, snapshot_id: str ) -> SnapshotReadOk | SnapshotReadNotFound | SnapshotReadCorrupt: """Load a snapshot with a typed result that distinguishes all outcomes. Returns one of: * ``{"status": "ok", "snapshot": SnapshotRecord}`` * ``{"status": "not_found"}`` * ``{"status": "corrupt", "path": str, "error": str}`` """ path = _snapshot_path(repo_root, snapshot_id) if not path.exists(): return SnapshotReadNotFound(status="not_found") try: record = SnapshotRecord.from_msgpack(_read_msgpack_dict(path)) _verify_snapshot_id(record, snapshot_id, path) return SnapshotReadOk(status="ok", snapshot=record) except Exception as exc: logger.critical("❌ Corrupt snapshot file %s: %s", path, exc) return SnapshotReadCorrupt(status="corrupt", path=str(path), error=str(exc)) def get_commit_snapshot_manifest( repo_root: pathlib.Path, commit_id: str ) -> Manifest | None: """Return the file manifest for the snapshot attached to *commit_id*, or ``None``.""" commit = read_commit(repo_root, commit_id) if commit is None: logger.warning("⚠️ Commit %s not found", commit_id[:8]) return None snapshot = read_snapshot(repo_root, commit.snapshot_id) if snapshot is None: logger.warning( "⚠️ Snapshot %s referenced by commit %s not found", commit.snapshot_id[:8], commit_id[:8], ) return None return dict(snapshot.manifest) def get_head_snapshot_manifest( repo_root: pathlib.Path, repo_id: str, branch: str ) -> Manifest | None: """Return the manifest of the most recent commit on *branch*, or ``None``.""" snapshot_id = get_head_snapshot_id(repo_root, repo_id, branch) if snapshot_id is None: return None snapshot = read_snapshot(repo_root, snapshot_id) if snapshot is None: return None return dict(snapshot.manifest) # --------------------------------------------------------------------------- # Tag operations # --------------------------------------------------------------------------- def write_tag(repo_root: pathlib.Path, tag: TagRecord) -> None: """Persist a tag record to ``.muse/tags//.msgpack``.""" tags_dir = _tags_dir(repo_root, tag.repo_id) tags_dir.mkdir(parents=True, exist_ok=True) path = tags_dir / f"{tag.tag_id}.msgpack" _write_msgpack_atomic(path, tag.to_dict()) logger.debug("✅ Stored tag %r on commit %s", tag.tag, tag.commit_id[:8]) def get_tags_for_commit( repo_root: pathlib.Path, repo_id: str, commit_id: str ) -> list[TagRecord]: """Return all tags attached to *commit_id*.""" tags_dir = _tags_dir(repo_root, repo_id) if not tags_dir.exists(): return [] results: list[TagRecord] = [] for path in tags_dir.glob("*.msgpack"): try: record = TagRecord.from_msgpack(_read_msgpack_dict(path)) if record.commit_id == commit_id: results.append(record) except Exception: continue return results def delete_tag(repo_root: pathlib.Path, repo_id: str, tag_id: str) -> bool: """Delete the tag file identified by *tag_id*; return True if it existed.""" tags_dir = _tags_dir(repo_root, repo_id) path = tags_dir / f"{tag_id}.msgpack" if path.exists(): path.unlink() logger.debug("🗑️ Deleted tag %s", tag_id) return True return False def get_all_tags(repo_root: pathlib.Path, repo_id: str) -> list[TagRecord]: """Return all tags in this repository. Corrupt tag files are skipped with a CRITICAL log entry — callers receive a best-effort list without the corrupt records. """ tags_dir = _tags_dir(repo_root, repo_id) if not tags_dir.exists(): return [] results: list[TagRecord] = [] for path in tags_dir.glob("*.msgpack"): try: results.append(TagRecord.from_msgpack(_read_msgpack_dict(path))) except Exception as exc: logger.critical("❌ Corrupt tag file %s — skipped in listing: %s", path, exc) return results # --------------------------------------------------------------------------- # Release operations # --------------------------------------------------------------------------- def write_release(repo_root: pathlib.Path, release: ReleaseRecord) -> None: """Persist a release record to ``.muse/releases//.msgpack``.""" releases_dir = _releases_dir(repo_root, release.repo_id) releases_dir.mkdir(parents=True, exist_ok=True) path = releases_dir / f"{release.release_id}.msgpack" _write_msgpack_atomic(path, release.to_dict()) logger.debug("✅ Stored release %r (%s)", release.tag, release.release_id[:8]) def read_release( repo_root: pathlib.Path, repo_id: str, release_id: str ) -> ReleaseRecord | None: """Load a release by its UUID, or ``None`` if it does not exist or is corrupt.""" path = _releases_dir(repo_root, repo_id) / f"{release_id}.msgpack" if not path.exists(): return None try: return ReleaseRecord.from_msgpack(_read_msgpack_dict(path)) except Exception as exc: logger.critical("❌ Corrupt release file %s: %s", path, exc) return None def get_release_for_tag( repo_root: pathlib.Path, repo_id: str, tag: str ) -> ReleaseRecord | None: """Return the release whose version tag matches *tag*, or ``None``. Searches all releases including drafts so callers can inspect or delete a draft before it is published. """ for release in list_releases(repo_root, repo_id, include_drafts=True): if release.tag == tag: return release return None def list_releases( repo_root: pathlib.Path, repo_id: str, channel: ReleaseChannel | None = None, include_drafts: bool = False, ) -> list[ReleaseRecord]: """Return all releases, newest first. Args: repo_root: Repository root. repo_id: Repository UUID. channel: Filter by channel; ``None`` returns all channels. include_drafts: When ``False`` (default) draft releases are hidden. """ releases_dir = _releases_dir(repo_root, repo_id) if not releases_dir.exists(): return [] results: list[ReleaseRecord] = [] for path in releases_dir.glob("*.msgpack"): try: r = ReleaseRecord.from_msgpack(_read_msgpack_dict(path)) except Exception as exc: logger.critical("❌ Corrupt release file %s — skipped in listing: %s", path, exc) continue if r.is_draft and not include_drafts: continue if channel is not None and r.channel != channel: continue results.append(r) results.sort(key=lambda r: r.created_at, reverse=True) return results def delete_release(repo_root: pathlib.Path, repo_id: str, release_id: str) -> bool: """Delete a release record. Returns ``True`` if it existed. Callers are responsible for enforcing that only draft releases may be deleted. This function performs no such guard — enforce at the CLI/API layer. """ path = _releases_dir(repo_root, repo_id) / f"{release_id}.msgpack" if path.exists(): path.unlink() logger.debug("🗑️ Deleted release %s", release_id) return True return False def build_changelog( repo_root: pathlib.Path, from_commit_id: str | None, to_commit_id: str, max_commits: int = 500, ) -> list[ChangelogEntry]: """Walk the commit graph from *to_commit_id* back to *from_commit_id*. Returns a list of :class:`ChangelogEntry` dicts, oldest first, excluding merge commits and ``sem_ver_bump="none"`` commits (they carry no user-visible change). *from_commit_id* is excluded; *to_commit_id* is included. Args: repo_root: Repository root. from_commit_id: The last release commit (exclusive start), or ``None`` for a first release (walks back to the initial commit). to_commit_id: The HEAD commit to release (inclusive end). max_commits: Safety cap — changelog never exceeds this many entries. """ raw = walk_commits_between(repo_root, to_commit_id, from_commit_id, max_commits) entries: list[ChangelogEntry] = [] for commit in reversed(raw): # oldest first entries.append(ChangelogEntry( commit_id=commit.commit_id, message=commit.message, sem_ver_bump=commit.sem_ver_bump, breaking_changes=list(commit.breaking_changes), author=commit.author, committed_at=commit.committed_at.isoformat(), agent_id=commit.agent_id, model_id=commit.model_id, )) return entries # --------------------------------------------------------------------------- # Remote sync helpers (push/pull) # ---------------------------------------------------------------------------