"""★★★HITL 0 — 사람 판정은 **production 사슬에 닿지 않는다** (사용자 2026-09-03 최상위 불변식).
production 입구(정책 lane · sidecar)는 **날것 CP** 를 읽고 자동 선택 한 장이 그대로 붙는다.
사람 행이 0건이어도 붙어야 하고, 판정·선택은 도구(`apply_reviews`)의 override 일 뿐이다.
★아래 옛 전제(「판정이 닿아야 한다」)는 뒤집혔다 — 검토 75건 뒤에도 usable 1/18 이던 결함.
--- 옛 문서 ---
★★★사람 판정이 **production 사슬**에 닿나 (Codex 조건 5 · 2026-09-02).

앞 판은 `apply_reviews` 를 만들어 놓고 **production 호출자가 0곳**이었다.
그래서 「판정 후 3장 부착」은 **시험이 직접 부른 값**이었지, 실제
`episode_reference_policy → sidecar → scene_detail` 사슬의 결과가 아니었다.

이 시험은 **`apply_reviews` 를 직접 부르지 않는다.** 실제 PostgreSQL 의
판정 행과 실제 저장된 체크포인트를 두고, 소비자 **입구**만 태운다.
"""
from __future__ import annotations

import json
import uuid
from pathlib import Path

import pytest

from app.models.project import Episode

from app.core.database import Base, SessionLocal, engine
from app.models.project import GroundingReferenceFidelityReview as M
from app.modules.pipeline import grounding_bundle_projection as bp
from app.modules.pipeline import grounding_fidelity_review as fr
from app.modules.pipeline import grounding_reference_bundle as rb
from app.modules.pipeline import grounding_sidecar_writer as sw
from app.modules.pipeline import reference_acquisition as ra

ERA, REGION = "가나다 무렵", "라마바 지방"


def _acq(sid, fid, purpose, owner, sha_name):
    """중앙 조사가 낸 줄 하나 — `selected` 이고 사진이 있다."""
    return {
        "research_subject_id": sid, "identity": f"acq-{fid}",
        "disposition": "acquired", "status": ra.STATUS_SELECTED,
        "outcome": ra.STATUS_SELECTED, "why": "", "why_unbought": None,
        "downstream_blocked": False,
        "acquisition": {
            "subject_id": sid, "status": ra.STATUS_SELECTED,
            "chosen": {"index": 1, "path": f"pics/{sha_name}.png",
                       "url": f"https://x.invalid/{sha_name}.jpg"},
            "rounds": [{"round_no": 1, "terms_native": [f"{ERA} {REGION} 것"],
                        "downloaded_candidates": [
                            {"index": 1, "path": f"pics/{sha_name}.png",
                             "url": f"https://x.invalid/{sha_name}.jpg"}]}]},
        "source_evidence": {},
        "ledger_row": {"research_subject_id": sid, "owner_type": owner,
                       "final_id": fid, "parent_final_id": None,
                       "purpose": purpose, "covers": [fid]},
    }


