"""GROUNDING-V2 §2-4b — **조사가 필요하면 참조를 만든다**. ★전에는 이 판단이 `visible_shot_count` 와 variant pole 로만 정해졌다 — 조사 결과가 **아예 안 들어갔다**. 그러면 한 컷에만 나오지만 시대 고증이 걸린 대상이 `text_only` 로 지나간다. """ import json import pytest from app.core.episode_reference_policy import compute_episode_reference_policy def _run(**over): base = dict(visible_shot_count={"C01": 1, "P01": 1}, entity_types={"C01": "character", "P01": "prop"}, variant_pole_short_ids=set()) return compute_episode_reference_policy(**{**base, **over}) class TestResearchForcesAReference: def test_a_researched_character_is_forced(self): got = _run(research_required_short_ids={"C01"}) assert got["policy"]["C01"]["mode"] == "reference_required" assert "조사 필요" in got["policy"]["C01"]["reason"] # ★원래 판단도 남긴다 — 왜 올라갔는지 못 찾으면 못 고친다 assert "원래 판단" in got["policy"]["C01"]["reason"] def test_a_researched_prop_is_not_provisional_any_more(self): """★후보가 아니라 **확정**이다 — 「나중에 정한다」로 두면 새어 나간다.""" got = _run(research_required_short_ids={"P01"}) assert got["policy"]["P01"]["mode"] == "reference_required" assert got["policy"]["P01"]["provisional"] is False def test_it_only_raises_never_lowers(self): """★이미 올라간 것을 표식이 없다고 **안 내린다** — 보호가 풀린다.""" got = _run(visible_shot_count={"C01": 5}, research_required_short_ids=set()) assert got["policy"]["C01"]["mode"] == "reference_required" def test_an_already_required_subject_keeps_its_reason(self): """★올라가 있던 것의 사유를 덮어쓰지 않는다.""" a = _run(visible_shot_count={"C01": 5}) b = _run(visible_shot_count={"C01": 5}, research_required_short_ids={"C01"}) assert a["policy"]["C01"] == b["policy"]["C01"] def test_nothing_changes_when_the_set_is_empty(self): """★★legacy 비회귀 — **바이트 단위로 같아야** 한다.""" before = compute_episode_reference_policy( visible_shot_count={"C01": 1, "C02": 3, "P01": 2, "P02": 1}, entity_types={"C01": "character", "C02": "character", "P01": "prop", "P02": "prop"}, variant_pole_short_ids={"C02"}) after = compute_episode_reference_policy( visible_shot_count={"C01": 1, "C02": 3, "P01": 2, "P02": 1}, entity_types={"C01": "character", "C02": "character", "P01": "prop", "P02": "prop"}, variant_pole_short_ids={"C02"}, research_required_short_ids=set()) assert json.dumps(before, sort_keys=True) == \ json.dumps(after, sort_keys=True) @pytest.mark.parametrize("marked", [None, set(), frozenset()]) def test_the_default_is_no_forcing(self, marked): got = (_run() if marked is None else _run(research_required_short_ids=marked)) assert got["policy"]["C01"]["mode"] == "text_only" class TestTheEndToEndGate: """★★★Codex 가 지정한 끝점 세 판. ``` completed · delta=yes → reference_required 로 **상승** completed · delta=no → 상승 없음 unresolved / retryable → 정책 스텝이 **막힌다**(→ scene_detail 도 못 감) ``` ★raw plan 의 `route="research"` 는 「살 대상」일 뿐 최종 강제값이 아니다. """ def _step(self, revs, *, override=False): from types import SimpleNamespace from unittest.mock import patch from app.core.steps.episode_reference_policy_step import ( EpisodeReferencePolicyStep) step = EpisodeReferencePolicyStep.__new__(EpisodeReferencePolicyStep) step.project_id = "p1" step.episode_id = "e1" step.project_config = ({"allow_missing_grounding": True} if override else {}) class _Q: def __init__(self, rows): self._rows = rows def filter(self, *a): return self def all(self): return self._rows step.db = SimpleNamespace(query=lambda model: _Q(revs)) # ★정책은 **이번 판이 기대하는 신원**만 본다 — 옛 revision 이 섞이면 # 입력이 바뀐 결과로 참조를 정하게 된다. cps = { "grounding_plan": {"data": {"decided": [ {"research_subject_id": "rs_a", "_short_id": "C01", "route": "research"}]}}, "grounding_research": {"data": { "expected_input_hashes": {r.research_subject_id: r.input_hash for r in revs}, "no_source": [], "unbuilt": []}}, } step._load_prev_checkpoint = lambda name: cps.get(name) return step def _rev(self, *, status, delta, sid="rs_a", created="2026-08-30T00:00:00Z"): from types import SimpleNamespace return SimpleNamespace(research_subject_id=sid, status=status, delta=delta, created_at=created, research_input_hash="ih_" + sid, input_hash="ih_" + sid) def test_completed_yes_forces_a_reference(self): step = self._step([self._rev(status="completed", delta="yes")]) forced, blocked = step._research_forced_short_ids() assert forced == {"C01"} and blocked == set() got = _run(research_required_short_ids=forced) assert got["policy"]["C01"]["mode"] == "reference_required" def test_completed_no_does_not_force(self): """★출처가 「차이 없다」고 말한 것까지 올리면 참조를 헛되이 굽는다.""" step = self._step([self._rev(status="completed", delta="no")]) forced, blocked = step._research_forced_short_ids() assert forced == set() and blocked == set() @pytest.mark.parametrize("status", ["unresolved_terminal", "retryable"]) def test_an_unfinished_subject_blocks_the_policy(self, status): """★빈 집합으로 바꾸면 「조사할 게 없었다」가 되어 그냥 지나간다.""" step = self._step([self._rev(status=status, delta="unresolved")]) _forced, blocked = step._research_forced_short_ids() assert blocked == {"rs_a"} def test_the_override_passes_but_only_completed_yes_is_used(self): """★알고도 진행할 때도 **완료된 yes 만** 쓴다.""" step = self._step([self._rev(status="retryable", delta="unresolved"), self._rev(status="completed", delta="yes", sid="rs_b")], override=True) forced, blocked = step._research_forced_short_ids() assert blocked == set() # ★rs_b 는 short_id 가 없어 강제 목록에 못 든다 — 그건 정상이다 assert "C01" not in forced def test_the_latest_attempt_wins(self): """★재개하면 새 행이 붙는다 — 첫 행만 보면 옛 실패에 영원히 묶인다.""" step = self._step([ self._rev(status="retryable", delta="unresolved", created="2026-08-30T00:00:00Z"), self._rev(status="completed", delta="yes", created="2026-08-30T01:00:00Z"), ]) forced, blocked = step._research_forced_short_ids() assert forced == {"C01"} and blocked == set() class TestOldRevisionsDoNotLeakIn: """★★★에피소드의 **모든 과거 revision** 중 최신을 쓰면, 입력이 바뀌었거나 이번 판이 `no_source` 인데도 **옛 결과가 섞인다** (Codex). 정책은 **이번 판이 기대하는 신원**과 맞는 행만 본다. """ def _step(self, revs, *, expected, no_source=(), unbuilt=()): from types import SimpleNamespace from app.core.steps.episode_reference_policy_step import ( EpisodeReferencePolicyStep) step = EpisodeReferencePolicyStep.__new__(EpisodeReferencePolicyStep) step.project_id, step.episode_id = "p1", "e1" step.project_config = {} class _Q: def filter(self, *a): return self def all(self): return revs step.db = SimpleNamespace(query=lambda m: _Q()) cps = { "grounding_plan": {"data": {"decided": [ {"research_subject_id": "rs_a", "_short_id": "C01", "route": "research"}]}}, "grounding_research": {"data": { "expected_input_hashes": dict(expected), "no_source": list(no_source), "unbuilt": list(unbuilt)}}, } step._load_prev_checkpoint = lambda name: cps.get(name) return step def _rev(self, *, ih, status="completed", delta="yes", sid="rs_a"): from types import SimpleNamespace return SimpleNamespace(research_subject_id=sid, status=status, delta=delta, created_at="2026-08-30T00:00:00Z", research_input_hash=ih) def test_a_stale_revision_is_ignored(self): """★입력이 바뀌면 옛 행은 **이번 판의 것이 아니다**.""" step = self._step([self._rev(ih="옛것")], expected={"rs_a": "이번것"}) forced, blocked = step._research_forced_short_ids() assert forced == set(), "옛 결과로 참조를 정했다" # ★기대한 대상의 결과가 없으니 **막는다** assert blocked == {"rs_a"} def test_a_matching_revision_is_used(self): """★positive control — 다 무시하면 조사가 아무 데도 안 닿는다.""" step = self._step([self._rev(ih="이번것")], expected={"rs_a": "이번것"}) forced, blocked = step._research_forced_short_ids() assert forced == {"C01"} and blocked == set() def test_a_no_source_subject_blocks(self): """★인용이 없어 못 산 것도 **막는다** — 유료 호출 0회여도 그렇다.""" step = self._step([], expected={}, no_source=["rs_a"]) _f, blocked = step._research_forced_short_ids() assert blocked == {"rs_a"} def test_an_unbuilt_subject_blocks(self): step = self._step([], expected={}, unbuilt=["rs_gone"]) _f, blocked = step._research_forced_short_ids() assert blocked == {"rs_gone"} class TestEveryOwnerNowGetsAReference: """★★★**뒤집은 시험** (2026-09-01 D 활성화). 앞에는 「`L##`·`LP##`·`O##` 는 만들 수 없으니 **막는다**」였다 — 그때는 §2-6.5 producer 가 없어서 참조 없이 통과시키면 「모른다를 아니다로 닫는」 자리였기 때문이다(Codex). 이제 그 producer 가 서고 중앙 조사기가 다섯 갈래를 다 내므로, **막는 것이 아니라 만든다**. ★지우지 않고 반대로 잠근다 — 되돌아가면 여기서 잡힌다. """ def _step(self, short_id): from types import SimpleNamespace from app.core.steps.episode_reference_policy_step import ( EpisodeReferencePolicyStep) step = EpisodeReferencePolicyStep.__new__(EpisodeReferencePolicyStep) step.project_id, step.episode_id = "p1", "e1" step.project_config = {} rev = SimpleNamespace(research_subject_id="rs_a", status="completed", delta="yes", created_at="2026-08-30T00:00:00Z", research_input_hash="ih") class _Q: def filter(self, *a): return self def all(self): return [rev] step.db = SimpleNamespace(query=lambda m: _Q()) cps = { "grounding_plan": {"data": {"decided": [ {"research_subject_id": "rs_a", "_short_id": short_id, "route": "research"}]}}, "grounding_research": {"data": { "expected_input_hashes": {"rs_a": "ih"}, "no_source": [], "unbuilt": []}}, } step._load_prev_checkpoint = lambda name: cps.get(name) return step def test_every_lane_is_forced_not_blocked(self): for short in ("C01", "P01", "L01", "LP01", "O07"): forced, blocked = self._step(short)._research_forced_short_ids() assert forced == {short}, f"★{short} 가 참조를 못 받는다" assert blocked == set(), f"★{short} 가 막혔다" def test_a_string_that_only_looks_like_an_id_still_blocks(self): """★★음성 대조 — 열었다고 **아무 글자나** 지나가면 안 된다. 접두표가 prefix-free 가 아니다 — `Pfoo` 도 `"P"` 로 시작한다. """ for junk in ("Pfoo", "Character01", "LP"): forced, blocked = self._step(junk)._research_forced_short_ids() assert forced == set(), f"★{junk} 로 참조를 만든다" assert blocked == {"rs_a"}, f"★{junk} 를 조용히 통과시켰다" class TestTheSupportedOwnersAreOneContract: """★★같은 규칙이 **두 곳**에 `startswith("C") or startswith("P")` 로 적혀 있었다 — 한쪽만 고쳐질 자리다 (2026-09-01). 계약 한 벌로 모은다. ★그리고 접두를 글자로 견주면 안 된다 — 접두표가 prefix-free 가 아니다: `LP01` 이 `"L"` 로도 시작하고, `Pfoo` 도 `"P"` 로 시작한다. """ def test_the_contract_names_who_can_get_a_reference(self): from app.modules.pipeline.grounding_entity_contract import ( MATERIALIZABLE_OWNER_TYPES, REFERENCE_SUPPORTED_OWNERS, reference_owner_of) # ★뒤집었다 — 이제 **다섯 갈래 전부**다(표에서 파생한다) assert set(REFERENCE_SUPPORTED_OWNERS) == set( MATERIALIZABLE_OWNER_TYPES) assert reference_owner_of("C01") == "character" assert reference_owner_of("P07") == "prop" def test_a_facet_owner_now_gets_one(self): """★producer 가 섰으므로 그 갈래도 참조를 만든다.""" from app.modules.pipeline.grounding_entity_contract import ( reference_owner_of) assert reference_owner_of("LP01") == "location_part" assert reference_owner_of("L01") == "location" assert reference_owner_of("O02") == "outlook" def test_a_string_that_only_looks_like_an_id_is_refused(self): """★양성 대조 — 접두만 보면 지나가던 것들.""" from app.modules.pipeline.grounding_entity_contract import ( reference_owner_of) for sid in ("Pfoo", "C", "", "P-1", "P١٢", "Character01"): assert reference_owner_of(sid) is None, sid def test_neither_consumer_compares_prefixes_on_its_own(self): """★★글자가 아니라 **AST 로** 본다 — 주석의 설명이 걸리면 안 된다.""" import ast import inspect from app.core.steps import episode_reference_policy_step as ep from app.core.steps import render_prompt_card as rp for mod in (ep, rp): tree = ast.parse(inspect.getsource(mod)) for n in ast.walk(tree): if not (isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "startswith"): continue for a in n.args: if isinstance(a, ast.Constant) and a.value in ( "C", "P", "L", "LP", "O"): raise AssertionError( f"{mod.__name__} 이 접두를 제 손으로 비교한다 " f"— 계약(`reference_owner_of`)을 쓴다")