"""form_reference 스텝의 **라운드 배선** 회귀 (A4). 계획 = docs/superpowers/plans/2026-08-02-a4-round-immutable-candidate-store.md ## 무엇을 지키는가 A4 의 실체는 "후보 파일 덮어쓰기"보다 크다 — **계보 소급 변조**다. 옛 `_upsert_ref_asset` 은 그룹당 한 row 를 찾아 `file_path` 를 갱신했고, 씨드의 `input_image_ids` 가 그 asset_id 를 가리키므로 **재검색 한 번이 과거 씨드의 입력 간선을 다른 이미지로 바꿨다.** 그래서 라운드 디렉터리와 자산 UUID 를 불변으로 두고, 기존 row 는 고치지 않고 검증만 한다. ## 왜 `_execute` 를 태우는가 지난 두 세션에 같은 실수를 했다 — private helper 만 부르는 테스트로 "이 경로가 이것을 막는다"를 주장했고, 앞단이 세어지지 않아 **스스로 반증되지 않았다.** 여기서는 스텝의 실질 진입점인 ``_execute`` 를 태우고 유료 경로 (``_run_group``)의 **호출 횟수**로 단언한다. ## 픽스처 규칙 **프로젝트 루트 안** 형태로 세운다(root/projects/...). 루트 밖 경로는 `to_relative_image_path` 가 입력을 그대로 돌려줘 경로 계약 결함을 가린다 (2026-08-01 실측). 그리고 유료 경로 대역은 `**kwargs` 로 받지 않는다 — 실제 시그니처가 바뀌면 **TypeError 로 터져야** 한다. 삼키면 라운드 디렉터리를 안 쓰는데도 통과한다. """ from __future__ import annotations import hashlib from pathlib import Path from typing import Any, Dict, List import pytest from app.core.errors import AppError from app.modules.pipeline.form_ref_rounds import ( LEGACY_CONTRACT_VERSION, LEGACY_ROUND_ID, ROUND_STORAGE_CONTRACT_VERSION, RoundState, abandon_round, list_rounds, load_round, open_round, transition_round, ) GID = "g_alpha" #: 새 저장 계약(v2)은 provenance header 를 **필수**로 요구한다. PROV = {"ref_pack_version": "sample-fixture-pack", "target_policy_version": "sample-fixture-policy"} # ── 최소 DB 대역 ──────────────────────────────────────────────────── class _FakeQuery: def __init__(self, rows: List[Any]): self._rows = rows self._filters: Dict[str, Any] = {} def filter_by(self, **kw): self._filters = kw return self def first(self): for row in self._rows: if all(getattr(row, k, None) == v for k, v in self._filters.items()): return row return None class _FakeDB: def __init__(self) -> None: self.rows: List[Any] = [] self.commits = 0 def query(self, *_a): return _FakeQuery(self.rows) def add(self, row): self.rows.append(row) def flush(self): pass def commit(self): self.commits += 1 def rollback(self): pass def _sha_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() class _Wired: """`_execute` 를 반복 호출할 수 있는 최소 배선 묶음.""" def __init__(self, step, root: Path, cps: Dict[str, Any] | None = None): self.step = step self.root = root self.cps: Dict[str, Any] = cps if cps is not None else {} self.cp: Dict[str, Any] | None = None self.calls: List[Dict[str, Any]] = [] @property def group_dir(self) -> Path: return self.step._ref_dir(GID) def dir_for(self, gid: str) -> Path: return self.step._ref_dir(gid) def add_group(self, gid: str) -> None: """대상 그룹을 하나 더 붙인다 (그룹 간 격리를 보려면 둘 이상 필요).""" self.cps["outdoor_lane_plan"]["data"]["groups"][gid] = { "status": "ok", "plan": {"shot_bindings": [{"lane": "structure_plate"}]}} self.cps["outdoor_place_spec"]["data"]["groups"][gid] = { "spec": {"items": [{"kind": "sample-fixture-item"}]}} def run(self, mode: str = "resume") -> Dict[str, Any]: return self.step._execute(mode=mode) def run_and_save_cp(self, mode: str = "resume") -> Dict[str, Any]: out = self.run(mode) self.cp = out return out @pytest.fixture() def wired(tmp_path: Path, monkeypatch) -> _Wired: from app.core.config import settings from app.core.steps.outdoor_structure_form_reference_step import ( OutdoorStructureFormReferenceStep, ) root = tmp_path / "the-road-root" projects = root / "projects" projects.mkdir(parents=True, exist_ok=True) monkeypatch.setattr("app.core.file_paths._resolve_root", lambda: root) monkeypatch.setattr(settings, "projects_dir", str(projects), raising=False) monkeypatch.setattr(settings, "outdoor_lane_pipe_enabled", True, raising=False) monkeypatch.setattr(settings, "outdoor_lane_plan_enabled", True, raising=False) monkeypatch.setattr(settings, "outdoor_seed_all_groups_enabled", False, raising=False) cps: Dict[str, Any] = { "outdoor_lane_plan": {"data": {"groups": { GID: {"status": "ok", "plan": { "shot_bindings": [{"lane": "structure_plate"}]}}}}}, "outdoor_place_spec": {"data": {"groups": { GID: {"spec": {"items": [{"kind": "sample-fixture-item"}]}}}}}, "background_classify": {"data": {"building_groups": []}}, "entity_merge": {"data": {"locations": []}}, "scene_save": {"data": {"segments": [ {"text": "sample fixture scene text"}]}}, "visual_world_rules": None, } step = OutdoorStructureFormReferenceStep.__new__( OutdoorStructureFormReferenceStep) step.project_id, step.episode_id = "P", "E" step.db = _FakeDB() step._load_prev_checkpoint = lambda sid: cps.get(sid) # type: ignore w = _Wired(step, root, cps) step.load_checkpoint = lambda: w.cp # type: ignore monkeypatch.setattr( "app.modules.pipeline.outdoor_structure_seed.derive_seed_inputs", lambda **_kw: {"structure_desc": "sample fixture structure"}) monkeypatch.setattr( "app.core.steps.shot_conti_light_step._resolve_openai_client", lambda: object()) monkeypatch.setattr( "app.core.world_context.build_world_facts_block", lambda _cp: "") return w def stub_run_group(w: _Wired, monkeypatch, *, fail_attempts: int = 0, candidates_per_attempt: int = 2): """유료 경로 대역 — 후보를 **실제로 디스크에 쓰고** 1장을 고른다. 호출마다 바이트를 다르게 만들어(라운드 번호+시도 번호) "라운드마다 다른 bytes 가 보존되는가"를 공허하지 않게 만든다. """ def _fake(*, group_id: str, structure_desc: str, world_facts_block: str, source_text: str, client: Any, candidate_dir: Path, start_index: int = 1, narrow: bool = False, typology: Dict[str, Any] | None = None, # 이 대역은 라운드 CP 로직만 시험한다 — 심판 머리말 인자 # (pick_structure_desc 등) 는 여기서 소비하지 않는다. **_ignored: Any) -> Dict[str, Any]: n = len(w.calls) + 1 w.calls.append({"group_id": group_id, "narrow": narrow, "candidate_dir": candidate_dir, "start_index": start_index}) candidate_dir.mkdir(parents=True, exist_ok=True) written: List[Path] = [] for i in range(candidates_per_attempt): dest = candidate_dir / f"cand_{start_index + i:02d}.png" dest.write_bytes(f"sample-fixture-png-call{n}-c{i}".encode()) written.append(dest) if n <= fail_attempts: return {"status": "failed", "group_id": group_id, "error": "sample fixture failure", "candidate_count": len(written), "audit": {"pack_version": "sample-fixture-pack"}} chosen = written[0] return {"status": "ok", "group_id": group_id, "form_ref_path": str(chosen), "form_ref_sha256": _sha_bytes(chosen.read_bytes()), "chosen_prompt_used": "sample fixture directive", "candidate_count": len(written), "audit": {"pack_version": "sample-fixture-pack"}} monkeypatch.setattr(w.step, "_run_group", _fake) # ══════════════════════════════════════════════════════════════════════ # 1. 지문 계산이 살아 있는가 # ══════════════════════════════════════════════════════════════════════ # ★19fb2e65 가 `_config_hash` 의 물리 모델 블록을 `_group_fingerprint` 에 # 복사하면서 `from app.core.config import settings` 를 빠뜨렸다. 모듈 전역 # 에도 settings 가 없어 **호출 즉시 NameError** 였고, 어떤 테스트도 그 # 함수를 태우지 않아 드러나지 않았다. 그 자리는 `_execute` 의 그룹 try 안 # 이라 결과는 "전 그룹 failed" — 유료 실행이 통째로 죽는다. def test_group_fingerprint_is_computable(): from app.core.steps.outdoor_structure_form_reference_step import ( OutdoorStructureFormReferenceStep as S, ) fp = S._group_fingerprint( None, structure_desc="sample fixture structure", world_facts_block="", source_text_sha="deadbeef", research_input_sha="cafebabe") assert isinstance(fp, str) and len(fp) == 16 def test_fingerprint_failure_does_not_silently_fail_every_group( wired, monkeypatch): """★유료 경로가 **실제로 불렸는지 횟수로** 확인한다.""" stub_run_group(wired, monkeypatch) out = wired.run() assert len(wired.calls) == 1, f"유료 경로가 {len(wired.calls)}회 불렸다" assert out["completed_count"] == 1 assert out["failed_count"] == 0 # ══════════════════════════════════════════════════════════════════════ # 2. 후보가 라운드 안에만 산다 # ══════════════════════════════════════════════════════════════════════ def test_candidates_are_written_inside_the_round_directory( wired, monkeypatch): stub_run_group(wired, monkeypatch) wired.run() used = wired.calls[0]["candidate_dir"] assert used == wired.group_dir / "rounds" / "r001" assert (used / "cand_01.png").is_file() # 그룹 디렉터리 바로 아래(구 flat 경로)에는 아무것도 쓰지 않는다 flat = [p for p in wired.group_dir.glob("cand_*.png")] assert flat == [], f"구 flat 경로에 썼다: {flat}" def test_run_group_itself_downloads_into_the_given_round_directory( wired, monkeypatch): """★위 회귀는 `_run_group` 을 대역으로 두므로 **실제 다운로드 경로**를 검증하지 못한다. 여기서는 그 함수를 직접 태워 후보가 정말 라운드 안에 떨어지는지 본다 — 검색·심판만 대역이고 저장 경로는 실물이다. """ from app.modules.pipeline import search_grounded_ref as sgr saved: List[Path] = [] def _fake_download(url, dest: Path, fallback=""): dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(f"sample-fixture-{dest.name}".encode()) saved.append(dest) return True def _fake_call_structured(step_key, _system, _user, _schema, **_kw): if step_key == "structure_form_ref_brief": # ★[v13] 대상별 **시각 권위**가 계약에 들어왔다 — 이 칸이 없으면 # `_run_group` 이 「사진을 살지 말지 정할 수 없다」로 선다. # 여기서 재는 것은 저장 경로지만, 대역이 프로덕션 계약을 못 # 채우면 그 앞에서 멈춰 **경로에 닿지도 못한다.** return {"visual_authority": "observable_real", "search_directive_native": "sample fixture directive", "search_terms_native": ["sample", "fixture"]} return {"verdicts": [ {"index": 1, "usable": True, "score": 40, "reason_ko": "ok"}, {"index": 2, "usable": True, "score": 90, "reason_ko": "better"}, ]} monkeypatch.setattr(sgr, "download_candidate", _fake_download) monkeypatch.setattr(sgr, "search_reference_images", lambda *_a, **_kw: {"images": [ {"image_url": f"https://example.invalid/{i}.jpg", "caption": "c", "source_website_url": "s"} for i in range(2)]}) monkeypatch.setattr(sgr, "load_brief_system", lambda: "sys") # ★[v14] 심판 지시문은 **권위마다 다르다** — 대역도 그 인자를 받아야 # 한다. 안 받으면 「저장 경로를 잰다」는 이 시험이 심판 앞에서 선다. monkeypatch.setattr(sgr, "load_pick_system", lambda **_kw: "sys") monkeypatch.setattr(sgr, "load_narrow_retry_hint", lambda: "hint") monkeypatch.setattr(sgr, "build_search_brief_user", lambda **_kw: "u") monkeypatch.setattr(sgr, "build_search_brief_schema", lambda: {}) monkeypatch.setattr(sgr, "build_pick_user_head", lambda **_kw: "u") monkeypatch.setattr(sgr, "build_pick_schema", lambda _n: {}) monkeypatch.setattr("app.modules.llm.llm_client.call_structured", _fake_call_structured) monkeypatch.setattr("app.modules.pipeline.multiroll_gemini.png_part", lambda _b: {"type": "image"}) round_dir = wired.group_dir / "rounds" / "r001" rec = wired.step._run_group( group_id=GID, structure_desc="d", world_facts_block="", source_text="t", client=object(), candidate_dir=round_dir, start_index=5) assert rec["status"] == "ok" assert [p.name for p in saved] == ["cand_05.png", "cand_06.png"] for p in saved: assert p.parent == round_dir # 구 flat 경로에는 아무것도 쓰지 않는다 assert list(wired.group_dir.glob("cand_*.png")) == [] # 점수가 높은 2번을 골랐고 그 파일이 라운드 안에 있다 assert Path(rec["form_ref_path"]).parent == round_dir assert rec["candidate_count"] == 2 def test_narrow_retry_appends_instead_of_overwriting(wired, monkeypatch): """★좁혀 재검색하는 2차 시도가 1차 후보를 덮어쓰면 감사 근거가 사라진다.""" stub_run_group(wired, monkeypatch, fail_attempts=1, candidates_per_attempt=2) wired.run() assert [c["start_index"] for c in wired.calls] == [1, 3] round_dir = wired.group_dir / "rounds" / "r001" names = sorted(p.name for p in round_dir.glob("cand_*.png")) assert names == ["cand_01.png", "cand_02.png", "cand_03.png", "cand_04.png"] # 1차 후보의 바이트가 그대로 남아 있다 assert (round_dir / "cand_01.png").read_bytes().endswith(b"call1-c0") # ══════════════════════════════════════════════════════════════════════ # 3. force 는 반드시 새 라운드 — 이전 라운드는 통째로 보존 # ══════════════════════════════════════════════════════════════════════ def test_force_opens_a_new_round_and_preserves_the_previous_one( wired, monkeypatch): """계획 §3.1 — force 2회가 서로 다른 bytes·path·UUID 를 보존한다.""" stub_run_group(wired, monkeypatch) first = wired.run_and_save_cp() second = wired.run_and_save_cp(mode="force") e1 = first["data"]["groups"][GID] e2 = second["data"]["groups"][GID] assert e1["round_id"] == "r001" and e2["round_id"] == "r002" assert e1["form_ref_path"] != e2["form_ref_path"] assert e1["form_ref_sha256"] != e2["form_ref_sha256"] assert e1["form_ref_asset_id"] != e2["form_ref_asset_id"] # ★1회차 파일이 **그대로 살아 있다** p1 = Path(e1["form_ref_path"]) assert p1.is_file() assert _sha_bytes(p1.read_bytes()) == e1["form_ref_sha256"] # 두 라운드가 각자 자기 디렉터리를 갖는다 assert sorted(r.round_id for r in list_rounds(wired.group_dir)) == [ "r001", "r002"] def test_checkpoint_projection_carries_the_full_contract(wired, monkeypatch): """계획 §2.2 — CP entry 는 {round_id, asset_id, path, sha256, pack, policy} 를 **typed 필드로** 갖는다. 감사 근거를 audit 안에 묻어 두면 되짚을 때 파싱이 필요하고, 그룹 공통 `data.target` 에만 두면 팩이 바뀐 뒤 이 산출이 무엇으로 만들어졌는지 알 수 없다. """ from app.modules.pipeline.search_grounded_ref import ( TARGET_POLICY_VERSION, resolve_ref_pack_version, ) stub_run_group(wired, monkeypatch) out = wired.run() e = out["data"]["groups"][GID] assert e["round_id"] == "r001" assert e["path_kind"] == "round" assert e["round_contract_version"] == ROUND_STORAGE_CONTRACT_VERSION assert e["form_ref_asset_id"] assert e["form_ref_path"] and e["form_ref_sha256"] assert e["ref_pack_version"] == resolve_ref_pack_version() assert e["target_policy_version"] == TARGET_POLICY_VERSION # journal 도 같은 산출을 갖는다 — 재투영이 CP 와 어긋나지 않는다 stored = load_round(wired.group_dir, "r001").result for key in ("form_ref_asset_id", "form_ref_path", "form_ref_sha256", "round_id", "ref_pack_version", "target_policy_version"): assert stored[key] == e[key] def test_a_second_run_does_not_mutate_the_first_assets_path( wired, monkeypatch): """★옛 upsert 의 소급 변조 — 같은 row 의 file_path 를 갱신하던 경로.""" stub_run_group(wired, monkeypatch) first = wired.run_and_save_cp() a1 = first["data"]["groups"][GID]["form_ref_asset_id"] path1 = first["data"]["groups"][GID]["form_ref_path"] wired.run_and_save_cp(mode="force") rows = {r.id: r for r in wired.step.db.rows} assert len(rows) == 2, "라운드마다 새 자산이어야 한다" from app.core.file_paths import resolve_image_path assert str(resolve_image_path(rows[a1].file_path)) == str( resolve_image_path(path1)), "1회차 자산의 경로가 바뀌었다" # ══════════════════════════════════════════════════════════════════════ # 4. 재개 — 열린 라운드를 이어가고 UUID 를 또 만들지 않는다 # ══════════════════════════════════════════════════════════════════════ def test_resume_continues_an_open_round_without_a_second_uuid( wired, monkeypatch): """계획 §3.4 — 재개가 같은 라운드에 UUID 를 중복 INSERT 하지 않는다.""" # 1차: 실패로 끝나 라운드가 열린 채 남는다 stub_run_group(wired, monkeypatch, fail_attempts=2) wired.run() opened = load_round(wired.group_dir, "r001") assert opened.state in (RoundState.OPENED, RoundState.CANDIDATES) preallocated = opened.preallocated_asset_id # 2차: 재개 — 새 라운드를 열지 않고 이어간다 wired.calls.clear() stub_run_group(wired, monkeypatch) out = wired.run() assert [r.round_id for r in list_rounds(wired.group_dir)] == ["r001"] entry = out["data"]["groups"][GID] assert entry["round_id"] == "r001" # ★OPENED 에서 채번한 UUID 를 그대로 쓴다 assert entry["form_ref_asset_id"] == preallocated assert len(wired.step.db.rows) == 1 def test_resume_does_not_overwrite_the_open_rounds_earlier_candidates( wired, monkeypatch): """★재개가 이어가는 라운드에서 후보 번호를 1부터 다시 쓰면 안 된다. narrow 재시도만 이어붙이기로 막아 두면 이 구멍이 남는다 — 실패로 열린 채 남은 라운드를 다음 실행이 이어갈 때 그 라운드의 이전 후보가 사라진다. """ stub_run_group(wired, monkeypatch, fail_attempts=2, candidates_per_attempt=2) wired.run() round_dir = wired.group_dir / "rounds" / "r001" before = sorted(p.name for p in round_dir.glob("cand_*.png")) assert before == ["cand_01.png", "cand_02.png", "cand_03.png", "cand_04.png"] first_bytes = (round_dir / "cand_01.png").read_bytes() wired.calls.clear() stub_run_group(wired, monkeypatch, candidates_per_attempt=2) out = wired.run() assert out["data"]["groups"][GID]["round_id"] == "r001" assert wired.calls[0]["start_index"] == 5, "이전 후보 뒤에서 시작해야 한다" after = sorted(p.name for p in round_dir.glob("cand_*.png")) assert after == before + ["cand_05.png", "cand_06.png"] assert (round_dir / "cand_01.png").read_bytes() == first_bytes def test_failed_entry_still_records_which_round_it_came_from( wired, monkeypatch): """실패해도 **어느 라운드의 산출인지**는 남는다 — 남은 후보의 출처.""" stub_run_group(wired, monkeypatch, fail_attempts=2) out = wired.run() entry = out["data"]["groups"][GID] assert entry["status"] == "failed" assert entry["round_id"] == "r001" assert entry["path_kind"] == "round" def test_an_exception_after_the_round_opened_keeps_the_round_marks( wired, monkeypatch): """★예외 경로도 라운드 표기를 잃지 않는다. 위 테스트는 `_run_group` 이 실패를 **반환**하는 정상 경로라 `except` 절을 타지 않는다. 자산 바인딩처럼 라운드가 이미 정해진 뒤의 예외에서 표기가 사라지면, 그 라운드에 남은 후보의 출처를 되짚을 수 없다. """ stub_run_group(wired, monkeypatch) def _boom(**_kw): raise RuntimeError("sample fixture bind failure") monkeypatch.setattr(wired.step, "_bind_ref_asset", _boom) out = wired.run() entry = out["data"]["groups"][GID] assert entry["status"] == "failed" assert entry["round_id"] == "r001" assert entry["path_kind"] == "round" def test_input_drift_abandons_the_open_round_and_opens_a_new_one( wired, monkeypatch): """계획 §3.5 — 지문이 어긋나면 이어가지 않는다(번호는 소비된 채).""" # 두 시도 모두 실패시켜 라운드를 **열린 채** 남긴다. 1회만 실패시키면 # narrow 재시도가 성공해 FINALIZED 로 닫히고, 그러면 이 회귀가 검사하려는 # "열린 라운드의 drift" 상황 자체가 만들어지지 않는다. stub_run_group(wired, monkeypatch, fail_attempts=2) wired.run() monkeypatch.setattr( "app.modules.pipeline.outdoor_structure_seed.derive_seed_inputs", lambda **_kw: {"structure_desc": "sample fixture structure CHANGED"}) wired.calls.clear() stub_run_group(wired, monkeypatch) out = wired.run() assert load_round(wired.group_dir, "r001").state == RoundState.ABANDONED assert out["data"]["groups"][GID]["round_id"] == "r002" # ══════════════════════════════════════════════════════════════════════ # 5. FINALIZED 이후 — 불변, 그리고 CP 만 날아갔을 때 유료 0콜 # ══════════════════════════════════════════════════════════════════════ def test_finalized_round_rejects_further_writes(wired, monkeypatch): """계획 §3.6 — 닫힌 라운드는 더 쓸 수 없다.""" stub_run_group(wired, monkeypatch) wired.run() assert load_round(wired.group_dir, "r001").state == RoundState.FINALIZED with pytest.raises(AppError) as exc: transition_round(wired.group_dir, "r001", RoundState.CANDIDATES) assert exc.value.code == "round_immutable" def test_crash_before_checkpoint_save_replays_with_zero_paid_calls( wired, monkeypatch): """★계획 §3.7 — `_execute` 는 끝났는데 CP 저장 직전에 끊긴 경우. 이 경로가 없으면 그 crash 가 **유료 전체 재실행**으로 바뀐다. """ stub_run_group(wired, monkeypatch) first = wired.run() # ← CP 를 저장하지 않는다 (crash 재현) assert wired.cp is None wired.calls.clear() out = wired.run() assert wired.calls == [], "유료 경로가 다시 불렸다" assert len(wired.step.db.rows) == 1, "자산을 또 INSERT 했다" entry = out["data"]["groups"][GID] assert entry["round_replayed"] is True assert out["round_replayed_count"] == 1 for key in ("form_ref_path", "form_ref_sha256", "form_ref_asset_id", "round_id"): assert entry[key] == first["data"]["groups"][GID][key] def test_force_does_not_replay_a_finalized_round(wired, monkeypatch): """명시 force 는 재사용을 우회한다 — 완료 라운드가 있어도 새로 연다.""" stub_run_group(wired, monkeypatch) wired.run() wired.calls.clear() out = wired.run(mode="force") assert len(wired.calls) == 1 assert out["data"]["groups"][GID]["round_id"] == "r002" assert out["round_replayed_count"] == 0 # ══════════════════════════════════════════════════════════════════════ # 6. journal 이 깨졌으면 fail-closed # ══════════════════════════════════════════════════════════════════════ def test_a_corrupt_journal_fails_the_group_without_paid_calls( wired, monkeypatch): """계획 §3.8 — 손상된 registry 를 디렉터리 스캔으로 복원하지 않는다.""" stub_run_group(wired, monkeypatch) wired.run() (wired.group_dir / "rounds" / "index.json").write_text( "{ not json", encoding="utf-8") wired.calls.clear() out = wired.run() assert wired.calls == [], "손상된 journal 위에서 유료 호출을 했다" assert out["failed_count"] == 1 assert "round_index_corrupt" in str( out["data"]["groups"][GID].get("error")) def test_multiple_open_rounds_fail_closed(wired, monkeypatch): """열린 라운드가 둘이면 어느 쪽이 권위인지 알 수 없다.""" import json as _json stub_run_group(wired, monkeypatch, fail_attempts=2) wired.run() # 정상 API 로는 둘을 동시에 열 수 없다(`open_round` 가 거부한다). 부분 # 실패로 그런 상태가 **디스크에 남은** 경우를 재현하려면 manifest 를 # 직접 만든다 — 이 회귀가 지키는 것은 그 상태를 만났을 때의 동작이다. abandon_round(wired.group_dir, "r001") open_round(wired.group_dir, input_fp="another-fingerprint", provenance=PROV) mp = wired.group_dir / "rounds" / "r001" / "manifest.json" data = _json.loads(mp.read_text(encoding="utf-8")) data["state"] = RoundState.OPENED.value mp.write_text(_json.dumps(data), encoding="utf-8") wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1 assert "round_multiple_open" in str( out["data"]["groups"][GID].get("error")) # ══════════════════════════════════════════════════════════════════════ # 7. 자산 바인딩 — insert-or-verify 이지 upsert 가 아니다 # ══════════════════════════════════════════════════════════════════════ def test_binding_verifies_and_never_rewrites_a_conflicting_row( wired, monkeypatch, tmp_path): """미리 채번한 UUID 에 **다른 파일**을 가리키는 row 가 이미 있으면 선다.""" from app.models.project import ImageAsset rnd = open_round(wired.group_dir, input_fp="fp", provenance=PROV) other = wired.group_dir / "other.png" other.parent.mkdir(parents=True, exist_ok=True) other.write_bytes(b"sample-fixture-other") wired.step.db.rows.append(ImageAsset( id=rnd.preallocated_asset_id, project_id="P", episode_id="E", asset_type="structure_form_ref", entity_id=GID, variant_type="form_ref", file_path=str(other), status="generated")) target = wired.group_dir / "rounds" / rnd.round_id / "cand_01.png" target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"sample-fixture-target") with pytest.raises(AppError) as exc: wired.step._bind_ref_asset( group_id=GID, asset_id=rnd.preallocated_asset_id, abs_png_path=str(target), sha256=_sha_bytes(b"sample-fixture-target"), prompt_used="x") assert exc.value.code == "form_reference.asset_bind_conflict" # ★기존 row 는 손대지 않는다 assert str(wired.step.db.rows[0].file_path) == str(other) def test_binding_is_idempotent_for_the_same_bytes(wired, monkeypatch): """같은 라운드를 두 번 묶어도 row 가 하나뿐이고 값이 그대로다.""" rnd = open_round(wired.group_dir, input_fp="fp", provenance=PROV) target = wired.group_dir / "rounds" / rnd.round_id / "cand_01.png" target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"sample-fixture-target") sha = _sha_bytes(b"sample-fixture-target") a = wired.step._bind_ref_asset( group_id=GID, asset_id=rnd.preallocated_asset_id, abs_png_path=str(target), sha256=sha, prompt_used="x") b = wired.step._bind_ref_asset( group_id=GID, asset_id=rnd.preallocated_asset_id, abs_png_path=str(target), sha256=sha, prompt_used="x") assert a == b == rnd.preallocated_asset_id assert len(wired.step.db.rows) == 1 def test_binding_detects_bytes_drift_under_the_same_path(wired): """★`ImageAsset` 에 sha 컬럼이 없으므로 **실제 파일 bytes** 로 본다.""" rnd = open_round(wired.group_dir, input_fp="fp", provenance=PROV) target = wired.group_dir / "rounds" / rnd.round_id / "cand_01.png" target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"sample-fixture-original") wired.step._bind_ref_asset( group_id=GID, asset_id=rnd.preallocated_asset_id, abs_png_path=str(target), sha256=_sha_bytes(b"sample-fixture-original"), prompt_used="x") target.write_bytes(b"sample-fixture-TAMPERED") with pytest.raises(AppError) as exc: wired.step._bind_ref_asset( group_id=GID, asset_id=rnd.preallocated_asset_id, abs_png_path=str(target), sha256=_sha_bytes(b"sample-fixture-original"), prompt_used="x") assert "sha" in str(exc.value.message) def test_contract_fields_do_not_depend_on_the_annotate_helper( wired, monkeypatch): """★계획 §3.9 — `annotate_generated_asset` 은 non-fatal helper 다. 그 helper 는 `except Exception:` 으로 삼킨 뒤 `logger.warning` 만 한다 (annotate.py:73). **계약 검증을 그것에 맡기면 실패가 조용히 성공으로 보인다.** 그래서 helper 가 **아무 일도 하지 않아도** 계약 필드는 직접 쓰여 있어야 하고, 재바인딩 검증도 그대로 동작해야 한다. """ calls: List[str] = [] def _noop(asset, **_kw): calls.append("called") return asset # 실제 helper 가 실패했을 때와 같은 상태 monkeypatch.setattr( "app.services.image_capture.annotate.annotate_generated_asset", _noop) rnd = open_round(wired.group_dir, input_fp="fp", provenance=PROV) target = wired.group_dir / "rounds" / rnd.round_id / "cand_01.png" target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"sample-fixture-target") sha = _sha_bytes(b"sample-fixture-target") asset_id = wired.step._bind_ref_asset( group_id=GID, asset_id=rnd.preallocated_asset_id, abs_png_path=str(target), sha256=sha, prompt_used="x") assert calls == ["called"], "annotate 가 실제로 불리지 않았다" row = wired.step.db.rows[0] # 계약 필드는 helper 가 아니라 **바인딩이 직접** 쓴다 assert asset_id == rnd.preallocated_asset_id assert row.entity_id == GID and row.variant_type == "form_ref" assert row.asset_type == "structure_form_ref" assert str(row.file_path) == str(target) # 그리고 검증도 helper 와 무관하게 성립한다 — 다른 파일이면 선다 other = target.with_name("cand_02.png") other.write_bytes(b"sample-fixture-different") with pytest.raises(AppError): wired.step._bind_ref_asset( group_id=GID, asset_id=rnd.preallocated_asset_id, abs_png_path=str(other), sha256=_sha_bytes(b"sample-fixture-different"), prompt_used="x") # ══════════════════════════════════════════════════════════════════════ # 7.5 두 저장소 사이 — journal 이 완료라 해도 자산이 있는 것은 아니다 # ══════════════════════════════════════════════════════════════════════ # ★라운드 상태는 파일에, 자산은 DB 트랜잭션에 산다. 둘 사이에 원자성이 없다. # 그래서 "journal 이 FINALIZED" 만으로 성공을 내보내면, commit 전 crash 나 # **뒤 그룹 실패의 rollback** 이 자산을 지운 상태가 성공으로 둔갑한다 # (Codex A4 리뷰 BLOCKING-1·2). def test_replay_fails_closed_when_the_bound_asset_is_gone( wired, monkeypatch): """★완료 라운드를 재투영하기 전에 **자산 실재**를 확인한다.""" stub_run_group(wired, monkeypatch) wired.run() # CP 는 저장하지 않는다(crash 재현) wired.step.db.rows.clear() # 자산만 사라진 상태 wired.calls.clear() out = wired.run() assert wired.calls == [], "자산이 없는데 유료 호출을 했다" assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == ( "form_reference.asset_missing") def test_reuse_fails_closed_when_the_journal_is_gone(wired, monkeypatch): """★CP 만 보고 재사용하면 journal 손상이 조용히 건너뛰어진다. 후보 파일은 남기고 **journal 만** 지운다 — 디렉터리째 지우면 `_reusable` 이 파일 검증에서 먼저 걸려 이 경로가 시험되지 않는다. """ stub_run_group(wired, monkeypatch) wired.run_and_save_cp() (wired.group_dir / "rounds" / "r001" / "manifest.json").unlink() wired.calls.clear() out = wired.run() assert wired.calls == [], "journal 이 없는데 유료 호출을 했다" assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == ( "round_manifest_missing") def test_reuse_fails_closed_when_the_asset_row_is_gone(wired, monkeypatch): """CP·journal 이 멀쩡해도 자산이 없으면 그 CP 를 재사용하지 않는다.""" stub_run_group(wired, monkeypatch) wired.run_and_save_cp() wired.step.db.rows.clear() wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == ( "form_reference.asset_missing") def test_reuse_fails_closed_when_the_journal_disagrees_with_the_checkpoint( wired, monkeypatch): """★CP 와 journal 이 다른 산출을 말하면 어느 쪽도 권위로 삼지 않는다. 바꾸는 것은 **asset_id** 다. `_reusable` 은 파일·sha·지문만 보고 자산 id 는 안 보므로, 여기가 CP 만 믿던 구멍이 실제로 열리는 자리다. """ stub_run_group(wired, monkeypatch) out1 = wired.run_and_save_cp() entry = out1["data"]["groups"][GID] entry["form_ref_asset_id"] = "11111111-2222-3333-4444-555555555555" wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == ( "form_reference.round_cp_mismatch") def test_resume_from_selected_binds_without_searching_again( wired, monkeypatch): """★선택이 durable 하면 재개는 **자산 바인딩부터** — 유료 0콜. 이 경로가 없으면 bind 실패 지점마다 검색이 되풀이된다. """ stub_run_group(wired, monkeypatch) real_bind = wired.step._bind_ref_asset state = {"n": 0} def _fail_once(**kw): state["n"] += 1 if state["n"] == 1: raise RuntimeError("sample fixture bind failure") return real_bind(**kw) monkeypatch.setattr(wired.step, "_bind_ref_asset", _fail_once) out1 = wired.run() assert out1["failed_count"] == 1 assert load_round(wired.group_dir, "r001").state == RoundState.SELECTED assert load_round(wired.group_dir, "r001").result, "선택이 durable 하지 않다" wired.calls.clear() out2 = wired.run() assert wired.calls == [], "선택이 durable 한데 다시 검색했다" assert out2["completed_count"] == 1 assert load_round(wired.group_dir, "r001").state == RoundState.FINALIZED def test_a_later_group_failure_does_not_undo_an_earlier_groups_asset( wired, monkeypatch): """★그룹 단위 commit — 뒤 그룹의 rollback 이 앞 그룹 자산을 되돌리면 journal 만 완료로 남아 phantom 이 된다. 여기서는 commit **호출 시점**으로 확인한다(대역 DB 라 실제 트랜잭션은 `test_form_reference_lineage_pg.py` 가 본다). """ wired.add_group("g_beta") def _fake(*, group_id, structure_desc, world_facts_block, source_text, typology=None, client, candidate_dir, start_index=1, narrow=False, **_ignored): wired.calls.append({"group_id": group_id}) if group_id == "g_beta": raise RuntimeError("sample fixture group failure") candidate_dir.mkdir(parents=True, exist_ok=True) dest = candidate_dir / f"cand_{start_index:02d}.png" dest.write_bytes(b"sample-fixture-alpha") return {"status": "ok", "group_id": group_id, "form_ref_path": str(dest), "form_ref_sha256": _sha_bytes(b"sample-fixture-alpha"), "chosen_prompt_used": "d", "candidate_count": 1, "audit": {}} monkeypatch.setattr(wired.step, "_run_group", _fake) out = wired.run() assert out["completed_count"] == 1 and out["failed_count"] == 1 # ★앞 그룹의 자산은 **그 그룹을 닫을 때** 이미 commit 됐다 assert wired.step.db.commits >= 1 assert load_round(wired.group_dir, "r001").state == RoundState.FINALIZED # 뒤 그룹은 라운드를 열어 둔 채 실패로 남는다 assert load_round(wired.dir_for("g_beta"), "r001").state in ( RoundState.OPENED, RoundState.CANDIDATES) # ══════════════════════════════════════════════════════════════════════ # 7.6 최초 바인딩도 입력을 검증한다 # ══════════════════════════════════════════════════════════════════════ def test_fresh_binding_rejects_a_missing_file(wired): """★신규 INSERT 경로에 파일 검증이 없으면 잘못된 row 가 확정된다.""" rnd = open_round(wired.group_dir, input_fp="fp", provenance=PROV) missing = wired.group_dir / "rounds" / rnd.round_id / "cand_01.png" with pytest.raises(AppError) as exc: wired.step._bind_ref_asset( group_id=GID, asset_id=rnd.preallocated_asset_id, abs_png_path=str(missing), sha256="deadbeef" * 8, prompt_used="x") assert exc.value.code == "form_reference.asset_bind_input_invalid" assert wired.step.db.rows == [], "파일이 없는데 row 를 만들었다" def test_fresh_binding_rejects_tampered_bytes(wired): rnd = open_round(wired.group_dir, input_fp="fp", provenance=PROV) target = wired.group_dir / "rounds" / rnd.round_id / "cand_01.png" target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"sample-fixture-actual") with pytest.raises(AppError) as exc: wired.step._bind_ref_asset( group_id=GID, asset_id=rnd.preallocated_asset_id, abs_png_path=str(target), sha256=_sha_bytes(b"sample-fixture-expected"), prompt_used="x") assert exc.value.code == "form_reference.asset_bind_input_invalid" assert wired.step.db.rows == [] def test_a_vanished_output_leaves_the_round_unfinished_and_no_row( wired, monkeypatch): """산출이 사라지면 라운드를 닫지 않는다 — 공개 진입점에서 확인.""" def _fake(*, group_id, structure_desc, world_facts_block, source_text, typology=None, client, candidate_dir, start_index=1, narrow=False): wired.calls.append({"group_id": group_id}) candidate_dir.mkdir(parents=True, exist_ok=True) dest = candidate_dir / f"cand_{start_index:02d}.png" dest.write_bytes(b"sample-fixture-png") sha = _sha_bytes(b"sample-fixture-png") dest.unlink() # 선택 뒤 사라진 상황 return {"status": "ok", "group_id": group_id, "form_ref_path": str(dest), "form_ref_sha256": sha, "chosen_prompt_used": "d", "candidate_count": 1, "audit": {}} monkeypatch.setattr(wired.step, "_run_group", _fake) out = wired.run() assert out["failed_count"] == 1 assert wired.step.db.rows == [] assert load_round(wired.group_dir, "r001").state != RoundState.FINALIZED def test_asset_bound_resume_never_mints_a_second_uuid(wired, monkeypatch): """★라운드당 예약 UUID 는 **불변**이다. manifest header 와 저장된 산출이 서로 묶여 있지 않으면, ASSET_BOUND 상태 에서 header 의 `preallocated_asset_id` 만 바꿔도 재개가 **새 UUID 로 자산을 또 INSERT** 한다(Codex 3차 재현: row 2개 · 새 UUID 로 FINALIZED). """ import json as _json import uuid as _uuid from app.core.steps import outdoor_structure_form_reference_step as mod stub_run_group(wired, monkeypatch) real_finalize = mod.finalize_round def _crash(*_a, **_kw): raise RuntimeError("sample fixture crash before finalize") monkeypatch.setattr(mod, "finalize_round", _crash) wired.run() assert load_round(wired.group_dir, "r001").state == RoundState.ASSET_BOUND assert len(wired.step.db.rows) == 1 original_id = wired.step.db.rows[0].id mp = wired.group_dir / "rounds" / "r001" / "manifest.json" data = _json.loads(mp.read_text(encoding="utf-8")) data["preallocated_asset_id"] = str(_uuid.uuid4()) mp.write_text(_json.dumps(data), encoding="utf-8") monkeypatch.setattr(mod, "finalize_round", real_finalize) wired.calls.clear() out = wired.run() assert wired.calls == [], "유료 호출이 났다" assert len(wired.step.db.rows) == 1, ( f"자산이 {len(wired.step.db.rows)}개 — header 를 바꾸자 새 UUID 로 " f"또 INSERT 했다") assert wired.step.db.rows[0].id == original_id assert out["failed_count"] == 1 @pytest.mark.parametrize("field,bogus", [ ("input_fp", "sample-fixture-other-fingerprint"), ("contract_version", 999), ]) def test_manifest_header_must_agree_with_the_stored_result( wired, monkeypatch, field, bogus): """★header 와 산출이 서로 다른 라운드를 말하면 재사용하지 않는다.""" import json as _json stub_run_group(wired, monkeypatch) wired.run_and_save_cp() mp = wired.group_dir / "rounds" / "r001" / "manifest.json" data = _json.loads(mp.read_text(encoding="utf-8")) data[field] = bogus mp.write_text(_json.dumps(data), encoding="utf-8") wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1, f"header 의 {field} 변조가 통과했다" def test_reuse_rejects_a_round_stored_under_another_contract( wired, monkeypatch): """★서로 일관되더라도 **현재 계약이 아니면** 재사용하지 않는다. header·산출·CP 를 모두 999 로 맞추면 결속 검증은 통과한다 — 지원 계약 자체를 따로 봐야 "계약 exact 일치" 상태 머신과 어긋나지 않는다. """ import json as _json stub_run_group(wired, monkeypatch) out1 = wired.run_and_save_cp() mp = wired.group_dir / "rounds" / "r001" / "manifest.json" data = _json.loads(mp.read_text(encoding="utf-8")) data["contract_version"] = 999 data["result"]["round_contract_version"] = 999 mp.write_text(_json.dumps(data), encoding="utf-8") out1["data"]["groups"][GID]["round_contract_version"] = 999 wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1 # 구 계약은 **지원하지 않으므로** parser 가 먼저 세운다(CP 대조까지 가지 # 않는다). 어느 층이든 유료 호출 전 typed failure 다. assert out["data"]["groups"][GID]["error_code"] == ( "round_manifest_unsupported_contract") def test_a_manifest_without_its_registry_entry_keeps_its_uuid( wired, monkeypatch): """★manifest 를 쓰고 index 를 쓰기 전에 끊긴 상태. "라운드 없음"으로 답하면 다음 실행이 **같은 번호를 다시 열어 예약 UUID 를 갈아치운다**(실측: UUID 보존 False). 라운드/UUID 불변 계약이 깨진다. """ stub_run_group(wired, monkeypatch, fail_attempts=2) wired.run() mp = wired.group_dir / "rounds" / "r001" / "manifest.json" before = mp.read_bytes() (wired.group_dir / "rounds" / "index.json").unlink() wired.calls.clear() out = wired.run() assert wired.calls == [], "registry 없는 라운드 위에서 유료 호출을 했다" assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == ( "round_registry_mismatch") assert mp.read_bytes() == before, "기존 manifest 를 덮어썼다" @pytest.mark.parametrize("mutate,label", [ (lambda r: r.update(status="failed"), "status=failed"), (lambda r: r.pop("status", None), "status 삭제"), (lambda r: r.update(path_kind="legacy_flat"), "path_kind 변조"), (lambda r: r.update(ref_pack_version="tampered-pack"), "pack 변조"), (lambda r: r.update(form_ref_sha256="not-a-sha"), "sha 형식 파괴"), (lambda r: r.pop("target_policy_version", None), "policy 삭제"), ]) def test_replay_rejects_a_corrupt_finalized_projection( wired, monkeypatch, mutate, label): """★CP 없는 재투영은 journal 산출을 **그대로 성공 CP** 로 내보낸다. shape 를 안 보면 손상된 journal 이 완료로 승격된다(실측: status='failed' 인 산출이 completed_count=1 로 올라갔다). """ import json as _json stub_run_group(wired, monkeypatch) wired.run() # CP 는 저장하지 않는다(crash 재현) mp = wired.group_dir / "rounds" / "r001" / "manifest.json" data = _json.loads(mp.read_text(encoding="utf-8")) mutate(data["result"]) mp.write_text(_json.dumps(data), encoding="utf-8") wired.calls.clear() out = wired.run() assert wired.calls == [], f"{label}: 유료 호출이 났다" assert out["failed_count"] == 1, f"{label}: 손상된 산출이 승격됐다" assert out["completed_count"] == 0 @pytest.mark.parametrize("key,bogus", [ ("ref_pack_version", "tampered-pack"), ("target_policy_version", "tampered-policy"), ]) def test_reuse_rejects_a_stamp_that_disagrees_with_the_round_header( wired, monkeypatch, key, bogus): """★CP 와 산출을 **둘 다** 변조해도 라운드 header 의 역사값이 잡는다. ★이것을 "현재 값과 같은가"로 검사하면 안 된다 — 지문은 대상 정책을 **의도적으로 제외**하고 팩도 이름이 아니라 검색 계약 sha 만 접으므로, 현재 값 대조는 **정상적인 역사 provenance 까지 거부**한다(Codex 5차 HIGH-2). 변조 방지는 header 결속의 몫, 무효화 판단은 지문의 몫이다. """ import json as _json stub_run_group(wired, monkeypatch) out1 = wired.run_and_save_cp() mp = wired.group_dir / "rounds" / "r001" / "manifest.json" data = _json.loads(mp.read_text(encoding="utf-8")) data["result"][key] = bogus mp.write_text(_json.dumps(data), encoding="utf-8") out1["data"]["groups"][GID][key] = bogus # CP 도 같은 거짓으로 wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1, f"{key} 를 둘 다 변조하니 통과했다" assert out["data"]["groups"][GID]["error_code"] == ( "round_manifest_header_mismatch") def test_replay_preserves_historical_stamps_across_a_label_only_change( wired, monkeypatch): """★정책 이름만 바뀌어도 **지문은 그대로**다 — 옛 산출을 그대로 재투영한다. 지문(`_group_fingerprint`)은 대상 정책을 의도적으로 제외한다. 그래서 "현재 값과 같은가"로 stamp 를 검사하면 이 정상 경로가 막히고, CP 가 남아 있으면 재사용되고 CP 만 없으면 거부되는 **비대칭**이 생긴다 (Codex 5차 HIGH-2 의 재현 시나리오). """ stub_run_group(wired, monkeypatch) first = wired.run() # CP 는 저장하지 않는다 old = first["data"]["groups"][GID]["target_policy_version"] monkeypatch.setattr( "app.modules.pipeline.search_grounded_ref.TARGET_POLICY_VERSION", "sample-fixture-relabelled-policy") wired.calls.clear() out = wired.run() assert wired.calls == [], "이름만 바뀌었는데 다시 검색했다" assert out["completed_count"] == 1 assert out["round_replayed_count"] == 1 # ★역사값이 **보존**된다 — 현재 값으로 갈아치우지 않는다 assert out["data"]["groups"][GID]["target_policy_version"] == old def test_resuming_a_bound_round_keeps_its_historical_stamps( wired, monkeypatch): """★재개가 **현재 값을 다시 읽으면** header 결속이 깨진다. 재투영 경로는 저장된 산출을 그대로 쓰지만, SELECTED 에서 이어가는 경로는 `_close_round` 가 표기를 다시 조립한다 — 거기서 현재 값을 읽으면 라운드가 열릴 때의 역사값과 어긋난다. """ stub_run_group(wired, monkeypatch) real_bind = wired.step._bind_ref_asset def _boom(**_kw): raise RuntimeError("sample fixture bind failure") monkeypatch.setattr(wired.step, "_bind_ref_asset", _boom) wired.run() rnd = load_round(wired.group_dir, "r001") assert rnd.state == RoundState.SELECTED old = rnd.provenance["target_policy_version"] monkeypatch.setattr( "app.modules.pipeline.search_grounded_ref.TARGET_POLICY_VERSION", "sample-fixture-relabelled-policy") monkeypatch.setattr(wired.step, "_bind_ref_asset", real_bind) wired.calls.clear() out = wired.run() assert wired.calls == [], "재개가 다시 검색했다" assert out["completed_count"] == 1, "재개가 역사값과 어긋나 실패했다" assert out["data"]["groups"][GID]["target_policy_version"] == old def test_replay_rejects_a_manifest_whose_provenance_header_was_removed( wired, monkeypatch): """★header 를 지우고 산출·CP 를 같은 거짓값으로 바꿔도 통과하면 안 된다. 실측(Codex 6차): provenance 가 **선택 필드**였을 때 이 조합이 `paid=0 · completed=1 · pack='tampered-pack'` 으로 통과했다. header shape 손상을 parser 가 정상으로 세탁한 것이다. """ import json as _json stub_run_group(wired, monkeypatch) wired.run() # CP 는 저장하지 않는다 mp = wired.group_dir / "rounds" / "r001" / "manifest.json" data = _json.loads(mp.read_text(encoding="utf-8")) del data["provenance"] data["result"]["ref_pack_version"] = "tampered-pack" data["result"]["target_policy_version"] = "tampered-policy" mp.write_text(_json.dumps(data), encoding="utf-8") wired.calls.clear() out = wired.run() assert wired.calls == [], "header 없는 tampered 산출로 유료 호출을 했다" assert out["completed_count"] == 0 assert out["failed_count"] == 1 def test_storage_contract_version_moves_both_hashes(wired, monkeypatch): """★저장 계약이 바뀌면 `config_hash` 와 그룹 지문이 **함께** 움직인다. 안 움직이면 manifest shape 이 바뀌었는데 완료 CP 가 그대로 재사용된다 (Codex 6차 수정 조건 1 — 이 상수가 두 해시에 접혀 있어야 실행 도달도 함께 해결된다). """ from app.core.steps import outdoor_structure_form_reference_step as mod args = dict(structure_desc="d", world_facts_block="", source_text_sha="s", research_input_sha="b") before_cfg = wired.step._config_hash() before_fp = wired.step._group_fingerprint(**args) monkeypatch.setattr(mod, "ROUND_STORAGE_CONTRACT_VERSION", 999) assert wired.step._config_hash() != before_cfg, "config_hash 가 안 움직였다" assert wired.step._group_fingerprint(**args) != before_fp, ( "그룹 지문이 안 움직였다") def test_an_explicit_null_schema_version_is_not_legacy(wired, monkeypatch): """★`dict.get()` 은 키 부재와 값 None 을 같은 것으로 만든다. 실측: 정상 v4 CP 의 `schema_version` 을 None 으로 두고 registry 를 깨뜨렸는데 `paid=0 · completed=1 · round_id='legacy'` 로 통과했다. """ stub_run_group(wired, monkeypatch) wired.run_and_save_cp() assert wired.cp is not None wired.cp["schema_version"] = None # 키는 있고 값이 null (wired.group_dir / "rounds" / "index.json").write_text( "{ not json", encoding="utf-8") wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == ( "form_reference.round_marks_invalid") def test_a_boolean_contract_version_does_not_pass_as_one(wired, monkeypatch): """★`True == 1` 이라 순수 값 비교는 bool 을 정수로 받아들인다. **SELECTED 구간**에서 시험한다 — 완료 산출은 shape 검증이 정수 필드를 `_exact_int` 로 따로 보지만, 중간 상태는 header 결속만 받는다. 즉 여기가 타입 비교가 **단독으로** 막아야 하는 자리다. """ import json as _json stub_run_group(wired, monkeypatch) real_bind = wired.step._bind_ref_asset def _boom(**_kw): raise RuntimeError("sample fixture bind failure") monkeypatch.setattr(wired.step, "_bind_ref_asset", _boom) wired.run() assert load_round(wired.group_dir, "r001").state == RoundState.SELECTED mp = wired.group_dir / "rounds" / "r001" / "manifest.json" data = _json.loads(mp.read_text(encoding="utf-8")) data["result"]["round_contract_version"] = True mp.write_text(_json.dumps(data), encoding="utf-8") # ★대역을 되돌린다 — 안 그러면 두 번째 실행도 bind 에서 터져 **타입 비교와 # 무관하게** 실패하고, 이 회귀가 공허해진다. monkeypatch.setattr(wired.step, "_bind_ref_asset", real_bind) wired.calls.clear() out = wired.run() assert wired.calls == [], "bool 이 정수로 통과해 재개가 진행됐다" assert out["failed_count"] == 1 assert out["completed_count"] == 0 def test_a_malformed_schema_version_is_not_treated_as_legacy( wired, monkeypatch): """★"부재" 와 "존재하지만 malformed" 를 합치면 손상이 legacy 로 내려간다. 실측(Codex 재현): `schema_version` 을 문자열 "4" 로 바꾸고 registry 를 깨뜨렸는데 `status=ok · round_id=legacy · paid=0` 으로 통과했다. """ stub_run_group(wired, monkeypatch) wired.run_and_save_cp() assert wired.cp is not None wired.cp["schema_version"] = "4" # 정수가 아니다 (wired.group_dir / "rounds" / "index.json").write_text( "{ not json", encoding="utf-8") wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == ( "form_reference.round_marks_invalid") # ══════════════════════════════════════════════════════════════════════ # 8. legacy — 구 flat 산출은 옮기지도 덮지도 않고 표기만 # ══════════════════════════════════════════════════════════════════════ def _legacy_cp(entry: Dict[str, Any]) -> Dict[str, Any]: """라운드 계약 **이전 스키마**의 CP 모양 — 그것만 legacy 로 인정한다.""" stripped = dict(entry) for key in ("round_id", "path_kind", "round_contract_version", "ref_pack_version", "target_policy_version"): stripped.pop(key, None) return {"schema_version": 3, "data": {"groups": {GID: stripped}}} def test_genuine_legacy_checkpoint_is_reused_with_legacy_marks( wired, monkeypatch): """라운드 계약 이전 **스키마**의 CP 를 재사용하면 `legacy` 로 명시한다.""" stub_run_group(wired, monkeypatch) first = wired.run() wired.cp = _legacy_cp(first["data"]["groups"][GID]) wired.calls.clear() out = wired.run() assert wired.calls == [] got = out["data"]["groups"][GID] assert got["reused"] is True assert got["round_id"] == LEGACY_ROUND_ID assert got["path_kind"] == "legacy_flat" assert got["round_contract_version"] == LEGACY_CONTRACT_VERSION def test_legacy_reuse_still_verifies_the_asset(wired, monkeypatch): """★legacy 예외는 **journal 부재**에만 적용된다 — 자산 검증은 항상 한다. legacy 라고 자산 확인까지 건너뛰면, 계보가 가리키는 row 가 사라진 CP 를 그대로 성공으로 내보낸다. """ stub_run_group(wired, monkeypatch) first = wired.run() wired.cp = _legacy_cp(first["data"]["groups"][GID]) wired.step.db.rows.clear() wired.calls.clear() out = wired.run() assert wired.calls == [], "자산이 없는데 유료 호출을 했다" assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == ( "form_reference.asset_missing") def test_current_schema_checkpoint_missing_round_marks_is_not_legacy( wired, monkeypatch): """★현행 스키마 CP 에서 라운드 표기만 사라진 것을 legacy 로 강등하면 검증이 통째로 건너뛰어진다. 실측(Codex 재현): 그 상태에서 DB row 까지 지웠는데 `status=ok` · 유료 0 · 자산 0 으로 "완료"됐다. legacy 여부는 **CP 스키마 버전**으로 판정해야 한다 — 필드 부재는 손상이지 legacy 가 아니다. """ from app.core.steps.outdoor_structure_form_reference_step import ( SCHEMA_VERSION, ) stub_run_group(wired, monkeypatch) first = wired.run() entry = dict(first["data"]["groups"][GID]) for key in ("round_id", "path_kind", "round_contract_version"): entry.pop(key, None) wired.cp = {"schema_version": SCHEMA_VERSION, "data": {"groups": {GID: entry}}} wired.step.db.rows.clear() wired.calls.clear() out = wired.run() assert wired.calls == [], "손상된 현행 CP 위에서 유료 호출을 했다" assert out["failed_count"] == 1 # ★**어떻게** 실패했는지까지 본다. legacy 로 강등되면 자산 검증에서 # `asset_missing` 으로 걸려 "실패했다"만으로는 구분되지 않는다. assert out["data"]["groups"][GID]["error_code"] == ( "form_reference.round_marks_invalid") def test_partial_round_marks_on_a_current_schema_checkpoint_fail_closed( wired, monkeypatch): """일부만 남은 표기도 legacy 가 아니다 — 부분 손상이다.""" from app.core.steps.outdoor_structure_form_reference_step import ( SCHEMA_VERSION, ) stub_run_group(wired, monkeypatch) first = wired.run() entry = dict(first["data"]["groups"][GID]) entry.pop("path_kind", None) # round_id 는 남긴다 wired.cp = {"schema_version": SCHEMA_VERSION, "data": {"groups": {GID: entry}}} wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == ( "form_reference.round_marks_invalid") # ── journal 이 권위임을 CP 경로에서도 지킨다 (BLOCKING-2) ────────────── def test_reuse_fails_closed_when_the_registry_is_corrupt( wired, monkeypatch): """★CP 가 있으면 registry 손상이 journal SOT 를 막지 못하던 경로. `load_round` 를 직접 부르면 index.json 을 우회한다 — 등록 여부를 먼저 확인해야 한다. """ stub_run_group(wired, monkeypatch) wired.run_and_save_cp() (wired.group_dir / "rounds" / "index.json").write_text( "{ not json", encoding="utf-8") wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1 assert out["data"]["groups"][GID]["error_code"] == "round_index_corrupt" @pytest.mark.parametrize("key,bogus", [ ("path_kind", "legacy_flat"), ("round_contract_version", 999), ("ref_pack_version", "bogus-pack"), ("target_policy_version", "bogus-policy"), ]) def test_reuse_compares_the_whole_projection_not_three_fields( wired, monkeypatch, key, bogus): """★asset_id/path/sha 3개만 보면 나머지 투영 필드 변조가 통과한다.""" stub_run_group(wired, monkeypatch) out1 = wired.run_and_save_cp() out1["data"]["groups"][GID][key] = bogus wired.calls.clear() out = wired.run() assert wired.calls == [] assert out["failed_count"] == 1, f"{key} 변조가 통과했다" # path_kind 변조는 표기 검증에서, 나머지는 journal 대조에서 걸린다 — # 어느 쪽이든 **유료 전에** typed 실패로 선다. assert out["data"]["groups"][GID]["error_code"] in ( "form_reference.round_cp_mismatch", "form_reference.round_marks_invalid") def test_a_bound_round_without_its_result_fails_closed(wired, monkeypatch): """★SELECTED 인데 산출이 비면 **유료 재검색으로 내려가면 안 된다**. 실측(Codex 재현): manifest 의 result 를 {} 로 손상시키고 재개하자 `status=ok · paid_calls=1` 로 다시 검색했다. 그 상태는 계약상 존재할 수 없으므로 유료 호출 전에 선다. """ import json as _json stub_run_group(wired, monkeypatch) real_bind = wired.step._bind_ref_asset def _boom(**_kw): raise RuntimeError("sample fixture bind failure") monkeypatch.setattr(wired.step, "_bind_ref_asset", _boom) wired.run() assert load_round(wired.group_dir, "r001").state == RoundState.SELECTED mp = wired.group_dir / "rounds" / "r001" / "manifest.json" data = _json.loads(mp.read_text(encoding="utf-8")) data["result"] = {} mp.write_text(_json.dumps(data), encoding="utf-8") # ★`monkeypatch.undo()` 를 쓰면 fixture 의 settings 패치까지 풀려 # `_execute` 가 no-op 으로 빠진다 — 그러면 아무것도 검증하지 못한다. monkeypatch.setattr(wired.step, "_bind_ref_asset", real_bind) wired.calls.clear() out = wired.run() assert wired.calls == [], "산출이 빈 라운드에서 다시 검색했다" assert out["failed_count"] == 1 def test_reused_round_entry_keeps_its_round_id(wired, monkeypatch): """라운드 계약으로 만들어진 산출을 재사용하면 그 round_id 를 보존한다.""" stub_run_group(wired, monkeypatch) wired.run_and_save_cp() wired.calls.clear() out = wired.run() got = out["data"]["groups"][GID] assert wired.calls == [] assert got["reused"] is True assert got["round_id"] == "r001" assert got["round_contract_version"] == ROUND_STORAGE_CONTRACT_VERSION # ══════════════════════════════════════════════════════════════════════ # 9. 지문 계약 — 저장 계약은 싣고 round_id 는 싣지 않는다 # ══════════════════════════════════════════════════════════════════════ def test_round_id_is_not_part_of_the_fingerprint(wired, monkeypatch): """★`round_id` 는 입력이 아니라 산출 identity 다. 지문에 넣으면 재개가 **항상** miss 되어 매번 새로 검색한다. """ import json as _json from app.core.steps import outdoor_structure_form_reference_step as mod payloads: List[Dict[str, Any]] = [] original = _json.dumps def _spy(obj, **kw): if isinstance(obj, dict): payloads.append(obj) return original(obj, **kw) monkeypatch.setattr(mod.json, "dumps", _spy) wired.step._group_fingerprint( structure_desc="d", world_facts_block="", source_text_sha="s", research_input_sha="b") wired.step._config_hash() assert payloads, "지문 payload 를 관측하지 못했다" for payload in payloads: assert payload.get("round_storage_contract") == ( ROUND_STORAGE_CONTRACT_VERSION) assert "round_id" not in payload # ── 사전조사·부속 참조 결속 (2026-08-03) ──────────────────────────── def test_pre_research_inputs_move_the_group_fingerprint(): """★조사가 읽는 것이 바뀌면 지문이 움직여야 한다. 실내·외 설명은 명세·씬이 아니라 **배경 분류 멤버 목록과 위치 서술**에서 나온다(`derive_seed_inputs`). 나머지가 그대로여도 혼자 바뀔 수 있고, 지문이 안 움직이면 조사 문항·답과 부속 검색 입력이 달라졌는데 이전 CP 를 조용히 재사용한다. """ from app.core.steps.outdoor_structure_form_reference_step import ( OutdoorStructureFormReferenceStep as S, ) base = dict(structure_desc="d", world_facts_block="w", source_text_sha="s", research_input_sha="a") assert (S._group_fingerprint(None, **base) != S._group_fingerprint(None, **{**base, "research_input_sha": "b"})) def test_reuse_rejects_a_checkpoint_whose_research_record_was_changed( wired, monkeypatch): """CP 의 조사·부속 기록이 라운드 기록과 다르면 재사용하지 않는다. ★조사 블록은 텍스트라 소비측 파일 검증에 걸리지 않는다. 여기서 막지 않으면 씨드가 journal 과 다른 사실을 물고 그린다. """ from app.core.errors import AppError stub_run_group(wired, monkeypatch) first = wired.run_and_save_cp() # ★CP 전체를 그대로 두고 그룹 entry 만 손댄다. `schema_version` 을 # 빠뜨리면 legacy 로 강등돼 라운드 대조 자체가 생략된다. entry = dict(first["data"]["groups"][GID]) entry["typology"] = {"block": "SAMPLE_FIXTURE_tampered", "items": []} wired.cp = {**first, "data": {**first["data"], "groups": {GID: entry}}} # 그룹 단위 격리라 스텝이 서지는 않는다 — 그 그룹이 실패로 떨어지고 # 사유가 남는다. 조용히 재사용되지 않는 것이 이 잠금의 요점이다. out = wired.run() entry_after = out["data"]["groups"][GID] assert entry_after["status"] == "failed" assert entry_after["error_code"] == "form_reference.round_cp_mismatch" assert "사전조사" in entry_after["error"]