@pytest.fixture
def world(tmp_path):
    """실제 저장된 CP 둘 + 실제 사진 파일들 + 실제 스텝 객체."""
    from app.core.config import settings
    from app.core.steps.episode_reference_policy_step import (
        EpisodeReferencePolicyStep as Policy)

    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]}"

    # ★사진은 **진짜 파일**이다 — 내용 지문을 서버가 직접 낸다
    pics = root / "pics"
    pics.mkdir()
    same = b"\x89PNG\r\n\x1a\n" + b"SAME" * 8
    for name, blob in (("a", same), ("b", same),      # ★a·b 는 **같은 사진**
                       ("c", b"\x89PNG\r\n\x1a\nCCCC"),
                       ("d", b"\x89PNG\r\n\x1a\nDDDD")):
        (pics / f"{name}.png").write_bytes(blob)

    rows = [_acq("rs_a", "LP01", "detail", "location_part", "a"),
            _acq("rs_b", "LP04", "detail", "location_part", "b"),
            _acq("rs_c", "LP03", "detail", "location_part", "c"),
            _acq("rs_d", "P01", "", "prop", "d")]
    ep = (Path(settings.projects_dir) / pid / "checkpoints" / "episodes"
          / eid)
    for step, data in (("reference_acquisition", {"rows": rows}),
                       ("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()
    # ★권한 문(`verify_project_access`)이 **실제 등록 행**을 본다
    from app.models.catalog import ProjectRegistry, UserAccount

    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) — 이 프로젝트의 에피소드여야 한다
    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, "rows": rows,
               "root": root, "uid": uid}
    finally:
        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 _hashes(w):
    """서버가 **스스로** 낸 판정 신원들. ★시험이 hash 를 안 지어낸다."""
    import hashlib

    from app.core.config import settings

    cp_path = (Path(settings.projects_dir) / w["pid"] / "checkpoints"
               / "episodes" / w["eid"] / "reference_acquisition"
               / "manifest.json")
    cp = json.loads(cp_path.read_text(encoding="utf-8"))

    def _sha(row):
        rel = str(((row.get("acquisition") or {}).get("chosen") or {}
                   ).get("path") or "")
        p = w["root"] / rel
        return hashlib.sha256(p.read_bytes()).hexdigest() if p.is_file() else ""

    return {one["payload"]["final_id"]: one["hash"]
            for one in fr.review_inputs_for(
                cp, project_id=w["pid"], episode_id=w["eid"],
                coordinates={"era": ERA, "region": REGION}, sha_of=_sha)}


def _record(w, fid, verdict):
    """**끝점으로** 판정을 넣는다 — 시험이 표를 손으로 안 채운다."""
    from app.api.v1 import grounding_fidelity as api

    h = _hashes(w)[fid]
    one, _root = api._one_candidate(w["pid"], w["eid"], h)
    p = one["payload"]
    body = api.VerdictIn(
        project_id=w["pid"], episode_id=w["eid"], review_input_hash=h,
        verdict=verdict, reason="", idempotency_key=f"{h}:{verdict}",
        observed=api.ObservedIn(image_sha256=p["image_sha256"],
                                era=p["era"], region=p["region"]))

    class _User:
        id, username, role = w["uid"], w["uid"], "admin"

    return api.record_verdict(body, db=w["db"], current_user=_User())


def _attached(w):
    """★**production 입구**로 읽는다 — 시험이 `apply_reviews` 를 안 부른다."""
    cp = w["step"]._load_prev_checkpoint("reference_acquisition")   # ★날것 CP — production 입구
    got = sw.members_for_shot(
        cp, ["LP01", "LP03", "LP04", "P01"],
        content_sha_of=lambda r: f"sha_{(r.get('ledger_row') or {}).get('final_id')}",
        coordinate_of=lambda r: {
            "source": rb.SOURCE_FILE,
            "path": f"refs/{(r.get('ledger_row') or {}).get('final_id')}.jpg"})
    return sorted(m["subject_final_id"] for m in got
                  if m.get("member_identity")), got


