"""TDD — push must verify objects exist in storage, not just in the DB. Root cause of the staging missing-objects incident (2026-04-28): The referential integrity check at end of push queried musehub_objects DB rows for externally-referenced objects, but did NOT verify those objects exist in R2/S3 storage. Result: DB row present, R2 bytes absent → push accepted → raw/{file} returns 404 (object missing from storage). These tests codify the three failure modes and the correct behaviour: I1 When backend.put() silently returns but stores nothing, subsequent GET must fail — and the push handler must detect this and emit ERROR. I2 When an externally-referenced object is in the DB but absent from actual storage, the referential integrity check must emit ERROR, not RESULT ok=True. I3 After a successful push, every object referenced by any snapshot manifest must be retrievable from storage (exists() == True). I4 Static check: the integrity check in musehub_wire.py must call backend.exists() for externally-referenced objects, not just query DB. """ from __future__ import annotations import inspect import msgpack import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from unittest.mock import AsyncMock, MagicMock, patch from muse.core.types import blob_id from muse.core.mpack import WIRE_CONTENT_TYPE, MuseWireFrameWriter from musehub.models.wire import ( SFRAME_COMMIT_PACK, SFRAME_END, SFRAME_HEADER, SFRAME_RESULT, SFRAME_OBJECT, ) from musehub.types.json_types import JSONObject, JSONValue, StrDict from tests.factories import create_repo _fw = MuseWireFrameWriter() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _pack(obj: JSONValue) -> bytes: return msgpack.packb(obj, use_bin_type=True) def _wrap(ft: str, data: JSONValue) -> bytes: return _fw.wrap(frame_type=ft, payload=_pack(data)) def _oid(data: bytes) -> str: return blob_id(data) def _header_frame(n_objects: int = 0, n_commits: int = 1) -> bytes: return _wrap(SFRAME_HEADER, { "t": SFRAME_HEADER, "branch": "main", "force": False, "have": [], "head": _oid(b"head"), "n_objects": n_objects, "n_commits": n_commits, }) def _object_frame(raw: bytes, path: str = "file.py") -> tuple[str, bytes]: oid = _oid(raw) frame = _wrap(SFRAME_OBJECT, { "t": SFRAME_OBJECT, "id": oid, "path": path, "enc": "raw", "content": raw, }) return oid, frame def _commit_pack_frame( commits: list[JSONObject], snapshots: list[JSONObject], ) -> bytes: return _wrap(SFRAME_COMMIT_PACK, { "t": SFRAME_COMMIT_PACK, "commits": commits, "snapshots": snapshots, }) def _end_frame(n_objects: int = 0, n_commits: int = 1) -> bytes: return _wrap(SFRAME_END, { "t": SFRAME_END, "n_objects": n_objects, "n_commits": n_commits, }) def _make_commit(snapshot_id: str, branch: str = "main") -> JSONObject: return { "commit_id": _oid(f"commit-{snapshot_id}".encode()), "parent_ids": [], "snapshot_id": snapshot_id, "branch": branch, "message": "integrity test commit", "author": "gabriel", "committed_at": "2026-04-28T00:00:00+00:00", "signature": "", "signer_key_id": "", "agent_id": "claude-code", "model_id": "claude-sonnet-4-6", "metadata": {}, } def _make_snapshot(snap_id: str, manifest: StrDict) -> JSONObject: return {"snapshot_id": snap_id, "manifest": manifest} # --------------------------------------------------------------------------- # I1 — when put() silently fails, storage must not report the object present # --------------------------------------------------------------------------- def test_i1_local_backend_put_is_durable(tmp_path: "Path") -> None: # type: ignore[name-defined] # noqa: F821 """LocalBackend.put() must persist bytes such that exists() returns True. This test pins the contract: if put() returns a URI, exists() must be True afterwards. A backend that violates this contract allows ghost DB rows. """ import asyncio from musehub.storage.backends import LocalBackend backend = LocalBackend() repo_root = tmp_path / "repos" / "gabriel" / "test" raw = b"hello world" oid = _oid(raw) asyncio.run(backend.put(oid, raw, repo_root=repo_root)) assert asyncio.run(backend.exists(oid, repo_root=repo_root)) is True # --------------------------------------------------------------------------- # I2 — externally-referenced object in DB but absent from storage → ERROR # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_i2_external_ref_in_db_but_missing_from_storage_is_rejected( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Push must be rejected when a snapshot references an object in the DB but absent from actual storage. Scenario: - A previous push stored object X in DB (musehub_objects row exists). - The R2/S3 bytes for X are gone — storage.exists() returns False. - A new push sends a snapshot referencing X as an external object (not included in this push's object frames). Expected: push emits ERROR, not RESULT ok=True. Note: request sessions are independent from the test session. Test data must be committed (not just flushed) to be visible to the push handler's session. """ import musehub.db.musehub_models as db_models repo = await create_repo(db_session, owner="test-user-wire", name="i2-external-missing") # Plant a ghost DB row: object in musehub_objects, but NOT in storage. # The tmp-path backend from conftest._tmp_objects_dir starts empty, so # any object in DB but not pushed through the backend is a ghost. ghost_raw = b"ghost object bytes - i2" ghost_oid = _oid(ghost_raw) db_session.add(db_models.MusehubObject( object_id=ghost_oid, path="ghost.py", size_bytes=len(ghost_raw), disk_path=f"objects/{ghost_oid[7:9]}/{ghost_oid[9:]}", storage_uri=f"local://objects/{ghost_oid[7:9]}/{ghost_oid[9:]}", )) # Commit so the push handler's independent session can see this row. await db_session.commit() # Build a snapshot that references the ghost object externally. snap_id = _oid(b"snap-i2") manifest = {"ghost.py": ghost_oid} commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, manifest) # Push with NO object frames — ghost_oid is external (not in this push). body = ( _header_frame(n_objects=0, n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=0, n_commits=1) ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**wire_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) # The response may be 200 with embedded error frames or a 4xx. assert resp.status_code in (200, 422), ( f"Unexpected status {resp.status_code}: {resp.text}" ) if resp.status_code == 200: unpacker = msgpack.Unpacker(raw=False) unpacker.feed(resp.content) frames = list(unpacker) ok_true = [f for f in frames if f.get("ok") is True] assert not ok_true, ( f"Push must not emit ok=True when a referenced object is in DB " f"but missing from storage. Got frames: {frames}" ) # 422 from the server is also a correct rejection. # --------------------------------------------------------------------------- # I3 — after successful push, all snapshot-referenced objects are in storage # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_i3_successful_push_all_objects_retrievable( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Every object referenced by a snapshot must be in storage after a push. This is the golden-path integration test: push a commit with a snapshot whose manifest references one object, then verify backend.exists() returns True for that object. Regression guard for the staging gap. """ import musehub.services.musehub_wire as wire_svc repo = await create_repo(db_session, owner="test-user-wire", name="i3-all-objects") raw = b"real file bytes" oid, obj_frame = _object_frame(raw, path="real.py") snap_id = _oid(b"snap-i3") manifest = {"real.py": oid} commit = _make_commit(snapshot_id=snap_id) snap = _make_snapshot(snap_id, manifest) body = ( _header_frame(n_objects=1, n_commits=1) + obj_frame + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=1, n_commits=1) ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**wire_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code == 200 unpacker = msgpack.Unpacker(raw=False) unpacker.feed(resp.content) frames = list(unpacker) ok_frames = [f for f in frames if f.get("ok") is True] assert ok_frames, f"Expected ok=True, got: {frames}" # All snapshot-referenced objects must now be in storage. from musehub.storage.backends import repo_root_for backend = wire_svc.get_backend() _repo_root = repo_root_for(repo.owner, repo.slug) for path, ref_oid in manifest.items(): assert await backend.exists(ref_oid, repo_root=_repo_root), ( f"Object {ref_oid} (path={path}) is referenced by snapshot manifest " f"but missing from storage after successful push." ) # --------------------------------------------------------------------------- # I4 — static check: integrity check must use backend.exists(), not just DB # --------------------------------------------------------------------------- def test_i4_integrity_check_uses_backend_exists() -> None: """The referential integrity check in wire_push_stream must call backend.exists() for externally-referenced objects, not just query the DB. Querying DB rows alone cannot detect the ghost-object scenario where a row exists in musehub_objects but the storage bytes are absent. This is a static assertion on the source of musehub_wire.py. """ from musehub.services import musehub_wire source = inspect.getsource(musehub_wire) assert "backend.exists" in source, ( "musehub_wire.py referential integrity check must call backend.exists() " "to verify objects are in actual storage, not just in the DB. " "DB rows can exist without corresponding R2/S3 bytes (ghost objects). " "Add: `if not await backend.exists(oid): ...` for externally-referenced objects." )