"""사람이 확인한 C/P 조사 사진이 canonical ref 생성 호출의 **실제 입력**에 실린다.

★Codex BLOCK 2 (2026-09-02): 중앙 정책은 verified C/P 를 `required_refs` 의 ID 로만
올렸고, Phase1 의 `extra_references` 는 의존 엔티티 ref map 만 봤다. 그래서 P01 의
사람이 확인한 사진은 「P01 참조가 필요하다」는 boolean 만 만들고 canonical ref 는
그 사진 **없이** 생성됐다.

여기서 재는 것 — 전부 production 입구:
  ① `verified_policy_references` 가 **실물** 중앙 CP(canary ① · 21줄) 위에서 P01 만
     낸다(LP 는 verified 라도 제외 · unavailable 은 0).
  ② `ReferenceImageService.generate_base_references` → orchestrator → Phase1 →
     `generate_and_validate_reference` 직전에서 P01 의 `extra_references` 에 그
     **정확한 bytes** 가 실리고, C01(사진 없음)은 종전 그대로 None, LP01 은 생성
     대상이 **아니다**(BLOCK 1 의 생산자 끝점). 호출 수는 늘지 않는다.
  ③ 저장된 `ImageAsset.pipeline_metadata_json.grounding_inputs` 에 그 sha 좌표.
  ④ 판정이 없으면 P01 도 None — 기존 동작 한 바이트 불변.
"""
from __future__ import annotations

import copy
import hashlib
import json
import uuid
from pathlib import Path

import pytest
from sqlalchemy import text as sql_text

from tests.grounding._review_seed import load_frozen_central_cp, seed_verdicts, write_photos

pytestmark = pytest.mark.pg

FULLTEXT = "테스트용 원고. 두 사람이 상 앞에 앉는다."
# ★project/episode 는 시험마다 새로 — 사진 경로는 CP 에 적힌 **상대 경로** 그대로라
#  (기준은 projects_dir 의 부모) 어느 project 로 읽어도 같은 파일에 닿는다.
PID = ""
EID = ""


def _rows_by_subject(cp):
    return {r["research_subject_id"]: r for r in cp["data"]["rows"]}


