"""★★★집행자가 **둘**인데 한쪽만 막고 「붙는 것 0」이라 보고했다.

Codex 재리뷰 2026-09-02: `episode_reference_policy` 는 `fidelity` 를 봤지만
`grounding_bundle_projection` 의 배경·아웃룩 두 갈래는 `outcome == selected`
만 보고 그대로 붙이고 있었다. 실물 `selected` LP01/03/04/05 는 배경 sidecar
경로에서 **여전히 멤버가 됐다**.

이 시험은 **실물 CP** 로 `grounding_sidecar_writer` 의 공개 입구 두 개를
그대로 태운다 — 조립부에서 재지 않는다.
"""
from __future__ import annotations

import copy
import json
from pathlib import Path

import pytest

from app.modules.pipeline import grounding_sidecar_writer as sw
from app.modules.pipeline import reference_acquisition as ra

CANARY = Path("/Users/manta/Documents/Projects/TheRoad-I1/artifact"
              "/canary_69e821758f3d/projects"
              "/8e2e65b7-e910-4b12-b081-c23de0affab5/checkpoints/episodes"
              "/9e64c302-82e6-4407-b560-8c70c2ab7192/reference_acquisition"
              "/manifest.json")


def _sha_of(row):
    return f"sha_{(row.get('ledger_row') or {}).get('final_id')}"


def _coord_of(row):
    from app.modules.pipeline import grounding_reference_bundle as rb

    return {"source": rb.SOURCE_FILE,
            "path": f"refs/{(row.get('ledger_row') or {}).get('final_id')}.jpg"}


@pytest.fixture
def real_cp():
    if not CANARY.is_file():
        pytest.skip("실물 canary CP 가 없다 — 이 기계에서만 도는 끝점")
    return json.loads(CANARY.read_text(encoding="utf-8"))


def _visible(cp):
    """이 CP 가 든 배경 갈래 final_id 전부."""
    out = []
    for r in (cp.get("data") or {}).get("rows") or ():
        led = r.get("ledger_row") or {}
        if led.get("final_id"):
            out.append(str(led["final_id"]))
    return sorted(set(out))


