"""고증 판정 — 사람이 정한 것을 **DB 에** 적는 유일한 문.
★★★계약 (Codex 2026-09-02):
· 정본은 이 표다. `artifact/` 의 HTML·JSON 은 화면 산출이지 결정이 아니다
· 판정자는 **서버의 인증 주체**에서 온다 — 클라이언트가 임의로 못 보낸다
· append-only. 정정은 `supersedes_id` 를 든 **새 행**
· 사람이 **실제로 본** 사진 SHA·좌표가 붙일 것과 다르면 **안 받는다**
· CP 나 산출 JSON 을 직접 고치는 길은 만들지 않는다
"""
from __future__ import annotations
import json
import logging
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional
from fastapi import APIRouter, Depends
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session as OrmSession
from app.api.deps import get_current_user, get_db, verify_project_access
from app.core.errors import AppError
from app.models.catalog import UserAccount
from app.models.project import GroundingReferenceFidelityReview
from app.modules.pipeline import grounding_fidelity_review as fr
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1/grounding-fidelity",
tags=["grounding-fidelity"])
KST = timezone(timedelta(hours=9))
class ObservedIn(BaseModel):
"""사람이 **실제로 화면에서 본 것**. ★이것과 붙일 것이 달라지면 안 받는다."""
image_sha256: str
era: str = ""
region: str = ""
class VerdictIn(BaseModel):
"""판정 하나.
★★★**입력 신원을 클라이언트가 못 만든다** (Codex BLOCK 2026-09-02).
앞 판은 canonical payload 를 body 에서 통째로 만들고 `observed` 도 **같은
body** 와만 견줬다 — 항진식이다. 지어낸 입력과 지어낸 `observed` 를 함께
보내면 정본 표에 그대로 들어갔다.
이제 서버가 **얼어붙은 CP·세계 규칙·지금 파일 SHA** 로 후보를 다시 만들고,
`review_input_hash` 로 **정확히 한 줄**을 찾는다. 없거나 사진·좌표가
달라졌으면 **쓰기 0** 으로 거절한다.
"""
project_id: str
episode_id: str
review_input_hash: str
verdict: str = Field(..., description="verified | rejected")
reason: str = ""
observed: ObservedIn
idempotency_key: str
supersedes_id: Optional[str] = None
# ★쓰기 문은 deps 로 올렸다 (Codex BLOCK 2026-09-02 밤: PUT 좌표가 같은 문을 써야 한다).
from app.api.deps import require_project_write # noqa: E402,F401
def _sha_of_file(path) -> str:
import hashlib
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1 << 20), b""):
h.update(chunk)
return h.hexdigest()
def _candidates(project_id: str, episode_id: str):
"""서버가 **얼어붙은 것에서 다시 만든** 후보들. ★클라이언트를 안 믿는다.
사진 SHA 는 **지금 디스크의 파일**에서 낸다 — 사진이 바뀌었으면 신원이
달라지고, 그러면 옛 결정도 새 결정도 그 사진에 안 붙는다.
"""
from pathlib import Path
cp, world, root = _frozen(project_id, episode_id)
data = (world or {}).get("data") or {}
coords = {"era": str(data.get("era") or "").strip(),
"region": str(data.get("region") or "").strip()}
def _sha(row):
# ★사이드카·CP 판정·probe 와 **같은 helper** — 화면에서 본 bytes 와 붙는
# bytes 가 갈리지 않는다 (Codex 2026-09-02)
from app.modules.pipeline.grounding_sidecar_writer import row_content_sha256
return row_content_sha256(row, root=root)
got = fr.review_inputs_for(cp, project_id=project_id,
episode_id=episode_id, coordinates=coords,
sha_of=_sha, strict=False)
# ★신원을 못 만든 줄은 **후보가 아니다** — 화면에도 안 나오고 판정도
# 못 받는다(그 까닭은 CP 의 `grounding_fidelity.fault` 에 남는다)
return [x for x in got if not x.get("fault")], Path(root)
def _one_candidate(project_id: str, episode_id: str, want_hash: str):
"""그 신원의 후보 **하나**. 없으면 선다."""
got, root = _candidates(project_id, episode_id)
mine = [x for x in got if x["hash"] == str(want_hash)]
if len(mine) != 1:
raise AppError(
code="grounding_fidelity.no_such_candidate",
message=("그 신원의 후보가 지금 없다 — 사진이나 좌표가 달라졌을 "
"수 있다. 안 본 것에 결정을 붙이지 않는다"),
status_code=409)
return mine[0], root
@router.post("/verdict")
def record_verdict(body: VerdictIn,
db: OrmSession = Depends(get_db),
current_user: UserAccount = Depends(get_current_user)):
"""사람의 판정을 **덧붙인다**. 같은 것을 두 번 보내면 같은 행이 나온다."""
require_project_write(body.project_id, db, current_user)
require_episode(db, body.project_id, body.episode_id)
if body.verdict not in fr.KNOWN_VERDICTS:
raise AppError(code="grounding_fidelity.unknown_verdict",
message=(f"모르는 판정 {body.verdict!r} — "
f"{list(fr.KNOWN_VERDICTS)} 중 하나여야 한다"),
status_code=422)
# ★★★서버가 **다시 만든다** — 클라이언트가 준 것은 신원 열쇠 하나뿐이다
one, _root = _one_candidate(body.project_id, body.episode_id,
body.review_input_hash)
payload = dict(one["payload"])
try:
# ★사람이 **본 것**이 서버가 지금 보는 것과 같은가
fr.assert_observed_matches(payload, body.observed.model_dump())
canonical = fr.canonical_input(payload)
want = fr.input_hash(payload)
except fr.ReviewInputBroken as exc:
raise AppError(code="grounding_fidelity.input_broken",
message=str(exc), status_code=422) from exc
# ★같은 것을 두 번 보내면 **같은 행**이 나온다 — 새로 안 적는다
seen = (db.query(GroundingReferenceFidelityReview)
.filter_by(project_id=body.project_id,
episode_id=body.episode_id,
idempotency_key=body.idempotency_key).one_or_none())
if seen is not None:
return {"id": seen.id, "review_input_hash": seen.review_input_hash,
"verdict": seen.verdict, "created": False}
if body.supersedes_id:
prev = db.get(GroundingReferenceFidelityReview, body.supersedes_id)
if prev is None or prev.review_input_hash != want:
raise AppError(
code="grounding_fidelity.supersedes_mismatch",
message=("대신하려는 앞 결정이 없거나 **다른 입력**의 것이다 — "
"무엇을 고치는지 모르는 채 안 적는다"),
status_code=409)
now = datetime.now(KST).isoformat(timespec="seconds")
row = GroundingReferenceFidelityReview(
id=uuid.uuid4().hex,
project_id=body.project_id, episode_id=body.episode_id,
research_subject_id=payload["research_subject_id"],
final_id=payload["final_id"], purpose=payload["purpose"],
acquisition_identity=payload["acquisition_identity"],
image_sha256=payload["image_sha256"], era=payload["era"],
region=payload["region"],
review_input_json=canonical, review_input_hash=want,
verdict=body.verdict, reason=body.reason,
# ★★서버가 정한다 — 클라이언트가 보낸 이름은 안 쓴다
reviewer_actor=str(getattr(current_user, "username", None)
or getattr(current_user, "id", "")),
reviewed_at=now,
review_contract_version=fr.REVIEW_CONTRACT_VERSION,
observed_json=body.observed.model_dump_json(),
supersedes_id=body.supersedes_id,
idempotency_key=body.idempotency_key,
created_at=now)
db.add(row)
db.commit()
logger.info("[grounding-fidelity] %s %s by %s", body.verdict,
want[:12], row.reviewer_actor)
return {"id": row.id, "review_input_hash": want,
"verdict": row.verdict, "created": True}
class SelectIn(BaseModel):
"""이 대상에 **쓸 사진 한 장** — 판정(맞다/아니다)과 다른 물음이다 (Codex BLOCK 2026-09-02 밤).
새 선택은 같은 대상의 앞 선택을 서버가 `supersedes_id` 로 대신한다(append-only · 원자적)."""
project_id: str
episode_id: str
review_input_hash: str
idempotency_key: str
@router.post("/select")
def select_photo(body: SelectIn,
db: OrmSession = Depends(get_db),
current_user: UserAccount = Depends(get_current_user)):
"""이 대상에 쓸 사진을 고른다. ★선택만으로는 안 붙는다 — 그 사진의 판정이 verified 여야 한다."""
from app.models.project import GroundingReferenceSelection as S
require_project_write(body.project_id, db, current_user)
require_episode(db, body.project_id, body.episode_id)
one, _root = _one_candidate(body.project_id, body.episode_id, body.review_input_hash)
payload = one["payload"]
subject = str(payload["research_subject_id"])
seen = (db.query(S).filter_by(project_id=body.project_id, episode_id=body.episode_id,
idempotency_key=body.idempotency_key).one_or_none())
if seen is not None:
return {"id": seen.id, "research_subject_id": seen.research_subject_id,
"review_input_hash": seen.review_input_hash, "created": False}
# ★★같은 대상의 선택을 **행 잠금으로 직렬화**한다 (Codex NON-BLOCK 2026-09-02 밤): 두 요청이
# 같이 들어오면 둘 다 prev 를 읽고 둘 다 살아남는다. 잠근 뒤 살아 있는 것을 **전부** 대신한다.
locked = (db.query(S).filter_by(project_id=body.project_id, episode_id=body.episode_id,
research_subject_id=subject).with_for_update().all())
live = fr.live_selections([{"id": r.id, "research_subject_id": r.research_subject_id,
"review_input_hash": r.review_input_hash, "selected_at": r.selected_at,
"supersedes_id": r.supersedes_id} for r in locked], subject_id=subject)
prev = live[-1] if live else None
now = datetime.now(KST).isoformat(timespec="seconds")
row = S(id=uuid.uuid4().hex, project_id=body.project_id, episode_id=body.episode_id,
research_subject_id=subject, review_input_hash=body.review_input_hash,
image_sha256=str(payload.get("image_sha256") or ""),
selected_by=str(getattr(current_user, "username", None) or getattr(current_user, "id", "")),
selected_at=now,
# ★살아 있는 앞 선택을 **전부, 같은 commit 에서** 대신한다(여럿이면 `;` 로 잇는다)
supersedes_id=(";".join(str(x["id"]) for x in live) if live else None),
idempotency_key=body.idempotency_key, created_at=now)
db.add(row)
db.commit()
verdict = fr.effective_verdict(
fr.read_reviews(db, project_id=body.project_id, episode_id=body.episode_id),
want_hash=body.review_input_hash)["verdict"]
logger.info("[grounding-fidelity] select %s for %s by %s (replaced %s)", body.review_input_hash[:12],
subject, row.selected_by, (prev or {}).get("review_input_hash", "")[:12])
return {"id": row.id, "research_subject_id": subject, "review_input_hash": body.review_input_hash,
"replaced": (prev or {}).get("review_input_hash"), "verified": verdict == fr.VERDICT_VERIFIED,
"created": True}
@router.get("/login", response_class=HTMLResponse)
def login_page(next: str = ""):
"""작은 로그인 화면. ★같은 origin 이라 쿠키가 그대로 남는다.
★★로그인은 `POST` 라 브라우저 주소창으로는 못 연다. 이 판정 화면을
**격리 canary backend** 에서 볼 때 들어갈 문이 없어서 둔다.
★인증 자체는 기존 `/api/v1/auth/login` 이 한다 — 여기는 그것을 부르는
양식일 뿐이고, 이 파일에 비밀번호가 한 글자도 없다.
"""
import re as _re
# ★★★HTML escape 를 **JS 문자열**에 쓰면 안 된다 (실측 09-02):
# `&` 가 `&` 가 되어 redirect 주소가 깨졌다. escape 로 씨름하지 말고
# **허용 목록**으로 본다 — 이 문의 주소만, URL 에 쓰는 글자만.
where = str(next or "") or f"{router.prefix}/verdicts"
if not (where.startswith(f"{router.prefix}/")
and _re.fullmatch(r"[A-Za-z0-9_\-./?=&%]+", where)):
raise AppError(code="grounding_fidelity.bad_next",
message="갈 곳이 이 문 안이 아니다", status_code=400)
return HTMLResponse(
'로그인'
''
''
'')
@router.get("/review-page", response_class=HTMLResponse)
def review_page(project_id: str, episode_id: str,
db: OrmSession = Depends(get_db),
current_user: UserAccount = Depends(get_current_user)):
"""판정 화면을 **backend 가 직접** 내준다. ★같은 origin 이어야 한다.
★★★왜 (실측 2026-09-02): 화면을 정적 서버(8940)에 두고 API(8000)로
보내면 브라우저가 **CORS 로 막는다** — 이 판에는 CORS 미들웨어가 아예
없다. 같은 문에서 내주면 세션 쿠키가 그대로 간다.
★사진은 `/photo?…&key=<신원>` 이 **서버가 다시 만든 후보만** 내준다 —
임의 경로를 안 받는다 (Codex BLOCK).
"""
verify_project_access(project_id, db=db, user=current_user)
require_episode(db, project_id, episode_id)
from tools.grounding_audit import fidelity_review_page as pg
got, _root = _candidates(project_id, episode_id)
_cp, world, _r = _frozen(project_id, episode_id)
data = (world or {}).get("data") or {}
# ★카드를 만드는 법은 **한 곳**이 안다 — 사진 열쇠만 신원으로 준다
rows = pg.cards_from(got, key_of=lambda x: x["hash"])
sels = fr.read_selections(db, project_id=project_id, episode_id=episode_id)
subjects = {str(s.get("research_subject_id") or "") for s in sels}
picked = {str((fr.effective_selection(sels, subject_id=sid) or {}).get("review_input_hash") or "")
for sid in subjects}
return HTMLResponse(pg.render(
rows, api_base=str(router.prefix),
image_base=(f"{router.prefix}/photo?project_id={project_id}"
f"&episode_id={episode_id}&key="),
coords={"era": str(data.get("era") or ""),
"region": str(data.get("region") or "")},
selected=picked))
@router.get("/photo")
def photo(project_id: str, episode_id: str, key: str,
db: OrmSession = Depends(get_db),
current_user: UserAccount = Depends(get_current_user)):
"""받아 둔 사진 한 장. ★**서버가 다시 만든 후보**만 열쇠로 찾는다.
★★★앞 판은 임의 `path` 를 받아 `PROJECTS_DIR` 의 **부모** 아래를 열어
줬다 — 프로젝트 하나에 권한만 있으면 같은 뿌리의 다른 프로젝트·장부·
산출까지 볼 수 있었다 (Codex BLOCK). 이제 열쇠는 **판정 신원**이고,
그 후보의 파일만, 그것도 **내용 SHA 가 맞을 때만** 연다.
"""
from fastapi.responses import FileResponse
verify_project_access(project_id, db=db, user=current_user)
require_episode(db, project_id, episode_id)
one, root = _one_candidate(project_id, episode_id, key)
want = (root / one["chosen_path"]).resolve()
# ★허용 뿌리는 **이 프로젝트 디렉토리**다 — `projects_dir.parent` 면 옆 프로젝트도 안이라 격리가 안 된다(Codex P1).
if not _inside(want, _project_root(project_id)) or not want.is_file():
raise AppError(code="grounding_fidelity.no_photo",
message="그 사진이 없다", status_code=404)
if _sha_of_file(want) != one["payload"]["image_sha256"]:
raise AppError(
code="grounding_fidelity.photo_changed",
message="사진 내용이 신원과 다르다 — 안 연다", status_code=409)
return FileResponse(path=str(want))
def require_episode(db: OrmSession, project_id: str, episode_id: str) -> None:
"""`episode_id` 가 **이 프로젝트의** 에피소드인가 — DB 로 본다 (Codex PR #82 P1: 인증은 project_id 에만 걸리고
episode_id 는 그대로 경로에 붙어 다른 프로젝트의 CP·사진을 읽을 수 있었다)."""
from app.models.project import Episode
row = (db.query(Episode)
.filter(Episode.id == str(episode_id), Episode.project_id == str(project_id))
.first())
if row is None:
raise AppError(code="episode.not_found", message="그 프로젝트의 에피소드가 아니다",
status_code=404)
def _project_root(project_id: str):
from pathlib import Path
from app.core.config import settings
return (Path(settings.projects_dir) / str(project_id)).resolve()
def _inside(path, base) -> bool:
p, b = path.resolve(), base.resolve()
return p == b or b in p.parents
def _frozen(project_id: str, episode_id: str):
"""이 에피소드의 **얼어붙은 것들**. ★디스크에서 읽기만 한다. ★경로는 이 프로젝트 디렉토리 안이어야 한다."""
import json as _json
from pathlib import Path
from app.core.config import settings
root = Path(settings.projects_dir).parent
ep = (Path(settings.projects_dir) / project_id / "checkpoints"
/ "episodes" / episode_id)
if not _inside(ep, _project_root(project_id)):
raise AppError(code="grounding_fidelity.no_checkpoint",
message="중앙 조사 체크포인트가 없다", status_code=404)
cp_p = ep / "reference_acquisition" / "manifest.json"
w_p = ep / "visual_world_rules" / "manifest.json"
if not cp_p.is_file():
raise AppError(code="grounding_fidelity.no_checkpoint",
message="중앙 조사 체크포인트가 없다", status_code=404)
world = (_json.loads(w_p.read_text(encoding="utf-8"))
if w_p.is_file() else {})
return _json.loads(cp_p.read_text(encoding="utf-8")), world, root
@router.get("/verdicts")
def list_verdicts(project_id: str, episode_id: str,
db: OrmSession = Depends(get_db),
current_user: UserAccount = Depends(get_current_user)):
"""이 에피소드의 판정 전부. ★CP 투영이 이것을 읽는다."""
verify_project_access(project_id, db=db, user=current_user) # ★인자 순서 — (project_id, db, user) (Codex P1: 앞 판은 500)
require_episode(db, project_id, episode_id)
rows = (db.query(GroundingReferenceFidelityReview)
.filter_by(project_id=project_id, episode_id=episode_id)
.order_by(GroundingReferenceFidelityReview.reviewed_at).all())
return {"rows": [{
"id": r.id, "review_input_hash": r.review_input_hash,
"verdict": r.verdict, "reason": r.reason,
"reviewer_actor": r.reviewer_actor, "reviewed_at": r.reviewed_at,
"supersedes_id": r.supersedes_id,
"review_contract_version": r.review_contract_version,
} for r in rows], "review_contract": fr.REVIEW_CONTRACT_VERSION}