@pytest.fixture
def world(pg_session, tmp_path: Path, monkeypatch):
    """실물 CP + 진짜 사진 파일 + 사람 판정(P01 verified · LP01#detail verified) + DB 행."""
    from app.core.config import settings

    global PID, EID
    PID, EID = f"p-{uuid.uuid4()}", str(uuid.uuid4())
    root = tmp_path
    monkeypatch.setattr(settings, "projects_dir", str(root / "projects"))
    monkeypatch.setattr("app.core.file_paths.PROJECT_ROOT", root)
    monkeypatch.setattr(settings, "gemini_api_key", "test-key-not-used")

    cp = load_frozen_central_cp()
    rows = _rows_by_subject(cp)
    assert rows["rs_bdfb2b7b981735bdf4309704"]["ledger_row"]["final_id"] == "P01"
    assert rows["LP01#detail"]["ledger_row"]["owner_type"] == "location_part"
    photos = write_photos(root, cp["data"]["rows"])
    ep = root / "projects" / PID / "checkpoints" / "episodes" / EID
    (ep / "reference_acquisition").mkdir(parents=True, exist_ok=True)
    (ep / "reference_acquisition" / "manifest.json").write_text(
        json.dumps(cp, ensure_ascii=False), encoding="utf-8")

    uid = f"photo-{uuid.uuid4()}"
    db = pg_session
    db.execute(sql_text(
        "INSERT INTO user_account (id, username, display_name, password_hash, role, "
        "is_active, created_at, updated_at) VALUES (:uid, :un, 't', 'x', 'creator', 1, "
        "'2026-01-01', '2026-01-01')"), {"uid": uid, "un": f"u_{uid}"})
    db.execute(sql_text(
        "INSERT INTO project_registry (id, name, created_by, created_at, updated_at) "
        "VALUES (:pid, 'photo', :uid, '2026-01-01', '2026-01-01')"), {"pid": PID, "uid": uid})
    db.execute(sql_text(
        "INSERT INTO episode (id, project_id, title, episode_number, source_filename, "
        "source_path, status, fulltext, language, created_at, updated_at) VALUES "
        "(:eid, :pid, '1화', 1, 'x.txt', 'x/x.txt', 'analyzed', :ft, 'ko', "
        "'2026-01-01', '2026-01-01')"), {"eid": EID, "pid": PID, "ft": FULLTEXT})
    ids = {}
    for short_id, etype in (("C01", "character"), ("P01", "prop"), ("LP01", "location_part")):
        cid = str(uuid.uuid4())
        db.execute(sql_text(
            "INSERT INTO entity_canon (id, project_id, short_id, name, entity_type, "
            "description, stable_traits, metadata_json, t2i_prompt, status, created_at, "
            "updated_at) VALUES (:cid, :pid, :s, :n, :e, 'desc', '[]', '{}', 'a prompt', "
            "'active', '2026-01-01', '2026-01-01')"),
            {"cid": cid, "pid": PID, "s": short_id, "n": f"{etype} {short_id}", "e": etype})
        db.execute(sql_text(
            "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id, "
            "t2i_appearance_count) VALUES (:lid, :cid, :pid, :eid, 5)"),
            {"lid": str(uuid.uuid4()), "cid": cid, "pid": PID, "eid": EID})
        ids[short_id] = cid
    # ★world guide 를 미리 둔다 — orchestrator 가 같은 hash 면 LLM 을 안 부른다
    wg_hash = hashlib.md5(f"{FULLTEXT}:{len(ids)}:0".encode()).hexdigest()
    db.execute(sql_text(
        "INSERT INTO world_guide (id, project_id, episode_id, guide_json, source_hash, "
        "created_at) VALUES (:id, :pid, :eid, '{}', :h, '2026-01-01')"),
        {"id": str(uuid.uuid4()), "pid": PID, "eid": EID, "h": wg_hash})
    db.commit()
    return {"root": root, "cp": cp, "photos": photos, "ids": ids, "db": db}


def _verify(world, *subjects):
    n = seed_verdicts(world["db"], cp=world["cp"], root=world["root"], project_id=PID,
                      episode_id=EID, want={s: "verified" for s in subjects})
    assert n == len(subjects), n


def _p01_photo(world) -> bytes:
    row = _rows_by_subject(world["cp"])["rs_bdfb2b7b981735bdf4309704"]
    return world["photos"][row["acquisition"]["chosen"]["path"]]


# ── ① 읽는 쪽 ────────────────────────────────────────────────────────
class TestTheReaderTakesOnlyVerifiedPolicyOwners:
    def test_p01_verified_lp01_verified_only_p01_comes_out(self, world):
        from app.modules.pipeline.grounding_canonical_ref_inputs import verified_policy_references

        _verify(world, "rs_bdfb2b7b981735bdf4309704", "LP01#detail")
        got = verified_policy_references(world["db"], project_id=PID, episode_id=EID)
        assert set(got) == {"P01"}, set(got)
        one = got["P01"]
        assert len(one) == 1
        photo = _p01_photo(world)
        assert one[0]["bytes"] == photo
        assert one[0]["content_sha256"] == hashlib.sha256(photo).hexdigest()
        assert one[0]["identity"] == "rs_bdfb2b7b981735bdf4309704"
        assert one[0]["acquisition_identity"]

    def test_without_a_verdict_nothing_comes_out(self, world):
        from app.modules.pipeline.grounding_canonical_ref_inputs import verified_policy_references

        assert verified_policy_references(world["db"], project_id=PID, episode_id=EID) == {}

    def test_rejected_is_zero(self, world):
        from app.modules.pipeline.grounding_canonical_ref_inputs import verified_policy_references

        seed_verdicts(world["db"], cp=world["cp"], root=world["root"], project_id=PID,
                      episode_id=EID, want={"rs_bdfb2b7b981735bdf4309704": "rejected"})
        assert verified_policy_references(world["db"], project_id=PID, episode_id=EID) == {}

    def test_no_central_cp_means_the_old_path(self, world):
        from app.modules.pipeline.grounding_canonical_ref_inputs import verified_policy_references

        assert verified_policy_references(world["db"], project_id="no-such", episode_id="x") == {}


