test_wire_snapshot_delta.py
python
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d
feat(pack): delta-encode snapshots in MPackBundle wire format
Sonnet 4.6
minor
⚠ breaking
121 days ago
| 1 | """Tests for snapshot delta encoding in _frame_generator. |
| 2 | |
| 3 | Verifies that _frame_generator: |
| 4 | A. Sends the first snapshot as a full manifest |
| 5 | B. Sends subsequent snapshots as deltas |
| 6 | C. Deltas chain correctly (each references the previous snapshot) |
| 7 | D. A stream with only one commit sends a full snapshot (no delta) |
| 8 | E. write_commit_pack includes snapshot_deltas in msgpack output |
| 9 | F. Applying every delta in stream order reconstructs all manifests |
| 10 | """ |
| 11 | from __future__ import annotations |
| 12 | |
| 13 | import struct |
| 14 | import msgpack |
| 15 | from muse.core.types import Manifest, MsgpackDict, blob_id, long_id |
| 16 | |
| 17 | |
| 18 | # --------------------------------------------------------------------------- |
| 19 | # Helpers |
| 20 | # --------------------------------------------------------------------------- |
| 21 | |
| 22 | def _parse_mwp_frames(data: bytes) -> list[dict]: |
| 23 | """Parse concatenated MWP-framed bytes into decoded payload dicts.""" |
| 24 | frames: list[dict] = [] |
| 25 | pos = 0 |
| 26 | magic = b"muse" |
| 27 | while pos < len(data): |
| 28 | assert data[pos:pos + 4] == magic, f"bad magic at {pos}" |
| 29 | pos += 4 |
| 30 | pos += 1 # version byte |
| 31 | header_len = struct.unpack_from(">I", data, pos)[0] |
| 32 | pos += 4 + header_len |
| 33 | payload_len = struct.unpack_from(">Q", data, pos)[0] |
| 34 | pos += 8 |
| 35 | payload_bytes = data[pos:pos + payload_len] |
| 36 | pos += payload_len |
| 37 | frames.append(msgpack.unpackb(payload_bytes, raw=False)) |
| 38 | return frames |
| 39 | |
| 40 | |
| 41 | def _parse_mwp_payload(mwp_frame: bytes) -> MsgpackDict: |
| 42 | """Extract msgpack payload from a raw MWP frame (past the binary envelope).""" |
| 43 | # MWP: b"muse" | 0x01 | uint32 header_len | msgpack(envelope) | uint64 payload_len | payload |
| 44 | assert mwp_frame[:4] == b"muse" |
| 45 | assert mwp_frame[4] == 0x01 |
| 46 | header_len = struct.unpack(">I", mwp_frame[5:9])[0] |
| 47 | payload_offset = 9 + header_len + 8 # skip envelope + 8-byte payload_len |
| 48 | payload = mwp_frame[payload_offset:] |
| 49 | return msgpack.unpackb(payload, raw=False) |
| 50 | |
| 51 | |
| 52 | def _build_manifest(n_files: int) -> Manifest: |
| 53 | return {f"src/module_{i}.py": long_id(f"{'a' * 60}{i:04d}") for i in range(n_files)} |
| 54 | |
| 55 | |
| 56 | def _make_commit(snapshot_id: str, i: int) -> MsgpackDict: |
| 57 | return { |
| 58 | "commit_id": long_id(f"{'c' * 60}{i:04d}"), |
| 59 | "snapshot_id": snapshot_id, |
| 60 | "message": f"commit {i}", |
| 61 | "branch": "dev", |
| 62 | "author": "gabriel", |
| 63 | "committed_at": "2026-04-23T00:00:00+00:00", |
| 64 | "parent_commit_id": None, |
| 65 | "agent_id": "", |
| 66 | "model_id": "", |
| 67 | "toolchain_id": "", |
| 68 | "signer_key": "", |
| 69 | "signature": "", |
| 70 | } |
| 71 | |
| 72 | |
| 73 | def _make_snapshot(manifest: Manifest, i: int) -> MsgpackDict: |
| 74 | snap_bytes = msgpack.packb(manifest, use_bin_type=True) |
| 75 | snap_id = blob_id(snap_bytes) |
| 76 | return { |
| 77 | "snapshot_id": snap_id, |
| 78 | "manifest": manifest, |
| 79 | "directories": [], |
| 80 | "created_at": "2026-04-23T00:00:00+00:00", |
| 81 | "note": "", |
| 82 | } |
| 83 | |
| 84 | |
| 85 | def _build_linear_history(n_commits: int, n_files: int) -> tuple[list, list, dict]: |
| 86 | """Build a sequence of commits each changing 3 files in a manifest. |
| 87 | |
| 88 | Returns: (commits, snapshots, snap_by_id) |
| 89 | """ |
| 90 | commits = [] |
| 91 | snapshots = [] |
| 92 | snap_by_id = {} |
| 93 | |
| 94 | manifest = _build_manifest(n_files) |
| 95 | for i in range(n_commits): |
| 96 | # Mutate 3 files per commit |
| 97 | new_manifest = dict(manifest) |
| 98 | for j in range(3): |
| 99 | path = f"src/module_{(i * 3 + j) % n_files}.py" |
| 100 | new_manifest[path] = long_id(f"{'b' * 60}{i * 3 + j:04d}") |
| 101 | snap = _make_snapshot(new_manifest, i) |
| 102 | commit = _make_commit(snap["snapshot_id"], i) |
| 103 | commits.append(commit) |
| 104 | snapshots.append(snap) |
| 105 | snap_by_id[snap["snapshot_id"]] = snap |
| 106 | manifest = new_manifest |
| 107 | |
| 108 | return commits, snapshots, snap_by_id |
| 109 | |
| 110 | |
| 111 | # --------------------------------------------------------------------------- |
| 112 | # A. First snapshot in stream is always a full manifest |
| 113 | # --------------------------------------------------------------------------- |
| 114 | |
| 115 | def test_first_snapshot_is_full() -> None: |
| 116 | from muse.core.transport import _frame_generator |
| 117 | |
| 118 | commits, snapshots, snap_by_id = _build_linear_history(n_commits=5, n_files=100) |
| 119 | |
| 120 | raw = b"".join(_frame_generator([], commits, snapshots, local_head=None)) |
| 121 | c_frames = [f for f in _parse_mwp_frames(raw) if f.get("t") == "C"] |
| 122 | assert c_frames, "expected at least one C frame" |
| 123 | assert len(c_frames[0].get("snapshots", [])) >= 1, "first C frame must contain at least one full snapshot" |
| 124 | |
| 125 | |
| 126 | # --------------------------------------------------------------------------- |
| 127 | # B. Subsequent snapshots sent as deltas |
| 128 | # --------------------------------------------------------------------------- |
| 129 | |
| 130 | def test_subsequent_snapshots_are_deltas() -> None: |
| 131 | from muse.core.transport import _frame_generator |
| 132 | |
| 133 | commits, snapshots, _ = _build_linear_history(n_commits=10, n_files=100) |
| 134 | |
| 135 | raw = b"".join(_frame_generator([], commits, snapshots, local_head=None)) |
| 136 | c_frames = [f for f in _parse_mwp_frames(raw) if f.get("t") == "C"] |
| 137 | |
| 138 | all_deltas = [] |
| 139 | for cf in c_frames: |
| 140 | all_deltas.extend(cf.get("snapshot_deltas") or []) |
| 141 | |
| 142 | assert len(all_deltas) > 0, "expected delta snapshots for commits 2+" |
| 143 | |
| 144 | |
| 145 | # --------------------------------------------------------------------------- |
| 146 | # C. Delta chain integrity — each delta references a previously sent snapshot |
| 147 | # --------------------------------------------------------------------------- |
| 148 | |
| 149 | def test_delta_chain_references_valid_base() -> None: |
| 150 | from muse.core.transport import _frame_generator |
| 151 | |
| 152 | commits, snapshots, _ = _build_linear_history(n_commits=20, n_files=100) |
| 153 | |
| 154 | raw = b"".join(_frame_generator([], commits, snapshots, local_head=None)) |
| 155 | c_frames = [f for f in _parse_mwp_frames(raw) if f.get("t") == "C"] |
| 156 | |
| 157 | seen_snapshot_ids: set[str] = set() |
| 158 | for cf in c_frames: |
| 159 | for snap in cf.get("snapshots") or []: |
| 160 | seen_snapshot_ids.add(snap["snapshot_id"]) |
| 161 | for delta in cf.get("snapshot_deltas") or []: |
| 162 | base_id = delta.get("base_id") |
| 163 | assert base_id in seen_snapshot_ids, ( |
| 164 | f"delta references base {base_id!r} not yet seen in stream" |
| 165 | ) |
| 166 | seen_snapshot_ids.add(delta["snapshot_id"]) |
| 167 | |
| 168 | |
| 169 | # --------------------------------------------------------------------------- |
| 170 | # D. Single commit — full snapshot, no deltas |
| 171 | # --------------------------------------------------------------------------- |
| 172 | |
| 173 | def test_single_commit_sends_full_snapshot_no_deltas() -> None: |
| 174 | from muse.core.transport import _frame_generator |
| 175 | |
| 176 | commits, snapshots, _ = _build_linear_history(n_commits=1, n_files=50) |
| 177 | |
| 178 | raw = b"".join(_frame_generator([], commits, snapshots, local_head=None)) |
| 179 | c_frames = [f for f in _parse_mwp_frames(raw) if f.get("t") == "C"] |
| 180 | |
| 181 | assert len(c_frames) == 1 |
| 182 | assert len(c_frames[0].get("snapshots", [])) == 1 |
| 183 | assert c_frames[0].get("snapshot_deltas") in (None, []) |
| 184 | |
| 185 | |
| 186 | # --------------------------------------------------------------------------- |
| 187 | # E. write_commit_pack includes snapshot_deltas key when provided |
| 188 | # --------------------------------------------------------------------------- |
| 189 | |
| 190 | def test_write_commit_pack_includes_snapshot_deltas() -> None: |
| 191 | from muse.core.mpack import MPackStreamWriter |
| 192 | writer = MPackStreamWriter() |
| 193 | |
| 194 | delta = { |
| 195 | "snapshot_id": long_id("a" * 64), |
| 196 | "base_id": long_id("b" * 64), |
| 197 | "added": {"src/foo.py": long_id("c" * 64)}, |
| 198 | "removed": [], |
| 199 | "directories": [], |
| 200 | "created_at": "2026-04-23T00:00:00+00:00", |
| 201 | "note": "", |
| 202 | } |
| 203 | raw = writer.write_commit_pack(commits=[], snapshots=[], snapshot_deltas=[delta]) |
| 204 | frame = msgpack.unpackb(raw, raw=False) |
| 205 | assert "snapshot_deltas" in frame |
| 206 | assert len(frame["snapshot_deltas"]) == 1 |
| 207 | assert frame["snapshot_deltas"][0]["base_id"] == long_id("b" * 64) |
| 208 | |
| 209 | |
| 210 | def test_write_commit_pack_no_deltas_key_when_empty() -> None: |
| 211 | from muse.core.mpack import MPackStreamWriter |
| 212 | writer = MPackStreamWriter() |
| 213 | raw = writer.write_commit_pack(commits=[], snapshots=[]) |
| 214 | frame = msgpack.unpackb(raw, raw=False) |
| 215 | assert "snapshot_deltas" not in frame |
| 216 | |
| 217 | |
| 218 | # --------------------------------------------------------------------------- |
| 219 | # F. Applying every delta in stream order reconstructs all manifests |
| 220 | # --------------------------------------------------------------------------- |
| 221 | |
| 222 | def test_delta_chain_full_reconstruction() -> None: |
| 223 | """All manifests reconstructed by walking the delta chain must equal |
| 224 | the original snapshots passed to _frame_generator.""" |
| 225 | from muse.core.transport import _frame_generator |
| 226 | from muse.core.mpack import apply_snapshot_delta |
| 227 | |
| 228 | n_commits = 30 |
| 229 | commits, snapshots, snap_by_id = _build_linear_history(n_commits=n_commits, n_files=200) |
| 230 | |
| 231 | raw = b"".join(_frame_generator([], commits, snapshots, local_head=None)) |
| 232 | c_frames = [f for f in _parse_mwp_frames(raw) if f.get("t") == "C"] |
| 233 | |
| 234 | # Walk stream, reconstructing every snapshot |
| 235 | reconstructed: dict[str, dict] = {} # snapshot_id → manifest |
| 236 | |
| 237 | for cf in c_frames: |
| 238 | for snap in cf.get("snapshots") or []: |
| 239 | reconstructed[snap["snapshot_id"]] = snap["manifest"] |
| 240 | for delta in cf.get("snapshot_deltas") or []: |
| 241 | base_manifest = reconstructed[delta["base_id"]] |
| 242 | full = apply_snapshot_delta(base_manifest, delta["added"], delta["removed"]) |
| 243 | reconstructed[delta["snapshot_id"]] = full |
| 244 | |
| 245 | for snap in snapshots: |
| 246 | sid = snap["snapshot_id"] |
| 247 | assert sid in reconstructed, f"snapshot {sid[:16]} not in reconstructed" |
| 248 | assert reconstructed[sid] == snap["manifest"], ( |
| 249 | f"reconstructed manifest for {sid[:16]} does not match original" |
| 250 | ) |
File History
1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d
feat(pack): delta-encode snapshots in MPackBundle wire format
Sonnet 4.6
minor
⚠
121 days ago