"""★★★사용자 5단계 획득 (2026-09-03) — 텍스트 조사 → 좁은 검색 → CRITERIA 판정 → 넓은 재검색 → **반드시 한 장**. HITL 0: 사람 없이 끝까지 간다. 무료 · 바깥 호출 0(conftest netprobe). 대역은 진짜 함수 서명을 따른다 — `research_target` 대신 같은 서명의 `research_call`, 심판은 `judge(cands, criteria=)`. """ from __future__ import annotations import json from pathlib import Path import pytest from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline import grounding_target_research as gtr from app.modules.pipeline import reference_acquisition as ra from app.modules.pipeline import reference_acquisition_rounds as rr ERA, REGION = "가나다 무렵", "라마바 지방" def _research(client, *, skeleton, evidence, opik_metadata=None, **_kw): """조사 대역 — 뼈대와 근거를 받아 좁은/넓은 질의와 CRITERIA 를 낸다.""" kind = skeleton["coarse_type_label"] return { "what_it_is": f"{kind} 는 무엇이다", "appearance_criteria": f"- {kind} 의 생김새 1\n- 생김새 2", "narrow_queries": [f"{REGION} {ERA} {kind} 소장품 사진", f"{REGION} {ERA} {kind} 유물"], "rough_queries": [f"{REGION} {ERA} {kind}"], "search_directive_native": f"{REGION} {ERA} 의 {kind} 사진을 찾아라", "language_lock_native": "한국어로 찾는다", "sources": ["https://example.invalid/a"], "provenance": {"provider": "fake"}, "_evidence_seen": list(evidence), } def _writer(calls): def _rc(client, **kw): calls.append(kw) return _research(client, **kw) return gtr.make_writer(world_facts="w", source_text="원문", era=ERA, region=REGION, research_call=_rc, client=object()) def _target(sid="rs_x"): return {"subject_id": sid, "directive_native": "됨직한 것", "terms_native": ["갑"], "language_lock_native": "한국어", "owner_type": "prop", "coarse_type_label": "물건", "source_quotes": ["원고 문장 하나", "원고 문장 둘"]} class TestTheResearchWriter: def test_round_one_is_narrow_round_two_is_rough_and_research_happens_once(self): calls = [] w = _writer(calls) b1 = w(_target(), narrow=False) b2 = w(_target(), narrow=True) assert b1["search_terms_native"] == _research(None, skeleton={"coarse_type_label": "물건"}, evidence=[])["narrow_queries"] assert b2["search_terms_native"] == [f"{REGION} {ERA} 물건"] assert b1["vlm_criteria"] and b1["vlm_criteria"] == b2["vlm_criteria"] assert len(calls) == 1, "★대상당 조사는 한 번이다" assert calls[0]["evidence"] == ["원고 문장 하나", "원고 문장 둘"], "★근거 문장이 조사에 안 실렸다" assert ERA in b1["search_directive_native"] and REGION in b1["search_directive_native"] def test_a_prior_research_on_the_target_is_reused_without_a_call(self): calls = [] w = _writer(calls) prior = _research(None, skeleton={"coarse_type_label": "물건"}, evidence=[]) prior["evidence_fingerprint"] = gtr.evidence_fingerprint(_target()["source_quotes"]) b = w({**_target("rs_y"), "_prior_research": prior}, narrow=True) assert calls == [], "★재개가 조사를 다시 샀다" assert b["search_terms_native"] == prior["rough_queries"] def test_identity_is_the_skeleton_the_packs_and_the_quotes_not_the_prose(self): w = _writer([]) a = w.target_identity(_target()) b = w.target_identity({**_target(), "visual_brief": "다름", "surface_form": "다름"}) c = w.target_identity({**_target(), "coarse_type_label": "다른 물건"}) d = w.target_identity({**_target(), "source_quotes": ["다른 문장"]}) assert a == b, "★묘사 산문은 신원이 아니다" assert a != c and a != d, "★뼈대·원고 인용은 신원이다 (Codex BLOCK 1)" assert w.identity_inputs["research_pack"] == gtr.RESEARCH_PACK_VERSION assert w.identity_inputs["pick_pack"] == ctp.PROMPT_PACK_VERSION def test_coordinates_are_prepended_when_the_model_left_them_out(self): got = gtr.brief_from_research( {"narrow_queries": ["질의 하나"], "rough_queries": ["질의 둘"], "search_directive_native": "그냥 찾아라", "appearance_criteria": "c"}, narrow=False, era=ERA, region=REGION, lock="한국어로 찾는다") assert got["search_directive_native"].startswith(f"{REGION} {ERA}") def test_the_packs_carry_no_fixture_nouns(self): """★팩은 일반 지시문 — 원고(fixture)의 낱말이 한 개도 없다.""" import sys sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from tests.grounding.fixtures import period_episode as fx words = set() for v in (getattr(fx, "EXPECTED_SCENE_PLACES", {}) or {}).values(): words.add(str(v[1])) pack = gtr.load_pack() text = str(pack["stems"][gtr.SYSTEM_STEM]["content"]) + json.dumps( pack["stems"][gtr.SCHEMA_STEM]["content"], ensure_ascii=False) pick = ctp.load_pack() text += str(pick["stems"][ctp.SYSTEM_STEM]["content"]) assert words, "★fixture 가 자리 낱말을 선언하지 않았다" assert not [w for w in words if w in text], "★팩에 원고 낱말이 들어갔다" def _search_returning(n): calls = {"n": 0} def _s(**kw): calls["n"] += 1 # ★라운드마다 다른 URL — 중복으로 버려지지 않게 return {"queries": [kw.get("terms_native")], "images": [{"image_url": f"https://x.invalid/r{calls['n']}_{i}.jpg", "thumbnail_url": "", "source_website_url": "https://x.invalid/s", "caption": ""} for i in range(1, n + 1)]} return _s def _download(url, dest, fallback_url=""): Path(dest).write_bytes(b"\x89PNG\r\n\x1a\n" + url.encode()) return True def _judge_none_match(sims): """기준에 맞는 것은 없지만 닮은 정도는 다르다 — 마지막엔 가장 닮은 것을 골라야 한다.""" seen = [] def _j(cands, criteria=""): seen.append(criteria) return {"j": {"verdicts": [ {"index": c["index"], "object_type_match": "unsure", "visible": True, "criteria_match": "no", "similarity": sims[(len(seen) - 1) % len(sims)][c["index"] - 1]} for c in cands]}} _j.seen = seen return _j class TestTheLastRoundAlwaysPicksOne: def test_no_criteria_match_in_two_rounds_still_selects_the_closest(self, tmp_path): judge = _judge_none_match([[10, 40, 20], [70, 30, 5]]) out = rr.acquire_one(_target(), workdir=tmp_path, rel_root=tmp_path, search=_search_returning(3), download=_download, judge=judge, write_brief=_writer([])) assert out["status"] == ra.STATUS_SELECTED, out.get("rounds") assert out["match_quality"] == "closest" and out["chosen"]["forced_pick"] is True assert out["chosen"]["round_no"] == 2 and out["chosen"]["index"] == 1, "★가장 닮은(70) 것이 아니다" assert out["rounds"][0]["criteria_used"] and out["rounds"][1]["criteria_used"] assert all(judge.seen), "★심판이 CRITERIA 를 못 받았다" assert out["rounds"][0]["research"]["appearance_criteria"] assert out["rounds"][1]["decision"]["next"] == ctp.NEXT_SELECT_CLOSEST assert out["outcome"] == ra.STATUS_SELECTED def test_a_criteria_match_in_round_one_ends_it(self, tmp_path): def _j(cands, criteria=""): return {"j": {"verdicts": [ {"index": c["index"], "object_type_match": "yes", "visible": True, "criteria_match": "yes" if c["index"] == 2 else "no", "similarity": 50} for c in cands]}} out = rr.acquire_one(_target(), workdir=tmp_path, rel_root=tmp_path, search=_search_returning(3), download=_download, judge=_j, write_brief=_writer([])) assert out["status"] == ra.STATUS_SELECTED and out["match_quality"] == "criteria" assert out["chosen"]["index"] == 2 and len(out["rounds"]) == 1 def test_nothing_downloaded_in_either_round_is_no_match(self, tmp_path): out = rr.acquire_one(_target(), workdir=tmp_path, rel_root=tmp_path, search=_search_returning(0), download=_download, judge=_judge_none_match([[1]]), write_brief=_writer([])) assert out["status"] == ra.STATUS_NO_MATCH and out["chosen"] is None def test_an_old_judge_without_the_criteria_argument_still_works(self, tmp_path): def _old(cands): return {"j": {"verdicts": [{"index": c["index"], "object_type_match": "yes", "visible": True} for c in cands]}} out = rr.acquire_one(_target(), workdir=tmp_path, rel_root=tmp_path, search=_search_returning(2), download=_download, judge=_old, write_brief=_writer([])) assert out["status"] == ra.STATUS_SELECTED and out["chosen"]["index"] == 1 class TestTheCombinerV2: def test_criteria_no_blocks_eligibility_but_closest_is_still_named(self): got = ctp.combine_coarse_verdicts({"j": {"verdicts": [ {"index": 1, "object_type_match": "yes", "visible": True, "criteria_match": "no", "similarity": 80}, {"index": 2, "object_type_match": "yes", "visible": True, "criteria_match": "unsure", "similarity": 20}, ]}}, 2) assert got["eligible"] == [] and got["chosen_index"] == 0 assert got["closest_index"] == 1 and got["closest_similarity"] == 80 def test_a_wrong_kind_by_all_judges_is_ranked_behind_the_rest(self): got = ctp.combine_coarse_verdicts({"j": {"verdicts": [ {"index": 1, "object_type_match": "no", "visible": True, "criteria_match": "no", "similarity": 95}, {"index": 2, "object_type_match": "unsure", "visible": True, "criteria_match": "no", "similarity": 10}, ]}}, 2) assert got["closest_index"] == 2 def test_v1_verdicts_without_the_new_fields_still_parse(self): got = ctp.combine_coarse_verdicts({"j": {"verdicts": [ {"index": 1, "object_type_match": "yes", "visible": True}]}}, 1) assert got["chosen_index"] == 1 and got["closest_index"] == 1 def test_the_last_round_decides_closest_only_when_candidates_exist(self): d = ctp.decide_next_round(ctp.MAX_ROUNDS, {"chosen_index": 0}, new_candidate_count=0, total_candidate_count=3) assert d["next"] == ctp.NEXT_SELECT_CLOSEST d0 = ctp.decide_next_round(ctp.MAX_ROUNDS, {"chosen_index": 0}, new_candidate_count=0, total_candidate_count=0) assert d0["next"] == ctp.NEXT_NO_MATCH class TestProductionWiring: def test_the_step_writer_is_the_research_writer(self): import ast import inspect import textwrap from app.core.steps import reference_acquisition_step as ras from app.core.steps.reference_acquisition_step import ReferenceAcquisitionStep as S # ★2026-09-03: 공장은 모듈 함수 `make_writer` — 스텝(19.3)과 야외 보충(21.915)이 같은 것을 부른다 step_src = textwrap.dedent(inspect.getsource(S._write_brief)) assert "make_writer(" in step_src and "gsb.make_writer(" not in step_src src = textwrap.dedent(inspect.getsource(ras.make_writer)) assert "gtr.make_writer(" in src and "gsb.make_writer(" not in src def test_the_production_judge_forwards_the_criteria(self): import inspect from app.core.steps import reference_acquisition_step as ras from app.core.steps.reference_acquisition_step import ReferenceAcquisitionStep as S assert "make_judge(" in inspect.getsource(S._judge) src = inspect.getsource(ras.make_judge) assert "criteria" in src and "CRITERIA" in src def test_the_research_prompt_carries_the_full_evidence(self): long = "긴 문장 " * 300 user = gtr.build_user({"owner_type": "prop", "coarse_type_label": "물건", "terms_native": ["갑"], "language_lock_native": "한국어", "era": ERA, "region": REGION}, [long]) assert long.strip() in user, "★근거를 잘랐다 — 절대 규칙 위반" class TestCodexCounterexamples: """★Codex BLOCK 1~3 (2026-09-03) 의 반례 끝점 넷.""" def test_a_changed_quote_changes_the_identity_and_reruns_the_research(self): """BLOCK 1 — 뼈대가 같아도 원고 인용이 바뀌면 신원이 바뀌고 조사가 다시 돈다.""" calls = [] w = _writer(calls) a = w.target_identity(_target()) b = w.target_identity({**_target(), "source_quotes": ["다른 뜻의 문장"]}) assert a != b, "★인용이 바뀌었는데 신원이 같다 — 옛 조사를 되쓴다" prior = _research(None, skeleton={"coarse_type_label": "물건"}, evidence=[]) prior["evidence_fingerprint"] = gtr.evidence_fingerprint(["원고 문장 하나", "원고 문장 둘"]) w({**_target("rs_a"), "_prior_research": prior}, narrow=True) assert calls == [], "★같은 근거면 되쓴다" w({**_target("rs_b"), "source_quotes": ["다른 뜻의 문장"], "_prior_research": prior}, narrow=True) assert len(calls) == 1, "★근거가 달라졌는데 옛 조사를 되썼다" @pytest.mark.parametrize("hole", ["what_it_is", "appearance_criteria", "search_directive_native", "rough_queries", "narrow_queries"]) def test_a_research_answer_with_an_empty_core_field_is_a_failure(self, hole): """BLOCK 2 — 핵심 칸이 비면 실패(재시도 가능)다. rough 를 narrow 로 대체하지 않는다.""" full = {"what_it_is": "a", "appearance_criteria": "b", "search_directive_native": "c", "narrow_queries": ["n"], "rough_queries": ["r"]} bad = {**full, hole: ("" if isinstance(full[hole], str) else [])} with pytest.raises(gtr.ResearchFailed): gtr.validate_research(bad) assert gtr.validate_research(full)["rough_queries"] == ["r"] def test_a_judge_that_raises_in_both_rounds_still_yields_one_photo(self, tmp_path): """BLOCK 3 — 심판이 두 번 다 터져도 받아 둔 bytes 가 있으면 selected 1장.""" def _boom(cands, criteria=""): raise RuntimeError("VLM 이 죽었다") out = rr.acquire_one(_target(), workdir=tmp_path, rel_root=tmp_path, search=_search_returning(2), download=_download, judge=_boom, write_brief=_writer([])) assert out["status"] == ra.STATUS_SELECTED assert out["chosen"]["forced_pick"] is True and out["chosen"]["forced_reason"] == "judge_unavailable" assert out["chosen"]["round_no"] == 1 and out["chosen"]["index"] == 1, "★검색 순위 첫 장이 아니다" assert "판정 실패" in out["rounds"][0]["decision"]["error"] assert len(out["rounds"]) == 2, "★1라운드 실패에서 멈췄다 — 넓혀 한 번 더 가야 한다" def test_a_failed_second_search_still_picks_from_the_first_round(self, tmp_path): """BLOCK 3 — 2라운드 검색이 죽어도 1라운드 후보 중 하나를 고른다.""" n = {"k": 0} def _search(**kw): n["k"] += 1 if n["k"] == 2: raise RuntimeError("검색 provider 가 죽었다") return _search_returning(2)(**kw) out = rr.acquire_one(_target(), workdir=tmp_path, rel_root=tmp_path, search=_search, download=_download, judge=_judge_none_match([[15, 60]]), write_brief=_writer([])) assert out["status"] == ra.STATUS_SELECTED and out["chosen"]["round_no"] == 1 assert out["chosen"]["index"] == 2, "★가장 닮은(60) 것이 아니다" def test_nothing_downloaded_and_a_failure_is_retryable_not_no_match(self, tmp_path): """받은 것이 0 이고 실패였으면 「못 봤다」— 「없다」가 아니다.""" def _search(**kw): raise RuntimeError("검색 provider 가 죽었다") out = rr.acquire_one(_target(), workdir=tmp_path, rel_root=tmp_path, search=_search, download=_download, judge=_judge_none_match([[1]]), write_brief=_writer([])) assert out["status"] == ra.STATUS_RETRYABLE and out["chosen"] is None class TestTheLanguageLockDoorIsGeneric: """★Codex ④ (2026-09-03): 우리 질의는 provider 앞에서 잠금 문장의 지배 문자 체계와 대조한다 — 언어 목록·regex 없음.""" def test_dominant_script_is_a_property_lookup_not_a_language_list(self): assert gtr.dominant_script("한글 낱말 셋") == "HANGUL" assert gtr.dominant_script("some latin words") == "LATIN" assert gtr.dominant_script("site:example.go.kr 됫박 곡물") == "HANGUL", "★연산자 글자를 세지 않는다" assert gtr.dominant_script("1960 !!") == "" def test_a_query_in_another_script_is_dropped_and_the_rest_kept(self): got = gtr.queries_in_lock("한국어로 찾는다", ["가나 다라", "quite english query", "site:a.b 마바"]) assert got["kept"] == ["가나 다라", "site:a.b 마바"] and got["dropped"] == ["quite english query"] def test_all_queries_in_another_script_is_a_retryable_failure(self): research = {"what_it_is": "w", "appearance_criteria": "c", "search_directive_native": "d", "narrow_queries": ["only english here"], "rough_queries": ["still english"], "sources": []} with pytest.raises(gtr.ResearchFailed): gtr.brief_from_research(research, narrow=False, era=ERA, region=REGION, lock="한국어로 찾는다") def test_no_readable_lock_means_no_door(self): got = gtr.queries_in_lock("", ["anything goes"]) assert got["kept"] == ["anything goes"] and got["dropped"] == [] # ★「ko」같은 부호는 문자 체계가 아니다 — 문을 안 건다 got2 = gtr.queries_in_lock("ko", ["가나다 질의"]) assert got2["kept"] == ["가나다 질의"] and got2["dropped"] == [] class TestTheLanguageDoorStandsOnTheFinalOutboundQuery: """Codex 2026-09-03: 기대 문자 체계는 언어 이름 표가 아니라 대상의 **구조화 잠금 + native terms** 에서, 검사는 **좌표를 지난 최종 outbound 질의**에.""" def test_native_terms_decide_the_script_when_the_lock_is_a_code(self): got = gtr.queries_in_lock("ko", ["a quite english query", "가나다 질의"], native_terms=["됫박", "되"]) assert got["kept"] == ["가나다 질의"] and got["dropped"] == ["a quite english query"] def test_the_door_checks_the_query_after_the_coordinates(self): research = {"narrow_queries": ["quite english query words here", "가나다 질의"], "rough_queries": [], "search_directive_native": "가나다 질의", "language_lock_native": "", "appearance_criteria": "x"} got = gtr.brief_from_research(research, narrow=False, era=ERA, region=REGION, lock="ko", native_terms=["됫박"]) assert got["search_terms_native"] == [f"{REGION} {ERA} — 가나다 질의"] assert got["language_lock_dropped"] == [f"{REGION} {ERA} — quite english query words here"] def test_no_basis_means_no_door(self): got = gtr.queries_in_lock("ko", ["anything goes"], native_terms=[]) assert got["kept"] == ["anything goes"] and got["dropped"] == [] class TestAStopRequestIsNotASearchFailure: """★실측 4398a55dc0bb (2026-09-03 05:08): 정지 요청(step.cancelled)이 「검색 실패」로 접혀 두 라운드가 실패로 적히고 대상 9개가 why 빈 retryable 로 남았다. 멈추라는 말은 그대로 올린다 — 규칙은 `run_control.is_abort` 한 곳.""" def test_a_cancel_from_the_search_propagates(self, tmp_path): from app.core.errors import AppError def _search(**_kw): raise AppError(code="step.cancelled", message="정지 요청", status_code=409) with pytest.raises(AppError) as e: rr.acquire_one(_target(), workdir=tmp_path, rel_root=tmp_path, search=_search, download=_download, judge=_judge_none_match([[10]]), write_brief=_writer([])) assert e.value.code == "step.cancelled" def test_an_ordinary_search_error_still_folds_into_a_retryable_row(self, tmp_path): def _search(**_kw): raise RuntimeError("provider 5xx") out = rr.acquire_one(_target(), workdir=tmp_path, rel_root=tmp_path, search=_search, download=_download, judge=_judge_none_match([[10]]), write_brief=_writer([])) assert out["status"] == ra.STATUS_RETRYABLE assert "검색 실패" in out["rounds"][-1]["decision"]["why"] def test_the_row_level_reason_comes_from_the_last_round_when_the_top_is_empty(self): from app.modules.pipeline import grounding_central_acquisition as ca got = {"why": "", "rounds": [{"decision": {"why": "첫 라운드"}}, {"decision": {"why": "다 못 봤다 — 검색 실패"}}]} assert ca.row_why(got) == "다 못 봤다 — 검색 실패" assert ca.row_why({"why": "위에 있음", "rounds": [{"decision": {"why": "x"}}]}) == "위에 있음" assert ca.row_why({"why": "", "rounds": []}) == ""