# ── ② 생성 호출 직전 ───────────────────────────────────────────────────
def _capture_generation(monkeypatch, calls, *, fail=False):
    from app.modules.pipeline import ref_image_pipeline as rp

    def _fake(**kw):
        calls.append(kw)
        if fail:
            raise RuntimeError("생성 실패(시험)")
        out = Path(kw["output_dir"]) / f"{uuid.uuid4().hex}.png"
        out.parent.mkdir(parents=True, exist_ok=True)
        out.write_bytes(b"\x89PNG generated")
        return {"file_path": str(out), "generation_model": "fake-model",
                "validation": {"score": 1.0}}

    monkeypatch.setattr(rp, "generate_and_validate_reference", _fake)


def _run(world):
    from app.services.reference_image_service import ReferenceImageService

    svc = ReferenceImageService(db=world["db"], project_id=PID, actor_id="사람")
    return svc.generate_base_references(episode_id=EID, mode="resume")


class TestTheBytesReachTheGenerationCall:
    def test_p01_gets_the_photo_c01_does_not_lp01_is_not_generated(self, world, monkeypatch):
        from app.modules.pipeline.grounding_canonical_ref_inputs import GROUNDING_REFERENCE_LABEL

        _verify(world, "rs_bdfb2b7b981735bdf4309704", "LP01#detail")
        calls = []
        _capture_generation(monkeypatch, calls)
        got = _run(world)

        by_name = {c["entity_name"]: c for c in calls}
        assert set(by_name) == {"prop P01", "character C01"}, set(by_name)     # ★LP01 없음
        assert got["generated"] == 2 and got["failed"] == 0, got
        p01 = by_name["prop P01"]
        assert p01["extra_references"] == [(GROUNDING_REFERENCE_LABEL, _p01_photo(world))]
        c01 = by_name["character C01"]
        assert c01["extra_references"] is None
        # ★variant 문구가 새지 않는다 — 사진 참조는 dep_refs 가 아니다
        assert "VARIANT" not in p01["entity_description"]
        assert "VARIANT" not in p01["t2i_prompt"]

    def test_the_call_count_does_not_grow(self, world, monkeypatch):
        calls_without, calls_with = [], []
        _capture_generation(monkeypatch, calls_without)
        _run(world)
        n0 = len(calls_without)
        # 두 번째 판 — 판정을 넣고 checkpoint 를 지워 다시 생성
        _verify(world, "rs_bdfb2b7b981735bdf4309704")
        cp_dir = world["root"] / "projects" / PID / "checkpoints" / "images" / EID
        for f in cp_dir.glob("*.json"):
            f.unlink()
        world["db"].execute(sql_text(
            "UPDATE image_asset SET is_primary = 0 WHERE project_id = :pid"), {"pid": PID})
        world["db"].commit()
        _capture_generation(monkeypatch, calls_with)
        _run(world)
        assert len(calls_with) == n0 == 2, (n0, len(calls_with))

    def test_the_asset_records_which_photo_went_in(self, world, monkeypatch):
        from app.models.project import ImageAsset

        _verify(world, "rs_bdfb2b7b981735bdf4309704")
        _capture_generation(monkeypatch, [])
        _run(world)
        photo_sha = hashlib.sha256(_p01_photo(world)).hexdigest()
        rows = {r.entity_id: r for r in world["db"].query(ImageAsset).filter(
            ImageAsset.project_id == PID, ImageAsset.asset_type == "reference",
            ImageAsset.is_primary == 1).all()}
        p01 = rows[world["ids"]["P01"]]
        meta = json.loads(p01.pipeline_metadata_json or "{}")
        assert meta["grounding_inputs"] == [{
            "final_id": "P01", "identity": "rs_bdfb2b7b981735bdf4309704",
            "acquisition_identity": meta["grounding_inputs"][0]["acquisition_identity"],
            "content_sha256": photo_sha}], meta
        assert meta["grounding_inputs"][0]["acquisition_identity"]
        c01 = rows[world["ids"]["C01"]]
        c01_meta = json.loads(c01.pipeline_metadata_json or "{}")
        # ★입력 없음도 **명시 값**으로 남는다 — resume 이 이 지문으로 신선도를 본다
        from app.modules.pipeline.grounding_canonical_ref_inputs import EMPTY_INPUT_DIGEST
        assert c01_meta["grounding_inputs"] == []
        assert c01_meta["grounding_input_digest"] == EMPTY_INPUT_DIGEST
        assert meta["grounding_input_digest"] != EMPTY_INPUT_DIGEST

    def test_without_a_verdict_p01_is_generated_as_before(self, world, monkeypatch):
        """★④ 기존 동작 불변 — 사진이 없으면 extra_references 는 None 그대로."""
        calls = []
        _capture_generation(monkeypatch, calls)
        _run(world)
        by_name = {c["entity_name"]: c for c in calls}
        assert set(by_name) == {"prop P01", "character C01"}
        assert by_name["prop P01"]["extra_references"] is None