class TestTheChainReadsTheDatabase:
    def test_no_verdict_attaches_every_selected_row(self, world):
        """★★★HITL 0 — 사람 행 0건: 자동 선택 세 장이 그대로 붙고 정책 lane 도 강제한다."""
        on, _m = _attached(world)
        assert on == ["LP01", "LP03", "LP04"], f"★{on}"
        forced, blocked, _a = world["step"]._central_forced_short_ids()
        assert forced == {"P01"} and blocked == set()

    def test_a_verified_row_changes_nothing_in_production(self, world):
        _record(world, "LP01", fr.VERDICT_VERIFIED)
        on, _m = _attached(world)
        assert on == ["LP01", "LP03", "LP04"], f"★{on}"

    def test_a_rejection_does_not_reach_production(self, world):
        """★production 은 판정 표를 안 읽는다 — 거절은 도구 산출에서만 보인다."""
        _record(world, "LP01", fr.VERDICT_VERIFIED)
        _record(world, "LP04", fr.VERDICT_REJECTED)
        on, _m = _attached(world)
        assert on == ["LP01", "LP03", "LP04"], f"★{on}"
        tool = fr.central_cp_with_reviews(world["step"], required=True)
        by = {(r.get("ledger_row") or {}).get("final_id"): r for r in tool["data"]["rows"]}
        assert by["LP04"]["outcome"] == ra.STATUS_UNAVAILABLE and by["LP01"]["outcome"] == ra.STATUS_SELECTED

    def test_the_same_photo_is_not_merged_across_subjects(self, world):
        """★★LP01 과 LP04 는 **같은 bytes** 다. 하나만 맞다고 해도
        **그쪽에만** 붙는다 — 합쳐지지 않는다."""
        _record(world, "LP01", fr.VERDICT_VERIFIED)
        _record(world, "LP04", fr.VERDICT_REJECTED)
        on, members = _attached(world)
        assert on == ["LP01", "LP03", "LP04"]
        ids = {m["subject_final_id"]: m["member_identity"] for m in members if m.get("member_identity")}
        assert ids["LP01"] != ids["LP04"], "★같은 bytes 라고 멤버가 합쳐졌다"

    def test_the_policy_lane_sees_the_prop(self, world):
        """★`P01` 은 prop 이라 sidecar 가 아니라 **정책 lane** 으로 간다."""
        _record(world, "P01", fr.VERDICT_VERIFIED)
        forced, blocked, audit = world["step"]._central_forced_short_ids()
        assert forced == {"P01"}
        assert blocked == set()
        on, _m = _attached(world)
        assert "P01" not in on

    def test_this_file_does_not_project_for_itself(self):
        """★★이 시험이 투영을 **직접 부르면** 제 답을 제가 만든다.

        ★글자로 막으면 **이 시험의 이름**이 걸린다(그래서 한 번 걸렸다).
        `ast` 로 **부르는 자리**만 본다.
        """
        import ast

        tree = ast.parse(Path(__file__).read_text(encoding="utf-8"))
        called = {getattr(n.func, "attr", "") or getattr(n.func, "id", "")
                  for n in ast.walk(tree) if isinstance(n, ast.Call)}
        assert "apply_reviews" not in called, "★시험이 직접 투영했다"
        assert "_load_prev_checkpoint" in called, (
            "★production 입구(날것 CP)를 안 태운다 — 이 시험이 아무것도 안 잠근다")


class TestChangingTheVerdictInvalidatesTheOldResult:
    """★사람이 마음을 바꾸면 끝난 판이 **옛 결과를 되쓰면** 안 된다."""

    def test_the_policy_audit_has_no_review_digest(self, world):
        """★HITL 0: production 감사·판단은 사람 표와 무관하다."""
        f0, _b, audit = world["step"]._central_forced_short_ids()
        assert "fidelity_reviews" not in audit, "★production 감사가 사람 표 지문을 든다"
        _record(world, "P01", fr.VERDICT_VERIFIED)
        _record(world, "LP01", fr.VERDICT_REJECTED)
        f1, _b1, audit1 = world["step"]._central_forced_short_ids()
        assert f1 == f0 == {"P01"} and audit1 == audit, "★사람 판정이 production 판단을 움직였다"

    def test_the_digest_is_not_in_the_purchase_identity(self):
        """★판정이 바뀌었다고 **사진을 다시 사지 않는다**."""
        import inspect

        from app.modules.pipeline import grounding_central_acquisition as ca

        src = inspect.getsource(ca.identity_of)
        assert "fidelity" not in src and "review" not in src


