"""**끝점**으로 — 판정 둘 다 verified + 선택 B → B 만 붙는다 · A 골랐다가 B 로 바꾸면 B 만 usable (Codex 재리뷰 2026-09-02 밤). API 함수·실제 DB 표·production reader(`central_cp_with_reviews`) 로 잰다. """ from __future__ import annotations import hashlib import json import uuid from pathlib import Path import pytest from app.core.database import Base, SessionLocal, engine from app.models.project import Episode from app.models.project import GroundingReferenceFidelityReview as M from app.models.project import GroundingReferenceSelection as S from app.modules.pipeline import grounding_fidelity_review as fr from app.modules.pipeline import reference_acquisition as ra ERA, REGION = "1960년대", "대한민국" def _row(sid, fid, cands): return {"research_subject_id": sid, "identity": f"id_{sid}", "outcome": ra.STATUS_UNAVAILABLE, "ledger_row": {"owner_type": "location", "final_id": fid, "purpose": "context"}, "acquisition": {"status": "no_match_after_retry", "chosen": None, "chosen_path": None, "rounds": [{"round_no": 1, "queries": [], "eligible": [], "downloaded_candidates": cands, "coordinates_carried": {"declared": ["era", "region"]}}]}} @pytest.fixture def world(tmp_path): from app.core.config import settings from app.core.steps.episode_reference_policy_step import EpisodeReferencePolicyStep as Policy from app.models.catalog import ProjectRegistry, UserAccount Base.metadata.create_all(bind=engine) root = tmp_path / "root"; (root / "projects").mkdir(parents=True) settings.projects_dir = str(root / "projects") pid, eid = f"p{uuid.uuid4().hex[:8]}", f"e{uuid.uuid4().hex[:8]}" cands = [] for i in (1, 2): rel = f"pics/rs_x_r1_0{i}.png" f = root / rel; f.parent.mkdir(parents=True, exist_ok=True) f.write_bytes(b"\x89PNG\r\n\x1a\n" + hashlib.sha256(rel.encode()).digest()) cands.append({"index": i, "path": rel, "url": f"https://x.invalid/{i}.jpg"}) ep = Path(settings.projects_dir) / pid / "checkpoints" / "episodes" / eid for step, data in (("reference_acquisition", {"rows": [_row("rs_x", "L01", cands)]}), ("visual_world_rules", {"era": ERA, "region": REGION})): d = ep / step; d.mkdir(parents=True) (d / "manifest.json").write_text(json.dumps({"status": "completed", "data": data}, ensure_ascii=False), encoding="utf-8") db = SessionLocal() now = "2026-09-02T00:00:00+09:00" uid = f"u{uuid.uuid4().hex[:8]}" db.add(UserAccount(id=uid, username=uid, display_name="고르는 사람", password_hash="x", role="admin", is_active=1, created_at=now, updated_at=now)); db.commit() db.add(ProjectRegistry(id=pid, name="선택 시험", status="active", created_by=uid, created_at=now, updated_at=now)); db.commit() db.add(Episode(id=eid, project_id=pid, episode_number=1, title="t", source_filename="s.pdf", source_path="s.pdf", created_at=now, updated_at=now)); db.commit() # ★require_episode(PR #82 · Codex P1) — 이 프로젝트의 에피소드여야 한다 user = db.get(UserAccount, uid) s = Policy.__new__(Policy); s.project_id, s.episode_id, s.db = pid, eid, db s.project_config = {"grounding_mode": "v2_chunk"} try: yield {"step": s, "db": db, "pid": pid, "eid": eid, "root": root, "user": user} finally: db.query(S).filter_by(project_id=pid).delete() db.query(M).filter_by(project_id=pid).delete() db.query(Episode).filter_by(id=eid).delete() # ★프로젝트보다 먼저 — FK db.query(ProjectRegistry).filter_by(id=pid).delete() db.query(UserAccount).filter_by(id=uid).delete() db.commit(); db.close() def _inputs(w): from app.api.v1 import grounding_fidelity as api got, _root = api._candidates(w["pid"], w["eid"]) return sorted(got, key=lambda x: x["candidate"]["index"]) def _verify(w, one): from app.api.v1 import grounding_fidelity as api p = one["payload"] body = api.VerdictIn(project_id=w["pid"], episode_id=w["eid"], review_input_hash=one["hash"], verdict="verified", reason="", idempotency_key=f"{one['hash']}:verified", observed=api.ObservedIn(image_sha256=p["image_sha256"], era=p["era"], region=p["region"])) return api.record_verdict(body, db=w["db"], current_user=w["user"]) def _select(w, one, key): from app.api.v1 import grounding_fidelity as api body = api.SelectIn(project_id=w["pid"], episode_id=w["eid"], review_input_hash=one["hash"], idempotency_key=key) return api.select_photo(body, db=w["db"], current_user=w["user"]) def _chosen_path(w): cp = fr.central_cp_with_reviews(w["step"], required=True) r = cp["data"]["rows"][0] return (r["acquisition"].get("chosen") or {}).get("path"), ra.usable_as_reference(r), r["grounding_fidelity"] def test_both_verified_then_select_b_then_switch_to_a(world): w = world a, b = _inputs(w) assert len(_inputs(w)) == 2, "★받아 둔 두 장이 다 사람 앞에 온다" _verify(w, a); _verify(w, b) # 둘 다 맞다 · 아직 안 골랐다 → 안 붙고, 사유가 「고르라」다 path, usable, fid = _chosen_path(w) # ★HITL 0: 맞다 둘 · 선택 없음 → 다시 묻지 않고 한 장을 자동으로 쓴다 assert path in (a["candidate"]["path"], b["candidate"]["path"]) and usable \ and fid["state"] == ra.FIDELITY_VERIFIED d0 = fr.reviews_digest_for(w["step"]) got = _select(w, b, "k1") assert got["created"] and got["replaced"] is None and got["verified"] is True path, usable, fid = _chosen_path(w) assert path == b["candidate"]["path"] and usable and fid["state"] == ra.FIDELITY_VERIFIED d1 = fr.reviews_digest_for(w["step"]) assert d1 != d0, "★선택이 소비자 지문을 움직인다" # A 로 바꾼다 — 서버가 앞 선택을 대신한다 got2 = _select(w, a, "k2") assert got2["replaced"] == b["hash"] rows = w["db"].query(S).filter_by(project_id=w["pid"]).all() assert len(rows) == 2 and {r.supersedes_id for r in rows} == {None, got["id"]}, "★UPDATE 가 아니라 대신하는 새 줄" path, usable, _ = _chosen_path(w) assert path == a["candidate"]["path"] and usable assert fr.reviews_digest_for(w["step"]) not in (d0, d1) # 같은 idempotency_key 는 같은 줄 assert _select(w, a, "k2")["created"] is False def test_selecting_before_any_verdict_records_but_does_not_attach(world): w = world a, _b = _inputs(w) got = _select(w, a, "k0") assert got["verified"] is False path, usable, fid = _chosen_path(w) assert path is None and not usable and "아직 verified" in fid["fault"] def test_the_page_words_follow_the_three_questions(world): """★Codex BLOCK (059a9963 재리뷰): 머리말이 옛 계약(「맞다 = 붙는다 · 아니다 = 대상은 참고 없이」)을 말했다. 세 물음으로 갈라 적고, 옛 문구가 렌더 끝점에 없음을 잠근다.""" from tools.grounding_audit import fidelity_review_page as pg from app.api.v1 import grounding_fidelity as api got, _root = api._candidates(world["pid"], world["eid"]) cards = pg.cards_from(got, key_of=lambda x: x["hash"]) html = pg.render(cards, api_base="/api", image_base="/img?key=", coords={"era": ERA, "region": REGION}) # ★2026-09-03 HITL 0 · 대상당 한 물음: 옛 세 물음 문구는 없고, 검증용 도구임과 한 물음이 적혀 있다 for old in ("이 사진이 그림의 참고로 붙는다", "그 대상은 참고 없이 간다", "이 사진은 안 붙는다", "물음은 셋입니다"): assert old not in html, old for new in ("검증용 도구", "대상마다 한 번", "이 사진을 쓴다", "없음"): assert new in html, new def test_two_live_selections_are_fail_closed_and_the_next_pick_replaces_both(world): w = world a, b = _inputs(w) _verify(w, a); _verify(w, b) # 동시 요청 흉내: 표에 살아 있는 선택 둘을 **직접** 심는다(정상 경로로는 못 만든다) now = "2026-09-02T23:40:00+09:00" for i, one in enumerate((a, b)): w["db"].add(S(id=f"dup{i}", project_id=w["pid"], episode_id=w["eid"], research_subject_id="rs_x", review_input_hash=one["hash"], image_sha256="", selected_by="t", selected_at=now, supersedes_id=None, idempotency_key=f"dup{i}", created_at=now)) w["db"].commit() path, usable, fid = _chosen_path(w) assert path is not None and usable and "둘 이상" in fid["fault"], \ "★기록이 어긋나도 자동 선택은 선다 (HITL 0) · 사유는 남는다" got = _select(w, b, "k9") rows = {r.id: r for r in w["db"].query(S).filter_by(project_id=w["pid"]).all()} assert set(rows[got["id"]].supersedes_id.split(";")) == {"dup0", "dup1"}, "★살아 있던 둘을 다 대신한다" path, usable, _ = _chosen_path(w) assert path == b["candidate"]["path"] and usable def test_select_serialises_on_the_subject_rows(): import inspect from app.api.v1 import grounding_fidelity as api assert "with_for_update()" in inspect.getsource(api.select_photo)