"""GROUNDING-V2 §2-3 — mode enum · shadow_plan 재생기 · 세계 맥락 전달. 통과 조건(계획 §2 표 3단계): 저장된 에피소드에서 **검색 호출 0** · **production 지문·산출 불변**(shadow CP 에만 기록). """ import json from pathlib import Path from unittest.mock import patch import pytest from app.core.errors import AppError from app.core.grounding_mode import ( DEFAULT_GROUNDING_MODE, GROUNDING_MODES, GROUNDING_MODE_LEGACY, GROUNDING_MODE_SHADOW_PLAN, GROUNDING_MODE_V2, buys_v2_research, resolve_grounding_mode, touches_production_fingerprint) from app.modules.pipeline.grounding_shadow import ( SHADOW_STEP_ID, ShadowSourceError, build_subjects_from_saved_episode, production_checkpoint_digest, replay_shadow_plan, write_shadow_checkpoint) def _episode(tmp_path: Path, *, entities=None, era="1983년", region="대한민국"): """저장된 에피소드 한 개를 흉내낸다 — 실제 CP 모양 그대로.""" ent = entities if entities is not None else { "characters": [{"name": "정임", "short_id": "C01", "description": "20대", "shot_count": 7}], "locations": [{"name": "버스 안", "short_id": "L01", "description": "낡은 좌석", "shot_count": 5}], "props": [{"name": "종이 승차권 뭉치", "short_id": "P03", "description": "고무줄로 묶인", "shot_count": 3}], } # ★`entity_merge` 다 — production 의 `grounding_plan` 이 읽는 그 자리. # `entity_filter` 로 읽으면 거기서 걸러진 것을 못 보는 모집단이 된다. d = tmp_path / "entity_merge" d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text(json.dumps( {"status": "completed", "data": ent}, ensure_ascii=False), encoding="utf-8") r = tmp_path / "visual_world_rules" r.mkdir(parents=True, exist_ok=True) (r / "manifest.json").write_text(json.dumps( {"status": "completed", "data": {"era": era, "region": region}}, ensure_ascii=False), encoding="utf-8") return tmp_path # ── mode enum (계약 §12) ────────────────────────────────────────────────── class TestModeEnum: def test_the_modes_are_exactly_these(self): """★**받는 값**은 넷이다 (2026-09-01 D 활성화). `v2_chunk` 를 여기 미리 넣었다가 되돌린 적이 있다 (Codex 2026-08-31) — 그때는 새 producer 가 manifest 에 없어서 「새 것은 안 돌고 옛 유료만 켜진 판」이 됐다. 이번에는 producer·판별·중앙 조사기·참조 갈래를 **같은 커밋에** 넣었고, 하나라도 빠지면 `grounding_activation_contract` 가 **조립 자리에서** 세운다. ★그리고 `buys_v2_research` 는 `mode == v2` 정확 비교라 이 판에서 옛 유료 셋이 자동으로 꺼진다. """ assert GROUNDING_MODES == {"legacy", "shadow_plan", "v2", "v2_chunk"} def test_the_value_moved_from_planned_to_accepted(self): """★★**뒤집은 시험** (2026-09-01 D 활성화). 앞에는 「계획 집합에만 있다」를 잠갔는데 그것이 이 커밋이 바꾸는 것이다. 지우지 않고 반대로 잠근다 — 두 집합의 **겹침 금지**는 그대로다. """ from app.core.grounding_mode import (GROUNDING_MODE_V2_CHUNK, KNOWN_MODES, PLANNED_MODES) assert GROUNDING_MODE_V2_CHUNK in GROUNDING_MODES assert PLANNED_MODES == frozenset() assert not (PLANNED_MODES & GROUNDING_MODES), \ "★계획된 값이 받는 값에 들어갔다 — 반만 켠 판이 생긴다" assert KNOWN_MODES == GROUNDING_MODES | PLANNED_MODES def test_the_catalog_only_names_it_inside_the_activation_gate(self): """★★예외를 뒀으니 **그 자리에서만** 쓰는지 잠근다. `step_catalog` 가 그 값을 쓰는 곳은 안전문 호출 하나여야 한다 — 다른 데서 쓰면 그것이 곧 **켜는 것**이다. """ import ast import inspect from app.core import step_catalog as sc tree = ast.parse(inspect.getsource(sc)) gate = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and ast.unparse(n.func).endswith("assert_activation")] assert len(gate) == 1, "★안전문 호출이 하나가 아니다" # ★모든 node 에 `lineno` 가 있는 것은 아니다(`Load` 등) — 있는 것만 inside = {n.lineno for g in gate for n in ast.walk(g) if hasattr(n, "lineno")} for n in ast.walk(tree): if (isinstance(n, ast.Name) and n.id == "GROUNDING_MODE_V2_CHUNK" and n.lineno not in inside): raise AssertionError( f"★안전문 밖에서 쓴다: step_catalog.py:{n.lineno}") def test_the_new_mode_is_wired_only_where_activation_needs_it(self): """★★**뒤집은 시험** — 이제 배선됐다. 다만 **어디에** 배선됐나를 잠근다. 앞에는 「아무도 안 쓴다」였다. 활성화했으므로 이제 묻는 것은 「모드 이름을 아는 자리가 **활성화 축**뿐인가」다 — 그 밖에서 쓰면 같은 규칙이 두 곳이 되어 한쪽만 고쳐진다. """ import ast from pathlib import Path as _P root = _P(__file__).resolve().parents[2] / "app" #: ★정의한 자리와, **아직 아무도 못 닿는** producer 스텝만 예외. #: 스텝이 이 술어를 묻는 것은 맞다 — 다만 그 스텝에 닿는 길이 #: 없어야 하고, 그 조건은 `test_chunk_producer_step.py` 가 잰다. #: ★`step_catalog.py` 는 **끄는 쪽**이다 — 활성화 안전문이 「아직 안 #: 켠 판인데 새 경로가 닿나」를 묻기 위해 그 값을 참조한다. 켜는 #: 것이 아니라 **반쪽 활성화를 막는 것**이므로 여기 둔다. #: ★그 자리에서만 쓰는지는 바로 아래 시험이 잰다. #: ★활성화 축 — 정의한 자리 · producer 스텝 · 조립 안전문 · #: 술어 등록부. 이 넷 밖에서 모드 이름을 쓰면 경계가 두 벌이 된다. SKIP = {"grounding_mode.py", "grounding_chunk_step.py", "step_catalog.py", "applicability.py", "grounding_activation_contract.py", # ★중앙 조사 스텝은 **갈래를 가르는 자리**다 — 옛 갈래와 # 새 갈래가 한 스텝에 있으므로 여기서 술어를 물어야 한다. "reference_acquisition_step.py", # ★판별도 갈래를 가르는 자리다 — C(c) 판이면 새 producer 의 # 산출을 읽고, 아니면 옛 것을 읽는다. "grounding_screen_step.py", # ★엔티티 병합도 갈래를 가른다 — C(c) 판이면 새 producer 의 # **호출 0 투영**이고, 아니면 옛 추출 셋을 읽는다. "entity_steps.py", # ★관계도 갈래를 가른다 — C(c) 판이면 `part_of` 를 호출 0 으로 # 투영하고, 아니면 옛 LLM 을 부른다. "entity_relation_step.py", # ★참조 정책도 갈래를 가른다 (2026-09-01 D cutover) — C(c) # 판이면 **중앙 조사 CP** 를 읽고, 아니면 옛 # `grounding_research` CP + revision 을 읽는다. 이 갈림이 # 없으면 `v2_chunk` 에서 조사가 고른 참조가 **하나도 의무가 # 안 된다**(실측 강제 0개). "episode_reference_policy_step.py", # ★씬 상세도 갈래를 가른다 (2026-09-02 D cutover) — C(c) # 판에서만 ①고증 sidecar 를 적고 ②그 산출 계약을 재개 # 지문에 접는다. 안 가르면 옛 CP 가 **sidecar 없는 카드**를 # 그대로 재사용해 cutover 가 기존 프로젝트에 안 닿고, # CP 부재를 「대상 0」으로 삼켜 참조 없이 그림까지 간다. "detail_steps.py", # ★canonical ref 입력도 갈래를 가른다 (2026-09-02 Codex E) — C(c) # 판이면 **중앙 조사 CP 가 반드시 있어야** 하고(부재·손상이면 # provider 앞에서 선다 — public API 가 중앙 조사 전에 눌리면 # 고증 없는 참조가 생겨 resume 이 영구 재사용한다), 아니면 CP # 없음 → 옛 길. 그 술어를 여기서 묻는다. "grounding_canonical_ref_inputs.py"} hits = [] for f in root.rglob("*.py"): if f.name in SKIP: continue tree = ast.parse(f.read_text(encoding="utf-8")) for n in ast.walk(tree): if isinstance(n, ast.Constant) and n.value == "v2_chunk": hits.append(f"{f}:{n.lineno}") if (isinstance(n, ast.Name) and n.id in ("GROUNDING_MODE_V2_CHUNK", "uses_chunk_producer", "CHUNK_PRODUCER_MODES")): hits.append(f"{f}:{n.lineno} {n.id}") assert hits == [], f"★새 모드가 배선됐다: {hits[:3]}" def test_no_project_or_env_actually_selects_the_new_mode(self): """★★값이 있다고 켜진 것이 아니다 — **아무도 안 고른다**.""" import os from app.core.database import SessionLocal from app.models.project import ProjectSettings assert os.environ.get("GROUNDING_MODE") != "v2_chunk" # ★★모델·칼럼을 **조립부에서 그대로** 가져온다. 앞 판은 표 이름을 # `projects` 로 **짐작**했다가 「없는 표」로 조용히 skip 됐다 — # 빈손이 축을 지나갔다. 설정은 `ProjectSettings.llm_config_json` 이다 # (`step_execution_service._load_project_config`). # ★★시험은 **빈 시험 DB** 를 본다 — 거기서 「0개」는 아무 뜻이 없다. # 그래서 여기서는 **조회가 성립하는지**(모델·칼럼이 맞는지)만 보고, # 실제 프로젝트가 고르지 않았다는 것은 **운영 DB 에서 손으로** 본다. # 빈 결과를 「없다」로 읽지 않는다. with SessionLocal() as db: rows = db.query(ProjectSettings.project_id, ProjectSettings.llm_config_json).all() bad = [pid for pid, cfg in rows if "v2_chunk" in (cfg or "")] assert bad == [], f"★{len(bad)}개가 새 모드를 골랐다: {bad[:3]}" if not rows: pytest.skip("시험 DB 가 비었다 — 운영 DB 는 따로 본다 " "(이 축은 여기서 못 잰다)") def test_default_is_legacy(self): assert DEFAULT_GROUNDING_MODE == GROUNDING_MODE_LEGACY assert resolve_grounding_mode() == GROUNDING_MODE_LEGACY def test_which_accepted_modes_buy_v2_research(self): """★★이 술어가 True 가 되는 순간 **옛 유료 스텝 넷**이 열린다. `if_grounding_v2` 가 이것을 묻고, 그 applicability 가 걸린 것이 `grounding_a0`·`grounding_plan`·`grounding_research`· `reference_acquisition` 이다. 받는 값 중에는 `v2` 뿐이어야 한다. """ assert {m for m in GROUNDING_MODES if buys_v2_research(m)} == { GROUNDING_MODE_V2} def test_exactly_one_accepted_mode_reaches_the_chunk_producer(self): """★★뒤집었다 — 이제 **하나만** 연다. 둘이면 경계가 흐려진다.""" from app.core.grounding_mode import (GROUNDING_MODE_V2_CHUNK, uses_chunk_producer) opens = [m for m in GROUNDING_MODES if uses_chunk_producer(m)] assert opens == [GROUNDING_MODE_V2_CHUNK] def test_the_old_and_new_producers_never_run_together(self): """★★한 판에서 둘 다 켜지면 **같은 것을 두 번 산다**.""" from app.core.grounding_mode import (buys_v2_research, uses_chunk_producer) for m in GROUNDING_MODES: assert not (buys_v2_research(m) and uses_chunk_producer(m)), m def test_the_accepted_value_uses_the_chunk_producer(self): """★받는 값이 되었고 새 producer 를 연다.""" from app.core.grounding_mode import (GROUNDING_MODE_V2_CHUNK, uses_chunk_producer) assert uses_chunk_producer(GROUNDING_MODE_V2_CHUNK) is True def test_the_two_v2_modes_have_different_fingerprints(self): """★★같은 지문이면 모드를 바꿔도 **하류가 옛 산출을 재사용한다** — 켰는데 아무것도 안 바뀐다.""" from app.core.grounding_mode import (GROUNDING_MODE_V2_CHUNK, fingerprint_value) a = fingerprint_value(GROUNDING_MODE_V2) b = fingerprint_value(GROUNDING_MODE_V2_CHUNK) assert a and b and a != b, f"★지문이 같다: {a!r} vs {b!r}" assert touches_production_fingerprint(GROUNDING_MODE_V2_CHUNK) is True def test_shadow_plan_never_touches_production_fingerprint(self): """★무료 관찰을 켠 것만으로 하류가 stale 되면 안 된다.""" assert touches_production_fingerprint(GROUNDING_MODE_SHADOW_PLAN) is False assert touches_production_fingerprint(GROUNDING_MODE_V2) is True @pytest.mark.parametrize("bad", ["shadowplan", "SHADOW", "true", "", " "]) def test_typo_fails_closed_not_silently_legacy(self, bad): """★오타 하나로 v2 주행이 legacy 로 돌면 며칠 쫓게 된다.""" with pytest.raises(AppError): resolve_grounding_mode({"grounding_mode": bad}) def test_config_beats_env(self, monkeypatch): monkeypatch.setenv("GROUNDING_MODE", "v2") assert resolve_grounding_mode({"grounding_mode": "shadow_plan"}) == "shadow_plan" assert resolve_grounding_mode() == "v2" # ── shadow 재생기 ───────────────────────────────────────────────────────── class TestShadowReplay: def test_builds_subjects_from_saved_checkpoint(self, tmp_path): ep = _episode(tmp_path) subs = build_subjects_from_saved_episode(ep, project_id="p", episode_id="e")["subjects"] assert {s["owner_type"] for s in subs} == {"character", "location", "prop"} assert all(s["canon_id"] is None for s in subs) assert all(s["bind_state"] == "unbound" for s in subs) def test_no_source_checkpoint_fails_closed(self, tmp_path): """★없는 것을 빈 목록으로 삼키면 깨진 에피소드가 섞여도 전체가 통과다.""" with pytest.raises(ShadowSourceError): build_subjects_from_saved_episode(tmp_path, project_id="p", episode_id="e") def test_corrupt_checkpoint_fails_closed(self, tmp_path): ep = _episode(tmp_path) (ep / "entity_merge" / "manifest.json").write_text("{깨짐", encoding="utf-8") with pytest.raises(ShadowSourceError): build_subjects_from_saved_episode(ep, project_id="p", episode_id="e") def test_replay_makes_no_search_and_changes_nothing(self, tmp_path): ep = _episode(tmp_path) out = replay_shadow_plan(ep, project_id="p", episode_id="e") assert out["search_calls"] == 0 assert out["production_unchanged"] is True assert out["subject_count"] == 3 # ★분류기를 안 부르면 호출도 0이다. assert out["classifier_logical_calls"] == 0 def test_classifier_calls_are_counted_separately_from_search(self, tmp_path): """★「검색 0」만 보고하면 provider 비용이 0으로 읽힌다.""" ep = _episode(tmp_path) def _fake(subjects, *, era="", region=""): return {"records": [], "fingerprint": None, "search_calls": 0} out = replay_shadow_plan(ep, project_id="p", episode_id="e", classify_fn=_fake) assert out["search_calls"] == 0 assert out["classifier_logical_calls"] == 1 def test_the_count_is_named_logical_not_transmissions(self, tmp_path): """★Router 재시도·fallback 때문에 실제 전송은 더 많을 수 있다. 이 값을 「전송 수」로 읽으면 비용을 과소 보고하게 되므로 이름으로 못박는다. """ ep = _episode(tmp_path) def _fake(subjects, *, era="", region=""): return {"records": [], "fingerprint": None, "search_calls": 0} out = replay_shadow_plan(ep, project_id="p", episode_id="e", classify_fn=_fake) assert "classifier_logical_calls" in out assert "classifier_provider_calls" not in out, "전송 수로 읽히는 이름이 남아 있다" def test_records_the_model_that_actually_judged(self, tmp_path): """★설정이 아니라 응답이 말한 모델을 남긴다 (Tier 3 fallback 대비).""" ep = _episode(tmp_path) def _fake(subjects, *, era="", region=""): return {"records": [{"research_subject_id": "x", "judge_model_alias": "gpt", "judge_physical_model": "openai/gpt-5.6-sol"}], "fingerprint": None, "search_calls": 0} out = replay_shadow_plan(ep, project_id="p", episode_id="e", classify_fn=_fake) assert out["classifier_judge"]["physical_model"] == "openai/gpt-5.6-sol" @pytest.mark.parametrize("bad", [GROUNDING_MODE_V2, GROUNDING_MODE_LEGACY]) def test_replay_takes_only_shadow_plan(self, tmp_path, bad): """★legacy 도 거부한다 — 「이 기구를 안 쓴다」는 뜻이지 「legacy 로 돌린다」가 아니다.""" ep = _episode(tmp_path) with pytest.raises(ValueError): replay_shadow_plan(ep, project_id="p", episode_id="e", mode=bad) def test_digest_ignores_the_shadow_step_itself(self, tmp_path): """shadow 산출이 바뀌는 것은 정상이다 — 그것이 이 단계의 산출이다.""" ep = _episode(tmp_path) before = production_checkpoint_digest(ep) d = ep / SHADOW_STEP_ID d.mkdir(parents=True, exist_ok=True) (d / "manifest.json").write_text("{}", encoding="utf-8") assert production_checkpoint_digest(ep) == before def test_digest_catches_a_real_production_change(self, tmp_path): """positive control — 진짜로 바뀌면 잡히는가.""" ep = _episode(tmp_path) before = production_checkpoint_digest(ep) p = ep / "entity_merge" / "manifest.json" p.write_text(p.read_text(encoding="utf-8") + " ", encoding="utf-8") assert production_checkpoint_digest(ep) != before def test_digest_covers_non_manifest_files_too(self, tmp_path): """★`*/manifest.json` 만 보면 실제 CP 의 json·html·png 변경을 못 잡는다. 실제 에피소드 하나에 비-manifest 파일이 11개 있었고, 그걸 바꿔도 `production_unchanged=True` 가 나왔다 (Codex 재현). """ ep = _episode(tmp_path) asset = ep / "entity_merge" / "assets" / "big.png" asset.parent.mkdir(parents=True, exist_ok=True) asset.write_bytes(b"\x89PNG-0") before = production_checkpoint_digest(ep) assert any(k.endswith("big.png") for k in before), "비-manifest 파일이 안 보인다" asset.write_bytes(b"\x89PNG-1") assert production_checkpoint_digest(ep) != before def test_writing_the_shadow_checkpoint_leaves_production_alone(self, tmp_path): """★shadow 를 쓴 뒤에도 production 은 그대로여야 한다.""" ep = _episode(tmp_path) before = production_checkpoint_digest(ep) out = replay_shadow_plan(ep, project_id="p", episode_id="e", write_checkpoint=True) assert out["production_unchanged"] is True assert production_checkpoint_digest(ep) == before written = Path(out["shadow_checkpoint_path"]) assert written.exists() and written.parent.name == SHADOW_STEP_ID assert json.loads(written.read_text(encoding="utf-8"))["mode"] == "shadow_plan" def test_no_classifier_leaves_every_subject_undecided(self, tmp_path): """★분류를 안 돌리면 skip 이 아니라 미확정이다.""" ep = _episode(tmp_path) out = replay_shadow_plan(ep, project_id="p", episode_id="e") assert out["counts"]["unresolved"] == 3 assert out["counts"]["skip"] == 0 # ── ★세계 맥락 전달 — 실측으로 드러난 결함 ──────────────────────────────── class TestWorldContextReachesTheJudge: """★분류기가 **조사 전에 상상으로 쓰인 묘사**만 보면 같은 실물이 갈린다. 실측: 같은 종이 승차권이 「시대가 적힌 에피소드」에서는 research, 「안 적힌 에피소드」에서는 generic→skip 이었다. `visual_world_rules` 에 era/region 이 **양쪽 다 있었는데** 분류기에 안 넘어가고 있었다. """ def test_era_and_region_appear_in_the_prompt(self): from app.modules.pipeline.grounding_classifier import build_user_prompt from app.modules.pipeline.grounding_subject import build_subject s = [build_subject(project_id="p", episode_id="e", source_anchor="S1", surface_form="가", owner_type="prop")] text = build_user_prompt(s, era="1983년", region="대한민국") assert "1983년" in text and "대한민국" in text def test_missing_era_is_marked_not_silently_empty(self): from app.modules.pipeline.grounding_classifier import build_user_prompt from app.modules.pipeline.grounding_subject import build_subject s = [build_subject(project_id="p", episode_id="e", source_anchor="S1", surface_form="가", owner_type="prop")] text = build_user_prompt(s) assert "알 수 없음" in text def test_prompt_tells_the_judge_not_to_infer_generic_from_a_silent_description(self): from app.modules.pipeline.grounding_classifier import build_user_prompt from app.modules.pipeline.grounding_subject import build_subject s = [build_subject(project_id="p", episode_id="e", source_anchor="S1", surface_form="가", owner_type="prop")] text = build_user_prompt(s, era="1983년") assert "generic" in text and "시대가 없다고 해서" in text def test_changing_era_moves_the_payload_hash(self): from app.modules.pipeline import grounding_classifier as gc from app.modules.pipeline.grounding_subject import build_subject s = [build_subject(project_id="p", episode_id="e", source_anchor="S1", surface_form="가", owner_type="prop")] rec = {"research_subject_id": s[0]["research_subject_id"], "grounding_class": "generic", "discriminability": "no", "referent_specificity": "generic_class", "difficulty": "easy", "confidence": 0.9, "visibility_intent": "yes", "locale": "KR", "generation": "1980s", "visible_discriminators": ["x"], "likely_failure_modes": ["y"], "generation_difficulty": "not_hard", "rationale": "z"} with patch.object(gc, "_call_structured", lambda **k: {"classifications": [rec]}): a = gc.classify(s, era="1983년") b = gc.classify(s, era="2020년") assert a["fingerprint"]["payload_hash"] != b["fingerprint"]["payload_hash"] assert a["fingerprint"]["era"] == "1983년" def test_replayer_reads_era_from_the_saved_rules_checkpoint(self, tmp_path): ep = _episode(tmp_path, era="1983년 늦가을", region="대한민국") seen = {} def _fake(subjects, *, era="", region=""): seen["era"], seen["region"] = era, region return {"records": [], "fingerprint": None, "search_calls": 0} replay_shadow_plan(ep, project_id="p", episode_id="e", classify_fn=_fake) assert seen == {"era": "1983년 늦가을", "region": "대한민국"} def test_missing_rules_checkpoint_fails_closed(self, tmp_path): """★빈 시대로 판정하면 방금 고친 결함이 그대로 재발한다.""" ep = _episode(tmp_path) (ep / "visual_world_rules" / "manifest.json").unlink() def _fake(subjects, *, era="", region=""): return {"records": [], "fingerprint": None, "search_calls": 0} with pytest.raises(ShadowSourceError): replay_shadow_plan(ep, project_id="p", episode_id="e", classify_fn=_fake) @pytest.mark.parametrize("era,region", [("", "대한민국"), ("1983년", ""), ("", "")]) def test_blank_era_or_region_fails_closed(self, tmp_path, era, region): ep = _episode(tmp_path, era=era, region=region) def _fake(subjects, *, era="", region=""): return {"records": [], "fingerprint": None, "search_calls": 0} with pytest.raises(ShadowSourceError): replay_shadow_plan(ep, project_id="p", episode_id="e", classify_fn=_fake) def test_no_classifier_does_not_need_the_rules_checkpoint(self, tmp_path): """분류를 안 돌리면 시대가 없어도 된다 — 시대는 판정에만 필요하다.""" ep = _episode(tmp_path) (ep / "visual_world_rules" / "manifest.json").unlink() out = replay_shadow_plan(ep, project_id="p", episode_id="e") assert out["counts"]["unresolved"] == 3 class TestShadowRerunKeepsEvidence: """★앞 판을 조용히 덮으면 **비교할 앞 판이 없어진다.** 계약 §8 이 이미 지목한 자리다 — 「검증 결과가 그 칸을 덮어 감사 기록이 사라졌다」. 두 판 대조가 이 단계의 유일한 무료 측정 수단이라 특히 그렇다. """ def test_second_run_archives_the_first(self, tmp_path): ep = _episode(tmp_path) first = replay_shadow_plan(ep, project_id="p", episode_id="e", write_checkpoint=True) first_bytes = Path(first["shadow_checkpoint_path"]).read_bytes() # 두 번째 판은 후보가 다르게 나오도록 입력을 바꾼다 _episode(tmp_path, entities={"characters": [], "locations": [], "props": [{"name": "다른 것", "short_id": "P09", "description": "x", "shot_count": 1}]}) second = replay_shadow_plan(ep, project_id="p", episode_id="e", write_checkpoint=True) cur = Path(second["shadow_checkpoint_path"]) archives = sorted(cur.parent.glob("manifest_*.json")) assert len(archives) == 1, "앞 판이 보관되지 않았다" assert archives[0].read_bytes() == first_bytes assert cur.read_bytes() != first_bytes assert json.loads(cur.read_text(encoding="utf-8"))["subject_count"] == 1 def test_archives_do_not_count_as_production_change(self, tmp_path): """보관본은 shadow 자리 안이라 production digest 를 안 움직인다.""" ep = _episode(tmp_path) before = production_checkpoint_digest(ep) replay_shadow_plan(ep, project_id="p", episode_id="e", write_checkpoint=True) out = replay_shadow_plan(ep, project_id="p", episode_id="e", write_checkpoint=True) assert out["production_unchanged"] is True assert production_checkpoint_digest(ep) == before class TestModeSemanticsAreNarrow: """★`legacy` 를 「무료」로 묶었던 것은 틀렸다. 실물: `.env` 에 `ERA_RESEARCH_ENABLED=true` 이고 `still_recipe_service.py:2159,4152` 가 `assess_and_research_cached` 로 **장소 시대 조사를 산다.** 이 모듈의 술어는 **v2 고증 조사**에 대해서만 말한다. """ def test_only_shadow_plan_buys_nothing(self): from app.core.grounding_mode import ( NO_RESEARCH_AT_ALL_MODES, buys_no_research_at_all) assert NO_RESEARCH_AT_ALL_MODES == {"shadow_plan"} assert buys_no_research_at_all("legacy") is False assert buys_no_research_at_all("shadow_plan") is True def test_v2_predicate_is_named_for_v2_research_only(self): from app.core import grounding_mode as gm assert gm.buys_v2_research("v2") is True assert gm.buys_v2_research("legacy") is False assert not hasattr(gm, "FREE_GROUNDING_MODES"), \ "「legacy 도 무료」로 읽히는 이름이 남아 있다" assert not hasattr(gm, "buys_research"), "넓은 옛 이름이 남아 있다" class TestShapeFailsClosed: @pytest.mark.parametrize("status", [None, "", "running", "failed", "partial"]) def test_non_completed_status_is_rejected(self, tmp_path, status): """★빈 status 를 완료로 봐 주면 미완 산출로 재생해 놓고 「후보가 적다」를 결함이 아니라 사실로 읽게 된다.""" ep = _episode(tmp_path) p = ep / "entity_merge" / "manifest.json" payload = json.loads(p.read_text(encoding="utf-8")) if status is None: payload.pop("status", None) else: payload["status"] = status p.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") with pytest.raises(ShadowSourceError): build_subjects_from_saved_episode(ep, project_id="p", episode_id="e") @pytest.mark.parametrize("bad", [None, [], "문자열", 3]) def test_wrong_filtered_entities_shape_is_rejected(self, tmp_path, bad): ep = _episode(tmp_path) p = ep / "entity_merge" / "manifest.json" p.write_text(json.dumps( {"status": "completed", "data": {"filtered_entities": bad}}, ensure_ascii=False), encoding="utf-8") with pytest.raises(ShadowSourceError): build_subjects_from_saved_episode(ep, project_id="p", episode_id="e") def test_rules_checkpoint_must_be_completed_too(self, tmp_path): ep = _episode(tmp_path) p = ep / "visual_world_rules" / "manifest.json" p.write_text(json.dumps({"data": {"era": "1983년", "region": "KR"}}, ensure_ascii=False), encoding="utf-8") def _fake(subjects, *, era="", region=""): return {"records": [], "fingerprint": None, "search_calls": 0} with pytest.raises(ShadowSourceError): replay_shadow_plan(ep, project_id="p", episode_id="e", classify_fn=_fake) class TestEffectiveTargetFollowsTheBackend: """★capability 는 **실제로 굽는 모델**에 귀속해야 한다 (계약 §3). 처음엔 backend 와 무관하게 언제나 gemini 를 돌려주고 backend alias 를 version 이라 불렀다. 둘 다 틀렸는데 **이 함수를 태우는 시험이 하나도 없어서** 126개가 전부 초록이었다. """ def _target(self, backend, monkeypatch): import importlib.util from app.core.config import settings spec = importlib.util.spec_from_file_location( "_shadow_tool", Path(__file__).resolve().parents[2] / "tools" / "prompt_measure" / "shadow_plan_replay.py") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) monkeypatch.setattr(settings, "still_image_backend", backend, raising=False) return mod._effective_target() def test_nb2_reports_gemini(self, monkeypatch): from app.core.config import settings provider, model, version, backend = self._target("nb2", monkeypatch) assert provider == "gemini" assert model == settings.gemini_image_model assert backend == "nb2" def test_grok2_reports_xai_not_gemini(self, monkeypatch): """★이게 핵심 — grok2 인데 gemini 를 기록하면 판정이 엉뚱한 모델에 붙는다.""" from app.core.config import settings provider, model, version, backend = self._target("grok2", monkeypatch) assert provider == "xai" assert model == settings.grok_image_model assert model != settings.gemini_image_model assert backend == "grok2" def test_unknown_backend_stops_instead_of_guessing(self, monkeypatch): with pytest.raises(SystemExit): self._target("nb3", monkeypatch) def test_backend_alias_is_not_reported_as_a_version(self, monkeypatch): """★`nb2`/`grok2` 는 **백엔드 이름**이지 모델 버전이 아니다.""" for backend in ("nb2", "grok2"): _provider, model, version, got_backend = self._target(backend, monkeypatch) assert version != got_backend, "backend alias 가 version 자리에 들어갔다" # 이 provider 들은 버전을 모델 id 에 담는다 — 없는 버전을 지어내지 않는다. assert version == model def test_blank_model_setting_stops(self, monkeypatch): from app.core.config import settings monkeypatch.setattr(settings, "gemini_image_model", "", raising=False) with pytest.raises(SystemExit): self._target("nb2", monkeypatch) class TestToolAndProductionShareOneCarry: """★도구가 프로덕션과 **다른 입력**을 보내면 잰 것이 뜻을 잃는다. 실제로 갈려 있었다 — production 은 A0 가 건진 **원문 문장**을 분류기에 넘기는데 측정 도구는 LLM 이 상상해 쓴 ``description`` 을 넘겼다. """ def test_both_call_the_same_function(self): """★결속을 두 곳에 적으면 갈린다. 한 함수를 쓰는지 본다.""" import app.core.steps.grounding_steps as steps import app.modules.pipeline.grounding_carry as carry import app.modules.pipeline.grounding_shadow as shadow assert steps.build_carry_index is carry.build_carry_index assert steps.match_candidate is carry.match_candidate assert "grounding_carry" in shadow.build_subjects_from_saved_episode.__doc__ def test_the_same_input_gives_the_same_subjects(self, tmp_path): """★끝점 — 같은 엔티티·같은 A0 후보면 **같은 subject id** 가 나온다.""" import json from app.modules.pipeline import grounding_carry as carry from app.modules.pipeline.grounding_shadow import ( build_subjects_from_saved_episode) ents = {"props": [{"short_id": "P01", "name": "요금통", "description": "상상 묘사"}], "characters": [], "locations": []} a0 = [{"surface_form": "쇠사슬로 묶인 요금통", "owner_type": "prop", "source_anchor": "SEG-001", "source_quote": "원문 문장이다.", "research_subject_id": "rs-1"}] ep = tmp_path / "e" d = ep / "entity_merge" d.mkdir(parents=True) (d / "manifest.json").write_text(json.dumps( {"status": "completed", "data": ents}, ensure_ascii=False), encoding="utf-8") prod = carry.build_subjects(ents, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=a0) tool = build_subjects_from_saved_episode( ep, project_id="p", episode_id="e", a0_candidates=a0) assert [s["research_subject_id"] for s in tool["subjects"]] == \ [s["research_subject_id"] for s in prod["subjects"]] assert tool["subjects"][0]["source_quote"] == "원문 문장이다." assert tool["quote_sources"] == {"manuscript": 1} def test_without_a0_the_basis_is_recorded_not_hidden(self, tmp_path): """★A0 가 없으면 상상 묘사로 잰다 — 그 사실이 **산출에 남아야** 한다.""" import json from app.modules.pipeline.grounding_shadow import ( build_subjects_from_saved_episode) ents = {"props": [{"short_id": "P01", "name": "요금통", "description": "상상 묘사"}], "characters": [], "locations": []} ep = tmp_path / "e" d = ep / "entity_merge" d.mkdir(parents=True) (d / "manifest.json").write_text(json.dumps( {"status": "completed", "data": ents}, ensure_ascii=False), encoding="utf-8") out = build_subjects_from_saved_episode(ep, project_id="p", episode_id="e") assert out["quote_sources"] == {"entity_description": 1} assert out["carried"] == 0 assert out["subjects"][0]["source_quote"] == "상상 묘사" class TestTheMeasurementBasisIsNeverLost: """★`quote_sources` 는 **모든 반환**에 있어야 한다. 하나라도 빠지면 그 경로로 잰 판은 「무엇을 근거로 쟀는지」를 모른 채 보고된다 — 그리고 그 보고는 「production 을 쟀다」로 읽힌다. """ @staticmethod def _ep(tmp_path, ents): import json ep = tmp_path / "e" d = ep / "entity_merge" d.mkdir(parents=True) (d / "manifest.json").write_text(json.dumps( {"status": "completed", "data": ents}, ensure_ascii=False), encoding="utf-8") return ep def test_an_empty_population_stops_instead_of_returning(self, tmp_path): """★엔티티 0을 「깨끗함」으로 읽지 않는다 — 근거 칸도 못 채운다.""" from app.modules.pipeline.grounding_shadow import ( ShadowSourceError, build_subjects_from_saved_episode) with pytest.raises(ShadowSourceError, match="하나도 없다"): build_subjects_from_saved_episode( self._ep(tmp_path, {"props": [], "characters": [], "locations": []}), project_id="p", episode_id="e") @pytest.mark.parametrize("ents", [ {"props": [{"short_id": "P01", "name": "요금통", "description": "d"}]}, {"props": [{"short_id": "P01", "name": ""}, {"short_id": "P02", "name": "요금통"}]}, {"props": ["문자열이 섞였다", {"short_id": "P01", "name": "요금통"}]}, ]) def test_every_return_carries_it(self, tmp_path, ents): from app.modules.pipeline.grounding_shadow import ( build_subjects_from_saved_episode) out = build_subjects_from_saved_episode( self._ep(tmp_path, ents), project_id="p", episode_id="e") for key in ("subjects", "unbound", "carry_reasons", "carried", "quote_sources"): assert key in out, f"{key} 가 빠졌다" def test_replay_output_carries_it_too(self, tmp_path, monkeypatch): """★재생기 산출에도 있어야 한다 — 보고가 거기서 나온다.""" import inspect as _inspect from app.modules.pipeline import grounding_shadow as sh src = _inspect.getsource(sh.replay_shadow_plan) for key in ("quote_sources", "carry_reasons", "a0_carried"): assert f'"{key}"' in src, f"재생기 산출에 {key} 가 없다" class TestBothPathsTakeTheSameCallable: """★「같은 함수를 쓴다」는 **객체가 같아야** 참이다.""" def test_the_step_and_the_tool_call_one_builder(self): import app.core.steps.grounding_steps as steps import app.modules.pipeline.grounding_carry as carry import app.modules.pipeline.grounding_shadow as shadow assert steps._carry.build_subjects is carry.build_subjects assert shadow.build_subjects_from_saved_episode.__module__ == \ "app.modules.pipeline.grounding_shadow" # ★도구가 자기 결속을 다시 만들면 안 된다 — `build_subject` 직접 호출 0 import inspect as _inspect src = _inspect.getsource(shadow.build_subjects_from_saved_episode) assert "build_subject(" not in src, "도구가 자기 subject 를 다시 만든다" assert "_carry.build_subjects(" in src def test_same_entities_and_candidates_give_identical_output(self, tmp_path): """★끝점 — subjects·unbound·사유가 **전부** 같아야 한다.""" import json from app.modules.pipeline import grounding_carry as carry from app.modules.pipeline.grounding_shadow import ( build_subjects_from_saved_episode) ents = {"props": [{"short_id": "P01", "name": "요금통", "description": "상상 묘사"}, {"short_id": "P02", "name": "승차권 뭉치", "description": "상상 묘사"}], "characters": [], "locations": []} a0 = [{"surface_form": "쇠사슬로 묶인 요금통", "owner_type": "prop", "source_anchor": "SEG-001", "source_quote": "원문 하나.", "research_subject_id": "rs-1"}, {"surface_form": "승차권", "owner_type": "prop", "source_anchor": "SEG-001", "source_quote": "원문 둘.", "research_subject_id": "rs-2"}, {"surface_form": "승차권 뭉치", "owner_type": "prop", "source_anchor": "SEG-002", "source_quote": "원문 셋.", "research_subject_id": "rs-3"}] ep = tmp_path / "e" d = ep / "entity_merge" d.mkdir(parents=True) (d / "manifest.json").write_text(json.dumps( {"status": "completed", "data": ents}, ensure_ascii=False), encoding="utf-8") prod = carry.build_subjects(ents, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=a0) tool = build_subjects_from_saved_episode( ep, project_id="p", episode_id="e", a0_candidates=a0) assert tool["subjects"] == prod["subjects"] assert tool["unbound"] == prod["unbound"] assert tool["carry_reasons"] == prod["carry_reasons"] # ★섞인 판이어야 시험이 뜻이 있다 — 붙은 것 하나 + 못 붙인 것 하나 assert prod["carry_reasons"]["matched"] == 1 assert prod["carry_reasons"]["ambiguous"] == 1 # ★★붙은 것 하나뿐이다. P02 가 두 후보 사이에서 못 정했고, 그 **두 # 후보는 승격 안 한다** — 「서로 다른 두 대상」인지 「같은 것을 두 번 # 적은 것」인지 기계적으로 못 가르고, 승격하면 같은 것을 두 번 조사할 # 수 있다(그건 돈이다). 못 정한 것은 `unresolved` 다. assert tool["quote_sources"] == {"manuscript": 1} led = prod["candidate_ledger"] assert led["by_disposition"]["carried"] == 1 assert led["by_disposition"]["promoted"] == 0 assert led["by_disposition"]["unresolved"] == 2 assert led["total"] == 3 class TestTheToolReadsTheSamePopulationProductionDoes: """★같은 함수를 써도 **읽는 자리가 다르면** 다른 것을 잰다. production 의 `grounding_plan`(13.65)은 `entity_merge`(13.5)를 읽는다. 도구가 `entity_filter`(13.7)를 읽으면 **거기서 걸러진 것을 아예 못 본다** — 그리고 걸러진 것이 곧 고증이 지키려는 대상이다. 실측(저장 에피소드 셋): 7→6 · 7→6 · **11→7**. """ @staticmethod def _ep_with_both(tmp_path): """merge 에는 4개, filter 에는 2개 — **다른** fixture 로 갈라 둔다.""" import json merged = {"characters": [], "locations": [], "props": [{"short_id": f"P0{i}", "name": f"소품{i}", "description": "d"} for i in range(1, 5)]} survived = {"characters": [], "locations": [], "props": merged["props"][:2]} ep = tmp_path / "e" for step, data in (("entity_merge", merged), ("entity_filter", {"filtered_entities": survived})): d = ep / step d.mkdir(parents=True) (d / "manifest.json").write_text(json.dumps( {"status": "completed", "data": data}, ensure_ascii=False), encoding="utf-8") return ep def test_it_reads_merge_not_filter(self, tmp_path): from app.modules.pipeline.grounding_shadow import ( build_subjects_from_saved_episode) out = build_subjects_from_saved_episode( self._ep_with_both(tmp_path), project_id="p", episode_id="e") got = sorted(s["surface_form"] for s in out["subjects"]) assert got == ["소품1", "소품2", "소품3", "소품4"], \ "걸러진 뒤를 읽는다 — production 이 보는 것과 다르다" def test_the_source_step_is_the_one_production_reads(self): import inspect from app.core.steps.grounding_steps import GroundingPlanStep from app.modules.pipeline import grounding_shadow as sh src = inspect.getsource(GroundingPlanStep._execute) assert f'_load_prev_checkpoint("{sh._SOURCE_STEP}")' in src, \ f"production 은 {sh._SOURCE_STEP} 를 안 읽는다" def test_missing_merge_stops_instead_of_falling_back(self, tmp_path): """★없으면 `entity_filter` 로 **안 물러선다** — 조용히 다른 것을 잰다.""" import json from app.modules.pipeline.grounding_shadow import ( ShadowSourceError, build_subjects_from_saved_episode) ep = tmp_path / "e" d = ep / "entity_filter" d.mkdir(parents=True) (d / "manifest.json").write_text(json.dumps( {"status": "completed", "data": {"filtered_entities": {"props": [{"short_id": "P01", "name": "요금통"}]}}}, ensure_ascii=False), encoding="utf-8") with pytest.raises(ShadowSourceError, match="entity_merge"): build_subjects_from_saved_episode(ep, project_id="p", episode_id="e") class TestIdsAreTheRealScopeNotThePrefix: """★8자 좌표로 발급하면 production 과 **다른 subject id** 가 나온다. `mint_subject_id` 가 project/episode id 를 해시에 넣는다. 좌표는 사람이 읽는 접두어지 범위가 아니다. """ def test_a_prefix_and_the_full_uuid_give_different_ids(self): from app.modules.pipeline.grounding_subject import mint_subject_id kw = dict(episode_id="97375a4b-1111-2222-3333-444444444444", source_anchor="P01", surface_form="요금통", owner_type="prop") assert (mint_subject_id(project_id="da049582", **kw) != mint_subject_id( project_id="da049582-2c6d-492c-979d-f468d61bab6e", **kw)) def test_the_tool_resolves_the_real_ids_from_the_path(self, tmp_path): import sys sys.path.insert(0, str( Path(__file__).resolve().parents[2] / "tools" / "prompt_measure")) import grounding_controls_acceptance as t ep = (tmp_path / "da049582-2c6d-492c-979d-f468d61bab6e" / "checkpoints" / "episodes" / "97375a4b-1111-2222-3333-444444444444") ep.mkdir(parents=True) pid, eid = t.real_ids(ep) assert pid == "da049582-2c6d-492c-979d-f468d61bab6e" assert eid == "97375a4b-1111-2222-3333-444444444444" class TestOwnerGroupsAndContestedCandidates: """★A0 의 owner 는 **잠정**이다 — 파이프라인 갈래와 늘 같지 않다. 실측(0aea12c2): A0 가 「요금통」을 `location_part`(쇠사슬로 묶여 못 옮긴다)로 적었는데 파이프라인은 `P01` 을 **prop** 으로 저장했다. 같은 원고 다른 화에서는 A0 가 그것을 `prop` 으로 적었다 — 경계에 걸친 대상이라 화마다 갈린다. owner 를 정확히 맞추라고 하면 §2-3c 양성 축이 **영원히 원문 근거를 못 받는다.** """ @staticmethod def _c(surface, owner, rid=None): return {"surface_form": surface, "owner_type": owner, "research_subject_id": rid or f"{owner}:{surface}", "source_anchor": "SEG-001", "source_quote": "원문."} def _build(self, ents, cands): from app.modules.pipeline.grounding_carry import build_subjects return build_subjects(ents, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=cands) def test_a_fixed_installation_never_binds_to_a_stored_prop(self): """★★**뒤집힌 시험**이다 (§2-6.5a · Codex BLOCK · 09-01). 앞에는 `location_part` 후보가 `prop` 행에 **이름으로** 붙는 것을 잠갔다. §2-6.5a 로 LP 행이 실제로 생기자 그 문이 **양방향**이 됐고, 다른 부분 대상의 원문 근거·참조 의무가 조용히 건너갔다. 이제 두 갈래를 **갈랐다**. """ out = self._build( {"props": [{"short_id": "P01", "name": "쇠사슬에 묶인 투명 요금통", "description": "상상"}]}, [self._c("요금통", "location_part")]) assert out["carried"] == 0, "★갈래를 넘어 붙었다" # ★엔티티는 제 subject 로 남고, 후보는 따로 `deferred` 로 남는다 assert [s["owner_type"] for s in out["subjects"]] == ["prop"] led = out["candidate_ledger"]["rows"] assert [r["disposition"] for r in led] == ["deferred"] @pytest.mark.parametrize("stored_owner,cand_owner", [ ("character", "outlook"), ("outlook", "character"), ("location", "prop"), ("prop", "outlook"), ("character", "prop"), ("location", "outlook"), # ★§2-6.5a 로 **이 둘도** 갈렸다 — 부분 설비와 소품은 다른 대상이다 ("prop", "location_part"), ("location_part", "prop"), ]) def test_the_person_and_what_they_wear_never_bind(self, stored_owner, cand_owner): """★owner 검사가 막으려던 것은 **이것**이다.""" from app.modules.pipeline.grounding_carry import ( build_carry_index, match_candidate) carry = build_carry_index([self._c("감색 차장 제복", cand_owner)]) assert match_candidate(carry, "감색 차장 제복", stored_owner) == (None, "none") def test_one_candidate_taken_by_two_entities_binds_to_neither(self): """★앞 판은 뒤에 온 엔티티를 **통째로 삼켰다** — 분류도 기록도 없이. 같은 subject id 라 「접는다」로 넘어갔는데, 물려받은 id 가 겹치는 것은 「같은 대상」이 아니라 **어느 쪽 것인지 못 정한 것**이다. """ out = self._build( {"props": [{"short_id": "P01", "name": "쇠사슬에 묶인 요금통", "description": "상상"}, {"short_id": "P02", "name": "요금통 속 동전", "description": "상상"}]}, [self._c("요금통", "prop")]) assert out["carried"] == 0 assert out["carry_reasons"].get("contested") == 2 assert out["subjects"] == [], "분류기에 보냈다" # ★둘 다 기록에 남아야 한다 — 하나도 사라지면 안 된다 assert sorted(u["_short_id"] for u in out["unbound"]) == ["P01", "P02"] for u in out["unbound"]: assert u["route"] == "unresolved" assert u["a0_candidates"][0]["source_quote"] == "원문." def test_two_entities_that_hit_different_candidates_both_bind(self): out = self._build( {"props": [{"short_id": "P01", "name": "쇠사슬에 묶인 요금통"}, {"short_id": "P03", "name": "고무줄로 묶인 회수권 뭉치"}]}, [self._c("요금통", "prop"), self._c("회수권 뭉치", "prop")]) assert out["carried"] == 2 assert len(out["subjects"]) == 2 assert {s["research_subject_id"] for s in out["subjects"]} == { "prop:요금통", "prop:회수권 뭉치"} def test_no_entity_is_ever_silently_dropped(self): """★들어간 이름 수 = 나온 행 수. 어느 갈래로 가든 하나도 안 사라진다.""" ents = {"props": [{"short_id": f"P0{i}", "name": f"요금통 {i}"} for i in range(1, 4)], "characters": [{"short_id": "C01", "name": "정임"}], "locations": [{"short_id": "L01", "name": "버스 안"}]} out = self._build(ents, [self._c("요금통", "prop")]) assert len(out["subjects"]) + len(out["unbound"]) == 5