"""**실제 사슬** 끝점 — 다섯 갈래가 공통 장부를 지나 묶음까지. ★유료 0. Codex (2026-09-01) — 「다섯 갈래가 **손으로 만든 시험값이 아니라 실제 producer 산출에서** 공통 장부로 들어가는지, 부모 장소 참고사진이 참조 조사 단계와 최종 이미지 입력 단계 **모두에서 한 번만** 처리되는지 확인하겠습니다.」 ★새 원고를 만들지 않는다 — 이미 있는 `period_episode`(1960년대) fixture 를 쓴다. 손으로 짓는 것은 **모델이 낸 행** 하나뿐이고, 그 뒤는 전부 production 함수다 — reduce_episode → project_rows_and_candidates → build_subjects → build_population → screen_subjects → 갈래별 resolver → merge → **의무 계획** → 중앙 조사기 → 묶음 투영 → plan_bundle ★★의무는 **조사 앞**에서 선다 (Codex BLOCK · 09-01). 투영은 결과를 옮기기만 한다 — 조사 끝난 뒤 맥락 멤버를 지어내면 「따로 조사한 맥락 사진」이 없다. """ from __future__ import annotations import pytest from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline import grounding_acquisition_ledger as gl from app.modules.pipeline import grounding_bundle_projection as bp from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_chunk_merge as cm from app.modules.pipeline import grounding_reference_bundle as rb from app.modules.pipeline import grounding_reference_obligations as ro from app.modules.pipeline import grounding_screen as gs from app.modules.pipeline.grounding_entity_contract import ( MATERIALIZABLE_OWNER_TYPES, REFERENCE_KIND_BY_OWNER) from tests.grounding.fixtures import period_episode as ep #: fixture 가 정한 부모 관계. ★여기서 짐작하지 않는다. PARENT_KEY = {"owner_location_part": "owner_location", "plain_location_part": "owner_location", "owner_outlook": "owner_character"} #: ★★부모 장소 행이 **없는** LP — 실제 갈래다(원문에 그 장소가 독립 대상으로 #: 안 잡혔다). 모델이 `host_context` 로 맥락을 낸다. 그 값은 fixture 의 #: 시대·지역과 그 낱말에서만 온다 — 지어낸 고유명사가 아니다. HOSTLESS = { "key": "hostless_location_part", "owner": "location_part", "at": [(1, "이발소 회전 간판", 1)], "host": {"state": "explicit_context_only", "parent_local_id": None, "search_subject": None, "evidence": None}, } def _model_rows(): """fixture 표적을 **모델이 낸 행** 모양으로. ★두 축은 fixture 것만 쓴다.""" rows, rel, key_of = [], [], {} for i, t in enumerate(ep.EXPECTED_TARGETS): scene, word, nth = t["at"][0] lid = f"c0#{i}" key_of[t["key"]] = lid axis = ep.AXIS_BASIS.get(t["key"]) or {} rows.append({ "local_id": lid, "owner_type": t["owner"], "surface_form": word, "occurrences": [{"source_span": ep.span_of(scene, word, nth), "source_quote": word}], "shot_binding_status": "bound_complete", "shot_appearance_ids": [f"s{scene}#1", f"s{scene}#2"], "hard_to_generate": bool(axis.get("hard")), "viewers_would_notice": bool(axis.get("notice")), # ★검색 재료 — **fixture 의 시대·지역과 그 낱말**에서만 온다. # producer 계약(`PRODUCER_PAYLOAD_FIELDS`)의 칸이다. "coarse_type_label": word, "visual_brief": f"{ep.ERA} {ep.REGION} {word}", "search_terms_native": [word, f"{ep.ERA} {word}"], "language_lock_native": "ko", }) for child, parent in PARENT_KEY.items(): rel.append({"remove_local_id": key_of[child], "keep_local_id": key_of[parent], "relation": cm.REL_PART_OF}) # ★부모 좌표가 있는 LP 는 `bound_parent` 를 낸다 — 그 장소가 이 판독에 # 독립 행으로 있기 때문이다 for child, parent in PARENT_KEY.items(): if not child.endswith("location_part"): continue for r in rows: if r["local_id"] == key_of[child]: r["host_context"] = {"state": "bound_parent", "parent_local_id": key_of[parent], "search_subject": None, "evidence": []} return rows, rel, key_of #: 낱말 → fixture 가 정한 두 축. ★대역이 **제 판단을 안 한다**. _AXIS_BY_WORD = {t["at"][0][1]: bool((ep.AXIS_BASIS.get(t["key"]) or {}) .get("hard")) for t in ep.EXPECTED_TARGETS} def _assess_like_the_fixture(surface_by_rsid): """판별 대역 — **fixture 의 두 축 그대로** 답한다. ★★앞 판은 무엇이든 `PLAN_OK` 를 줘서 **음성 대조(가위·창)까지 조사 대상**이 됐다. 그러면 판별이 아무 일도 안 한 판을 재는 것이다. """ from app.modules.pipeline import era_research as era def _fn(**kw): word = surface_by_rsid.get(str(kw.get("canonical_scope_id") or ""), "") if _AXIS_BY_WORD.get(word): return {"status": era.PLAN_OK, "plan": {"assess_sha": f"sha_{word}", "subjects": [{"q": word}]}} return {"status": era.PLAN_NO_SUBJECT, "plan": None} return _fn @pytest.fixture(scope="module") def chain(): """★**production 함수만** 태운다. 손으로 짓는 것은 모델 행뿐이다.""" from app.modules.pipeline import grounding_carry as gc rows, rel, key_of = _model_rows() red = cm.reduce_episode(rows, rel, segments=ep.segment_texts()) proj = ad.project_rows_and_candidates(red, project_id="p", episode_id="e") built = gc.build_subjects({k: list(v) for k, v in proj["rows"].items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=proj["candidates"]) ledger = built.get("candidate_ledger") or {} disp = dict(built.get("subject_dispositions") or {}) disp.update({str(r.get("research_subject_id") or ""): str(r.get("disposition") or "") for r in (ledger.get("rows") or [])}) pop = gs.build_population(built["subjects"], proj["candidates"], disp) # ★★대역의 표는 **판별이 실제로 보는 목록**에서 온다. 앞 판은 # `built["subjects"]` 만 썼는데 A0 후보로 들어온 아웃룩이 거기 없어서 # 낱말을 못 찾고 조용히 「비대상」이 됐다 — 다섯 갈래 중 하나가 빠진 # 것을 못 볼 뻔했다. surface_by_rsid = {str(x.get("research_subject_id") or ""): str(x.get("surface_form") or "") for x in pop} screened = gs.screen_subjects( pop, dispositions=disp, world_facts_block=ep.WORLD_FACTS, cache_get=lambda k: None, cache_put=lambda k, v: v, assess_fn=_assess_like_the_fixture(surface_by_rsid)) # ★★subject → 등록된 정본 ID. **production 이 낸 것**에서 온다 — # 후보 장부의 `producer_link.final_id` 가 그것이다(발급자·계약판까지 # 검증된 링크). 앞 판은 `_short_id` 를 읽었는데 그런 칸이 **없어서** # 표가 통째로 비었고, 세 갈래가 조용히 unresolved 로 빠졌다. final_by_subject = { str(r.get("research_subject_id") or ""): str(((r.get("producer_link") or {}).get("final_id")) or "") for r in (ledger.get("rows") or []) if (r.get("producer_link") or {}).get("final_id")} return {"reduced": red, "proj": proj, "built": built, "screen": screened, "final_by_subject": final_by_subject, "key_of": key_of} @pytest.fixture(scope="module") def merged(chain): """갈래별 resolver → **공통 장부 하나**. ★손으로 만든 줄이 없다.""" from app.modules.pipeline import grounding_outlook_binding as ob rows = chain["screen"]["rows"] by_owner = gl.split_by_owner(rows) base = gl.resolve_entity_backed( [r for o in ("character", "location", "prop") for r in by_owner[o]], final_id_by_subject=chain["final_by_subject"]) facet = gl.resolve_facet_rows(by_owner["location_part"], final_id_by_subject=chain["final_by_subject"]) # ★outlook 은 phase3 뒤에 결속된다 — fixture 가 정한 부모로 배정을 만든다 parent_of = {} for r in by_owner["outlook"]: fb = r.get("grounding_facet_binding") or {} parent_of[str(fb.get("final_id") or "")] = str( fb.get("parent_final_id") or "") p3 = {"outlooks": [{"outlook_id": o} for o in parent_of if o], "scene_assignments": [{"scene_index": 2, "assignments": [ {"character_id": c, "outlook_id": o} for o, c in parent_of.items() if o and c]}]} return gl.merge(base, facet, ob.bind(by_owner["outlook"], p3)) class TestTheRealChainCarriesEveryLane: def test_the_screen_rows_come_from_the_real_producer(self, chain): """★★손으로 만든 값이 아니다 — emitter 가 낸 것이다.""" assert chain["proj"]["candidates"], "★emitter 가 후보를 하나도 안 냈다" owners = {str(r.get("owner_type") or "") for r in chain["screen"]["rows"]} assert owners == set(MATERIALIZABLE_OWNER_TYPES) def test_all_five_owners_reach_the_common_ledger(self, merged): assert set(gl.owners_present(merged)) == set(MATERIALIZABLE_OWNER_TYPES) def test_every_row_lands_in_exactly_one_lane(self, merged): acc = gl.accounting(merged) assert acc["rows"] == sum(acc["counts"].values()) assert acc["rows"] == len(merged["rows"]) def test_the_negative_controls_are_not_targets(self, merged): """★음성 대조(가위·창)는 조사 대상이 아니다 — 전부 사면 의미가 없다.""" buy = {r["research_subject_id"] for r in gl.acquisition_targets(merged)} assert buy and len(buy) < len(merged["rows"]), \ "★모든 줄이 조사 대상이면 판별이 아무 일도 안 한 것이다" @pytest.fixture(scope="module") def obligations(merged): """★조사 **앞**에서 서는 의무 장부.""" return ro.plan(merged) class TestObligationsStandBeforeAnyResearch: def test_each_part_gets_its_own_detail_obligation(self, obligations): by = ro.purposes_of(obligations) lp = [k for k in by if k.startswith("LP")] assert lp, "★장소 부분 의무가 없다" for k in lp: assert "detail" in by[k] def test_the_parent_owns_the_context(self, obligations): own = ro.context_owner_of(obligations) assert own, "★맥락 의무가 하나도 없다" assert set(own.values()) <= {ro.OWNER_PARENT, ro.OWNER_SELF} def test_siblings_share_one_context_obligation(self, obligations): ctx = [r for r in obligations["rows"] if r.get("purpose") == "context"] assert len(ctx) == len({r["final_id"] for r in ctx}), \ "★같은 주인에 맥락 의무가 둘이다" def test_context_and_detail_ask_different_things(self, obligations): """★★쓰임만 다시 적은 것이 아니라 **다른 질의**여야 한다.""" from app.modules.pipeline import grounding_acquisition_adapter as aa got = aa.inputs_from_ledger(obligations) q = {t["target"]["subject_id"]: t["target"]["directive_native"] for t in got["targets"]} pairs = [(k, v) for k, v in q.items() if "#" in k] assert pairs, "★의무 질의가 하나도 없다" assert len(set(v for _k, v in pairs)) == len(pairs), \ f"★두 의무가 **같은 것을 묻는다**: {pairs}" def test_nothing_disappears_from_the_ledger(self, merged, obligations): """★조사 대상이 아닌 줄도 그대로 남는다 — 회계가 온전해야 한다.""" kept = {str(r.get("research_subject_id") or "") for r in obligations["rows"]} for r in merged["rows"]: rsid = str(r.get("research_subject_id") or "") fid = str(r.get("final_id") or "") assert rsid in kept or any( k.startswith(f"{fid}#") for k in kept), f"★{rsid} 가 사라졌다" class TestTheCentralResearcherTakesThemAll: @pytest.fixture def researched(self, obligations, tmp_path): from app.modules.pipeline import grounding_chunk_journal as cj class Spy: def __init__(s): s.n = 0 def search(s, **kw): s.n += 1 return {"queries": ["q"], "images": [{"image_url": f"https://x/{i}.jpg", "thumbnail_url": "", "source_website_url": "", "caption": ""} for i in range(3)]} def download(s, url, dest, fallback_url=""): from pathlib import Path Path(dest).write_bytes(b"x") return True def judge(s, got): return {"j1": {"verdicts": [ {"index": i, "object_type_match": ctp.TYPE_MATCH[0], "visible": True} for i in range(1, len(got) + 1)]}} spy = Spy() out = ca.run(obligations, journal=cj.ChunkJournal(tmp_path / "j.json", contract={"v": 1}), cap=99, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) return {"out": out, "spy": spy} def test_no_obligation_disappears(self, obligations, researched): cov = ca.ledger_coverage(obligations, researched["out"]) assert cov["ok"], cov def test_nobody_waits_for_a_person(self, researched): assert ca.unfinished_rows(researched["out"]) == [] def test_each_lane_is_projected_to_a_reference_place(self, researched): seen = {} for r in researched["out"]["rows"]: owner = str((r["ledger_row"] or {}).get("owner_type") or "") seen.setdefault(owner, REFERENCE_KIND_BY_OWNER[owner]) assert set(seen) == set(MATERIALIZABLE_OWNER_TYPES) assert seen["location"] == seen["location_part"] == rb.BUNDLE_KIND class TestAPartWithNoPlaceRowStillGetsBoth: """★★Codex BLOCK (09-01) — 부모 행 없는 LP 하나가 **context/detail 두 target** 을 실제로 내는 양성 대조. 실제 producer 사슬에서 잰다. """ @pytest.fixture(scope="class") def hostless(self): """부모 장소가 **행으로 안 잡힌** LP 하나. 모델이 맥락을 낸다.""" from app.modules.pipeline import grounding_carry as gc word, place = "이발소 회전 간판", "이발소" quote = ep.segment_texts()["scene-1"] assert place in quote, "★fixture 원문에 그 장소 말이 없다" rows = [{ "local_id": "c0#0", "owner_type": "location_part", "surface_form": word, "occurrences": [{"source_span": ep.span_of(1, word), "source_quote": word}], "shot_binding_status": "bound_complete", "shot_appearance_ids": ["s1#1", "s1#2"], "hard_to_generate": True, "viewers_would_notice": True, "coarse_type_label": word, "visual_brief": f"{ep.ERA} {ep.REGION} {word}", "search_terms_native": [word], "language_lock_native": "ko", # ★모델이 낸 맥락 — 그 장소는 **행으로 안 잡혔다** # ★★`search_subject` 는 **근거 구절 안에 있는 말**이어야 한다. # 앞 판은 시대·지역을 앞에 붙였는데, 그것은 원문에 없는 말이라 # 검증이 잡았다 — 맞는 동작이다. 시대 좌표는 질의를 저작하는 # 자리가 붙인다(`test_era_coordinate_in_queries` 가 그것을 잰다). "host_context": { "state": "explicit_context_only", "parent_local_id": None, "search_subject": place, "evidence": [quote[quote.index(place): quote.index(place) + len(place)]], }, }] red = cm.reduce_episode(rows, [], segments=ep.segment_texts()) proj = ad.project_rows_and_candidates(red, project_id="p", episode_id="e") built = gc.build_subjects({k: list(v) for k, v in proj["rows"].items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=proj["candidates"]) led = built.get("candidate_ledger") or {} disp = dict(built.get("subject_dispositions") or {}) disp.update({str(r.get("research_subject_id") or ""): str(r.get("disposition") or "") for r in (led.get("rows") or [])}) pop = gs.build_population(built["subjects"], proj["candidates"], disp) surf = {str(x.get("research_subject_id") or ""): str(x.get("surface_form") or "") for x in pop} screened = gs.screen_subjects( pop, dispositions=disp, world_facts_block=ep.WORLD_FACTS, cache_get=lambda k: None, cache_put=lambda k, v: v, assess_fn=_assess_like_the_fixture(surf)) by_owner = gl.split_by_owner(screened["rows"]) final_by = {str(r.get("research_subject_id") or ""): str(((r.get("producer_link") or {}).get("final_id")) or "") for r in (led.get("rows") or []) if (r.get("producer_link") or {}).get("final_id")} return ro.plan(gl.merge(gl.resolve_facet_rows( by_owner["location_part"], final_id_by_subject=final_by))) def test_the_model_context_survives_the_whole_chain(self, hostless): """★중간에 한 곳만 안 옮겨도 여기서 잡힌다.""" ctx = [r for r in hostless["rows"] if r.get("purpose") == "context"] assert len(ctx) == 1 assert ctx[0].get("why") != ro.WHY_NO_CONTEXT_SOURCE, \ "★모델이 낸 맥락이 사슬 어딘가에서 사라졌다" def test_no_place_id_was_invented(self, hostless): assert ro.context_owner_of(hostless) == {"LP01": ro.OWNER_SELF} ids = {r["final_id"] for r in hostless["rows"]} assert ids == {"LP01"} def test_both_targets_really_go_out(self, hostless): from app.modules.pipeline import grounding_acquisition_adapter as aa got = aa.inputs_from_ledger(hostless) subs = sorted(t["target"]["subject_id"] for t in got["targets"]) assert subs == ["LP01#context", "LP01#detail"] assert got["skipped"] == [] def test_they_ask_for_different_things(self, hostless): from app.modules.pipeline import grounding_acquisition_adapter as aa from app.modules.pipeline import grounding_central_acquisition as ca2 got = aa.inputs_from_ledger(hostless) q = {t["target"]["subject_id"]: t["target"]["directive_native"] for t in got["targets"]} assert q["LP01#context"] != q["LP01#detail"] ids = {ca2.identity_of(t, contract_sha="s") for t in got["targets"]} assert len(ids) == 2, "★두 의무가 같은 신원이다" class TestAFabricatedContextNeverGoesOut: """★★Codex BLOCK (09-01) — 「원문 그대로」라고 **말하는 것**은 계약이 아니다. 실제 재현이 통과했다 — search_subject 「원문에 없는 고급 호텔」 ← 원문 어디에도 없다 evidence 「그는 회전 간판 앞에 섰다」 ← 원문 **어딘가**엔 있다 → 통과 그러면 반복 장소가 있는 원고에서 **다른 장소의 인용**이나 **지어낸 장소 이름**으로 참조 조사가 나간다. 여기서 그 셋을 다 막는지 잰다. """ WORD = "이발소 회전 간판" def _plan(self, host): from app.modules.pipeline import grounding_carry as gc segs = ep.segment_texts() rows = [{ "local_id": "c0#0", "owner_type": "location_part", "surface_form": self.WORD, "occurrences": [{"source_span": ep.span_of(1, self.WORD), "source_quote": self.WORD}], "shot_binding_status": "bound_complete", "shot_appearance_ids": ["s1#1", "s1#2"], "hard_to_generate": True, "viewers_would_notice": True, "coarse_type_label": self.WORD, "visual_brief": f"{ep.ERA} {ep.REGION} {self.WORD}", "search_terms_native": [self.WORD], "language_lock_native": "ko", "host_context": host, }] red = cm.reduce_episode(rows, [], segments=segs) proj = ad.project_rows_and_candidates(red, project_id="p", episode_id="e") built = gc.build_subjects({k: list(v) for k, v in proj["rows"].items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=proj["candidates"]) led = built.get("candidate_ledger") or {} disp = dict(built.get("subject_dispositions") or {}) disp.update({str(r.get("research_subject_id") or ""): str(r.get("disposition") or "") for r in (led.get("rows") or [])}) pop = gs.build_population(built["subjects"], proj["candidates"], disp) surf = {str(x.get("research_subject_id") or ""): str(x.get("surface_form") or "") for x in pop} screened = gs.screen_subjects( pop, dispositions=disp, world_facts_block=ep.WORLD_FACTS, cache_get=lambda k: None, cache_put=lambda k, v: v, assess_fn=_assess_like_the_fixture({**surf, **{k: self.WORD for k in surf}})) by_owner = gl.split_by_owner(screened["rows"]) final_by = {str(r.get("research_subject_id") or ""): str(((r.get("producer_link") or {}).get("final_id")) or "") for r in (led.get("rows") or []) if (r.get("producer_link") or {}).get("final_id")} return ro.plan(gl.merge(gl.resolve_facet_rows( by_owner["location_part"], final_id_by_subject=final_by))) def _targets(self, host): from app.modules.pipeline import grounding_acquisition_adapter as aa return aa.inputs_from_ledger(self._plan(host)) def _quote_from(self, scene, word): t = ep.segment_texts()[f"scene-{scene}"] i = t.index(word) return t[i:i + len(word)] def test_a_real_quote_with_an_invented_subject_is_refused(self): """★진짜 근거 + **지어낸 이름** — 그 이름으로 조사하러 가면 안 된다.""" host = {"state": "explicit_context_only", "parent_local_id": None, "search_subject": "원문에 없는 고급 호텔", "evidence": [self._quote_from(1, "이발소")]} got = self._targets(host) subs = [t["target"]["subject_id"] for t in got["targets"]] assert "LP01#context" not in subs assert [s["research_subject_id"] for s in got["skipped"]] == \ ["LP01#context"] def test_a_quote_from_another_scene_is_refused(self): """★진짜 인용이지만 **이 대상이 나온 씬이 아니다**.""" segs = ep.segment_texts() here = segs["scene-1"] other = next((t for sid, t in segs.items() if sid != "scene-1" and len(t) > 12), "") assert other, "★다른 씬이 없다 — 시험이 죽었다" picked = next((other[i:i + 6] for i in range(len(other) - 6) if other[i:i + 6] not in here), "") assert picked, "★다른 씬에만 있는 구절을 못 찾았다" host = {"state": "explicit_context_only", "parent_local_id": None, "search_subject": picked, "evidence": [picked]} got = self._targets(host) assert "LP01#context" not in [t["target"]["subject_id"] for t in got["targets"]] def test_a_quote_that_is_not_in_the_text_at_all_is_refused(self): host = {"state": "explicit_context_only", "parent_local_id": None, "search_subject": "어떤 장소", "evidence": ["원고 어디에도 없는 구절입니다"]} got = self._targets(host) assert "LP01#context" not in [t["target"]["subject_id"] for t in got["targets"]] def test_the_verified_one_still_goes_out(self): """★★양성 대조 — 막기만 하면 기능이 죽은 것이다.""" place = self._quote_from(1, "이발소") host = {"state": "explicit_context_only", "parent_local_id": None, "search_subject": place, "evidence": [place]} got = self._targets(host) subs = sorted(t["target"]["subject_id"] for t in got["targets"]) assert subs == ["LP01#context", "LP01#detail"] ctx = next(t for t in got["targets"] if t["target"]["subject_id"] == "LP01#context") occ = ctx["source_evidence"]["occurrences"] assert occ[0]["source_span"]["segment_id"] == "scene-1" assert occ[0]["source_quote"] == place def _verified(rows): """★★고증을 **확인한 판**으로 바꿔 본다 (양성 대조). 2026-09-02 부터 coarse `selected` 만으로는 안 붙는다 — 소비자 문이 두 축을 다 본다. 이 helper 는 「사람이 봤다」를 흉내 내어, 확인된 뒤에는 **옛 배치가 한 글자도 안 바뀌고 그대로 붙는지**를 잠근다. """ import copy from app.modules.pipeline import reference_acquisition as ra out = copy.deepcopy(list(rows)) for r in out: r["grounding_fidelity"] = {"state": ra.FIDELITY_VERIFIED} return out class TestTheRecordReachesTheDurableSidecar: """★★Codex BLOCK (09-01) — 손으로 만든 행이 아니라 **중앙 조사기 실제 결과**로 `write_sidecar → members_from_rpc` 를 걸어야 한다.""" @pytest.fixture def sidecar(self, obligations, tmp_path): """★맥락은 못 구하고 상세는 구한 판 — 실제 조사기를 태운다.""" from app.modules.pipeline import grounding_chunk_journal as cj # ★★어느 질의가 **맥락 의무**의 것인지 의무 장부에서 얻는다 — # 글자로 가르지 않는다 (그러면 문안을 고칠 때마다 어긋난다). from app.modules.pipeline import grounding_acquisition_adapter as aa plan_in = aa.inputs_from_ledger(obligations) # ★검색어는 **라운드가 바뀌어도 그대로**다 — 지시문은 좁힘 때 바뀐다 ctx_q = {tuple(t["target"]["terms_native"]) for t in plan_in["targets"] if (t["ledger_row"] or {}).get("purpose") == "context"} assert ctx_q, "★맥락 의무가 없다 — 시험이 죽었다" def _judge_only_details(got): return {"j1": {"verdicts": [ {"index": i, "object_type_match": ctp.TYPE_MATCH[0], "visible": True} for i in range(1, len(got) + 1)]}} def _search(**kw): # ★★맥락 의무는 **못 찾은 판**으로 만든다 — 그래야 # `reference_unavailable` 이 실제로 하나 생기고, 「둘 다 남는다」가 # 빈손으로 지나가지 않는다. if tuple(kw.get("terms_native") or ()) in ctx_q: return {"queries": ["q"], "images": []} return {"queries": ["q"], "images": [{"image_url": "https://x/0.jpg", "thumbnail_url": "", "source_website_url": "", "caption": ""}]} def _dl(url, dest, fallback_url=""): from pathlib import Path as _P _P(dest).write_bytes(b"x") return True rows = [r for r in obligations["rows"] if r.get("purpose")] assert rows, "★배경 갈래 의무가 없다 — 시험이 죽었다" out = ca.run(obligations, journal=cj.ChunkJournal(tmp_path / "sc.json", contract={"v": 1}), cap=99, workdir=tmp_path, rel_root=tmp_path, search=_search, download=_dl, judge=_judge_only_details) def _sha_of(r): return f"sha_{(r['ledger_row'] or {}).get('research_subject_id')}" def _coord_of(r): return {"source": rb.SOURCE_FILE, "path": (f"refs/" f"{(r['ledger_row'] or {}).get('final_id')}.jpg")} members = bp.members_from_rows(out["rows"], content_sha_of=_sha_of, coordinate_of=_coord_of) rpc: dict = {} required = rb.write_sidecar(rpc, members) return {"out": out, "rows": out["rows"], "members": members, "rpc": rpc, "required": required, "sha_of": _sha_of, "coord_of": _coord_of} def test_every_background_obligation_is_written(self, obligations, sidecar): want = {str(r["research_subject_id"]) for r in obligations["rows"] if r.get("purpose")} got = {str(m["obligation_subject_id"]) for m in sidecar["rpc"][rb.RPC_MEMBERS_KEY]} assert got == want, "★의무가 sidecar 에서 사라졌다" def test_reading_it_back_keeps_both_outcomes(self, sidecar): """★★**둘 다 실제로 있어야** 이 시험이 뜻이 있다 — 빈손이면 어떤 축이든 지나간다.""" back = rb.members_from_rpc(sidecar["rpc"]) assert back and len(back) == len(sidecar["members"]) kinds = {str(m.get("outcome") or "") for m in back} assert kinds == {rb.OUTCOME_SELECTED, rb.OUTCOME_UNAVAILABLE}, \ f"★한쪽만 나왔다: {kinds}" def test_the_missing_one_is_the_context(self, sidecar): miss = bp.unavailable_members(rb.members_from_rpc(sidecar["rpc"])) assert [m["purpose"] for m in miss] == ["context"] assert miss[0]["member_identity"] is None assert miss[0]["content_sha256"] is None assert miss[0]["covers"], "★덮던 의무 목록이 사라졌다" def test_a_coarse_selection_alone_becomes_required_nothing(self, sidecar): """★★★거친 종류로 골랐다고 **붙지 않는다** (Codex BLOCK 09-02). 그 사진이 그 시대·그 지역 것인지는 아무도 안 봤다. """ found = [m for m in sidecar["members"] if m["outcome"] == rb.OUTCOME_SELECTED] assert found, "★고른 것이 없으면 이 시험이 아무것도 안 잠근다" # ★HITL 0 (2026-09-03): 거친 종류로 고른 한 장이 **그대로 required** 다 assert sidecar["required"], "★자동 선택이 required 에 안 실렸다 (HITL 0)" for m in found: assert m["member_identity"] and m["content_sha256"] assert "not_attached" not in m def test_a_verified_selection_attaches_exactly_as_before(self, sidecar): """★★★양성 대조 — 확인된 뒤에는 **옛 배치 그대로** 붙는다.""" members = bp.members_from_rows( _verified(sidecar["rows"]), content_sha_of=sidecar["sha_of"], coordinate_of=sidecar["coord_of"]) found = [m for m in members if m["outcome"] == rb.OUTCOME_SELECTED] _plan, required = rb.plan_bundle(members) assert found and len(required) == len(found) for m in found: assert m["member_identity"] and m["content_sha256"] def test_the_sidecar_is_json(self, sidecar): import json assert json.loads(json.dumps(sidecar["rpc"], ensure_ascii=False)) def test_nobody_waits_for_a_person(self, sidecar): assert ca.unfinished_rows(sidecar["out"]) == [] class TestTheParentPhotoLandsOnce: """★★조사 단계와 최종 이미지 입력 단계 **모두에서 한 번**.""" def test_the_parent_context_is_researched_once(self, obligations): rows = gl.acquisition_targets(obligations) locs = [r for r in rows if r["owner_type"] == "location"] assert len(locs) == len({r["final_id"] for r in locs}), \ "★같은 장소를 두 번 조사한다" def test_siblings_share_one_context_member(self, obligations, tmp_path): from app.modules.pipeline import grounding_chunk_journal as cj def _search(**kw): return {"queries": ["q"], "images": [{"image_url": "https://x/0.jpg", "thumbnail_url": "", "source_website_url": "", "caption": ""}]} def _dl(url, dest, fallback_url=""): from pathlib import Path Path(dest).write_bytes(b"x") return True out = ca.run(obligations, journal=cj.ChunkJournal(tmp_path / "j2.json", contract={"v": 1}), cap=99, workdir=tmp_path, rel_root=tmp_path, search=_search, download=_dl, judge=lambda g: {"j1": {"verdicts": [ {"index": i, "object_type_match": ctp.TYPE_MATCH[0], "visible": True} for i in range(1, len(g) + 1)]}}) members = bp.members_from_rows( out["rows"], content_sha_of=lambda r: f"sha_{(r['ledger_row'] or {}).get('final_id')}", coordinate_of=lambda r: { "source": rb.SOURCE_FILE, "path": f"refs/{(r['ledger_row'] or {}).get('final_id')}.jpg"}) # ★HITL 0: 확인 전에도 자동 선택은 붙는다 assert rb.plan_bundle(members)[1] != [], "★자동 선택이 안 붙었다 (HITL 0)" members = bp.members_from_rows( _verified(out["rows"]), content_sha_of=lambda r: f"sha_{(r['ledger_row'] or {}).get('final_id')}", coordinate_of=lambda r: { "source": rb.SOURCE_FILE, "path": f"refs/{(r['ledger_row'] or {}).get('final_id')}.jpg"}) ctx = [m for m in members if m["purpose"] == "context"] assert len(ctx) == len({m["subject_final_id"] for m in ctx}), \ "★맥락 멤버가 주인마다 하나가 아니다" plan, _req = rb.plan_bundle(members) # ★같은 부모를 둔 장소 부분들이 **한 장**을 함께 덮는다 shared = [p for p in plan if len(p["covered_final_ids"]) > 1] assert shared, "★형제가 부모 사진을 함께 안 쓴다" for p in shared: assert "context" in p["purposes"] assert p["role"] == rb.ROLE_BY_PURPOSES[("context",)]