"""`v2_chunk` 에서 **참조 정책이 중앙 조사 결과를 읽나**. ★유료 0.

Codex 2026-09-01 acceptance —
  ①fresh `v2_chunk`: 중앙 `selected` → 올바른 하류 의무
  ②못 구함 → 막힘 0 · 강제 0 · raw 사유 보존
  ③`not_applicable` → 「참조 없음」 통계에 안 섞임
  ④legacy 는 **무변**
  ⑤`L`/`LP`/`O` 를 모르는 kind 로 안 낸다
  ⑥갈래마다 집행자가 **하나**

★★줄을 손으로 안 짓는다 — 실제 `ca.run()`·`ca._passthrough` 가 낸 것을
그대로 CP 에 넣고 **프로덕션 스텝 객체**로 읽는다.
"""
from __future__ import annotations

import json
from pathlib import Path

import pytest

from app.modules.pipeline import grounding_central_acquisition as ca
from app.modules.pipeline import grounding_entity_contract as gc
from app.modules.pipeline import reference_acquisition as ra
from app.core.steps.episode_reference_policy_step import (
    EpisodeReferencePolicyStep as Step)


def _session():
    """판정 표를 읽을 실제 session. ★production 스텝은 늘 갖고 있다."""
    from app.core.database import Base, SessionLocal, engine

    Base.metadata.create_all(bind=engine)
    return SessionLocal()


def _register(db, pid):
    """판정 행이 참조하는 **프로젝트 등록**. ★외래 열쇠가 실제로 있다."""
    import uuid as _uuid

    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()


def _step(tmp_path, cp_rows, *, mode="v2_chunk"):
    """프로덕션 스텝 객체 + 중앙 CP 하나. ★`_load_prev_checkpoint` 그대로 쓴다.

    ★★★2026-09-02: 소비자가 **DB 의 사람 판정을 얹어서** CP 를 읽는다.
    그래서 ①사진이 **진짜 파일**이어야 하고 ②`verified` 는 **표에 행이
    있어야** 한다 — 줄에 손으로 적어 둔 값은 투영이 덮어쓴다. 그것이 계약이다.
    """
    import hashlib
    import uuid as _uuid

    from app.core.config import settings
    from app.modules.pipeline import grounding_fidelity_review as _fr
    from app.models.project import GroundingReferenceFidelityReview as _M

    root = tmp_path / "root"
    (root / "projects").mkdir(parents=True, exist_ok=True)
    settings.projects_dir = str(root / "projects")
    pid, eid = f"p{_uuid.uuid4().hex[:8]}", "epi"
    ep = Path(settings.projects_dir) / pid / "checkpoints" / "episodes" / eid
    (ep / "reference_acquisition").mkdir(parents=True, exist_ok=True)
    (ep / "reference_acquisition" / "manifest.json").write_text(
        json.dumps({"status": "completed", "data": {"rows": cp_rows}},
                   ensure_ascii=False), encoding="utf-8")

    # ★사진은 진짜 파일이다 — 투영이 그 내용으로 신원을 만든다
    pics = root / "pics"
    pics.mkdir(exist_ok=True)
    for r in cp_rows:
        rel = str(((r.get("acquisition") or {}).get("chosen") or {}
                   ).get("path") or "")
        if rel:
            f = root / rel
            f.parent.mkdir(parents=True, exist_ok=True)
            f.write_bytes(b"\x89PNG\r\n\x1a\n" + rel.encode("utf-8"))

    s = Step.__new__(Step)
    s.project_id, s.episode_id = pid, eid
    s.db = _session()
    _register(s.db, pid)
    s.project_config = {"grounding_mode": mode}

    # ★줄에 적힌 `verified`/`rejected` 를 **표의 행으로** 옮긴다 — 정본은 표다
    _verdict = {ra.FIDELITY_VERIFIED: _fr.VERDICT_VERIFIED,
                ra.FIDELITY_REJECTED: _fr.VERDICT_REJECTED}
    want = {r.get("research_subject_id"):
            _verdict[str((r.get("grounding_fidelity") or {}).get("state"))]
            for r in cp_rows
            if str((r.get("grounding_fidelity") or {}).get("state") or "")
            in _verdict}
    if want:
        def _sha(row):
            rel = str(((row.get("acquisition") or {}).get("chosen") or {}
                       ).get("path") or "")
            f = root / rel
            return (hashlib.sha256(f.read_bytes()).hexdigest()
                    if rel and f.is_file() else "")

        cp = json.loads((ep / "reference_acquisition" / "manifest.json"
                         ).read_text(encoding="utf-8"))
        now = "2026-09-02T10:00:00+09:00"
        for one in _fr.review_inputs_for(
                cp, project_id=pid, episode_id=eid,
                coordinates={"era": "", "region": ""}, sha_of=_sha,
                strict=False):
            if one.get("fault") or one["payload"][
                    "research_subject_id"] not in want:
                continue
            s.db.add(_M(id=_uuid.uuid4().hex, project_id=pid, episode_id=eid,
                        research_subject_id=one["payload"][
                            "research_subject_id"],
                        final_id=one["payload"]["final_id"],
                        purpose=one["payload"]["purpose"],
                        acquisition_identity=one["payload"][
                            "acquisition_identity"],
                        image_sha256=one["payload"]["image_sha256"],
                        era="", region="",
                        review_input_json=_fr.canonical_input(one["payload"]),
                        review_input_hash=one["hash"],
                        verdict=want[one["payload"]["research_subject_id"]],
                        reason="",
                        reviewer_actor="사람", reviewed_at=now,
                        review_contract_version=_fr.REVIEW_CONTRACT_VERSION,
                        observed_json="{}", supersedes_id=None,
                        idempotency_key=(f"{one['hash']}:"
                                         f"{want[one['payload']['research_subject_id']]}"),
                        created_at=now))
        s.db.commit()
    return s