class TestItFailsClosedWhenTheRecordIsBroken:
    def test_an_unreadable_table_stops(self, world):
        class _Broken:
            def query(self, *_a, **_k):
                raise RuntimeError("표가 없다")

        world["step"].db = _Broken()
        with pytest.raises(fr.ReviewsUnreadable):
            fr.central_cp_with_reviews(world["step"], required=True)

    def test_a_missing_photo_only_loses_that_row(self, world):
        """★★사진 **하나**가 없다고 판 전체를 죽이지 않는다.

        그 줄은 어차피 **안 붙는다**(미확인) — 위험한 것은 「확인된 것이
        조용히 안 붙는 것」과 「안 본 것이 붙는 것」인데 둘 다 안 일어난다.
        ★조용하지도 않다 — 왜 안 붙는지가 CP 에 남는다.
        """
        _record(world, "LP01", fr.VERDICT_VERIFIED)
        (world["root"] / "pics" / "c.png").unlink()      # ★LP03 의 사진
        cp = fr.central_cp_with_reviews(world["step"], required=True)
        by = {(r.get("ledger_row") or {}).get("final_id"): r
              for r in cp["data"]["rows"]}
        assert by["LP03"]["grounding_fidelity"]["state"] == \
            ra.FIDELITY_UNVERIFIED
        assert by["LP03"]["grounding_fidelity"]["fault"], "★까닭이 안 남았다"
        # ★HITL 0: production 의 붙는 조건은 판정 칸을 안 읽는다 — 파일이 없는 줄은 붙이는
        #  자리(content sha)가 선다. 여기서는 도구가 까닭을 남겼는지만 본다.
        assert ra.usable_as_reference(by["LP01"])

    def test_a_missing_checkpoint_stops_at_the_consumer(self, world):
        """★CP 부재는 **소비자의 문**이 말한다 — 투영이 그 자리다.

        읽는 쪽에서 또 세우면 「무엇이 없었나」가 바뀐다.
        """
        from app.core.config import settings
        from app.modules.pipeline import grounding_central_acquisition as ca

        (Path(settings.projects_dir) / world["pid"] / "checkpoints"
         / "episodes" / world["eid"] / "reference_acquisition"
         / "manifest.json").unlink()
        assert fr.central_cp_with_reviews(world["step"], required=True) is None
        with pytest.raises(ca.ProjectionContractError):
            world["step"]._central_forced_short_ids()


class TestChangingTheVerdictReopensThePolicy:
    """★★★정책 스텝에 `_config_hash` 가 **없었다** (Codex BLOCK 09-02).

    그러면 `step_runner` 가 `compute_config_hash(project_config)` 로
    떨어지는데, **사람 판정은 `project_config` 를 안 바꾼다** — 완료된 정책
    CP 가 그대로 current 로 읽히고 옛 `research_required_short_ids` 가
    되쓰인다. `scene_detail` 은 다시 도는데 그 **입력**이 안 도는,
    두 소비자 중 한쪽만 깨지는 자리였다.

    ★`_central_forced_short_ids` 를 직접 부르는 시험으로 대신하지 않는다 —
    **재개가 무엇을 보는가**가 이 판의 질문이다.
    """

    def test_legacy_and_v2_hashes_do_not_move(self):
        """★★켠 판이 아닌 에피소드는 **한 바이트도** 안 바뀐다."""
        from app.core.step_runner import compute_config_hash
        from app.core.steps.episode_reference_policy_step import (
            EpisodeReferencePolicyStep as S)

        for mode in ("legacy", "v2", ""):
            s = S.__new__(S)
            s.project_config = {"grounding_mode": mode} if mode else {}
            assert s._config_hash() == compute_config_hash(s.project_config), \
                mode

    def test_the_resume_hash_does_not_move_when_a_verdict_lands(self, world):
        """★★★HITL 0 — 재개 지문은 사람 표와 무관하다. 움직이면 사람 표가 운영 절차가 된다."""
        s = world["step"]
        before = s._config_hash()
        _record(world, "P01", fr.VERDICT_VERIFIED)
        after = s._config_hash()
        assert after == before, "★사람 판정이 production 재개 지문을 움직였다 (HITL 0 위반)"
        # ★그리고 **되돌리면** 또 움직인다
        h = _hashes(world)["P01"]
        prev = (world["db"].query(
            __import__("app.models.project", fromlist=["M"]
                       ).GroundingReferenceFidelityReview)
            .filter_by(review_input_hash=h).one())
        _supersede(world, "P01", fr.VERDICT_REJECTED, prev.id)
        assert s._config_hash() == before

    def test_the_step_runner_uses_this_hash_not_the_fallback(self):
        """★★`step_runner` 가 **step-local** 을 쓰는지 — 없으면 fallback 이다."""
        import inspect

        from app.core import step_runner as sr
        from app.core.steps.episode_reference_policy_step import (
            EpisodeReferencePolicyStep as S)

        assert callable(getattr(S, "_config_hash", None)), (
            "★없으면 `compute_config_hash(project_config)` 로 떨어진다")
        src = inspect.getsource(sr.StepRunner)
        assert 'local_hash_fn = getattr(self, "_config_hash", None)' in src

    def test_the_digest_stays_out_of_the_purchase_identity(self):
        import inspect

        from app.core.steps.episode_reference_policy_step import (
            EpisodeReferencePolicyStep as S)
        from app.modules.pipeline import grounding_central_acquisition as ca

        assert "fidelity" not in inspect.getsource(ca.identity_of)
        assert "reviews_digest_for" not in inspect.getsource(S._config_hash), "★HITL 0 — 지문에 사람 표 금지"


