"""TDD -- muse#83 Phase 2: cascading commit rewrite + re-sign. migrate_commit_ids_cascade() must, given a snapshot_id_map from Phase 1: - rewrite the commit_id of any commit whose OWN snapshot_id is a key in the map, AND every descendant (transitively, since a descendant's commit_id embeds its parent's commit_id), - leave every commit outside that blast radius completely untouched, - re-sign every rewritten commit with the current signing identity (a stale signature over a superseded commit_id is a silent authenticity regression -- this is the one real gap in reusing muse.core.migrate.migrate_commit_ids as-is, which never re-signs). """ from __future__ import annotations import datetime import json import pathlib import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from muse.core.ids import hash_commit, hash_snapshot from muse.core.object_store import object_path from muse.core.paths import objects_dir from muse.core.provenance import ( encode_public_key, provenance_payload, sign_commit_ed25519, verify_commit_ed25519, ) from muse.core.snapshot_content_migrate import migrate_commit_ids_cascade from muse.core.transport import SigningIdentity from muse.core.types import decode_pubkey _AT_ISO = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc).isoformat() # --------------------------------------------------------------------------- # Fixtures / helpers # --------------------------------------------------------------------------- @pytest.fixture def repo(tmp_path: pathlib.Path) -> pathlib.Path: objects_dir(tmp_path).mkdir(parents=True, exist_ok=True) return tmp_path def _make_identity(handle: str = "gabriel") -> SigningIdentity: return SigningIdentity(handle=handle, private_key=Ed25519PrivateKey.generate()) def _sign(commit_id: str, private_key: Ed25519PrivateKey, author: str = "gabriel") -> str: payload = provenance_payload(commit_id=commit_id, author=author, committed_at=_AT_ISO) return sign_commit_ed25519(payload, private_key) def _write_unified_commit( repo: pathlib.Path, commit_id: str, *, parent_commit_id: str | None, snapshot_id: str, message: str, signer_public_key: str = "", signature: str = "", ) -> None: raw = { "commit_id": commit_id, "branch": "main", "snapshot_id": snapshot_id, "message": message, "committed_at": _AT_ISO, "parent_commit_id": parent_commit_id, "parent2_commit_id": None, "author": "gabriel", "agent_id": "", "model_id": "", "toolchain_id": "", "prompt_hash": "", "signature": signature, "signer_public_key": signer_public_key, "signer_key_id": "", "format_version": 8, "sem_ver_bump": "none", "breaking_changes": [], } payload = json.dumps(raw, separators=(",", ":")).encode() header = f"commit {len(payload)}\0".encode() path = object_path(repo, commit_id) path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(header + payload) def _read_unified_commit(repo: pathlib.Path, commit_id: str) -> dict: path = object_path(repo, commit_id) data = path.read_bytes() null_idx = data.index(b"\x00") return json.loads(data[null_idx + 1:]) # --------------------------------------------------------------------------- # Build the A -> B(bad snapshot) -> C -> D fixture chain from the plan doc # --------------------------------------------------------------------------- def _build_chain(repo: pathlib.Path, identity: SigningIdentity) -> dict[str, str]: manifest_a = {"a.txt": "sha256:" + "1" * 64} snap_a = hash_snapshot(manifest_a, None) manifest_b = {"a.txt": "sha256:" + "1" * 64, "b.txt": "sha256:" + "2" * 64} wrong_snap_b = "sha256:" + "0" * 64 # deliberately wrong, like the real bug manifest_c = {**manifest_b, "c.txt": "sha256:" + "3" * 64} snap_c = hash_snapshot(manifest_c, None) manifest_d = {**manifest_c, "d.txt": "sha256:" + "4" * 64} snap_d = hash_snapshot(manifest_d, None) id_a = hash_commit(parent_ids=[], snapshot_id=snap_a, message="A", committed_at_iso=_AT_ISO, author="gabriel", signer_public_key="") id_b = hash_commit(parent_ids=[id_a], snapshot_id=wrong_snap_b, message="B", committed_at_iso=_AT_ISO, author="gabriel", signer_public_key="") id_c = hash_commit(parent_ids=[id_b], snapshot_id=snap_c, message="C", committed_at_iso=_AT_ISO, author="gabriel", signer_public_key="") id_d = hash_commit(parent_ids=[id_c], snapshot_id=snap_d, message="D", committed_at_iso=_AT_ISO, author="gabriel", signer_public_key="") _write_unified_commit(repo, id_a, parent_commit_id=None, snapshot_id=snap_a, message="A", signature=_sign(id_a, identity.private_key)) _write_unified_commit(repo, id_b, parent_commit_id=id_a, snapshot_id=wrong_snap_b, message="B", signature=_sign(id_b, identity.private_key)) _write_unified_commit(repo, id_c, parent_commit_id=id_b, snapshot_id=snap_c, message="C", signature=_sign(id_c, identity.private_key)) _write_unified_commit(repo, id_d, parent_commit_id=id_c, snapshot_id=snap_d, message="D", signature=_sign(id_d, identity.private_key)) return { "id_a": id_a, "id_b": id_b, "id_c": id_c, "id_d": id_d, "wrong_snap_b": wrong_snap_b, "correct_snap_b": hash_snapshot(manifest_b, None), } # --------------------------------------------------------------------------- # CC_01 -- the cascade rewrites B, C, D and leaves A untouched # --------------------------------------------------------------------------- def test_cc_01_cascade_rewrites_descendants_and_spares_ancestor(repo: pathlib.Path) -> None: identity = _make_identity() ids = _build_chain(repo, identity) snapshot_id_map = {ids["wrong_snap_b"]: ids["correct_snap_b"]} a_bytes_before = object_path(repo, ids["id_a"]).read_bytes() result = migrate_commit_ids_cascade(repo, snapshot_id_map, identity, dry_run=False) assert ids["id_a"] not in result.id_map, "A has no bad snapshot and no rewritten parent -- must be untouched" assert ids["id_b"] in result.id_map assert ids["id_c"] in result.id_map assert ids["id_d"] in result.id_map assert len({result.id_map[ids["id_b"]], result.id_map[ids["id_c"]], result.id_map[ids["id_d"]]}) == 3 # A is byte-identical -- provably outside the blast radius. assert object_path(repo, ids["id_a"]).read_bytes() == a_bytes_before # --------------------------------------------------------------------------- # CC_02 -- the new B correctly carries the corrected snapshot_id, and C/D's # new records point at B/C's NEW commit_id (the cascade actually threads # through, not just independently rewritten in isolation) # --------------------------------------------------------------------------- def test_cc_02_cascade_threads_new_parent_ids_through(repo: pathlib.Path) -> None: identity = _make_identity() ids = _build_chain(repo, identity) snapshot_id_map = {ids["wrong_snap_b"]: ids["correct_snap_b"]} result = migrate_commit_ids_cascade(repo, snapshot_id_map, identity, dry_run=False) new_b = result.id_map[ids["id_b"]] new_c = result.id_map[ids["id_c"]] new_d = result.id_map[ids["id_d"]] b_record = _read_unified_commit(repo, new_b) assert b_record["snapshot_id"] == ids["correct_snap_b"] assert b_record["parent_commit_id"] == ids["id_a"], "A was not rewritten, so B's parent pointer is unchanged" c_record = _read_unified_commit(repo, new_c) assert c_record["parent_commit_id"] == new_b, "C must point at B's NEW commit_id, not the old one" d_record = _read_unified_commit(repo, new_d) assert d_record["parent_commit_id"] == new_c, "D must point at C's NEW commit_id, not the old one" # --------------------------------------------------------------------------- # CC_03 -- every rewritten commit is re-signed and the new signature verifies # against the NEW commit_id (not the old one) # --------------------------------------------------------------------------- def test_cc_03_rewritten_commits_are_resigned_and_verify(repo: pathlib.Path) -> None: identity = _make_identity() ids = _build_chain(repo, identity) snapshot_id_map = {ids["wrong_snap_b"]: ids["correct_snap_b"]} result = migrate_commit_ids_cascade(repo, snapshot_id_map, identity, dry_run=False) _, pub_str = encode_public_key(identity.private_key) _, pub_bytes = decode_pubkey(pub_str) for old_id in (ids["id_b"], ids["id_c"], ids["id_d"]): new_id = result.id_map[old_id] record = _read_unified_commit(repo, new_id) assert record["signer_public_key"] == pub_str payload = provenance_payload( commit_id=new_id, author=record["author"], agent_id=record.get("agent_id", ""), model_id=record.get("model_id", ""), toolchain_id=record.get("toolchain_id", ""), prompt_hash=record.get("prompt_hash", ""), committed_at=record["committed_at"], ) assert verify_commit_ed25519(payload, record["signature"], pub_bytes), ( f"re-signed commit {new_id[:18]} does not verify against its own new commit_id" ) # --------------------------------------------------------------------------- # CC_04 -- a sibling branch untouched by the corruption is provably unaffected # --------------------------------------------------------------------------- def test_cc_04_unrelated_sibling_commit_is_byte_identical(repo: pathlib.Path) -> None: identity = _make_identity() ids = _build_chain(repo, identity) # An unrelated commit forking from A, nothing to do with B/C/D at all. manifest_e = {"a.txt": "sha256:" + "1" * 64, "e.txt": "sha256:" + "9" * 64} snap_e = hash_snapshot(manifest_e, None) id_e = hash_commit(parent_ids=[ids["id_a"]], snapshot_id=snap_e, message="E", committed_at_iso=_AT_ISO, author="gabriel", signer_public_key="") _write_unified_commit(repo, id_e, parent_commit_id=ids["id_a"], snapshot_id=snap_e, message="E", signature=_sign(id_e, identity.private_key)) e_bytes_before = object_path(repo, id_e).read_bytes() snapshot_id_map = {ids["wrong_snap_b"]: ids["correct_snap_b"]} result = migrate_commit_ids_cascade(repo, snapshot_id_map, identity, dry_run=False) assert id_e not in result.id_map assert object_path(repo, id_e).read_bytes() == e_bytes_before # --------------------------------------------------------------------------- # CC_05 -- dry_run computes the id_map but writes nothing # --------------------------------------------------------------------------- def test_cc_05_dry_run_writes_nothing(repo: pathlib.Path) -> None: identity = _make_identity() ids = _build_chain(repo, identity) snapshot_id_map = {ids["wrong_snap_b"]: ids["correct_snap_b"]} result = migrate_commit_ids_cascade(repo, snapshot_id_map, identity, dry_run=True) assert set(result.id_map) == {ids["id_b"], ids["id_c"], ids["id_d"]} for old_id in (ids["id_b"], ids["id_c"], ids["id_d"]): assert not object_path(repo, result.id_map[old_id]).exists() # --------------------------------------------------------------------------- # CC_06 -- running the cascade twice with the same snapshot_id_map is # deterministic (same id_map both times). Old (superseded) objects are # never deleted per the non-destructive invariant, so a second run over the # same raw objects necessarily rediscovers the same mapping -- this is # stability, not "second run finds nothing" (that would require scoping the # scan to ref-reachable objects only, which is a Phase 3/4 concern, not # this function's). # --------------------------------------------------------------------------- def test_cc_06_running_twice_is_deterministic(repo: pathlib.Path) -> None: identity = _make_identity() ids = _build_chain(repo, identity) snapshot_id_map = {ids["wrong_snap_b"]: ids["correct_snap_b"]} result1 = migrate_commit_ids_cascade(repo, snapshot_id_map, identity, dry_run=False) result2 = migrate_commit_ids_cascade(repo, snapshot_id_map, identity, dry_run=False) assert result1.id_map == result2.id_map assert result2.commits_written == 0, "second run must skip -- the new objects already exist" assert result2.commits_skipped == len(result1.id_map)