class TestTheRealCheckpointAttachesNothingUnverified:
    def test_the_selected_five_are_all_unverified(self, real_cp):
        rows = (real_cp.get("data") or {}).get("rows") or []
        sel = [r for r in rows if r.get("outcome") == ra.STATUS_SELECTED]
        assert len(sel) == 5, f"★고른 것이 {len(sel)}장이다 — 시험 전제가 바뀜"
        assert all(ra.fidelity_of(r) == ra.FIDELITY_UNVERIFIED for r in sel)
        assert all(ra.usable_as_reference(r) for r in sel), "★HITL 0 — 자동 선택은 그대로 붙는다"

    def test_the_background_entry_requires_nothing(self, real_cp):
        """★★공개 입구 — `members_for_shot`. 조립부에서 안 잰다."""
        got = sw.members_for_shot(real_cp, _visible(real_cp),
                                  content_sha_of=_sha_of,
                                  coordinate_of=_coord_of)
        assert got, "★멤버가 하나도 안 나왔다 — 이 시험이 아무것도 안 잠근다"
        attached = sorted(m["subject_final_id"] for m in got if m.get("member_identity"))
        want = sorted({str((r.get("ledger_row") or {}).get("final_id"))
                       for r in real_cp["data"]["rows"]
                       if r.get("outcome") == ra.STATUS_SELECTED
                       and (r.get("ledger_row") or {}).get("purpose")})
        assert want and attached == want, f"★자동 선택이 안 붙었다 (HITL 0): {attached} vs {want}"

    def test_the_outlook_entry_requires_nothing(self, real_cp):
        """★★아웃룩 갈래도 같은 문을 지난다."""
        from app.modules.pipeline import grounding_bundle_projection as bp
        from app.modules.pipeline import grounding_reference_bundle as rb

        rows = (real_cp.get("data") or {}).get("rows") or []
        ids = [bp.outlook_ref_id(r) for r in rows]
        rpc = {rb.RPC_MEMBERS_KEY: []}
        got = sw.outlook_members_for_shot(real_cp, rpc,
                                          content_sha_of=_sha_of,
                                          coordinate_of=_coord_of)
        assert [m for m in got if m.get("member_identity")] == []
        assert ids is not None

    def test_a_verified_copy_attaches_by_the_canonical_contract(self,
                                                               real_cp):
        """★★★양성 대조 — LP 하나를 확인하면 **그것만** 정본대로 붙는다."""
        cp = copy.deepcopy(real_cp)
        rows = (cp.get("data") or {}).get("rows") or []
        target = next(r for r in rows
                      if r.get("outcome") == ra.STATUS_SELECTED
                      and (r.get("ledger_row") or {}).get("owner_type")
                      == "location_part")
        target["grounding_fidelity"] = {"state": ra.FIDELITY_VERIFIED}
        fid = str((target["ledger_row"] or {}).get("final_id"))

        got = sw.members_for_shot(cp, _visible(cp), content_sha_of=_sha_of,
                                  coordinate_of=_coord_of)
        attached = [m for m in got if m.get("member_identity")]
        plain = [m for m in sw.members_for_shot(real_cp, _visible(real_cp), content_sha_of=_sha_of,
                                                coordinate_of=_coord_of) if m.get("member_identity")]
        assert len(attached) == len(plain) and fid in [m["subject_final_id"] for m in attached], (
            "★사람 확인이 production 부착 수를 바꿨다 (HITL 0)")
        one = next(m for m in attached if m["subject_final_id"] == fid)
        assert one["content_sha256"] == _sha_of(target)
        assert one["purpose"], "★쓰임이 비었다 — 정본 계약이 아니다"

    def test_the_rest_stay_unverified_not_unavailable(self, real_cp):
        """★「못 구한 것」과 **「안 본 것」**을 섞지 않는다."""
        from app.modules.pipeline import grounding_bundle_projection as bp

        got = sw.members_for_shot(real_cp, _visible(real_cp),
                                  content_sha_of=_sha_of,
                                  coordinate_of=_coord_of)
        unver = bp.unverified_members(got)
        unavail = bp.unavailable_members(got)
        # ★고른 5장 중 **배경 갈래는 4장**이다(LP01·03·04·05). 남은 하나는
        #  `prop` 이라 이 입구가 아니라 정책 lane 으로 간다 — 데이터의 답이다.
        assert unver == [], f"★HITL 0 인데 「안 봐서 안 붙은 것」이 {len(unver)}개다"
        # ★같은 주인이 **쓰임마다** 멤버를 갖는다(맥락·상세). 그러니 주인
        #  이름으로 견주면 안 되고 **멤버 단위**로 겹치는지를 본다.
        def _key(m):
            return (m["subject_final_id"], m["purpose"])

        assert {_key(m) for m in unver} & {_key(m) for m in unavail} == set(), (
            "★못 구한 것과 안 본 것이 같은 멤버로 겹친다")
        assert unavail, "★못 구한 것이 하나도 없다 — 두 갈래 대조가 안 된다"


class TestTheConsumerHashMovesWithTheContracts:
    """★★옛 소비자 CP 가 **새 계약에서 그대로 재사용되면** 안 된다
    (Codex 조건 5 · 2026-09-02).

    `scene_detail` 의 `config_hash` 가 네 계약 판을 접는다 — 그중 하나만
    올려도 hash 가 움직여 옛 CP 가 재사용되지 않는다. 그 **결속**을 잠근다;
    판 번호는 못박지 않는다(그러면 다음에 올릴 때 이 시험이 막는다).
    """

    def test_the_hash_folds_the_four_contracts(self):
        import ast
        import inspect
        import textwrap

        from app.core.steps import detail_steps

        src = textwrap.dedent(inspect.getsource(detail_steps))
        i = src.index("grounding_sidecar_contract")
        block = src[i:i + 700]
        for name in ("SIDECAR_WRITER_CONTRACT_VERSION",
                     "PROJECTION_CONTRACT_VERSION",
                     "ACQUISITION_PROJECTION_VERSION",
                     "BUNDLE_CONTRACT_VERSION"):
            assert name in block, f"★{name} 이 config hash 에 안 접힌다"
        assert ast.parse(src) is not None

    def test_changing_one_contract_changes_the_hash(self, monkeypatch):
        """★★★끝점 — 판을 하나 바꾸면 hash 가 **실제로** 달라진다."""
        from app.core.steps import detail_steps as ds
        from app.modules.pipeline import grounding_sidecar_writer as sw2

        from app.core.database import Base, SessionLocal, engine

        Base.metadata.create_all(bind=engine)
        s = ds.SceneDetailStep.__new__(ds.SceneDetailStep)
        s.project_config = {"grounding_mode": "v2_chunk"}
        s.project_id, s.episode_id = "p", "e"
        # ★판정 묶음의 지문도 접히므로 session 이 있어야 한다
        s.db = SessionLocal()
        # ★hash 가 보는 **다른 재료**는 이 시험의 대상이 아니다 — 비워 둔다
        s._vca_printed_prop_context = lambda: {}
        s._vca_immobilized_subject_context = lambda: {}
        before = s._config_hash()
        monkeypatch.setattr(sw2, "SIDECAR_WRITER_CONTRACT_VERSION",
                            "9.999999999999")
        assert s._config_hash() != before, (
            "★계약을 올려도 hash 가 그대로다 — 옛 CP 가 재사용된다")