def _supersede(w, fid, verdict, prev_id):
    """앞 결정을 **대신하는** 새 행 — 끝점으로 넣는다."""
    from app.api.v1 import grounding_fidelity as api

    h = _hashes(w)[fid]
    one, _root = api._one_candidate(w["pid"], w["eid"], h)
    p = one["payload"]
    body = api.VerdictIn(
        project_id=w["pid"], episode_id=w["eid"], review_input_hash=h,
        verdict=verdict, reason="다시 보니 아니다",
        idempotency_key=f"{h}:{verdict}:2", supersedes_id=prev_id,
        observed=api.ObservedIn(image_sha256=p["image_sha256"],
                                era=p["era"], region=p["region"]))

    class _User:
        id, username, role = w["uid"], w["uid"], "admin"

    return api.record_verdict(body, db=w["db"], current_user=_User())


class TestASupersededVerdictChangesWhatAttaches:
    def test_reverting_a_verdict_changes_nothing_in_production(self, world):
        """★★★HITL 0 — 사람이 마음을 바꿔도 production 부착은 그대로다. 기록만 쌓인다."""
        from app.models.project import GroundingReferenceFidelityReview as M

        _record(world, "LP01", fr.VERDICT_VERIFIED)
        on, _m = _attached(world)
        assert on == ["LP01", "LP03", "LP04"]

        h = _hashes(world)["LP01"]
        prev = world["db"].query(M).filter_by(review_input_hash=h).one()
        _supersede(world, "LP01", fr.VERDICT_REJECTED, prev.id)

        on2, _m2 = _attached(world)
        assert on2 == on, "★사람 판정이 production 부착을 움직였다"
        # ★앞 행은 **안 지워진다** — 무엇이 언제 바뀌었는지가 남는다
        assert world["db"].query(M).filter_by(
            review_input_hash=h).count() == 2

    def test_the_policy_forced_set_follows_too(self, world):
        _record(world, "P01", fr.VERDICT_VERIFIED)
        forced, _b, _a = world["step"]._central_forced_short_ids()
        assert forced == {"P01"}

        from app.models.project import GroundingReferenceFidelityReview as M

        h = _hashes(world)["P01"]
        prev = world["db"].query(M).filter_by(review_input_hash=h).one()
        _supersede(world, "P01", fr.VERDICT_REJECTED, prev.id)
        forced2, _b2, _a2 = world["step"]._central_forced_short_ids()
        # ★HITL 0: 사람의 거절은 production 정책 lane 을 움직이지 않는다 — 도구 산출에서만 보인다
        assert forced2 == {"P01"}, "★사람 거절이 production 강제를 움직였다 (HITL 0 위반)"
        tool = fr.central_cp_with_reviews(world["step"], required=True)
        p01 = next(r for r in tool["data"]["rows"] if (r.get("ledger_row") or {}).get("final_id") == "P01")
        assert p01["outcome"] == ra.STATUS_UNAVAILABLE