"""조사 BLOCK 넷 (Codex/사용자 2026-09-02): ① 선언된 시대·지역은 **글자 그대로** — 모델이 다시 읽어 바꾸지 못한다 ② 재판정에서 새 검색 0 이면 `no_match_after_retry` 로 접지 않는다 ③ 1차를 모든 대상에게 먼저, 2차는 못 구한 대상만 · 1차 위에 잇는다(재구매 0) ④ 팩 pin: 판독기 6 · 지시문 5 — 뼈대·장소 분리·2차 넓힘 문안이 실제 파일에 있다 ⑤ (Codex BLOCK 1/5, 09-02 밤) 상한은 pass 수만큼 · 지시문 5 는 고정 명사 예시를 뺐다 """ from __future__ import annotations from pathlib import Path import pytest class TestDeclaredCoordinatesAreCarriedExactly: def test_declared_values_override_the_model(self): from app.modules.pipeline.grounding_coordinates import apply_declared_coordinates got = apply_declared_coordinates( {"era": "20세기 전반, 일제강점기 무렵", "region": "한반도 조선", "rules": []}, {"grounding_era": "1960년대", "grounding_region": "대한민국"}) assert got["era"] == "1960년대" and got["region"] == "대한민국" assert got["inferred_era"].startswith("20세기") and got["inferred_region"] == "한반도 조선" assert got["coordinate_source"] == {"era": "declared", "region": "declared"} def test_without_a_declaration_the_model_value_stays_and_nothing_is_invented(self): from app.modules.pipeline.grounding_coordinates import apply_declared_coordinates got = apply_declared_coordinates({"era": "", "region": "대한민국의 국도변"}, {}) assert got["era"] == "" and got["region"] == "대한민국의 국도변" assert got["coordinate_source"] == {"era": "inferred", "region": "inferred"} def test_region_only_declaration_leaves_era_to_the_model(self): from app.modules.pipeline.grounding_coordinates import apply_declared_coordinates got = apply_declared_coordinates({"era": "현대", "region": "x"}, {"grounding_region": "대한민국"}) assert got["era"] == "현대" and got["region"] == "대한민국" assert got["coordinate_source"] == {"era": "inferred", "region": "declared"} def test_the_world_rules_step_applies_it(self, monkeypatch): """★production 스텝 끝점 — 추출기가 무엇을 내든 선언이 이긴다.""" from app.core.steps import summary_steps as ss from app.modules.pipeline import visual_world_rules as vwr monkeypatch.setattr(vwr, "extract_visual_rules", lambda **kw: {"era": "모델이 지어낸 시대", "region": "모델 지역", "rules": []}) s = ss.VisualWorldRulesStep.__new__(ss.VisualWorldRulesStep) s.project_id, s.episode_id, s.db = "p", "e", None s.project_config = {"grounding_era": "1960년대", "grounding_region": "대한민국"} monkeypatch.setattr(s, "_load_prev_checkpoint", lambda sid: {"data": {"cleaned_text": "원고", "summary": "요약"}}, raising=False) monkeypatch.setattr(s, "build_opik_metadata", lambda **k: {}, raising=False) monkeypatch.setattr(ss, "get_planning_context", lambda *a, **k: type("P", (), { "inject_if_available": lambda self, *a: "", "has_planning_doc": False})(), raising=False) got = s._execute(mode="resume") # ★넓은 except→skip 없음 (Codex) assert got["data"]["era"] == "1960년대" and got["data"]["region"] == "대한민국" assert got["data"]["coordinate_source"]["era"] == "declared" def test_the_canary_passes_the_fixture_declaration(self): from tools.grounding_audit import canary_run as cr got = cr._owner_requirement("period_episode") assert got["grounding_era"] == "1960년대" and got["grounding_region"] == "대한민국" modern = cr._owner_requirement("modern_episode") assert "grounding_era" not in modern and modern["grounding_region"] == "대한민국" class TestRejudgeDoesNotClaimNoMatchWithoutRealSearches: def _record(self, searched_rounds): rounds = [] for i in range(searched_rounds): rounds.append({"round_no": i + 1, "queries": [["q"]], "downloaded_candidates": [ {"index": 1, "path": "x.png", "url": "http://x/1.png"}]}) return {"rounds": rounds, "candidates": [], "status": "retryable"} def test_one_searched_round_stays_retryable(self, monkeypatch, tmp_path): from app.modules.pipeline import reference_acquisition_rounds as rr from app.modules.pipeline import coarse_type_pick as ctp monkeypatch.setattr(rr, "resolve_cached_candidates", lambda cands, root: [ {"index": 1, "path": str(tmp_path / "x.png"), "rel": "x.png", "sha256": "0"}]) monkeypatch.setattr(ctp, "combine_coarse_verdicts", lambda per, n: {"chosen_index": 0}) monkeypatch.setattr(ctp, "decide_next_round", lambda *a, **k: {"next": ctp.NEXT_NO_MATCH, "round_no": 2, "why": "x"}) got = rr.rejudge_cached(self._record(1), root=tmp_path, judge=lambda c: {}) assert got["status"] == "retryable" assert got["replay"]["searched_rounds"] == 1 def test_two_searched_rounds_may_be_no_match(self, monkeypatch, tmp_path): from app.modules.pipeline import reference_acquisition_rounds as rr from app.modules.pipeline import coarse_type_pick as ctp monkeypatch.setattr(rr, "resolve_cached_candidates", lambda cands, root: [ {"index": 1, "path": str(tmp_path / "x.png"), "rel": "x.png", "sha256": "0"}]) monkeypatch.setattr(ctp, "combine_coarse_verdicts", lambda per, n: {"chosen_index": 0}) monkeypatch.setattr(ctp, "decide_next_round", lambda *a, **k: {"next": ctp.NEXT_NO_MATCH, "round_no": 2, "why": "x"}) got = rr.rejudge_cached(self._record(2), root=tmp_path, judge=lambda c: {}) assert got["status"] == "no_match_after_retry" class TestRoundOneGoesToEveryoneFirst: """★실 `ChunkJournal` · 실 상한 (Codex BLOCK 1, 2026-09-02): 앞 판은 `buy_or_reuse` 를 대역으로 바꿔 상한을 지나쳤다 — 실제로는 상한 = 대상 수 N 이라 **2차가 문 앞에서 전부 막혔다**.""" def _run(self, monkeypatch, tmp_path, calls, *, cap): from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import grounding_chunk_journal as cj from app.modules.pipeline import reference_acquisition_rounds as rr def fake_acquire_one(target, *, rounds, resume_from=None, **kw): sid = target["subject_id"] prev = list((resume_from or {}).get("rounds") or []) for r in range(len(prev) + 1, int(rounds) + 1): calls.append(f"{sid}{r}") prev.append({"round_no": r, "queries": [["q"]], "downloaded_candidates": []}) return {"subject_id": sid, "status": "retryable", "chosen": None, "rounds": prev, "candidates": []} monkeypatch.setattr(rr, "acquire_one", fake_acquire_one) monkeypatch.setattr(ca, "resolve_rounds", lambda r: 2) monkeypatch.setattr(ca._aa, "inputs_from_ledger", lambda ledger: { "targets": [{"target": {"subject_id": "A", "terms_native": []}, "source_evidence": {}, "ledger_row": {"final_id": "A"}}, {"target": {"subject_id": "B", "terms_native": []}, "source_evidence": {}, "ledger_row": {"final_id": "B"}}], "skipped": [], "refused": [], "auto_completed_rows": [], "not_applicable_rows": [], "ledger_rows": []}) journal = cj.ChunkJournal(tmp_path / "journal.json", contract={"wiring": ca.CONTRACT_VERSION}) got = ca.run({"rows": []}, journal=journal, cap=cap, workdir=tmp_path, search=lambda *a, **k: {}, download=lambda *a, **k: True, judge=lambda c: {}, write_brief=None, rounds=2) return got, journal def test_pass_order_with_the_real_journal_and_cap(self, monkeypatch, tmp_path): """대상 2 · 라운드 2 · 상한 4(=N×pass): [A1, B1, A2, B2] 가 **실 상한 안에서** 다 산다.""" calls = [] got, journal = self._run(monkeypatch, tmp_path, calls, cap=4) assert calls == ["A1", "B1", "A2", "B2"], calls p = got["purchases"] assert p["cap"] == 4 and p["cap_passes"] == 2 assert journal.bought() == 4 assert [r["disposition"] for r in got["rows"]] == ["acquired", "acquired"] def test_a_cap_of_n_blocks_the_second_pass(self, monkeypatch, tmp_path): """★양성 대조 — 상한을 대상 수 N 으로 주면 2차가 문 앞에서 막힌다(앞 판의 결함 그대로).""" calls = [] got, journal = self._run(monkeypatch, tmp_path, calls, cap=2) assert calls == ["A1", "B1"], calls assert got["purchases"]["cap_reached"] is True def test_the_step_cap_is_targets_times_passes(self, monkeypatch): """★production 스텝의 `_cap` 이 곱한다 — run() 이 아니라.""" from app.core.steps import reference_acquisition_step as rs from app.modules.pipeline import grounding_acquisition_ledger as gl from app.modules.pipeline import grounding_central_acquisition as ca monkeypatch.setattr(gl, "acquisition_targets", lambda ob: [1, 2, 3]) monkeypatch.setattr(ca, "resolve_rounds", lambda r: 2) s = rs.ReferenceAcquisitionStep.__new__(rs.ReferenceAcquisitionStep) s.project_config = {} assert s._cap({}) == 6 s.project_config = {"grounding_reference_cap": 1} assert s._cap({}) == 2 def test_resume_buys_nothing_again(self, monkeypatch, tmp_path): """같은 장부로 다시 돌리면 네 신원이 전부 재사용 — 새 호출 0.""" calls = [] self._run(monkeypatch, tmp_path, calls, cap=4) n = len(calls) _, journal = self._run(monkeypatch, tmp_path, calls, cap=4) assert len(calls) == n and journal.bought() == 4 class TestThePacksSayIt: ROOT = Path(__file__).resolve().parents[3] / "prompts" / "_base" def test_chunk_pack_six_is_pinned_and_has_the_rules(self): from app.modules.pipeline import grounding_chunk as gc # 2026-09-02 밤: 7 은 6 의 판독기 그대로 + merge 지시문에 「입은 것은 입은 사람의 part_of」 assert gc.CHUNK_PACK_VERSION == "7.202609022250" s = (self.ROOT / "grounding_chunk" / gc.CHUNK_PACK_VERSION / "system.md").read_text(encoding="utf-8") assert "씬이 벌어지는" in s and "bound_parent" in s and "뼈대" in s m = (self.ROOT / "grounding_chunk" / gc.CHUNK_PACK_VERSION / "merge_system.md").read_text(encoding="utf-8") assert "입은 사람" in m and "`part_of` 입니다" in m old_m = (self.ROOT / "grounding_chunk" / "6.202609022200" / "merge_system.md").read_text(encoding="utf-8") assert "입은 사람" not in old_m, "★옛 팩은 그대로" def test_brief_pack_five_is_pinned_and_has_the_rules(self): from app.modules.pipeline import grounding_ref_brief as grb assert grb.PACK_VERSION == "5" and grb.resolve_pack_version() == "5.202609022330" d = self.ROOT / "grounding_ref_brief" / "5.202609022330" b = (d / "brief_system.md").read_text(encoding="utf-8") h = (d / "narrow_retry_hint.md").read_text(encoding="utf-8") assert "SKELETON" in b and "reconstruction" in b assert "WIDER" in h and "PERIOD AND THE REGION ARE NOT QUALIFIERS" in h def test_pack_five_dropped_the_fixed_noun_examples_of_four(self): """★Codex BLOCK 5: 4 의 「현대 장소」문단이 고정 명사 예시를 들었다 — 5 는 원리만. 글자 대조는 **4 의 그 문장**이 5 에 없다는 것 하나뿐(뜻 판단 아님).""" four = (self.ROOT / "grounding_ref_brief" / "4.202609022200" / "brief_system.md").read_text(encoding="utf-8") five = (self.ROOT / "grounding_ref_brief" / "5.202609022330" / "brief_system.md").read_text(encoding="utf-8") example_sentence = "what a filling station" assert example_sentence in four and example_sentence not in five assert "someone who lives there notices" in five def test_old_packs_are_untouched(self): old = self.ROOT / "grounding_chunk" / "5.202609012800" / "system.md" assert "씬이 벌어지는" not in old.read_text(encoding="utf-8")