def _made(rsid, owner, fid, *, disposition, status=None, outcome=None):
    """★실제 `_passthrough` 가 만든 줄 위에서만 손댄다."""
    row = ca._passthrough({"research_subject_id": rsid, "owner_type": owner,
                           "final_id": fid},
                          disposition=disposition, why="시험")
    if status is not None:
        row["status"] = status
    if outcome is not None:
        row["outcome"] = outcome
    return row


def _selected(rsid, owner, fid, *, fidelity=ra.FIDELITY_VERIFIED):
    """★★2026-09-02: 고른 것과 **고증이 확인된 것**은 다른 축이다.

    앞 판은 `selected` 만 주면 붙었다. 이제 소비자가 **둘 다** 본다 —
    coarse 심판은 「무엇인가·보이는가」만 봤고 그 사진이 그 시대·그 지역
    것인지는 아무도 안 봤기 때문이다 (Codex BLOCK).
    """
    row = _made(rsid, owner, fid, disposition=ca.DISP_ACQUIRED,
                status=ra.STATUS_SELECTED, outcome=ra.STATUS_SELECTED)
    row["grounding_fidelity"] = {"state": fidelity}
    # ★★production 의 `ca.run` 은 산 줄마다 **구매 신원**을 적는다. 그것이
    #  없으면 사람 판정을 그 줄에 못 묶는다 — 여기서도 있어야 한다.
    row["identity"] = f"acq-{fid}"
    row["acquisition"] = {"subject_id": rsid, "status": ra.STATUS_SELECTED,
                          "chosen": {"index": 1, "path": f"pics/{fid}.png",
                                     "url": f"https://x.invalid/{fid}.jpg"},
                          "rounds": []}
    return row