class TestTheBindingIsByTypedIdOnly:
    def test_the_reader_reads_no_name(self):
        import ast
        import inspect
        import textwrap

        from app.modules.pipeline import grounding_canonical_ref_inputs as m

        tree = ast.parse(textwrap.dedent(inspect.getsource(m.verified_policy_references)))
        consts = {n.value for n in ast.walk(tree) if isinstance(n, ast.Constant)
                  and isinstance(n.value, str)}
        keys_read = {n.value for n in ast.walk(tree) if isinstance(n, ast.Constant)
                     and isinstance(n.value, str) and n.value in ("name", "entity_name", "surface_form")}
        assert not keys_read, keys_read
        assert "final_id" in consts and "owner_type" in consts

    def test_the_label_names_no_subject(self):
        from app.modules.pipeline.grounding_canonical_ref_inputs import GROUNDING_REFERENCE_LABEL

        assert "{" not in GROUNDING_REFERENCE_LABEL and "%" not in GROUNDING_REFERENCE_LABEL


# ── 재개 신선도 (Codex BLOCK 2026-09-02 · 재개 계약) ─────────────────────
def _primary(world, short_id):
    from app.models.project import ImageAsset

    return (world["db"].query(ImageAsset)
            .filter(ImageAsset.project_id == PID, ImageAsset.asset_type == "reference",
                    ImageAsset.entity_id == world["ids"][short_id], ImageAsset.is_primary == 1)
            .first())


def _all_refs(world, short_id):
    from app.models.project import ImageAsset

    return (world["db"].query(ImageAsset)
            .filter(ImageAsset.project_id == PID, ImageAsset.asset_type == "reference",
                    ImageAsset.entity_id == world["ids"][short_id])
            .order_by(ImageAsset.created_at).all())


def _cp_file(world) -> Path:
    return world["root"] / "projects" / PID / "checkpoints" / "images" / EID / "reference_checkpoint.json"


def _cp_rows(world) -> dict:
    return json.loads(_cp_file(world).read_text(encoding="utf-8")).get("completed", {})


def _rewrite_cp_rows(world, fn):
    p = _cp_file(world)
    d = json.loads(p.read_text(encoding="utf-8"))
    for k, row in d.get("completed", {}).items():
        fn(k, row)
    p.write_text(json.dumps(d, ensure_ascii=False), encoding="utf-8")


def _set_asset_meta(world, short_id, fn):
    a = _primary(world, short_id)
    meta = json.loads(a.pipeline_metadata_json or "{}")
    fn(meta)
    a.pipeline_metadata_json = json.dumps(meta, ensure_ascii=False)
    world["db"].commit()


def _names(calls):
    return sorted(c["entity_name"] for c in calls)


