"""GROUNDING-V2 §2-3.5 — 분류·계획 스텝.
★`classify_samples` 를 **실제로 부르는 첫 자리**다. 그 전까지는
「production 용 API 를 정의한 것」이었다.
"""
import inspect
import pathlib
from unittest.mock import patch
import pytest
from app.core.errors import AppError
from app.core.steps.grounding_steps import (
DEFAULT_SAMPLES, GroundingPlanStep, check_provenance)
from app.modules.pipeline.grounding_classifier import DEFAULT_JUDGES
_ENT = {"props": [{"short_id": "P01", "name": "회수권 뭉치", "description": "종이"}],
"characters": [], "locations": []}
_RULES = {"era": "1983년", "region": "대한민국"}
def _fp(**over):
base = {"payload_hash": "h1", "prompt_version": "v1",
"requested_judge_alias": "gpt",
"judge_model_alias": "gpt", "judge_physical_model": "openai/gpt-5.6-sol"}
base.update(over)
# 부탁한 판정자를 따로 안 적으면 실제 것과 같다고 본다 (손 fixture 편의)
if "judge_model_alias" in over and "requested_judge_alias" not in over:
base["requested_judge_alias"] = over["judge_model_alias"]
return base
def _panel_return(recs):
"""★대역 반환도 **계약대로** 만든다 — 판정자 수·표본 수·지문이 다 맞아야 한다.
손으로 한 판정자짜리를 돌려주면 「패널이 무너졌다」로 막힌다. 계약이
바뀌면 이 대역도 따라가게 `DEFAULT_JUDGES` 에서 만든다.
"""
runs = [_fp(requested_judge_alias=a, judge_model_alias=a,
judge_physical_model=f"phys/{a}")
for a in DEFAULT_JUDGES for _ in range(DEFAULT_SAMPLES)]
return {"by_subject": recs, "runs": runs, "judges": list(DEFAULT_JUDGES),
"logical_calls": DEFAULT_SAMPLES * len(DEFAULT_JUDGES),
"search_calls": 0}
def _downstream(out):
"""★하류가 **실제로 보는 것**을 돌려준다.
`run()` 은 ``save_checkpoint({"status": ..., **result})`` 로 펼치고,
`_load_prev_checkpoint` 는 그 manifest 를 그대로 읽는다. 그래서 하류가
읽는 자리는 ``cp["data"]`` 다. `_execute` 반환을 그대로 들여다보면
**평평하게 돌려줘서 `data` 가 없는 결함**을 못 잡는다 — 실제로 그랬다.
"""
cp = {"status": "completed", **out}
return cp.get("data", {})
def _step(mode, *, entities=_ENT, rules=_RULES):
s = GroundingPlanStep.__new__(GroundingPlanStep)
s.project_config = {"project_id": "p", "grounding_mode": mode}
s.project_id, s.episode_id = "p", "e"
s.build_opik_metadata = lambda: {}
s._load_prev_checkpoint = lambda sid: (
{"data": entities} if sid == "entity_merge"
else {"data": rules} if sid == "visual_world_rules" else None)
return s
class TestProvenanceGate:
"""★기록만 하면 「같은 것을 N회 쟀다」가 성립하는지 아무도 안 본다."""
def test_matching_runs_pass(self):
assert check_provenance([_fp(), _fp(), _fp()]) == []
def test_no_runs_fails(self):
assert check_provenance([]) != []
@pytest.mark.parametrize("col", ["payload_hash", "prompt_version",
"judge_physical_model"])
def test_a_differing_column_fails(self, col):
bad = check_provenance([_fp(), _fp(**{col: "다름"})])
assert any(col in b for b in bad)
def test_two_judges_are_normal_but_each_must_be_consistent(self):
"""★판정자가 **여럿**일 수 있다(패널) — alias 가 다른 것은 정상이다.
대신 **판정자마다** 나머지가 한 값이어야 한다. 안 그러면 「같은 것을
N회 쟀다」가 그 판정자 안에서 성립하지 않는다.
"""
panel = [_fp(judge_model_alias="gpt", judge_physical_model="openai/x"),
_fp(judge_model_alias="gpt", judge_physical_model="openai/x"),
_fp(judge_model_alias="gemini-pro",
judge_physical_model="google/y"),
_fp(judge_model_alias="gemini-pro",
judge_physical_model="google/y")]
assert check_provenance(panel) == []
broken = list(panel)
broken[1] = _fp(judge_model_alias="gpt", judge_physical_model="openai/z")
bad = check_provenance(broken)
assert any("gpt" in b and "judge_physical_model" in b for b in bad), bad
@pytest.mark.parametrize("col", ["payload_hash", "prompt_version",
"requested_judge_alias",
"judge_model_alias", "judge_physical_model"])
def test_an_empty_column_fails(self, col):
bad = check_provenance([_fp(**{col: None})])
assert any(col in b for b in bad)
class TestLegacyDoesNothing:
"""★legacy 는 이 스텝이 **없는 것과 같아야** 한다."""
def test_no_llm_call_and_no_output(self):
calls = {"n": 0}
import app.modules.pipeline.grounding_classifier as gc
def _cs(*a, **k):
calls["n"] += 1
return {"by_subject": {}, "runs": [], "logical_calls": 0,
"search_calls": 0}
with patch.object(gc, "classify_samples", _cs):
out = _downstream(_step("legacy")._execute())
assert calls["n"] == 0, "legacy 인데 분류기를 불렀다"
assert out["decided"] == [] and out["skipped"] is True
def test_legacy_does_not_even_need_world_context(self):
"""★legacy 는 시대가 없어도 안 선다 — 아무것도 안 하니까."""
out = _downstream(_step("legacy", rules={})._execute())
assert out["skipped"] is True
class TestV2WritesWhatTheFilterReads:
"""★`v2` 에서만 `decided` 를 쓴다.
★`shadow_plan` 갈래 시험은 걷었다 — manifest 의 `if_grounding_v2` 가
shadow 에서 이 스텝을 not_applicable 로 걸러 **프로덕션에서 도달할 수
없는 경로**가 됐다. 도달 못 하는 것을 시험하면 「막았다」는 거짓 안심이
된다. shadow 가 진짜로 아무것도 안 건드리는지는
`tests/core/test_grounding_v2_production_chain.py` 가 공개 `run()` 으로
파일·step_run·프롬프트 셋 다 잰다.
"""
def _run(self, mode):
import app.modules.pipeline.grounding_classifier as gc
def _cs(subjects, **k):
recs = {s["research_subject_id"]: [
{"research_subject_id": s["research_subject_id"],
"grounding_class": "externally_grounded", "discriminability": "yes",
"referent_specificity": "exact_variant", "difficulty": "hard",
"confidence": 0.9, "visibility_intent": "yes", "locale": "KR",
"generation": "1980s", "visible_discriminators": ["x"],
"likely_failure_modes": ["y"], "generation_difficulty": "not_hard", "target_image_provider": "gemini",
"target_image_model": "m", "target_image_model_version": "1"}
] * (DEFAULT_SAMPLES * len(DEFAULT_JUDGES)) for s in subjects}
return _panel_return(recs)
with patch.object(gc, "classify_samples", _cs):
return _downstream(_step(mode)._execute())
def test_v2_writes_the_key_the_filter_reads(self):
out = self._run("v2")
assert len(out["decided"]) == 1
assert out["decided"][0]["route"] == "research"
assert out["decided"][0]["_short_id"] == "P01"
def test_no_search_is_bought(self):
assert self._run("v2")["search_calls"] == 0
def test_samples_are_folded_not_single(self):
out = self._run("v2")
# ★판정자가 여럿이면 호출 수는 **표본 수 × 판정자 수**다
assert out["logical_calls"] == DEFAULT_SAMPLES * len(DEFAULT_JUDGES)
# ★한 대상의 표본은 **판정자마다 N개**를 다 합친 것이다
assert (out["decided"][0]["sample_count"]
== DEFAULT_SAMPLES * len(DEFAULT_JUDGES))
class TestFailClosed:
def test_missing_world_context_stops(self):
"""★빈 시대로 판정하면 분류기가 상상 묘사만 보고 답한다 — 실측 결함."""
with pytest.raises(AppError) as e:
_step("v2", rules={})._execute()
assert e.value.code == "grounding_plan.missing_world_context"
def test_provenance_mismatch_stops(self):
import app.modules.pipeline.grounding_classifier as gc
def _cs(subjects, **k):
return {"by_subject": {}, "runs": [_fp(), _fp(payload_hash="다름")],
"logical_calls": 2, "search_calls": 0}
with patch.object(gc, "classify_samples", _cs):
with pytest.raises(AppError) as e:
_step("v2")._execute()
assert e.value.code == "grounding_plan.provenance_mismatch"
def test_no_subjects_is_not_an_error(self):
out = _downstream(_step("v2", entities={"props": [], "characters": [],
"locations": []})._execute())
assert out["subject_count"] == 0 and out["decided"] == []
class TestStepIsActuallyDispatchable:
"""★manifest 에만 넣고 **실행기 표에 안 이으면** 불러도 404 다.
실제로 그랬다 — `STEP_CLASSES` 에 없어서
`get_step_runner("grounding_plan", ...)` 가 `step.not_found` 를 냈다.
사전(dict) 을 들여다보는 검사는 이걸 못 잡는다. 프로덕션이 부르는
**그 함수**를 태운다.
"""
def test_production_dispatch_returns_the_step(self):
from app.services.analysis_dispatch_service import get_step_runner
runner = get_step_runner(
step_id="grounding_plan", project_id="p", episode_id="e",
db=None, project_config={})
assert isinstance(runner, GroundingPlanStep)
assert runner.step_id == "grounding_plan"
def test_manifest_and_dispatch_agree(self):
"""manifest 에 있는 active step 은 전부 부를 수 있어야 한다."""
from app.core.step_manifest import STEP_MANIFEST
from app.services.analysis_dispatch_service import get_step_runner
assert STEP_MANIFEST["grounding_plan"]["lifecycle"] == "active"
get_step_runner(step_id="grounding_plan", project_id="p",
episode_id="e", db=None, project_config={})
class TestFilterFailsClosedWithoutAPlan:
"""★`v2` 인데 계획이 없으면 **거르면 안 된다.**
「없으면 안 더한다」는 legacy 에서는 맞지만 `v2` 에서는 fail-open 이다 —
계획이 안 돌았을 뿐인데 조사 대상이 조용히 지워진다.
"""
@staticmethod
def _filter_step(mode, *, plan_cp):
from app.core.steps.entity_steps import EntityFilterStep
s = EntityFilterStep.__new__(EntityFilterStep)
s.project_config = {"grounding_mode": mode} if mode else {}
s.project_id, s.episode_id = "p", "e"
s.build_opik_metadata = lambda: {}
s._load_cleaned_text = lambda: "본문"
cps = {"entity_merge": {"data": _ENT}, "grounding_plan": plan_cp}
s._load_prev_checkpoint = lambda sid: cps.get(sid)
return s
def test_v2_without_plan_checkpoint_raises(self):
s = self._filter_step("v2", plan_cp=None)
with pytest.raises(AppError) as e:
s._execute()
assert e.value.code == "entity_filter.grounding_plan_missing"
def test_v2_with_a_legacy_plan_checkpoint_raises(self):
"""모드를 바꿔 놓고 옛 계획을 그대로 읽으면 안 된다."""
s = self._filter_step("v2", plan_cp={"data": {"mode": "legacy",
"decided": []}})
with pytest.raises(AppError) as e:
s._execute()
assert e.value.code == "entity_filter.grounding_plan_stale"
@pytest.mark.parametrize("mode", [None, "legacy", "shadow_plan"])
def test_other_modes_do_not_require_a_plan(self, mode):
"""★legacy 는 이 게이트가 **없는 것과 같아야** 한다."""
s = self._filter_step(mode, plan_cp=None)
with patch("app.modules.pipeline.entity_filter"
".filter_low_frequency_entities") as f:
f.return_value = {"filtered": _ENT}
s._execute()
assert f.call_count == 1
assert f.call_args.kwargs["protected_short_ids"] is None
class TestApplicabilityIsResolvedBothWays:
"""★모드 판정 자리가 **둘**이다 — 실행 중(runner)과 정적 평가(stub).
★``shadow_plan`` 도 **안 돈다** — production step id 를 쓰는 스텝이라
한 번 돌리면 DAG 하류가 무효화된다. shadow 관측은 저장 CP 만 읽는
offline 재생기(§2-3a)가 한다.
stub 에는 ``project_config`` 가 없다. ENV 만 보면 project_config 로 켠
프로젝트를 legacy 로 오판하고, 프로젝트 설정을 못 읽으면 dispatcher 가
선행으로 안 세운다. 두 갈래를 다 잰다.
"""
@pytest.mark.parametrize("mode,want", [
("legacy", False), ("shadow_plan", False), ("v2", True), (None, False),
])
def test_runner_path(self, mode, want):
from app.core.applicability import _if_grounding_v2
class _R:
project_config = {"grounding_mode": mode} if mode else {}
project_id = "p"
assert _if_grounding_v2(_R()) is want
@pytest.mark.parametrize("env,want", [
("v2", "applicable"), ("shadow_plan", "not_applicable"),
("legacy", "not_applicable"), (None, "not_applicable"),
])
def test_static_path_reads_env_when_there_is_no_config(
self, env, want, monkeypatch):
"""★`_StubRunner` 에는 ``project_config`` 가 없다 — 여기서 죽으면 안 된다."""
from app.core.applicability import evaluate_step_applicability
if env is None:
monkeypatch.delenv("GROUNDING_MODE", raising=False)
else:
monkeypatch.setenv("GROUNDING_MODE", env)
for sid in ("grounding_a0", "grounding_plan"):
assert evaluate_step_applicability(sid, "p-no-such", "e") == want
def test_a_broken_value_is_treated_as_on(self, monkeypatch):
"""★오타를 legacy 로 삼키면 「켰는데 안 바뀐다」를 며칠 쫓는다."""
from app.core.applicability import _if_grounding_v2
class _R:
project_config = {"grounding_mode": "v2 "} # 없는 값
project_id = "p"
monkeypatch.setattr(
"app.core.grounding_mode.resolve_grounding_mode",
lambda cfg: (_ for _ in ()).throw(ValueError("없는 값")))
assert _if_grounding_v2(_R()) is True
def test_the_rule_name_is_registered(self):
"""★미등록 규칙이면 `resolve_applicability` 가 ValueError 를 낸다."""
from app.core.applicability import APPLICABILITY_VALIDATORS
from app.core.step_manifest import STEP_MANIFEST
for sid in ("grounding_a0", "grounding_plan"):
rule = STEP_MANIFEST[sid]["applicability"]
assert rule in APPLICABILITY_VALIDATORS, f"{sid}: {rule} 미등록"
class TestCarryIsOwnerSafeAndOneToOne:
"""★잘못 결속한 id 는 **다른 대상의 근거를 물려받는다.**
표면형만 보고 붙이면 사람과 그 사람이 입은 것이 같은 subject 가 되고,
분류기는 엉뚱한 원문 문장을 근거로 판정한다.
"""
@staticmethod
def _c(surface, owner="prop"):
return {"surface_form": surface, "owner_type": owner,
"research_subject_id": f"{owner}:{surface}",
"source_quote": "원문", "source_anchor": "SEG-001"}
def _carry(self, *cands):
from app.core.steps.grounding_steps import build_carry_index
return build_carry_index(list(cands))
def test_same_owner_matches(self):
from app.core.steps.grounding_steps import match_candidate
c = self._carry(self._c("고무줄로 묶인 회수권 뭉치"))
got, why = match_candidate(c, "회수권 뭉치", "prop")
assert why == "matched" and got["research_subject_id"] == \
"prop:고무줄로 묶인 회수권 뭉치"
def test_a_different_owner_never_matches(self):
"""★같은 말이라도 owner 가 다르면 다른 대상이다."""
from app.core.steps.grounding_steps import match_candidate
c = self._carry(self._c("감색 차장 제복", owner="outlook"))
assert match_candidate(c, "감색 차장 제복", "character") == (None, "none")
assert match_candidate(c, "감색 차장 제복", "prop") == (None, "none")
assert match_candidate(c, "감색 차장 제복", "outlook")[1] == "matched"
def test_two_candidates_matching_one_name_bind_to_neither(self):
"""★「가장 긴 것을 고른다」는 임의 선택이다 — 안 붙인다."""
from app.core.steps.grounding_steps import match_candidate
c = self._carry(self._c("종이 승차권"), self._c("종이 승차권 뭉치"))
assert match_candidate(c, "종이 승차권 뭉치", "prop") == (None, "ambiguous")
def test_a_repeated_surface_form_binds_to_neither(self):
"""★A0 가 같은 말을 두 번 적으면 색인이 아무 쪽이나 갖는다."""
from app.core.steps.grounding_steps import match_candidate
a, b = self._c("요금통"), self._c("요금통")
b["research_subject_id"] = "prop:요금통#2"
c = self._carry(a, b)
assert match_candidate(c, "요금통", "prop") == (None, "duplicate_surface")
def test_no_match_is_not_an_error(self):
from app.core.steps.grounding_steps import match_candidate
assert match_candidate(self._carry(self._c("요금통")),
"손수건", "prop") == (None, "none")
def test_a_candidate_without_owner_or_surface_is_not_indexed(self):
c = self._carry({"surface_form": "", "owner_type": "prop"},
{"surface_form": "x", "owner_type": ""})
assert c["index"] == {}
class TestStaleGateCoversPackAndModel:
"""★지문이 안 움직이면 **resume 이 옛 체크포인트를 그대로 건너뛴다.**
고친 것이 아무 데도 안 닿는다 — 이 저장소에서 두 번 낸 결함이다
(`test_published_pack_ratchet` 참조).
"""
@staticmethod
def _step(cls, sid, cfg=None):
s = cls.__new__(cls)
s.project_config = cfg or {"grounding_mode": "v2"}
s.project_id, s.episode_id, s.step_id = "p", "e", sid
return s
def _a0(self, **kw):
from app.core.steps.grounding_steps import GroundingA0Step
return self._step(GroundingA0Step, "grounding_a0", **kw)
def _plan(self, **kw):
from app.core.steps.grounding_steps import GroundingPlanStep
return self._step(GroundingPlanStep, "grounding_plan", **kw)
def test_both_steps_produce_a_hash(self):
assert len(self._a0()._config_hash()) == 16
assert len(self._plan()._config_hash()) == 16
assert self._a0()._config_hash() != self._plan()._config_hash()
def test_pack_bytes_are_folded_in(self, monkeypatch):
"""★버전 문자열만 접으면 **같은 디렉토리를 고쳤을 때** 못 잡는다."""
import app.core.steps.grounding_steps as g
before = self._a0()._config_hash()
real = g._pack_fingerprint
def _shifted(module, version):
out = dict(real(module, version))
out["system"] = "달라진바이트"
return out
monkeypatch.setattr(g, "_pack_fingerprint", _shifted)
assert self._a0()._config_hash() != before
def test_an_unreadable_pack_stops_instead_of_going_constant(self):
"""★삼키면 지문이 **상수**가 된다 — 고치려던 결함 그 자체다."""
import app.core.steps.grounding_steps as g
with pytest.raises(AppError) as e:
g._pack_fingerprint("grounding_a0", "그런버전없음")
assert e.value.code == "grounding.pack_unreadable"
def test_schemas_are_read_as_schemas(self):
"""★스키마를 ``kind="prompt"`` 로 읽으면 못 찾고, 그걸 삼키면 상수가 된다."""
import app.core.steps.grounding_steps as g
from app.modules.pipeline import grounding_a0 as a0
from app.modules.pipeline import grounding_classifier as gc
for mod, ver, stem in (("grounding_a0", a0.PROMPT_PACK_VERSION,
"a0_schema"),
("grounding_classify", gc.PROMPT_PACK_VERSION,
"classify_schema")):
fp = g._pack_fingerprint(mod, ver)
assert fp[stem] and "unreadable" not in fp
def test_sample_count_is_folded_in(self, monkeypatch):
"""★표본 수를 바꾸면 판정이 달라진다."""
import app.core.steps.grounding_steps as g
before = self._plan()._config_hash()
monkeypatch.setattr(g, "DEFAULT_SAMPLES", DEFAULT_SAMPLES + 2)
assert self._plan()._config_hash() != before
def test_the_image_model_is_no_longer_folded_in(self, monkeypatch):
"""★★**이미지 모델을 바꿔도 분류를 다시 사지 않는다** (2026-08-30).
분류가 그 좌표를 아예 안 보게 됐다(`difficulty` 폐기). 그런데 지문에
남겨 두면 backend 를 바꾼 것만으로 에피소드 전체 분류를 다시 산다.
"""
import app.core.steps.grounding_steps as g
assert not hasattr(g, "_target_image_coords"), \
"좌표 helper 가 남아 있으면 누군가 다시 지문에 넣는다"
h = self._plan()._config_hash()
assert isinstance(h, str) and h
def test_planner_contract_is_folded_in(self, monkeypatch):
import app.modules.pipeline.grounding_planner as gp
before = self._plan()._config_hash()
monkeypatch.setattr(gp, "PLANNER_CONTRACT_VERSION",
gp.PLANNER_CONTRACT_VERSION + 1)
assert self._plan()._config_hash() != before
def test_grounding_mode_still_moves_it(self):
"""★기존 project_config 지문도 그대로 접혀 있어야 한다."""
assert (self._plan(cfg={"grounding_mode": "v2"})._config_hash()
!= self._plan(cfg={"grounding_mode": "shadow_plan"})._config_hash())
class TestAPanelThatCollapsedIsNotAPanel:
"""★production 은 ``strict_single_attempt=False`` 라 tier fallback 으로
`gpt` 슬롯이 gemini 로 넘어갈 수 있다. 그러면 **실제** alias 는 둘 다
gemini 이고, 「판정자마다 한 값」만 보는 gate 는 그걸 통과시킨다 (Codex).
한 모델 패널을 두 모델 패널이라고 세면, 이 판에서 얻은 「어긋남 0」이
통째로 뜻을 잃는다. 그래서 **부탁한 것**과 대조하고, 무너지면 막는다.
"""
@staticmethod
def _panel(**over):
gpt = dict(requested_judge_alias="gpt", judge_model_alias="gpt",
judge_physical_model="openai/gpt-5.6-sol")
gem = dict(requested_judge_alias="gemini-pro",
judge_model_alias="gemini-pro",
judge_physical_model="google/gemini-3.1-pro")
gem.update(over)
return [_fp(**gpt), _fp(**gpt), _fp(**gem), _fp(**gem)]
def test_a_healthy_panel_passes(self):
assert check_provenance(self._panel(),
expected_judges=["gpt", "gemini-pro"],
expected_samples=2) == []
def test_a_collapse_to_one_physical_model_is_blocked(self):
"""gemini 슬롯이 fallback 으로 **gpt 물리 모델**을 썼다."""
bad = check_provenance(
self._panel(judge_model_alias="gpt",
judge_physical_model="openai/gpt-5.6-sol"),
expected_judges=["gpt", "gemini-pro"], expected_samples=2)
assert any("한 모델로 무너졌다" in b for b in bad), bad
def test_a_missing_judge_is_blocked(self):
runs = [r for r in self._panel()
if r["requested_judge_alias"] == "gpt"]
bad = check_provenance(runs, expected_judges=["gpt", "gemini-pro"],
expected_samples=2)
assert any("판정자 집합이 다르다" in b for b in bad), bad
def test_an_uneven_sample_count_is_blocked(self):
runs = self._panel()[:3]
bad = check_provenance(runs, expected_judges=["gpt", "gemini-pro"],
expected_samples=2)
assert any("표본이 1개" in b for b in bad), bad
def test_the_real_fingerprint_has_the_column_the_gate_reads(self):
"""★손으로 만든 fixture 가 아니라 **분류기가 실제로 내는 지문**을 먹인다.
이 판에서 이미 한 번 당했다 — fixture 에 옛 칸을 넣어 둬서 앞뒤가
끊긴 것을 63개 시험이 전부 놓쳤다.
"""
from unittest.mock import patch
from app.modules.pipeline import grounding_classifier as gc
subj = {"research_subject_id": "rs_1", "surface_form": "x",
"owner_type": "prop", "source_quote": "q"}
def _call(**kw):
sink = kw.get("usage_sink")
if sink is not None:
sink.update({"alias": "gpt", "physical_model": "openai/x"})
return {"classifications": [{
"research_subject_id": "rs_1", "grounding_class": "generic",
"discriminability": "no", "referent_specificity": "generic_class",
"difficulty": "easy", "confidence": 0.9,
"visibility_intent": "yes", "locale": "KR",
"generation": "1980s", "visible_discriminators": ["a"],
"likely_failure_modes": ["b"], "generation_difficulty": "not_hard", "rationale": "c"}]}
with patch.object(gc, "_call_structured", _call):
out = gc.classify_samples(
[subj], samples=1, judges=["gpt"], strict_single_attempt=True)
runs = out["runs"]
assert runs and "requested_judge_alias" in runs[0], runs
assert runs[0]["requested_judge_alias"] == "gpt"
# ★gate 가 읽는 칸이 실제 지문에 **다 있다** — 비었다는 사유가 안 나온다
bad = check_provenance(runs, expected_judges=out["judges"],
expected_samples=1)
assert not [b for b in bad if "비었다" in b], bad
def test_every_judge_column_the_gate_reads_reaches_the_fingerprint(self):
"""★칸을 하나 더해 놓고 **지문에 복사하는 줄**을 안 고치면 조용히 빈다.
지문은 `judge` dict 를 통째로 넣지 않고 **칸을 골라 복사**한다.
실제로 그래서 `requested_judge_alias` 가 지문에 안 실렸다.
"""
import inspect
from app.modules.pipeline import grounding_classifier as gc
src = inspect.getsource(gc.classify)
gate = inspect.getsource(check_provenance)
for col in ("requested_judge_alias", "judge_model_alias",
"judge_physical_model"):
assert f'"{col}": judge[' in src, f"{col} 이 지문에 안 실린다"
assert col in gate
def test_the_expected_panel_comes_from_the_contract_not_the_return(self):
"""★기대 집합을 호출이 돌려준 `smp["judges"]` 에서 받으면 **자기보고**다.
한 판정자만 돌고 그 목록도 한 판정자로 줄면 그대로 통과한다 (Codex).
계약(`DEFAULT_JUDGES`)에서 받아야 한다.
"""
import inspect
from app.core.steps.grounding_steps import GroundingPlanStep
src = inspect.getsource(GroundingPlanStep)
i = src.index("check_provenance(")
seg = src[i:i + 500]
assert "expected_judges=list(DEFAULT_JUDGES)" in seg, seg
assert 'smp.get("judges")' not in seg, "자기보고를 계약으로 쓴다"
# 돌아온 목록이 계약과 다르면 그것도 막는다
assert "계약과 다르다" in src
class TestOneJudgeSelfReportIsFailClosed:
"""★끝점에서 잰다 — 분류기가 「나는 한 판정자로 돌았다」고 스스로 보고해도
스텝이 **막아야** 한다. 기대 집합은 계약(`DEFAULT_JUDGES`)이 정본이다.
"""
def _run_with(self, ret):
import app.modules.pipeline.grounding_classifier as gc
with patch.object(gc, "classify_samples", lambda subjects, **k: ret):
return _step("v2")._execute()
@staticmethod
def _recs(sid="P01"):
return {sid: [{"research_subject_id": sid,
"grounding_class": "externally_grounded",
"discriminability": "yes",
"referent_specificity": "exact_variant",
"difficulty": "hard", "confidence": 0.9,
"visibility_intent": "yes", "locale": "KR",
"generation": "1980s", "visible_discriminators": ["x"],
"likely_failure_modes": ["y"], "generation_difficulty": "not_hard",
"target_image_provider": "gemini",
"target_image_model": "m",
"target_image_model_version": "1"}]
* (DEFAULT_SAMPLES * len(DEFAULT_JUDGES))}
def test_one_judge_that_reports_itself_as_the_whole_panel_is_blocked(self):
"""★`judges` 를 스스로 한 판정자로 줄여 보고해도 통과하면 안 된다."""
one = [_fp(requested_judge_alias="gpt", judge_model_alias="gpt",
judge_physical_model="phys/gpt")] * DEFAULT_SAMPLES
with pytest.raises(AppError) as e:
self._run_with({"by_subject": self._recs(), "runs": one,
"judges": ["gpt"],
"logical_calls": DEFAULT_SAMPLES,
"search_calls": 0})
assert "판정자" in str(e.value)
def test_a_contract_shaped_panel_passes(self):
"""★positive control — 계약대로면 이 gate 가 안 문다."""
out = self._run_with(_panel_return(self._recs()))
assert out["data"]["decided"]
class TestTheJudgePanelIsInTheFingerprint:
"""★판정자를 바꿔도 지문이 안 움직이면 resume 이 **옛 판정을 그대로
건너뛴다.** 누가 답했는지가 바뀌었는데 다시 안 묻는 것이다 (Codex).
"""
def _hash(self, judges):
from unittest.mock import patch
from app.modules.pipeline import grounding_classifier as gc
s = _step("v2")
with patch.object(gc, "DEFAULT_JUDGES", tuple(judges)):
return s._config_hash()
def test_changing_the_panel_moves_the_fingerprint(self):
a = self._hash(("gpt", "grok"))
b = self._hash(("gpt", "gemini-pro"))
assert a != b, "판정자를 바꿨는데 지문이 그대로다"
def test_the_same_panel_gives_the_same_fingerprint(self):
assert self._hash(("gpt", "grok")) == self._hash(("gpt", "grok"))
def test_the_panel_order_is_not_an_identity(self):
"""★접기가 순서에 안 흔들리는데 지문만 순서를 세면, **같은 판정이
나오는데 지문이 달라 다시 산다.**
`classify_samples` 는 판정자를 순서대로 다 부르고 결과를 합칠 뿐이라
첫 판정자에게 우선권이 없다. `decide_route_from_samples` 도 실측에서
순서에 안 흔들렸다.
"""
assert self._hash(("gpt", "grok")) == self._hash(("grok", "gpt"))
def test_folding_really_is_order_independent(self):
"""★위 주장의 근거 — 접기가 정말 순서에 안 흔들리는지 태워서 본다."""
from app.modules.pipeline import grounding_planner as gp
def _rec(j):
return {"grounding_class": "externally_grounded",
"discriminability": "yes",
"referent_specificity": "exact_variant",
"difficulty": "hard", "confidence": 0.9,
"visibility_intent": "yes", "locale": "KR",
"generation": "1980s", "visible_discriminators": ["x"],
"likely_failure_modes": ["y"], "generation_difficulty": "not_hard",
"target_image_provider": "g", "target_image_model": "m",
"target_image_model_version": "1", "judge_alias": j}
recs = [_rec("gpt"), _rec("grok")]
assert gp.decide_route_from_samples(recs)["route"] == \
gp.decide_route_from_samples(list(reversed(recs)))["route"]
def test_the_classifier_contract_is_folded_too(self):
import inspect
from app.core.steps.grounding_steps import GroundingPlanStep
src = inspect.getsource(GroundingPlanStep._config_hash)
assert "classifier_contract" in src
assert "DEFAULT_JUDGES" in src
class TestTheContractNamesFieldsThatActuallyExist:
"""★계약 문서가 **없는 칸 이름**을 적어 두면, 다음 사람이 그 이름으로 코드를
쓰고 조용히 `None` 을 읽는다. 이 판에서 실제로 그랬다 — 문서는
`judge_model/version` 이라 적혀 있었는데 코드는 셋으로 갈라져 있었다.
★문서 전체를 시험하지 않는다. **판정자 좌표 세 칸**만 본다 — 이름이 갈리면
route 근거를 못 되짚는 자리라서.
"""
_DOC = (pathlib.Path(__file__).resolve().parents[3]
/ "docs" / "design" / "2026-08-29-grounding-v2-contract.md")
def test_the_judge_coordinates_in_the_doc_are_the_ones_the_code_writes(self):
from app.modules.pipeline import grounding_classifier as gc
doc = self._DOC.read_text(encoding="utf-8")
for field in ("judge_step", "judge_model_alias", "judge_physical_model"):
assert f"`{field}`" in doc, f"계약에 {field} 가 없다"
assert field in inspect.getsource(gc), f"코드에 {field} 가 없다"
def test_the_old_names_are_gone_from_the_doc(self):
"""★positive control — 옛 이름이 남아 있으면 이 시험이 잡는다."""
doc = self._DOC.read_text(encoding="utf-8")
assert "`judge_model/version`" not in doc
#: ★폐기한 규칙이 **살아 있는 문장**으로 남으면, 다음 사람이 그것을 현행으로
#: 읽는다. 이 판에서 실제로 그랬다 — 계약 앞쪽은 A AND B 를 확정해 놓고
#: 뒤쪽 세 자리가 capability bypass·Sol 단독·`쉬움까지 건너뜀` 을 그대로
#: 규정하고 있었다. ★문서 전체를 재지 않는다 — **폐기한 절단선 문구**만.
_RETIRED_PHRASES = ("쉬움까지 건너뜀", "쉬움 → 건너뜀",
"초기 판정자는 **Sol**", "팩 · Sol 버전")
def test_the_retired_cutoff_is_not_stated_as_current(self):
doc = self._DOC.read_text(encoding="utf-8")
live = doc.split("")[0] + "".join(
part.split(" ")[-1] for part in doc.split("")[1:])
found = [ph for ph in self._RETIRED_PHRASES if ph in live]
assert found == [], f"폐기한 규칙이 살아 있는 문장으로 남았다: {found}"
def test_the_history_block_still_keeps_them(self):
"""★positive control — 지운 게 아니라 **접어 둔 것**이어야 한다.
통째로 지우면 왜 폐기했는지가 사라져 같은 안이 다시 올라온다.
"""
doc = self._DOC.read_text(encoding="utf-8")
assert "" in doc and "폐기한 옛 규칙" in doc
assert "capability" in doc.split("")[1].split(" ")[0]