class TestOnlyThePolicyLaneIsForced:

    def test_character_and_prop_become_obligations(self, tmp_path):
        s = _step(tmp_path, [_selected("r1", "character", "C01"),
                             _selected("r2", "prop", "P03")])
        forced, blocked, audit = s._central_forced_short_ids()
        assert forced == {"C01", "P03"}
        assert blocked == set()
        assert audit["enforced_elsewhere"] == []

    def test_places_and_outlooks_go_to_their_own_gate(self, tmp_path):
        """★★배경 묶음·아웃룩 자리가 붙이는 것을 **두 벌로** 요구하지 않는다."""
        s = _step(tmp_path, [_selected("r1", "location", "L01"),
                             _selected("r2", "location_part", "LP02"),
                             _selected("r3", "outlook", "O07")])
        forced, blocked, audit = s._central_forced_short_ids()
        assert forced == set(), f"★정책이 남의 갈래를 강제했다: {forced}"
        assert blocked == set()
        got = {a["final_id"]: a["enforced_by"]
               for a in audit["enforced_elsewhere"]}
        assert got == {"L01": gc.ENFORCE_BY_BACKGROUND_SIDECAR,
                       "LP02": gc.ENFORCE_BY_BACKGROUND_SIDECAR,
                       "O07": gc.ENFORCE_BY_OUTLOOK_ATTACHMENT}

    def test_every_owner_has_exactly_one_enforcer(self):
        seen = {}
        for owner in gc.MATERIALIZABLE_OWNER_TYPES:
            gate = gc.enforcement_gate_of(owner)
            assert gate in gc.ENFORCEMENT_GATES, f"★{owner} 에 집행자가 없다"
            seen.setdefault(gate, []).append(owner)
        assert set(seen) == set(gc.ENFORCEMENT_GATES), (
            "★선언만 있고 맡은 갈래가 없는 집행자가 있다")


class TestWhatWasNotFoundNeitherBlocksNorForces:

    def test_unavailable_keeps_its_raw_reason(self, tmp_path):
        s = _step(tmp_path, [
            _made("r1", "character", "C01", disposition=ca.DISP_CAP_REACHED,
                  status=ra.STATUS_RETRYABLE,
                  outcome=ra.STATUS_UNAVAILABLE)])
        forced, blocked, audit = s._central_forced_short_ids()
        assert forced == set() and blocked == set()
        row = audit["reference_unavailable"][0]
        assert row["status"] == ra.STATUS_RETRYABLE, "★raw 상태가 사라졌다"
        assert row["disposition"] == ca.DISP_CAP_REACHED
        assert row["outcome"] == ra.STATUS_UNAVAILABLE

    def test_a_not_applicable_row_is_counted_apart(self, tmp_path):
        s = _step(tmp_path, [
            _made("r1", "prop", "P01",
                  disposition=ca.DISP_NOT_APPLICABLE)])
        _f, _b, audit = s._central_forced_short_ids()
        assert audit["reference_unavailable"] == [], (
            "★애초에 살 것이 아니었던 줄이 「못 구했다」로 섞였다")
        assert [a["final_id"] for a in audit["not_applicable"]] == ["P01"]
        assert audit["not_applicable"][0]["outcome"] is None

    def test_an_explicitly_empty_run_is_a_normal_empty_run(self, tmp_path):
        """★양성 대조 — **명시적인 `rows: []`** 는 정상이다."""
        s = _step(tmp_path, [])
        forced, blocked, audit = s._central_forced_short_ids()
        assert (forced, blocked) == (set(), set())
        assert audit["enforced_elsewhere"] == []


