"""★야외 구조물 형태 참조의 **보충 획득** (설계 §8 · HITL 0 · 2026-09-03 · Codex BLOCK 2·3 반영). 의무는 결정적(ID 로만) · 기존 보충 CP 는 의무를 막지 않는다(되쓰기·재시도·신원은 ca.run/장부) · 구조 오류는 provider 앞에서 · 별도 CP append-only(유효 결과가 달라지면 supersede) · merge view 하나 · 사람 대기 없음. 무료 · 바깥 호출 0.""" from __future__ import annotations import ast import hashlib import inspect import json import textwrap from pathlib import Path import pytest from app.modules.pipeline import grounding_outdoor_supplement as gos from app.modules.pipeline import reference_acquisition as ra from app.modules.pipeline.grounding_entity_contract import PRODUCER_PAYLOAD from app.modules.pipeline.grounding_reference_obligations import OBLIGATION_STRUCTURE_FORM def _loc_row(fid, *, kind="context", outcome=ra.STATUS_SELECTED, path=None, ident=None): led = {"research_subject_id": f"{fid}#{kind}", "owner_type": "location", "final_id": fid, "parent_final_id": None, "purpose": None if kind == OBLIGATION_STRUCTURE_FORM else kind, "obligation_kind": kind if kind == OBLIGATION_STRUCTURE_FORM else None, "covers": [fid], "status": "bound", "screen": "obligation", PRODUCER_PAYLOAD: {"coarse_type_label": "어떤 곳", "search_terms_native": ["어떤 곳"], "language_lock_native": "한국어"}, "source_evidence": {"surface_form": "어떤 곳", "source_quote": "원고 문장"}} acq = {"status": outcome, "chosen": {"path": path} if path else None, "chosen_path": path or "", "candidates": [], "rounds": []} return {"research_subject_id": led["research_subject_id"], "identity": ident or f"id-{led['research_subject_id']}", "disposition": "acquired", "status": outcome, "outcome": ra.acquisition_outcome(outcome), "ledger_row": led, "acquisition": acq} def _front(rows): return {"status": "completed", "data": {"rows": rows}} def _t(gid, loc, desc="이층 목조 앞면"): return {"gid": gid, "loc_id": loc, "loc_ids": [loc], "structure_desc": desc} class TestObligationsAreDeterministic: def test_one_obligation_per_location_without_a_base_structure_form_row(self): front = _front([_loc_row("L01"), _loc_row("L02")]) led = gos.structure_form_obligations(front, [_t("g1", "L01"), _t("g2", "L01"), _t("g3", "L02", "단층 벽돌")]) sids = [r["research_subject_id"] for r in led["rows"]] assert sids == ["L01#structure_form", "L02#structure_form"] r = led["rows"][0] assert r["obligation_kind"] == OBLIGATION_STRUCTURE_FORM and r["purpose"] is None assert r[PRODUCER_PAYLOAD]["coarse_type_label"] == "이층 목조 앞면" assert r["source_evidence"]["source_quote"] == "원고 문장", "★원고 근거를 잃었다" assert r["covers"] == ["L01"] and r["supplement_for_group"] == "g1" def test_a_base_structure_form_row_means_no_obligation(self): front = _front([_loc_row("L01"), _loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM)]) assert gos.structure_form_obligations(front, [_t("g1", "L01")])["rows"] == [] def test_an_existing_supplement_row_does_not_block_the_obligation(self): """★Codex BLOCK 2: 옛 보충 줄(retryable 이든 selected 든)은 의무를 막지 않는다 — ca.run/장부가 판단한다.""" front = _front([_loc_row("L01")]) led = gos.structure_form_obligations(front, [_t("g1", "L01")]) assert [r["research_subject_id"] for r in led["rows"]] == ["L01#structure_form"] def test_structural_errors_are_found_before_any_purchase(self): """★Codex BLOCK 3: 장소 0/여럿 · 서술 결손 · 중앙에 장소 줄 없음 — 전부 한 번에, provider 앞에서.""" front = _front([_loc_row("L01")]) problems = gos.validate_targets(front, [ {"gid": "g0", "loc_ids": [], "loc_id": "", "structure_desc": "x"}, {"gid": "g2", "loc_ids": ["L01", "L02"], "loc_id": "", "structure_desc": "x"}, {**_t("g3", "L01"), "structure_desc": ""}, _t("g9", "L09")]) assert len(problems) == 4 and all("g" in p for p in problems) with pytest.raises(ValueError): gos.structure_form_obligations(front, [_t("g9", "L09")]) class TestTheSupplementIsAppendOnly: def test_same_identity_and_same_result_is_a_noop(self): old = _loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM) first = gos.merge_supplement(None, [old], config_hash="h") same = gos.merge_supplement(first, [old], config_hash="h") assert len(same["data"]["rows"]) == 1 and same["data"]["live_count"] == 1 def test_same_identity_but_a_recovered_result_supersedes(self): """★Codex BLOCK 2: 같은 신원의 incomplete/retryable 이 selected 로 회복되면 옛 줄을 살리지 않는다.""" old = _loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM, outcome=ra.STATUS_RETRYABLE) first = gos.merge_supplement(None, [old], config_hash="h") new = _loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM, path="p.png") # 같은 identity · selected second = gos.merge_supplement(first, [new], config_hash="h") rows = second["data"]["rows"] assert len(rows) == 2 and rows[0]["superseded_by"] and "superseded_by" not in rows[1] assert second["data"]["live_count"] == 1 def test_a_new_identity_supersedes(self): old = _loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM) first = gos.merge_supplement(None, [old], config_hash="h") third = gos.merge_supplement(first, [{**old, "identity": "id-다른"}], config_hash="h") assert [bool(r.get("superseded_by")) for r in third["data"]["rows"]] == [True, False] def test_two_live_rows_for_one_subject_stop(self): a = _loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM) broken = {"data": {"rows": [a, {**a, "identity": "id-2"}]}} with pytest.raises(gos.SupplementConflict): gos.merge_supplement(broken, [a], config_hash="h") class TestTheMergeView: def test_base_plus_live_supplement_and_a_conflict_stops(self): front = _front([_loc_row("L01")]) supp = gos.merge_supplement(None, [_loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM)], config_hash="h") rows = gos.outdoor_reference_rows(front, supp) assert [r["research_subject_id"] for r in rows] == ["L01#context", "L01#structure_form"] clash = _front([_loc_row("L01"), _loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM)]) with pytest.raises(gos.SupplementConflict): gos.outdoor_reference_rows(clash, supp) def _step(tmp_path, monkeypatch): from app.core.config import settings import app.core.steps.outdoor_structure_form_reference_step as m monkeypatch.setattr(settings, "projects_dir", str(tmp_path / "projects")) st = m.OutdoorStructureFormReferenceStep.__new__(m.OutdoorStructureFormReferenceStep) st.project_id, st.episode_id = "p", "e" st.project_config = {"grounding_mode": "v2_chunk"} st._config_hash = lambda: "cfg" st._load_prev_checkpoint = lambda sid: {} st.db = _FakeDB() return st class _FakeDB: """★투영이 자산 행을 **묶는지** 보려고 — id 로 찾고(insert-or-verify) add·commit 만 센다.""" def __init__(self): self.rows = {} self.commits = 0 def query(self, model): db = self class _Q: def __init__(self): self.f = {} def filter_by(self, **kw): self.f = kw return self def first(self): cands = [r for r in db.rows.values() if all(getattr(r, k, None) == v for k, v in self.f.items())] return cands[0] if cands else None return _Q() def add(self, row): self.rows[row.id] = row def commit(self): self.commits += 1 def flush(self): pass class TestTheProjectionReadsTheMergeView: def test_a_selected_supplement_row_becomes_the_form_reference(self, tmp_path, monkeypatch): st = _step(tmp_path, monkeypatch) rel = "projects/p/references/grounding/e/L01#structure_form_r1_01.png" f = tmp_path / rel f.parent.mkdir(parents=True); f.write_bytes(b"\x89PNG\r\n\x1a\nFORM") supp = gos.merge_supplement(None, [_loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM, path=rel)], config_hash="h") got = st._project_front_checkpoint( _front([_loc_row("L01")]), target_gids=["g1"], universe=["g1"], plate_gids=["g1"], no_spec=[], building_groups={"g1": {"members": [{"loc_id": "L01", "is_indoor": False}]}}, supplement_cp=supp) one = got["data"]["groups"]["g1"] assert one["status"] == "ok" and one["form_ref_sha256"] == hashlib.sha256(b"\x89PNG\r\n\x1a\nFORM").hexdigest() assert one["audit"]["research_subject_id"] == "L01#structure_form" # ★2026-09-03 09:06 실측: 자산 id 가 없으면 씨드가 fail-closed 로 선다 — 투영도 legacy 와 같은 행을 묶는다 row = st.db.rows[one["form_ref_asset_id"]] assert (row.asset_type, row.entity_id, row.variant_type, row.project_id, row.episode_id) == ( "structure_form_ref", "g1", "form_ref", "p", "e") assert row.file_path == one["form_ref_path"] and st.db.commits >= 1 def test_the_same_selection_binds_the_same_asset_row_on_resume(self, tmp_path, monkeypatch): """★재개마다 새 UUID 를 만들면 씨드의 input_image_ids 계보가 갈린다 — 같은 sha 는 같은 id·행 하나.""" st = _step(tmp_path, monkeypatch) rel = "projects/p/references/grounding/e/L01#structure_form_r1_01.png" f = tmp_path / rel f.parent.mkdir(parents=True); f.write_bytes(b"\x89PNG\r\n\x1a\nFORM") supp = gos.merge_supplement(None, [_loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM, path=rel)], config_hash="h") kw = dict(target_gids=["g1"], universe=["g1"], plate_gids=["g1"], no_spec=[], building_groups={"g1": {"members": [{"loc_id": "L01", "is_indoor": False}]}}, supplement_cp=supp) a = st._project_front_checkpoint(_front([_loc_row("L01")]), **kw)["data"]["groups"]["g1"]["form_ref_asset_id"] b = st._project_front_checkpoint(_front([_loc_row("L01")]), **kw)["data"]["groups"]["g1"]["form_ref_asset_id"] assert a == b and len(st.db.rows) == 1 # 다른 사진(sha)이면 새 행 — 기존 행은 그대로 f.write_bytes(b"\x89PNG\r\n\x1a\nOTHER") supp2 = gos.merge_supplement(None, [_loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM, path=rel)], config_hash="h2") c = st._project_front_checkpoint(_front([_loc_row("L01")]), **{**kw, "supplement_cp": supp2})["data"]["groups"]["g1"]["form_ref_asset_id"] assert c != a and len(st.db.rows) == 2 and st.db.rows[a].file_path == str(tmp_path / rel) def test_without_a_structure_form_row_it_stops_naming_the_location(self, tmp_path, monkeypatch): from app.core.errors import AppError st = _step(tmp_path, monkeypatch) with pytest.raises(AppError) as exc: st._project_front_checkpoint( _front([_loc_row("L01")]), target_gids=["g1"], universe=["g1"], plate_gids=["g1"], no_spec=[], building_groups={"g1": {"members": [{"loc_id": "L01", "is_indoor": False}]}}, supplement_cp=None) assert "L01" in str(exc.value.message) and "structure_form" in str(exc.value.message) class TestTheSupplementBuysThroughTheCentralBoundary: def test_it_calls_ca_run_with_the_same_factories_and_the_stop_check(self): import app.core.steps.outdoor_structure_form_reference_step as m src = textwrap.dedent(inspect.getsource(m.OutdoorStructureFormReferenceStep._supplement_structure_forms)) tree = ast.parse(src) called = {ast.unparse(n.func) for n in ast.walk(tree) if isinstance(n, ast.Call)} for want in ("ca.run", "ras.make_search", "ras.make_download", "ras.make_judge", "ras.make_writer", "gos.validate_targets", "gos.structure_form_obligations", "gos.merge_supplement", "atomic_write_json"): assert want in called, f"★{want} 을 안 부른다" assert "search_reference_images" not in called and "_run_group" not in called, "★야외가 제 손으로 산다" run_call = next(n for n in ast.walk(tree) if isinstance(n, ast.Call) and ast.unparse(n.func) == "ca.run") assert "stop_check" in {k.arg for k in run_call.keywords}, "★사용자 중단이 워커에 안 닿는다" assert "write_text" not in src, "★비원자 덮어쓰기" def test_the_central_step_uses_the_same_factories_and_keeps_its_replay_journal(self): from app.core.steps import reference_acquisition_step as ras for name in ("_judge", "_write_brief", "_search", "_download"): src = textwrap.dedent(inspect.getsource(getattr(ras.ReferenceAcquisitionStep, name))) called = {ast.unparse(n.func) for n in ast.walk(ast.parse(src)) if isinstance(n, ast.Call)} assert any(c.startswith("make_") for c in called), f"★{name} 이 공장을 안 쓴다: {called}" # ★Codex BLOCK 1 (2026-09-03): 공장 추출 때 이 상수를 잃었다 — 재판정 경로가 AttributeError 로 죽는다 assert ras.ReferenceAcquisitionStep.REPLAY_JOURNAL == "journal_replay.json" def test_structural_errors_stop_before_ca_run(self, tmp_path, monkeypatch): from app.core.errors import AppError from app.modules.pipeline import grounding_central_acquisition as ca monkeypatch.setattr(ca, "run", lambda *a, **k: (_ for _ in ()).throw(AssertionError("★ca.run 이 불렸다"))) st = _step(tmp_path, monkeypatch) with pytest.raises(AppError) as exc: st._supplement_structure_forms( _front([_loc_row("L01")]), target_gids=["g1", "g2"], building_groups={"g1": {"members": [{"loc_id": "L01", "is_indoor": False}]}, "g2": {"members": [{"loc_id": "L01", "is_indoor": False}, {"loc_id": "L05", "is_indoor": False}]}}, spec_groups={"g1": {"spec": {}}, "g2": {"spec": {}}}, locations_by_id={}) assert exc.value.code == "outdoor_form_reference.supplement_targets_invalid" def test_a_base_structure_form_row_means_provider_zero(self, tmp_path, monkeypatch): from app.modules.pipeline import grounding_central_acquisition as ca import app.core.steps.outdoor_structure_form_reference_step as m monkeypatch.setattr(ca, "run", lambda *a, **k: (_ for _ in ()).throw(AssertionError("★ca.run 이 불렸다"))) monkeypatch.setattr(m, "derive_seed_inputs", lambda **k: {"structure_desc": "이층 목조"}, raising=False) st = _step(tmp_path, monkeypatch) import app.modules.pipeline.outdoor_structure_seed as oss monkeypatch.setattr(oss, "derive_seed_inputs", lambda **k: {"structure_desc": "이층 목조"}) got = st._supplement_structure_forms( _front([_loc_row("L01"), _loc_row("L01", kind=OBLIGATION_STRUCTURE_FORM)]), target_gids=["g1"], building_groups={"g1": {"members": [{"loc_id": "L01", "is_indoor": False}]}}, spec_groups={"g1": {"spec": {}}}, locations_by_id={}) assert got is None class TestTheEndpointsCodexNamed: """실제 ca.run(대역 provider) 으로 보충 CP 가 어떻게 움직이나.""" def _run_supp(self, tmp_path, front, existing, *, match, desc="이층 목조 앞면", search_fails=False): 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 grounding_target_research as gtr from tests.grounding.test_central_acquisition import _Spy led = gos.structure_form_obligations(front, [_t("g1", "L01", desc)]) spy = _Spy(match=match) if search_fails: # ★provider 실패 — 두 라운드 다 받은 것 0 이면 retryable(못 봤다) 이고 장부에 incomplete 로 남는다. # (`_Spy(match=False)` 는 5단계 계약상 closest 로 selected 가 된다 — Codex NON-BLOCK 2026-09-03) def _boom(**kw): spy.searched += 1 raise RuntimeError("검색 provider 가 죽었다") spy.search = _boom def _rc(client, *, skeleton, evidence, **_kw): k = skeleton["coarse_type_label"] return {"what_it_is": k, "appearance_criteria": f"- {k}", "narrow_queries": [f"r e {k}"], "rough_queries": [f"r e {k}"], "search_directive_native": f"r e {k}", "language_lock_native": "한국어", "sources": [], "provenance": {}} w = gtr.make_writer(world_facts="w", source_text="s", era="e", region="r", research_call=_rc, client=object()) got = ca.run(led, journal=cj.ChunkJournal(tmp_path / "supp_j.json", contract={"v": 1}), cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge, write_brief=w, workers=1) return gos.merge_supplement(existing, got["rows"], config_hash="h"), spy def test_a_retryable_row_is_retried_and_the_recovered_selected_supersedes_it(self, tmp_path): front = _front([_loc_row("L01")]) first, spy1 = self._run_supp(tmp_path, front, None, match=True, search_fails=True) live = [r for r in first["data"]["rows"] if not r.get("superseded_by")] assert len(live) == 1 and spy1.searched >= 1 assert live[0]["status"] == ra.STATUS_RETRYABLE, "★첫 판이 정말 retryable 이어야 이 시험이 뜻이 있다" second, spy2 = self._run_supp(tmp_path, front, first, match=True) rows = second["data"]["rows"] assert spy2.searched > 0, "★retryable 을 다시 안 샀다" assert second["data"]["live_count"] == 1 and rows[-1]["outcome"] == ra.STATUS_SELECTED assert rows[0]["superseded_by"] and "superseded_by" not in rows[-1] def test_an_unchanged_selected_row_costs_nothing_and_appends_nothing(self, tmp_path): front = _front([_loc_row("L01")]) first, _ = self._run_supp(tmp_path, front, None, match=True) second, spy2 = self._run_supp(tmp_path, front, first, match=True) assert spy2.searched == 0, "★같은 신원을 다시 샀다" assert len(second["data"]["rows"]) == len(first["data"]["rows"]), "★같은 결과를 또 덧붙였다" def test_a_changed_description_is_a_new_identity_and_a_new_purchase(self, tmp_path): front = _front([_loc_row("L01")]) first, _ = self._run_supp(tmp_path, front, None, match=True) second, spy2 = self._run_supp(tmp_path, front, first, match=True, desc="단층 벽돌 앞면") assert spy2.searched >= 1 assert [bool(r.get("superseded_by")) for r in second["data"]["rows"]] == [True, False]