class TestResumeFreshnessFollowsTheGroundingInput:
    def test_a_pre_feature_ungrounded_ref_is_reused_until_a_verdict_arrives(self, world, monkeypatch):
        """★Codex 끝점 1·2: 옛(지문 없는) P01 primary+CP → verified → **정확히 1회** 재생성,
        새 metadata/CP 지문 일치; 같은 상태 resume → 0."""
        from app.modules.pipeline.grounding_canonical_ref_inputs import (
            GROUNDING_INPUT_DIGEST_KEY, grounding_input_digest, verified_policy_references)

        first = []
        _capture_generation(monkeypatch, first)
        _run(world)
        assert _names(first) == ["character C01", "prop P01"]
        # ★이 기능 전에 구운 자산처럼 — 자산·CP 에서 지문 칸을 뺀다
        for sid in ("C01", "P01"):
            _set_asset_meta(world, sid, lambda m: m.pop(GROUNDING_INPUT_DIGEST_KEY, None))
        _rewrite_cp_rows(world, lambda k, r: r.pop(GROUNDING_INPUT_DIGEST_KEY, None))
        again = []
        _capture_generation(monkeypatch, again)
        _run(world)
        assert again == [], "★입력이 없으면 옛 자산은 신선하다 — legacy 를 다시 굽지 않는다"

        old_p01 = _primary(world, "P01")
        _verify(world, "rs_bdfb2b7b981735bdf4309704")
        third = []
        _capture_generation(monkeypatch, third)
        _run(world)
        assert _names(third) == ["prop P01"], _names(third)          # ★정확히 1회 · P01 만
        assert third[0]["extra_references"] and third[0]["extra_references"][0][1] == _p01_photo(world)
        want = grounding_input_digest(
            verified_policy_references(world["db"], project_id=PID, episode_id=EID)["P01"])
        new_p01 = _primary(world, "P01")
        assert new_p01.id != old_p01.id
        assert json.loads(new_p01.pipeline_metadata_json)[GROUNDING_INPUT_DIGEST_KEY] == want
        assert _cp_rows(world)[world["ids"]["P01"]][GROUNDING_INPUT_DIGEST_KEY] == want
        # ★비파괴 — 옛 자산은 남고 primary 만 내려갔다
        olds = [a for a in _all_refs(world, "P01") if a.id == old_p01.id]
        assert olds and olds[0].is_primary == 0

    def test_same_state_resume_buys_nothing(self, world, monkeypatch):
        _verify(world, "rs_bdfb2b7b981735bdf4309704")
        first = []
        _capture_generation(monkeypatch, first)
        _run(world)
        assert _names(first) == ["character C01", "prop P01"]
        again = []
        _capture_generation(monkeypatch, again)
        _run(world)
        assert again == []

    def test_a_changed_photo_regenerates_once_and_a_failure_keeps_the_old_primary(self, world, monkeypatch):
        """★Codex 끝점 3: 사진 SHA 가 바뀌면(판정이 그 bytes 위에 섰으므로 unverified 로
        내려감) 다시 1회; 실패하면 옛 primary 그대로."""
        _verify(world, "rs_bdfb2b7b981735bdf4309704")
        _capture_generation(monkeypatch, [])
        _run(world)
        grounded = _primary(world, "P01")
        row = _rows_by_subject(world["cp"])["rs_bdfb2b7b981735bdf4309704"]
        (world["root"] / row["acquisition"]["chosen"]["path"]).write_bytes(b"\x89PNG other bytes")

        failing = []
        _capture_generation(monkeypatch, failing, fail=True)
        _run(world)
        assert _names(failing) == ["prop P01"]
        assert _primary(world, "P01").id == grounded.id, "★실패면 옛 primary 를 지키지 내리지 않는다"

        ok = []
        _capture_generation(monkeypatch, ok)
        _run(world)
        assert _names(ok) == ["prop P01"]
        assert ok[0]["extra_references"] is None, "★더는 verified 가 아니다 — 사진 없이"
        assert _primary(world, "P01").id != grounded.id

    def test_cp_only_old_digest_is_not_reused(self, world, monkeypatch):
        from app.modules.pipeline.grounding_canonical_ref_inputs import GROUNDING_INPUT_DIGEST_KEY

        _capture_generation(monkeypatch, [])
        _run(world)
        c01 = world["ids"]["C01"]
        _rewrite_cp_rows(world, lambda k, r: r.__setitem__(GROUNDING_INPUT_DIGEST_KEY, "0ld0ld0ld0ld0ld0")
                         if k == c01 else None)
        again = []
        _capture_generation(monkeypatch, again)
        _run(world)
        assert _names(again) == ["character C01"]

    def test_asset_only_old_digest_is_not_reused(self, world, monkeypatch):
        from app.modules.pipeline.grounding_canonical_ref_inputs import GROUNDING_INPUT_DIGEST_KEY

        _capture_generation(monkeypatch, [])
        _run(world)
        _set_asset_meta(world, "C01", lambda m: m.__setitem__(GROUNDING_INPUT_DIGEST_KEY, "0ld0ld0ld0ld0ld0"))
        again = []
        _capture_generation(monkeypatch, again)
        _run(world)
        assert _names(again) == ["character C01"]