class TestOneVerdictMakesExactlyOnePhotoUsable:
    """★★★Codex 화면 acceptance의 마지막 축 — **실물 CP** 로 잰다.

        결정 0건        → 붙는 것 0
        그 한 장 verified → **그 한 장만**
        rejected        → 0
        다른 SHA / 다른 좌표 → 0
        모순된 결정 둘   → 0 (fail-safe)
    """

    @staticmethod
    def _bits(real_cp):
        import hashlib

        from app.modules.pipeline import grounding_fidelity_review as fr

        base = Path("/Users/manta/Documents/Projects/TheRoad-I1/artifact"
                    "/canary_69e821758f3d")
        world = json.loads((base / "projects"
                            "/8e2e65b7-e910-4b12-b081-c23de0affab5"
                            "/checkpoints/episodes"
                            "/9e64c302-82e6-4407-b560-8c70c2ab7192"
                            "/visual_world_rules/manifest.json"
                            ).read_text(encoding="utf-8"))["data"]
        coords = {"era": str(world.get("era") or ""),
                  "region": str(world.get("region") or "")}

        def sha_of(row):
            rel = str(((row.get("acquisition") or {}).get("chosen") or {}
                       ).get("path") or "")
            p = base / rel
            if not (rel and p.is_file()):
                return ""
            h = hashlib.sha256()
            with open(p, "rb") as f:
                for c in iter(lambda: f.read(1 << 20), b""):
                    h.update(c)
            return h.hexdigest()

        return fr, coords, sha_of

    @staticmethod
    def _attached(cp):
        got = sw.members_for_shot(cp, _visible(cp), content_sha_of=_sha_of,
                                  coordinate_of=_coord_of)
        return sorted(m["subject_final_id"] for m in got
                      if m.get("member_identity"))

    def test_the_six_axes(self, real_cp):
        fr, coords, sha_of = self._bits(real_cp)
        pid = "8e2e65b7-e910-4b12-b081-c23de0affab5"
        eid = "9e64c302-82e6-4407-b560-8c70c2ab7192"
        inputs = fr.review_inputs_for(real_cp, project_id=pid, episode_id=eid,
                                      coordinates=coords, sha_of=sha_of)
        # ★거절은 **모델이 고른 장**에 해야 그 줄을 움직인다 — 다른 후보의 거절은 선택과 무관하다
        lp = next(x for x in inputs
                  if x["payload"]["final_id"].startswith("LP") and x.get("model_pick"))
        fid = lp["payload"]["final_id"]

        def _apply(rows, *, sha=sha_of, co=coords):
            return fr.apply_reviews(real_cp, rows, project_id=pid,
                                    episode_id=eid, coordinates=co,
                                    sha_of=sha)

        def _row(verdict, rid="r1", at="2026-09-02T12:00"):
            return {"id": rid, "review_input_hash": lp["hash"],
                    "verdict": verdict, "reviewed_at": at,
                    "reviewer_actor": "사람", "supersedes_id": None}

        yes = [_row("verified")]
        # ★HITL 0: 자동 선택은 판정 전에 붙는다 · 「맞다」는 아무것도 안 바꾼다 · 「아니다」는
        #  **도구 산출**에서만 그 줄을 뺀다 · 다른 사진/좌표의 판정은 안 따라간다(그대로 붙는다)
        base = self._attached(real_cp)
        assert fid in base, "★자동 선택이 붙지 않았다 (HITL 0)"
        assert self._attached(_apply(yes)) == base
        assert self._attached(_apply([_row("rejected")])) == [x for x in base if x != fid]
        assert self._attached(_apply(yes, sha=lambda _r: "다른-sha")) == base
        assert self._attached(_apply(yes, co={**coords, "era": "다른 시기"})) == base
        # ★대신(supersede)하지 않은 두 판정은 **어긋난 기록**이다 — override 가 아니라 미확인 → 그대로 붙는다
        assert self._attached(_apply(
            yes + [_row("rejected", rid="r2",
                        at="2026-09-02T13:00")])) == base