"""묶음이 **끝에서 끝까지** 가나. ★유료 0. Codex 목록 5 (2026-08-31) — > 실제 general scene production 조립 → `LabeledRefPayload` → > `resolve_ref_roles` → validator 를 **한 끝점**으로 태워 > `plate/prev + context + detail`, close framing, unavailable, > stale member identity 를 무료로 잠금. ★조각을 따로 부르는 것과 **줄줄이 태우는 것**은 다르다. 앞서 나는 「그릇이 셋을 받는다」까지만 재고 「붙이는 코드」를 안 봤다가 지적받았다. """ from __future__ import annotations import pytest from app.core.errors import StaleUpstreamError from app.core.ref_contract_validator import (RefContractError, validate_attached_refs) from app.modules.pipeline import grounding_reference_bundle as gb from app.modules.pipeline.grounding_host_context import (PURPOSE_CONTEXT, PURPOSE_DETAIL) from app.services.prompt_service import (make_labeled_ref_payload, resolve_ref_roles) BG = "L01B01" #: 좌표 → bytes. ★durable sidecar 는 **좌표**를 적는다(CP 는 JSON). BLOBS = {"a-img": b"img", "a-a": b"a", "a-b": b"b", "a-same": b"same"} def _sha(b): import hashlib return hashlib.sha256(b).hexdigest() def _load(coord): aid = str((coord or {}).get("asset_id") or "") return BLOBS.get(aid, b""), {"source": "asset", "asset_id": aid} def _m(purpose, *, outcome=gb.OUTCOME_SELECTED, ident=None, img=b"img"): d = {"subject_final_id": "LP01", "purpose": purpose, "outcome": outcome, "member_identity": ident or f"id-{purpose}", "label": f"{purpose} 사진"} if img is not None: key = next(k for k, v in BLOBS.items() if v == img) d["source"] = "asset" d["asset_id"] = key d["content_sha256"] = _sha(img) return d def _rpc_sidecar(members): """★만드는 쪽이 하는 일 — members **와** required 를 **둘 다** 적는다.""" rpc = {} gb.write_sidecar(rpc, members) # ★원자적 producer 한 벌 return rpc def _plate(): """기존 배경판 — `background_chain_ref` 가 넣는 모양.""" return ([("배경판", b"plate")], [("background", BG)], ["background_chain_ref"], [{"pipeline_role": "background_render"}]) def _rpc(): return {"asset_requirements": { "required_refs": [{"kind": "background", "id": BG, "policy": "required"}], "readiness_policy": "na"}} def run_chain(members, *, close=False, plate=True): """★조립 → payload → 역할 → validator → 조사 검사. **줄줄이**.""" lr, am, rr, rm = _plate() if plate else ([], [], [], []) ph, required = gb.plan_bundle(members) gb.insert_bundle_refs(lr, am, rr, rm, ph, load_bytes=_load, where="t") payload = make_labeled_ref_payload( labeled_refs=lr, ref_roles=rr, ref_role_metadata=rm, attached_meta=am) res = resolve_ref_roles(payload) validate_attached_refs(_rpc(), lr, am, "프롬프트", close, chain_bg_lookup=lambda _b: None, reference_phrase_kinds=["background"]) gb.assert_grounding_attached(required, am) return {"lists": (lr, am, rr, rm), "roles": res.ref_roles, "instructions": res.ref_instructions, "required": required} class TestPlateAndContextAndDetailAllArrive: def test_three_photos_reach_the_prompt(self): got = run_chain([_m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, img=b"b")]) lr, am, rr, rm = got["lists"] assert len(lr) == len(am) == len(rr) == len(rm) == 3 assert len(got["roles"]) == 3, f"★프롬프트에 {len(got['roles'])}장" for i in (1, 2, 3): assert any(f"image {i}" in t.lower() for t in got["roles"]), \ f"★Image {i} 가 없다" def test_the_plate_is_still_there(self): _lr, am, _rr, _rm = run_chain([_m(PURPOSE_DETAIL)])["lists"] assert ("background", BG) in am def test_each_role_says_a_different_thing(self): """★맥락은 **분위기**, 상세는 **그 물건**. 뭉치면 안 된다.""" got = run_chain([_m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, img=b"b")]) joined = " ".join(got["instructions"]).lower() assert "mood" in joined, "★맥락의 분위기 지시가 없다" assert "reproduce the component" in joined, "★상세 지시가 없다" def test_the_merged_one_carries_both(self): """★같은 사진이면 한 장이 **둘 다** 말한다.""" got = run_chain([_m(PURPOSE_CONTEXT, img=b"same"), _m(PURPOSE_DETAIL, img=b"same")]) lr, _am, rr, _rm = got["lists"] assert len(lr) == 2, "★합쳐지지 않았다" assert "grounding_context_detail_ref" in rr joined = " ".join(got["instructions"]).lower() assert "place's architecture" in joined and "component" in joined class TestCloseFramingDoesNotHideAMissingDetail: """★★★지금 validator 는 클로즈업이면 배경 검사를 **통째로** 건너뛴다. 그 구멍을 **조사 검사가 메운다** — 클로즈업이든 아니든 `selected` 는 붙어야 한다. 부분 클로즈업이야말로 상세 참조가 가장 필요한 자리다. """ def test_it_passes_when_attached_even_in_close_framing(self): run_chain([_m(PURPOSE_DETAIL)], close=True) def test_a_missing_detail_still_stops_in_close_framing(self): _ph, required = gb.plan_bundle([_m(PURPOSE_DETAIL)]) lr, am, _rr, _rm = _plate() # ★클로즈업이라 validator 는 지나간다 — validate_attached_refs(_rpc(), lr, am, "프롬프트", True, chain_bg_lookup=lambda _b: None, reference_phrase_kinds=[]) # ★그런데 조사 검사는 **선다** with pytest.raises(gb.BundleContractError): gb.assert_grounding_attached(required, am) class TestUnavailableFlowsThroughWithoutBlocking: """★못 구한 것이 **하류를 막으면 안 된다** — HITL 없음.""" def test_both_unavailable_leaves_only_the_plate(self): got = run_chain([ _m(PURPOSE_CONTEXT, outcome=gb.OUTCOME_UNAVAILABLE, img=None), _m(PURPOSE_DETAIL, outcome=gb.OUTCOME_UNAVAILABLE, img=None)]) lr, _am, _rr, _rm = got["lists"] assert len(lr) == 1 and got["required"] == [] def test_one_selected_one_unavailable(self): got = run_chain([ _m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, outcome=gb.OUTCOME_UNAVAILABLE, img=None)]) lr, _am, _rr, _rm = got["lists"] assert len(lr) == 2 and len(got["required"]) == 1 class TestAStaleMemberIdentityIsCaught: """★★획득 계약이 바뀌면 **옛 사진으로 통과하면 안 된다**.""" def test_the_old_attachment_does_not_satisfy_the_new_obligation(self): old = run_chain([_m(PURPOSE_DETAIL, ident="A")]) _lr, old_am, _rr, _rm = old["lists"] _ph, new_required = gb.plan_bundle([_m(PURPOSE_DETAIL, ident="B")]) with pytest.raises(gb.BundleContractError): gb.assert_grounding_attached(new_required, old_am) def test_the_same_identity_still_passes(self): got = run_chain([_m(PURPOSE_DETAIL, ident="A")]) _lr, am, _rr, _rm = got["lists"] gb.assert_grounding_attached(got["required"], am) class TestTheChainStopsOnRealBreakage: """★양성 대조 — 줄이 끊기면 **선다**.""" def test_dropping_the_plate_stops(self): """★조사 참조가 기존 배경판을 **대신하지 못한다**.""" with pytest.raises((RefContractError, StaleUpstreamError)): run_chain([_m(PURPOSE_DETAIL)], plate=False) def test_an_undeclared_role_would_stop(self): """★역할을 선언 안 했으면 payload 에서 즉시 선다.""" from app.services.prompt_service import RefRoleError with pytest.raises(RefRoleError): make_labeled_ref_payload( labeled_refs=[("a", b"x")], ref_roles=["지어낸역할"], ref_role_metadata=[{}], attached_meta=[("background", "b")]) class TestTheSidecarGateIsByteIdenticalWhenAbsent: """★★★내가 Codex 에 물었던 것의 **답** — 비회귀를 어떻게 증명하나. 답은 **등가 시험**이다. sidecar 가 없으면 네 평행 목록이 **글자까지 그대로**여야 한다. 그러면 general scene 조립에 이 문을 넣어도 기존 CP 는 한 글자도 안 바뀐다. """ @staticmethod def _lists(): return ([("배경판", b"plate"), ("인물", b"c")], [("background", BG), ("character", "C01")], ["background_chain_ref", "character_ref"], [{"pipeline_role": "background_render"}, {"x": 1}]) @pytest.mark.parametrize("rpc", [ None, {}, "글자", 7, {"asset_requirements": {}}, {"other_key": [1, 2]}, ]) def test_no_sidecar_changes_nothing(self, rpc): import copy lists = self._lists() before = copy.deepcopy(lists) n = gb.attach_from_rpc(*lists, rpc, load_bytes=_load) assert n == 0, f"★{n}장을 붙였다" assert lists == before, "★목록이 바뀌었다" def test_no_sidecar_means_no_check_either(self): """★검사도 안 한다 — 요구가 없으니 막을 것도 없다.""" gb.assert_from_rpc({}, []) gb.assert_from_rpc(None, []) def test_an_empty_list_is_an_explicit_verdict(self): """★★빈 목록과 **없음**은 다르다. 빈 목록 = 「이 샷엔 묶음이 없다」는 **명시 판정**. 없음 = 「이 CP 는 새 모양이 아니다」. """ assert gb.members_from_rpc({gb.RPC_MEMBERS_KEY: []}) == [] assert gb.members_from_rpc({}) is None def test_a_broken_sidecar_stops(self): with pytest.raises(gb.BundleContractError): gb.members_from_rpc({gb.RPC_MEMBERS_KEY: "목록 아님"}) class TestTheValidatorReadsWhatWasWritten: """★★붙이는 쪽 목록을 그대로 넘기면 **둘 다 빠뜨려도 통과**한다 (Codex).""" def _rpc_with(self, members): return _rpc_sidecar(members) def test_attaching_writes_the_required_list(self): rpc = self._rpc_with([_m(PURPOSE_DETAIL)]) lr, am, rr, rm = _plate() gb.attach_from_rpc(lr, am, rr, rm, rpc, load_bytes=_load) assert gb.required_from_rpc(rpc), "★요구가 안 적혔다" gb.assert_from_rpc(rpc, am) def test_a_dropped_attachment_is_caught_by_the_written_list(self): """★★붙이다 빠뜨려도 **적힌 것**이 잡는다.""" rpc = self._rpc_with([_m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, img=b"b")]) lr, am, rr, rm = _plate() gb.attach_from_rpc(lr, am, rr, rm, rpc, load_bytes=_load) # ★붙인 것 하나를 **빼 본다** — 실제로 새는 상황 dropped = [x for x in am if not str(x[1]).startswith("grounding:")] with pytest.raises(gb.BundleContractError): gb.assert_from_rpc(rpc, dropped + [am[1]]) def test_the_written_list_survives_a_round_trip(self): import json rpc = self._rpc_with([_m(PURPOSE_DETAIL)]) lr, am, rr, rm = _plate() gb.attach_from_rpc(lr, am, rr, rm, rpc, load_bytes=_load) # ★CP 는 JSON 으로 오간다 — 되읽어도 같은 요구여야 한다 again = json.loads(json.dumps( {gb.RPC_REQUIRED_KEY: rpc[gb.RPC_REQUIRED_KEY]}, ensure_ascii=False)) assert gb.required_from_rpc(again) == gb.required_from_rpc(rpc) gb.assert_from_rpc(again, am) @pytest.mark.parametrize("bad", ["글자", [1], [{"kind": "a"}, "x"]]) def test_a_broken_required_list_stops(self, bad): if bad == "글자": with pytest.raises(gb.BundleContractError): gb.required_from_rpc({gb.RPC_REQUIRED_KEY: bad}) else: with pytest.raises(gb.BundleContractError): gb.required_from_rpc({gb.RPC_REQUIRED_KEY: bad}) # ───────────────────────────────────────────────────────────────────── # ★★★같은 사진을 둘이 가리킬 때 (Codex BLOCK 1, 2026-08-31) # # > A 파일은 claimed sha 와 일치, B 파일은 다른 bytes 인데 같은 sha 를 # > 주장하게 하면 inserted=1 로 통과하고 B 경로도 coordinates 에 남습니다. # # 앞 판은 `coordinates[0]` **하나만** 확인하면서 계보에는 **좌표 전부**를 # 적었다. 그래서 확인 안 한 자산이 「확인된 출처」 노릇을 했다. # ───────────────────────────────────────────────────────────────────── SAME = "두 갈래가 같은 사진을 가리킨다".encode("utf-8") def _two_pointing_at(*, a_bytes, b_bytes): """맥락·상세가 **각각 다른 좌표**로 **같은 sha 를 주장**한다.""" claimed = _sha(SAME) store = {"코드-A": a_bytes, "코드-B": b_bytes} def load(coord): aid = str((coord or {}).get("asset_id") or "") return store.get(aid, b""), {"source": "asset", "asset_id": aid} def member(purpose, aid): return {"subject_final_id": "LP01", "purpose": purpose, "outcome": gb.OUTCOME_SELECTED, "member_identity": f"id-{purpose}", "label": f"{purpose} 사진", "source": "asset", "asset_id": aid, "content_sha256": claimed} return load, [member(PURPOSE_CONTEXT, "코드-A"), member(PURPOSE_DETAIL, "코드-B")] def _run_dedupe(members, load): """★`run_chain` 과 **같은 끝점** — 읽는 함수만 갈아 끼운다.""" lr, am, rr, rm = _plate() plan, required = gb.plan_bundle(members) n = gb.insert_bundle_refs(lr, am, rr, rm, plan, load_bytes=load, where="t") payload = make_labeled_ref_payload(labeled_refs=lr, ref_roles=rr, ref_role_metadata=rm, attached_meta=am) resolve_ref_roles(payload) validate_attached_refs(_rpc(), lr, am, "프롬프트", False, chain_bg_lookup=lambda _b: None, reference_phrase_kinds=["background"]) gb.assert_grounding_attached(required, am) return n, lr, rm class TestBothCoordinatesAreVerified: def test_two_pointers_to_one_photo_become_one_image(self): """둘 다 진짜 같은 사진이면 **1장**으로 접힌다.""" load, members = _two_pointing_at(a_bytes=SAME, b_bytes=SAME) n, lr, rm = _run_dedupe(members, load) assert n == 1, f"★같은 사진인데 {n}장 붙었다" assert len(lr) == 2, "★배경판 + 접힌 1장" got = rm[0] # ★묶음은 **앞쪽**에 꽂힌다(배경판이 뒤) assert sorted(got.get("purposes") or []) == [PURPOSE_CONTEXT, PURPOSE_DETAIL] # ★두 좌표를 **다 읽어 다 확인**했다는 기록이 남는다 assert len(got.get("verified_sources") or []) == 2, \ f"★확인된 출처가 {got.get('verified_sources')}" @pytest.mark.parametrize("order", [(0, 1), (1, 0)]) def test_the_order_does_not_change_the_result(self, order): """★AB 와 BA 가 **같아야** 한다 — 앞 판은 갈렸다.""" load, members = _two_pointing_at(a_bytes=SAME, b_bytes=SAME) n, lr, rm = _run_dedupe([members[i] for i in order], load) assert n == 1 assert sorted(rm[0]["purposes"]) == [PURPOSE_CONTEXT, PURPOSE_DETAIL] assert sorted(s["asset_id"] for s in rm[0]["verified_sources"]) == \ ["코드-A", "코드-B"], "★한쪽 좌표의 계보가 사라졌다" @pytest.mark.parametrize("bad_at", [0, 1]) def test_a_contaminated_coordinate_stops_it(self, bad_at): """★★**첫째든 둘째든** 사진이 주장과 다르면 선다. 앞 판은 **둘째만 오염되면 통과**했다 — 그게 이 시험의 자리다. """ blobs = [SAME, SAME] blobs[bad_at] = "전혀 다른 사진".encode("utf-8") load, members = _two_pointing_at(a_bytes=blobs[0], b_bytes=blobs[1]) with pytest.raises(gb.BundleContractError) as e: _run_dedupe(members, load) assert "코드-A" in str(e.value) or "코드-B" in str(e.value) def test_an_unreadable_second_coordinate_stops_it(self): """★못 읽은 좌표를 계보로 남기지 않는다.""" load, members = _two_pointing_at(a_bytes=SAME, b_bytes=b"") with pytest.raises(gb.BundleContractError): _run_dedupe(members, load)