"""TDD — apply_mpack partial failure atomicity. Invariant: a commit must never be written that creates a dangling reference chain. Write order is objects → snapshots → commits. This ordering is necessary but not sufficient. If an object fails to write (content/ID mismatch, OSError), any snapshot referencing that object must also be skipped, and any commit referencing that snapshot must also be skipped. Without this, a poisoned object entry in an mpack produces: commit (written) → snapshot (written) → object (NOT written) — a dangling reference that silently corrupts the local store. Tests ----- PF-1 Poisoned object (content/ID mismatch) → its snapshot and commit are not written. PF-2 OSError mid-object-write → no snapshots or commits are written. PF-3 Unaffected commits (different object chain) still write successfully when one object in the mpack is poisoned. PF-4 All objects succeed → all snapshots and commits write normally (regression guard — the fix must not break the happy path). """ from __future__ import annotations import datetime import json import pathlib from unittest.mock import patch import pytest from muse.core.mpack import MPack, apply_mpack, build_mpack from muse.core.object_store import has_object, write_object from muse.core.paths import muse_dir from muse.core.ids import hash_commit, hash_snapshot from muse.core.refs import write_branch_ref from muse.core.commits import ( CommitRecord, read_commit, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, read_snapshot, write_snapshot, ) from muse.core.types import blob_id _DT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) def _init_repo(root: pathlib.Path) -> pathlib.Path: dot = muse_dir(root) dot.mkdir(parents=True) (dot / "repo.json").write_text(json.dumps({"repo_id": "pf-test"})) for d in ("commits", "snapshots", "objects", "refs/heads"): (dot / d).mkdir(parents=True, exist_ok=True) (dot / "HEAD").write_text("ref: refs/heads/main\n") (dot / "config.toml").write_text("") return root def _make_mpack_with_poisoned_object( good_content: bytes, poisoned_oid: str, poisoned_content: bytes, ) -> tuple[MPack, str, str]: """Build a minimal MPack where one object has a content/ID mismatch. Returns (mpack, snapshot_id, commit_id). The poisoned object has poisoned_oid as its declared ID but poisoned_content whose sha256 does NOT equal poisoned_oid. """ good_oid = blob_id(good_content) manifest = {"src/good.py": good_oid, "src/bad.py": poisoned_oid} sid = hash_snapshot(manifest) cid = hash_commit( parent_ids=[], snapshot_id=sid, message="poisoned", committed_at_iso=_DT.isoformat(), author="gabriel", ) mpack: MPack = { "commits": [CommitRecord( commit_id=cid, branch="main", snapshot_id=sid, message="poisoned", committed_at=_DT, parent_commit_id=None, parent2_commit_id=None, author="gabriel", metadata={}, structured_delta=None, sem_ver_bump="none", breaking_changes=[], agent_id="", model_id="", toolchain_id="", prompt_hash="", signature="", signer_key_id="", ).to_dict()], "snapshots": [{ "snapshot_id": sid, "parent_snapshot_id": None, "delta_upsert": manifest, "delta_remove": [], }], "blobs": [ {"object_id": good_oid, "content": good_content}, {"object_id": poisoned_oid, "content": poisoned_content}, ], } return mpack, sid, cid # --------------------------------------------------------------------------- # PF-1 Poisoned object → snapshot and commit not written # --------------------------------------------------------------------------- def test_pf1_poisoned_object_prevents_commit_write(tmp_path: pathlib.Path) -> None: """A content/ID mismatch on an object must prevent its commit from being written. When write_object raises ValueError (hash mismatch), any snapshot that references that object_id must be skipped, and any commit referencing that snapshot must also be skipped — no dangling references. """ dst = _init_repo(tmp_path / "dst") good_content = b"def good(): pass\n" # poisoned: declare a sha256:aaa... ID but send different bytes poisoned_oid = "sha256:" + "a" * 64 poisoned_content = b"this does not hash to aaa..." mpack, sid, cid = _make_mpack_with_poisoned_object( good_content, poisoned_oid, poisoned_content ) apply_mpack(dst, mpack) assert read_commit(dst, cid) is None, ( "commit was written despite its snapshot referencing a poisoned object — " "dangling reference chain created" ) assert read_snapshot(dst, sid) is None, ( "snapshot was written despite referencing a poisoned object" ) # --------------------------------------------------------------------------- # PF-2 OSError mid-object-write → no snapshots or commits written # --------------------------------------------------------------------------- def test_pf2_oserror_on_object_write_aborts_cleanly(tmp_path: pathlib.Path) -> None: """OSError during object write must not leave any commits or snapshots written.""" dst = _init_repo(tmp_path / "dst") content = b"print('hello')\n" oid = blob_id(content) manifest = {"src/hello.py": oid} sid = hash_snapshot(manifest) cid = hash_commit( parent_ids=[], snapshot_id=sid, message="oserror test", committed_at_iso=_DT.isoformat(), author="gabriel", ) mpack: MPack = { "commits": [CommitRecord( commit_id=cid, branch="main", snapshot_id=sid, message="oserror test", committed_at=_DT, parent_commit_id=None, parent2_commit_id=None, author="gabriel", metadata={}, structured_delta=None, sem_ver_bump="none", breaking_changes=[], agent_id="", model_id="", toolchain_id="", prompt_hash="", signature="", signer_key_id="", ).to_dict()], "snapshots": [{ "snapshot_id": sid, "parent_snapshot_id": None, "delta_upsert": manifest, "delta_remove": [], }], "blobs": [{"object_id": oid, "content": content}], } with patch("muse.core.mpack.write_pack", side_effect=OSError("disk full")): with pytest.raises(OSError, match="disk full"): apply_mpack(dst, mpack) assert read_commit(dst, cid) is None, "commit written after OSError on object write" assert read_snapshot(dst, sid) is None, "snapshot written after OSError on object write" # --------------------------------------------------------------------------- # PF-3 Unaffected commits still write when one object chain is poisoned # --------------------------------------------------------------------------- def test_pf3_clean_commits_write_when_one_chain_is_poisoned(tmp_path: pathlib.Path) -> None: """A poisoned object must not prevent unrelated commits from being written.""" dst = _init_repo(tmp_path / "dst") # Clean chain clean_content = b"def clean(): pass\n" clean_oid = blob_id(clean_content) clean_manifest = {"src/clean.py": clean_oid} clean_sid = hash_snapshot(clean_manifest) clean_cid = hash_commit( parent_ids=[], snapshot_id=clean_sid, message="clean", committed_at_iso=_DT.isoformat(), author="gabriel", ) # Poisoned chain poisoned_oid = "sha256:" + "b" * 64 poisoned_content = b"wrong bytes" poisoned_manifest = {"src/bad.py": poisoned_oid} poisoned_sid = hash_snapshot(poisoned_manifest) poisoned_cid = hash_commit( parent_ids=[], snapshot_id=poisoned_sid, message="poisoned", committed_at_iso=_DT.isoformat(), author="gabriel", ) mpack: MPack = { "commits": [ CommitRecord( commit_id=clean_cid, branch="main", snapshot_id=clean_sid, message="clean", committed_at=_DT, parent_commit_id=None, parent2_commit_id=None, author="gabriel", metadata={}, structured_delta=None, sem_ver_bump="none", breaking_changes=[], agent_id="", model_id="", toolchain_id="", prompt_hash="", signature="", signer_key_id="", ).to_dict(), CommitRecord( commit_id=poisoned_cid, branch="main", snapshot_id=poisoned_sid, message="poisoned", committed_at=_DT, parent_commit_id=None, parent2_commit_id=None, author="gabriel", metadata={}, structured_delta=None, sem_ver_bump="none", breaking_changes=[], agent_id="", model_id="", toolchain_id="", prompt_hash="", signature="", signer_key_id="", ).to_dict(), ], "snapshots": [ {"snapshot_id": clean_sid, "parent_snapshot_id": None, "delta_upsert": clean_manifest, "delta_remove": []}, {"snapshot_id": poisoned_sid, "parent_snapshot_id": None, "delta_upsert": poisoned_manifest, "delta_remove": []}, ], "blobs": [ {"object_id": clean_oid, "content": clean_content}, {"object_id": poisoned_oid, "content": poisoned_content}, ], } apply_mpack(dst, mpack) assert read_commit(dst, clean_cid) is not None, "clean commit was not written" assert read_commit(dst, poisoned_cid) is None, ( "poisoned commit was written despite referencing a missing object" ) # --------------------------------------------------------------------------- # PF-4 Happy path regression guard # --------------------------------------------------------------------------- def test_pf4_happy_path_unaffected(tmp_path: pathlib.Path) -> None: """All objects succeed → all snapshots and commits write normally.""" src = _init_repo(tmp_path / "src") dst = _init_repo(tmp_path / "dst") content = b"def hello(): return 42\n" oid = blob_id(content) write_object(src, oid, content) manifest = {"src/hello.py": oid} sid = hash_snapshot(manifest) write_snapshot(src, SnapshotRecord(snapshot_id=sid, manifest=manifest)) cid = hash_commit( parent_ids=[], snapshot_id=sid, message="happy", committed_at_iso=_DT.isoformat(), author="gabriel", ) write_commit(src, CommitRecord( commit_id=cid, branch="main", snapshot_id=sid, message="happy", committed_at=_DT, parent_commit_id=None, parent2_commit_id=None, author="gabriel", metadata={}, structured_delta=None, sem_ver_bump="none", breaking_changes=[], agent_id="", model_id="", toolchain_id="", prompt_hash="", signature="", signer_key_id="", )) write_branch_ref(src, "main", cid) mpack = build_mpack(src, [cid], have=[]) result = apply_mpack(dst, mpack) assert result["commits_written"] == 1 assert result["snapshots_written"] == 1 assert result["blobs_written"] == 1 assert read_commit(dst, cid) is not None assert read_snapshot(dst, sid) is not None assert has_object(dst, oid)