"""REC-240 — the composition_ref SSOT-flip acceptance tests. Proves the ONE-level indirection seam (SYNTHESIS: pointer → manifest → rendered view) is wired end-to-end WITHOUT changing spend behavior for legacy flat boards: * a board written WITH a ``composition_ref`` resolves through the manifest → view (the new indirection path), and the resolved artifact is the manifest's view, NOT the flat path; * a LEGACY flat board (no ``composition_ref``) resolves through ``resolve_board_for_spend`` byte-identically (back-compat — first-class path); * the producer ``composition_ref_from_board`` builds a real ``CompositionManifest`` from a board path and round-trips through ``board_record_to_cache``; * non-vacuousness: a ``composition_ref`` whose ``view`` differs from the flat ``artifact`` resolves to the VIEW (the manifest), proving the indirection actually fires. Reuses the live resolver fixtures/helpers from ``test_board_builder`` so the SSOT-stamp + freshness + version-discrimination path is the REAL paid-gate path, not a stub. """ from __future__ import annotations import pytest from recoil.api.schemas.composition import CompositionManifest from recoil.pipeline._lib import board_builder as bb from recoil.pipeline.core.take import Beat from recoil.pipeline.tests.test_board_builder import ( # live resolver harness project_paths, # noqa: F401 (pytest fixture) _resolver_approved_record, _resolver_beat, _resolver_board, _resolver_grouping, _stamp_resolver_record, ) _FLAT_ARTIFACT = "prep/ep_001/storyboards/EP001_CONT_004_v01.png" _VIEW_ARTIFACT = "prep/ep_001/storyboards/EP001_CONT_004_composed.png" # --------------------------------------------------------------------------- # Producer: composition_ref_from_board builds a real CompositionManifest and # board_record_to_cache round-trips the projected composition_ref. # --------------------------------------------------------------------------- def test_producer_builds_manifest_and_cache_round_trips() -> None: board = {"status": "approved", "artifact": _FLAT_ARTIFACT} # Real CompositionManifest construction (atom-version URN grammar enforced). members = ["atom:EP001/beat/EP001_SH10@t0"] ref = bb.composition_ref_from_board(board, kind="CONT", members=members) # The producer's ref is a faithful projection of a constructible manifest. manifest = CompositionManifest(kind=ref["kind"], members=ref["members"], layout=ref["layout"]) assert manifest.kind == "CONT" assert manifest.members == members # v1 trivial wrap: the rendered view IS today's board PNG. assert ref["view"] == _FLAT_ARTIFACT # Consumer: the SSOT record → cache projection carries composition_ref through. record = {**board, "composition_ref": ref} cache = bb.board_record_to_cache(record) assert cache["composition_ref"] == ref # And the cache is a valid Beat.board (the validator accepts the seam). Beat(beat_id="EP001_CONT_004", board={**cache, "source_sha256": "s", "approved_by": None, "updated_at": "2026-06-23T00:00:00Z"}) def test_producer_rejects_board_without_artifact() -> None: with pytest.raises(bb.BoardBuilderError): bb.composition_ref_from_board({"status": "approved"}) # --------------------------------------------------------------------------- # set_board_proposed seam: composition_ref is additive + round-trips; legacy # (no composition_ref) is unchanged. # --------------------------------------------------------------------------- def test_set_board_proposed_carries_composition_ref_alongside_artifact() -> None: beat = Beat("EP001_CONT_004") ref = {"kind": "CONT", "members": [], "layout": {}, "view": _VIEW_ARTIFACT} beat.set_board_proposed(_FLAT_ARTIFACT, "a" * 64, composition_ref=ref) # The flat artifact is preserved ALONGSIDE the composition_ref (not replaced). assert beat.board["artifact"] == _FLAT_ARTIFACT assert beat.board["composition_ref"] == ref # Round-trips through to_dict/from_dict (the validator accepts it). assert Beat.from_dict(beat.to_dict()).board == beat.board def test_set_board_proposed_legacy_has_no_composition_ref_key() -> None: beat = Beat("EP001_CONT_004") beat.set_board_proposed(_FLAT_ARTIFACT, "a" * 64) assert "composition_ref" not in beat.board # back-compat: legacy shape unchanged def test_validate_board_rejects_malformed_composition_ref() -> None: beat = Beat("EP001_CONT_004") with pytest.raises(ValueError): beat.set_board_proposed( _FLAT_ARTIFACT, "a" * 64, composition_ref={"kind": "NOPE", "members": [], "layout": {}, "view": _VIEW_ARTIFACT} ) # --------------------------------------------------------------------------- # resolve_board_for_spend: composition_ref present → manifest indirection; # absent → legacy byte-identical. # --------------------------------------------------------------------------- def test_resolve_board_with_composition_ref_resolves_through_manifest_view(project_paths): # noqa: F811 """A version-discriminated SSOT record carrying a composition_ref resolves the spend artifact THROUGH the manifest (pointer → manifest → view), not the flat path.""" project = project_paths.project beat = _resolver_beat(board=None, grouping=_resolver_grouping(["EP001_SH10"])) ref = {"kind": "CONT", "members": [], "layout": {}, "view": _VIEW_ARTIFACT} record = _resolver_approved_record(project, beat, composition_ref=ref) _stamp_resolver_record(project, beat, record) approved, board = bb.resolve_board_for_spend(project, 1, beat) assert approved is True # The resolved spend artifact is the manifest's VIEW, NOT the record's flat artifact. assert board["artifact"] == _VIEW_ARTIFACT assert record["artifact"] != _VIEW_ARTIFACT # non-vacuous: they genuinely differ assert board["composition_ref"] == ref def test_resolve_legacy_flat_board_unchanged_byte_identical(project_paths): # noqa: F811 """Back-compat anchor: a LEGACY flat record (no composition_ref) resolves EXACTLY as before — board == board_record_to_cache(record), artifact == the flat path. This test must pass on BOTH old and new code (the seam is purely additive).""" project = project_paths.project beat = _resolver_beat(board=None, grouping=_resolver_grouping(["EP001_SH10"])) record = _resolver_approved_record(project, beat) # NO composition_ref _stamp_resolver_record(project, beat, record) approved, board = bb.resolve_board_for_spend(project, 1, beat) assert approved is True assert board == bb.board_record_to_cache(record) # byte-identical projection assert board["artifact"] == record["artifact"] # flat path, unchanged assert "composition_ref" not in board def test_resolve_legacy_no_grouping_cache_fallback_unchanged(project_paths): # noqa: F811 """The non-grouped legacy cache-fallback branch also preserves byte-identical behavior when there is no composition_ref (the second return site of the indirection helper).""" cache = _resolver_board(status="approved") # no composition_ref beat = _resolver_beat(board=cache, grouping=None) approved, board = bb.resolve_board_for_spend(project_paths.project, 1, beat) assert approved is True assert board == cache # unchanged — legacy first-class path def test_resolve_legacy_no_grouping_cache_with_composition_ref_indirects(project_paths): # noqa: F811 """Even a non-grouped legacy beat whose cache carries a composition_ref indirects through the manifest view (the fallback return site applies the same seam).""" ref = {"kind": "CONT", "members": [], "layout": {}, "view": _VIEW_ARTIFACT} cache = {**_resolver_board(status="approved"), "composition_ref": ref} beat = _resolver_beat(board=cache, grouping=None) approved, board = bb.resolve_board_for_spend(project_paths.project, 1, beat) assert approved is True assert board["artifact"] == _VIEW_ARTIFACT # indirection fired assert cache["artifact"] != _VIEW_ARTIFACT # non-vacuous # --------------------------------------------------------------------------- # LIVE writer path (codex finding): _stamp_board_ssot builds the SSOT record # from Beat.board fields — it must carry composition_ref into that record, else # the indirection is dormant on the paid path (resolve_board_for_spend reads the # SSOT record, not Beat.board). # --------------------------------------------------------------------------- def test_stamp_board_ssot_carries_composition_ref_into_record(project_paths, monkeypatch): # noqa: F811 from recoil.pipeline.cli import generate as gen from recoil.pipeline._lib import derivation_manifest as dm captured: dict = {} monkeypatch.setattr( dm, "stamp_board", lambda project, ep, h, record, **kw: captured.update(record=record), ) # Avoid plan/script-span scaffolding — orthogonal to the seam under test. monkeypatch.setattr(gen, "_board_script_span_provenance", lambda *a, **k: ([], None)) ref = {"kind": "CONT", "members": [], "layout": {}, "view": _VIEW_ARTIFACT} board = {**_resolver_board(status="approved"), "composition_ref": ref} beat = _resolver_beat(board=board, grouping=_resolver_grouping(["EP001_SH10"])) gen._stamp_board_ssot(project_paths.project, 1, beat, scene_version=2) assert captured["record"].get("composition_ref") == ref # Non-vacuous: a board WITHOUT the seam omits the key (legacy byte-identical). captured.clear() board2 = {k: v for k, v in board.items() if k != "composition_ref"} beat2 = _resolver_beat(board=board2, grouping=_resolver_grouping(["EP001_SH10"])) gen._stamp_board_ssot(project_paths.project, 1, beat2, scene_version=2) assert "composition_ref" not in captured["record"]