def _set_mode(world, mode: str):
    world["db"].execute(sql_text(
        "INSERT INTO project_settings (id, project_id, llm_config_json, updated_at) "
        "VALUES (:id, :pid, :cfg, '2026-01-01')"),
        {"id": str(uuid.uuid4()), "pid": PID, "cfg": json.dumps({"grounding_mode": mode})})
    world["db"].commit()


class TestTheDirectApiFailsClosedInV2Chunk:
    def test_v2_chunk_without_a_central_cp_buys_nothing(self, world, monkeypatch):
        """★Codex E: public 경로가 중앙 조사 전에 눌리면 고증 없는 canonical ref 가
        생기고 resume 이 영구 재사용한다 — provider 앞에서 선다."""
        _set_mode(world, "v2_chunk")
        cp = world["root"] / "projects" / PID / "checkpoints" / "episodes" / EID / "reference_acquisition" / "manifest.json"
        cp.unlink()
        calls = []
        _capture_generation(monkeypatch, calls)
        with pytest.raises(Exception) as ei:
            _run(world)
        assert calls == [], calls
        assert _primary(world, "P01") is None and _primary(world, "C01") is None
        assert "reference_acquisition" in str(ei.value) or "중앙" in str(ei.value) or "CP" in str(ei.value)

    def test_v2_chunk_with_the_central_cp_runs(self, world, monkeypatch):
        _set_mode(world, "v2_chunk")
        calls = []
        _capture_generation(monkeypatch, calls)
        _run(world)
        assert _names(calls) == ["character C01", "prop P01"]

    def test_legacy_without_a_central_cp_is_unchanged(self, world, monkeypatch):
        _set_mode(world, "legacy")
        cp = world["root"] / "projects" / PID / "checkpoints" / "episodes" / EID / "reference_acquisition" / "manifest.json"
        cp.unlink()
        calls = []
        _capture_generation(monkeypatch, calls)
        _run(world)
        assert _names(calls) == ["character C01", "prop P01"]


class TestTheStepHashFoldsTheInputsOnlyInV2Chunk:
    def _step(self, world, mode):
        from app.core.steps import image_steps as im

        s = im.RefImageGenStep.__new__(im.RefImageGenStep)
        s.project_id, s.episode_id, s.db = PID, EID, world["db"]
        s.project_config = {"grounding_mode": mode}
        return s

    def test_legacy_hash_is_the_base_hash(self, world):
        from app.core.step_runner import compute_config_hash

        s = self._step(world, "legacy")
        assert s._config_hash() == compute_config_hash(s.project_config)

    def test_v2_chunk_hash_moves_when_a_verdict_arrives(self, world):
        from app.core.step_runner import compute_config_hash

        s = self._step(world, "v2_chunk")
        before = s._config_hash()
        assert before != compute_config_hash(s.project_config)
        _verify(world, "rs_bdfb2b7b981735bdf4309704")
        after = s._config_hash()
        assert after != before
        assert s._config_hash() == after, "★결정적"