class TestALostRecordIsNotAnEmptyRun:
    """★★★켠 판에서 CP 부재·손상은 「대상 0」이 아니라 **기록 손실**이다.

    Codex BLOCK (2026-09-02): 「지금은 정책도 sidecar 도 동시에 사라져 참조
    없이 최종 이미지 생성까지 갈 수 있습니다.」
    """

    def _bare(self, tmp_path, mode="v2_chunk"):
        from app.core.config import settings

        settings.projects_dir = str(tmp_path)
        s = Step.__new__(Step)
        s.project_id, s.episode_id = "proj", "epi"
        s.db = _session()
        s.project_config = {"grounding_mode": mode}
        return s

    def test_a_missing_checkpoint_stops(self, tmp_path):
        s = self._bare(tmp_path)
        with pytest.raises(ca.ProjectionContractError):
            s._central_forced_short_ids()

    @pytest.mark.parametrize("cp,why", [
        ({"status": "running", "data": {"rows": []}}, "덜 돈 판"),
        ({"status": "completed"}, "`data` 없음"),
        ({"status": "completed", "data": {}}, "`rows` 키 없음"),
        ({"status": "completed", "data": {"rows": None}}, "rows=None"),
        ({"status": "completed", "data": {"rows": {"a": 1}}}, "rows=dict"),
    ])
    def test_a_broken_checkpoint_stops(self, cp, why):
        with pytest.raises(ca.ProjectionContractError):
            ca.acquisition_projection(cp, required=True)

    def test_the_old_road_still_folds_quietly(self):
        """★음성 대조 — 안 켠 판은 **예전 그대로** 빈 목록이다."""
        for cp in (None, {}, {"data": {}}, {"data": {"rows": []}}):
            assert ca.acquisition_projection(cp) == []

    def test_a_contradictory_row_stops(self):
        """★raw 상태와 처분이 어긋나면 선다 — 둘을 각각 믿지 않는다."""
        row = _made("r1", "prop", "P01", disposition=ca.DISP_ACQUIRED,
                    status=ra.STATUS_SELECTED,
                    outcome=ra.STATUS_UNAVAILABLE)
        with pytest.raises(ca.ProjectionContractError) as e:
            ca.acquisition_projection({"status": "completed",
                                       "data": {"rows": [row]}},
                                      required=True)
        assert "어긋난다" in str(e.value)

    def test_the_sidecar_door_uses_the_same_rule(self):
        """★sidecar 쪽도 **같은 한 곳**이 판정한다 — 규칙이 두 벌이 아니다."""
        from app.modules.pipeline import grounding_sidecar_writer as sw

        with pytest.raises(ca.ProjectionContractError):
            sw.assert_central_checkpoint(None)
        sw.assert_central_checkpoint({"status": "completed",
                                      "data": {"rows": []}})


class TestTheKindThatReachesTheCard:
    """★★끝점 — 정책이 낸 것이 실제 `required_refs` 에서 **아는 kind** 인가."""

    def _refs(self, forced):
        from app.core.steps.render_prompt_card import (
            _research_forced_from_policy, research_forced_refs)

        manifest = {"grounding": {"research_required_short_ids":
                                  sorted(forced)}}
        got = _research_forced_from_policy(manifest)
        return research_forced_refs(got, [], visible_entities=sorted(forced))

    def test_policy_lane_ids_become_known_kinds(self):
        out = self._refs({"C01", "P03"})
        assert {r["kind"] for r in out} == {"character", "prop"}

    def test_a_place_never_reaches_the_card_as_a_kind(self):
        """★`location` 이라는 kind 는 카드 어휘에 **없다**."""
        assert self._refs({"L01", "LP02", "O07"}) == []

    def test_the_kind_comes_from_the_contract_not_the_owner(self):
        """★갈래 이름을 kind 로 그대로 내면 여기서 어긋난다."""
        for owner in gc.owners_enforced_by(gc.ENFORCE_BY_POLICY):
            kind = gc.reference_kind_of_owner(owner)
            assert kind is not None
            sid = gc.OWNER_PREFIX[owner] + "09"
            got = self._refs({sid})
            assert [r["kind"] for r in got] == [kind]


class TestTheRealExecuteTakesTheBranch:
    """★★★조립부 말고 **`_execute` 끝점**에서 잰다.

    `_central_forced_short_ids` 를 직접 부르면 「갈래가 배선됐나」를 못 본다 —
    앞에 그 부류로 여러 번 데었다.
    """

    def _run(self, tmp_path, cp_rows, *, mode="v2_chunk"):
        from app.core.config import settings

        s = _step(tmp_path, cp_rows, mode=mode)
        # ★`_step` 이 옮긴 **실제 자리**에 넣는다 — 손으로 다시 조립하면
        #  둘이 갈려 스텝이 못 찾는다
        root = (Path(settings.projects_dir) / s.project_id / "checkpoints"
                / "episodes" / s.episode_id)

        def put(step, data):
            d = root / step
            d.mkdir(parents=True, exist_ok=True)
            (d / "manifest.json").write_text(
                json.dumps({"status": "completed", "data": data},
                           ensure_ascii=False), encoding="utf-8")

        put("shot_director", {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 0,
             "visible_entity_ids": ["C01", "P03", "L01"]}]}]})
        put("shot_selection", {"scenes": [{"scene_index": 1,
                                           "selected_shot_indices": [0]}]})

        # ★★★session 을 **대역으로 갈아 끼우지 않는다** (2026-09-02).
        #  앞 판은 모든 조회에 빈 목록을 주는 `_DB` 를 끼웠는데, 이제
        #  소비자가 **사람 판정 표**를 그 session 으로 읽는다 — 갈아 끼우면
        #  `_step` 이 넣어 둔 판정이 통째로 안 보이고, 「의무가 안 됐다」로
        #  잘못 읽힌다. 실제 session 은 이 프로젝트에 다른 행이 없으므로
        #  엔티티 조회는 어차피 빈 목록이다.
        return s._execute()

    def test_the_manifest_carries_the_central_result(self, tmp_path):
        got = self._run(tmp_path, [_selected("r1", "character", "C01"),
                                   _selected("r2", "prop", "P03"),
                                   _selected("r3", "location", "L01")])
        g = got["data"]["grounding"]
        assert g["research_required_short_ids"] == ["C01", "P03"], (
            "★중앙이 고른 것이 의무가 안 됐다 — 갈래가 안 배선됐다")
        assert [a["final_id"] for a in g["enforced_elsewhere"]] == ["L01"]

    def test_the_forced_subject_is_reference_required_in_the_policy(
            self, tmp_path):
        """★끝점 — 강제가 **정책 칸까지** 간다."""
        got = self._run(tmp_path, [_selected("r1", "character", "C01")])
        assert got["data"]["policy"]["C01"]["mode"] == "reference_required"

    def test_an_unavailable_run_does_not_stand(self, tmp_path):
        """★★못 구했다고 **사람을 기다리지 않는다** (HITL 0)."""
        got = self._run(tmp_path, [
            _made("r1", "character", "C01",
                  disposition=ca.DISP_CAP_REACHED,
                  status=ra.STATUS_RETRYABLE,
                  outcome=ra.STATUS_UNAVAILABLE)])
        g = got["data"]["grounding"]
        assert g["research_required_short_ids"] == []
        row = g["reference_unavailable"][0]
        assert row["final_id"] == "C01"
        assert row["status"] == ra.STATUS_RETRYABLE, "★raw 상태가 사라졌다"
        assert row["disposition"] == ca.DISP_CAP_REACHED
        assert got["data"]["policy"]["C01"]["mode"] != "reference_required"

    def test_legacy_execute_writes_no_audit_block(self, tmp_path):
        """★음성 대조 — 옛 판 산출에는 새 칸이 **아예 없다**."""
        got = self._run(tmp_path, [_selected("r1", "character", "C01")],
                        mode="v2")
        g = got["data"]["grounding"]
        assert set(g) == {"research_required_short_ids", "override_used"}, (
            f"★legacy 산출이 달라졌다: {sorted(g)}")
        assert g["research_required_short_ids"] == []


class TestLegacyIsUntouched:
    """★음성 대조 — 옛 판은 **한 글자도** 안 달라진다."""

    def test_legacy_never_reads_the_central_checkpoint(self, tmp_path):
        s = _step(tmp_path, [_selected("r1", "character", "C01")],
                  mode="v2")
        assert s._uses_central() is False
        # ★옛 갈래는 db 가 없으면 빈손이다 — 중앙 CP 가 있어도 안 본다
        assert s._research_forced_short_ids() == (set(), set())

    def test_the_audit_block_does_not_exist_in_legacy(self):
        """★legacy manifest 에 새 칸이 생기면 byte 가 달라진다."""
        import inspect

        src = inspect.getsource(Step._execute)
        assert "if audit:" in src, "★감사 칸이 조건 없이 붙는다"


class TestSelectedIsNotYetGroundingVerified:
    """★★★coarse `selected` 가 그대로 붙던 것 (Codex BLOCK 2026-09-02).

    심판은 「무엇인가 · 보이는가」만 본다 — 시대·나라·정확성은 **안 묻는다**
    (사용자 확정 08-31). 그러니 `selected` 는 「그 종류의 사진을 하나 골랐다」
    일 뿐 「그 시대·그 지역 것이 맞다」가 아니다.
    """

    def test_an_unverified_selection_is_not_attached(self, tmp_path):
        s = _step(tmp_path, [_selected("r1", "character", "C01",
                                       fidelity=ra.FIDELITY_UNVERIFIED)])
        forced, blocked, audit = s._central_forced_short_ids()
        assert forced == {"C01"}, "★자동 선택이 안 붙었다 (HITL 0)"
        assert blocked == set(), "★막으면 안 된다"
        assert [r["research_subject_id"]
                for r in audit["fidelity_unverified"]] == ["r1"]
        assert audit["reference_unavailable"] == [], (
            "★못 구한 것과 **안 본 것**을 섞으면 안 된다")

    def test_a_row_with_no_fidelity_field_is_unverified(self, tmp_path):
        """★칸이 아예 없으면 **아직 아니다** — 「모르는 것」이 통과가 되면 안 된다."""
        row = _made("r1", "character", "C01", disposition=ca.DISP_ACQUIRED,
                    status=ra.STATUS_SELECTED, outcome=ra.STATUS_SELECTED)
        forced, _blocked, audit = _step(tmp_path,
                                        [row])._central_forced_short_ids()
        assert forced == {"C01"}, "★HITL 0 — 판정 칸이 없어도 붙는다"
        assert len(audit["fidelity_unverified"]) == 1

    def test_a_verified_selection_is_attached(self, tmp_path):
        """★음성 대조 — 확인된 것은 **여전히 붙는다**."""
        s = _step(tmp_path, [_selected("r1", "character", "C01")])
        forced, _b, audit = s._central_forced_short_ids()
        assert forced == {"C01"}
        assert audit["fidelity_unverified"] == []

    def test_a_human_rejection_is_not_reported_as_unverified(self, tmp_path):
        """★사람이 **보고 거절한 것**은 「아무도 안 본 것」과 다른 칸이다.

        canary ② 실측: LP04 는 사람이 거절했는데 감사가 `fidelity_unverified`
        로 보고해 「아직 안 봤다」로 읽혔다. 둘 다 안 붙고 안 막는 것은 같다.
        """
        s = _step(tmp_path, [_selected("r1", "character", "C01",
                                       fidelity=ra.FIDELITY_REJECTED)])
        forced, blocked, audit = s._central_forced_short_ids()
        assert forced == set() and blocked == set()
        assert [r["research_subject_id"]
                for r in audit["fidelity_rejected"]] == ["r1"]
        assert audit["fidelity_unverified"] == [], "★거절을 미확인으로 셌다"
        assert audit["reference_unavailable"] == []