"""참조 이미지 게이트 — 「대상은 episode, 자산은 canon/project」 (2026-08-29).

## 무엇이 결함이었나 — 실측 episode 97375a4b

`scene_image_pipeline` 이 이 한 줄로 죽었다:

    참조 이미지 미완성: 2개 누락 (정임(character), 고무줄로 묶인 낡은 종이 회수권 뭉치(prop))

둘 다 **앞 에피소드에서 참조를 이미 만든 canon** 이었다. 두 쪽이 범위를
다르게 봤다:

    reference_pipeline_orchestrator.py:202-208   「이미 있다」를 project 범위로 → 건너뛴다
    pipeline_gate.py (before)                     「있어야 한다」를 episode 범위로 → 막는다

그래서 다시 태워도 안 풀리는 **교착**이었다.

## 왜 게이트 쪽이 틀렸나

참조의 primary 는 프로젝트 안에서 canon 당 **하나**다 —
`reference_phase1_service.py:198-202` 가 새 참조를 쓸 때 `project_id +
entity_id` 범위로 옛 primary 를 전부 내린다. 같은 게이트 함수의 합성
(composite) 검사도 이미 project 범위로 본다. episode 범위를 요구한 참조
검사만 혼자 달랐다.

★그래서 「행을 하나 더 쓴다」는 답이 아니다 — 한 canon 에 primary 가 둘이
되거나, 옛 행을 내려 앞 에피소드가 대신 막힌다.

Lane: ``-m pg``.
"""
from __future__ import annotations

import uuid
from pathlib import Path

import pytest
from sqlalchemy import text as sql_text

pytestmark = pytest.mark.pg


def _seed(db, pid: str, *episode_ids: str) -> None:
    uid = f"refgate-{uuid.uuid4()}"
    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, 'refgate', :uid, '2026-01-01', '2026-01-01')"
    ), {"pid": pid, "uid": uid})
    for n, eid in enumerate(episode_ids, start=1):
        db.execute(sql_text(
            "INSERT INTO episode (id, project_id, title, episode_number, "
            "source_filename, source_path, status, created_at, updated_at) VALUES "
            "(:eid, :pid, :t, :n, 'x.txt', 'x/x.txt', 'analyzed', "
            "'2026-01-01', '2026-01-01')"
        ), {"eid": eid, "pid": pid, "t": f"{n}화", "n": n})


def _canon(db, pid: str, short_id: str, name: str, etype: str) -> str:
    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, '', '[]', '{}', '', 'active', "
        "'2026-01-01', '2026-01-01')"
    ), {"cid": cid, "pid": pid, "s": short_id, "n": name, "e": etype})
    return cid


def _link(db, pid: str, eid: str, cid: str) -> None:
    db.execute(sql_text(
        "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id) "
        "VALUES (:lid, :cid, :pid, :eid)"
    ), {"lid": str(uuid.uuid4()), "cid": cid, "pid": pid, "eid": eid})


def _ref_asset(db, pid: str, eid: str, cid: str, rel_path: str,
               *, primary: int = 1) -> str:
    aid = str(uuid.uuid4())
    db.execute(sql_text(
        "INSERT INTO image_asset (id, project_id, episode_id, entity_id, "
        "asset_type, file_path, is_primary, status, created_at) VALUES "
        "(:aid, :pid, :eid, :cid, 'reference', :fp, :pr, 'ok', '2026-01-01')"
    ), {"aid": aid, "pid": pid, "eid": eid, "cid": cid, "fp": rel_path,
        "pr": primary})
    return aid


@pytest.fixture
def world(pg_session, tmp_path: Path, monkeypatch):
    """1화에서 참조를 만든 canon 이 2화에 다시 나오는 상황."""
    monkeypatch.setattr("app.core.config.settings.projects_dir",
                        str(tmp_path / "projects"))
    # 상대 file_path 해소 root — `file_paths._resolve_root` 가 보는 자리.
    monkeypatch.setattr("app.core.file_paths.PROJECT_ROOT", tmp_path)

    pid = f"p-{uuid.uuid4()}"
    ep1, ep2 = str(uuid.uuid4()), str(uuid.uuid4())
    _seed(pg_session, pid, ep1, ep2)

    char = _canon(pg_session, pid, "C01", "정임", "character")
    _link(pg_session, pid, ep1, char)
    _link(pg_session, pid, ep2, char)     # 2화에도 나온다
    pg_session.commit()
    return pid, ep1, ep2, char, tmp_path


def _write_file(root: Path, rel: str) -> None:
    f = root / rel
    f.parent.mkdir(parents=True, exist_ok=True)
    f.write_bytes(b"\x89PNG fake")


def _missing(db, pid: str, eid: str):
    from app.core.pipeline_gate import missing_reference_entities

    return [e.name for e in missing_reference_entities(db, pid, eid)]


def test_앞_에피소드에만_있는_참조로도_이번_에피소드가_통과한다(pg_session, world):
    """★이번 결함의 본체 — 재사용된 canon 이 영영 막히던 자리."""
    pid, ep1, ep2, char, root = world
    rel = "refs/c01.png"
    _write_file(root, rel)
    _ref_asset(pg_session, pid, ep1, char, rel)     # 1화에서 만든 정본
    pg_session.commit()

    assert _missing(pg_session, pid, ep2) == [], (
        "앞 에피소드에서 만든 정본 참조를 두고 2화가 막혔다")


def test_참조가_아예_없으면_막는다(pg_session, world):
    """양성 확인 — 진짜 미완성은 그대로 잡아야 한다."""
    pid, ep1, ep2, char, root = world
    assert _missing(pg_session, pid, ep2) == ["정임"]


def test_DB행은_있는데_파일이_없으면_막는다(pg_session, world):
    """행만 보고 통과시키지 않는다."""
    pid, ep1, ep2, char, root = world
    _ref_asset(pg_session, pid, ep1, char, "refs/사라진.png")   # 파일 안 만듦
    pg_session.commit()

    assert _missing(pg_session, pid, ep2) == ["정임"]


def test_primary_가_아니면_막는다(pg_session, world):
    """정본(primary)만 인정한다 — 옛 후보를 준비된 것으로 읽지 않는다."""
    pid, ep1, ep2, char, root = world
    rel = "refs/old.png"
    _write_file(root, rel)
    _ref_asset(pg_session, pid, ep1, char, rel, primary=0)
    pg_session.commit()

    assert _missing(pg_session, pid, ep2) == ["정임"]


def test_다른_프로젝트의_자산은_안_친다(pg_session, world):
    """자산 범위는 canon/project 지 전역이 아니다."""
    pid, ep1, ep2, char, root = world
    other_pid = f"p-{uuid.uuid4()}"
    other_ep = str(uuid.uuid4())
    _seed(pg_session, other_pid, other_ep)
    rel = "refs/남의것.png"
    _write_file(root, rel)
    # 같은 canon id 를 남의 프로젝트 행이 들고 있어도 인정하면 안 된다
    _ref_asset(pg_session, other_pid, other_ep, char, rel)
    pg_session.commit()

    assert _missing(pg_session, pid, ep2) == ["정임"]


def test_게이트와_진행률이_같은_말을_한다(pg_session, world):
    """게이트는 통과인데 화면은 미완성인 오독을 막는다."""
    from app.core.pipeline_gate import get_pipeline_status

    pid, ep1, ep2, char, root = world
    rel = "refs/c01.png"
    _write_file(root, rel)
    _ref_asset(pg_session, pid, ep1, char, rel)
    pg_session.commit()

    st = get_pipeline_status(pg_session, pid, ep2)["reference_images"]
    assert (st["ref_done"], st["entity_total"], st["complete"]) == (1, 1, True), st


def test_장소는_참조_대상이_아니다(pg_session, world):
    """배경 참조 생성이 꺼져 있다 — 분모에 넣으면 영영 미완성이 된다."""
    pid, ep1, ep2, char, root = world
    loc = _canon(pg_session, pid, "L01", "버스 안", "location")
    _link(pg_session, pid, ep2, loc)
    rel = "refs/c01.png"
    _write_file(root, rel)
    _ref_asset(pg_session, pid, ep1, char, rel)
    pg_session.commit()

    from app.core.pipeline_gate import get_pipeline_status

    assert _missing(pg_session, pid, ep2) == []
    st = get_pipeline_status(pg_session, pid, ep2)["reference_images"]
    assert st["entity_total"] == 1 and st["complete"] is True, st


# ── ★게이트 **본체**를 태운다 ─────────────────────────────────────────
#
# 2026-08-29 Codex BLOCK: 위 시험들은 helper 와 진행률만 불러 **조립을
# 안 태웠다.** 그래서 게이트 안에서 지워진 이름을 읽는 NameError 두 건을
# 못 잡았다. 「조립하는 자리 말고 끝점에서 재라」를 여기서 어겼다.


def _still(db, pid: str, eid: str, idx: int = 0) -> None:
    db.execute(sql_text(
        "INSERT INTO scene_still (id, project_id, episode_id, still_index, "
        "scene_index, shot_index, is_selected, status, created_at) VALUES "
        "(:id, :pid, :eid, :i, 1, 1, true, 'active', '2026-01-01')"
    ), {"id": str(uuid.uuid4()), "pid": pid, "eid": eid, "i": idx})


def _outlook_pair(db, pid: str, char_id: str, outlook_id: str,
                  eid: str | None = None) -> None:
    """★배정에 **화를 적는다** (alembic 014, 2026-09-04).

    안 적으면 legacy(화 모름) 갈래로 떨어지고, 그 갈래는 「인물·아웃룩이
    **둘 다** 이 화에 active 로 걸려 있을 때만」 인정한다. 실환경에서는
    sync 가 늘 화를 적으므로 여기가 하한이다 — 안 적으면 이 시험은
    「배정이 아예 없는」 판을 재게 된다.
    """
    db.execute(sql_text(
        "INSERT INTO character_outlook (id, project_id, episode_id, "
        "character_id, outlook_id, created_at) "
        "VALUES (:id, :pid, :e, :c, :o, '2026-01-01')"
    ), {"id": str(uuid.uuid4()), "pid": pid, "e": eid,
        "c": char_id, "o": outlook_id})


def _composite_asset(db, pid: str, eid: str, char_id: str, outlook_id: str,
                     rel_path: str) -> None:
    db.execute(sql_text(
        "INSERT INTO image_asset (id, project_id, episode_id, entity_id, "
        "asset_type, file_path, prompt_used, is_primary, status, created_at) "
        "VALUES (:aid, :pid, :eid, :cid, 'reference', :fp, :pu, 1, 'ok', "
        "'2026-01-01')"
    ), {"aid": str(uuid.uuid4()), "pid": pid, "eid": eid, "cid": char_id,
        "fp": rel_path, "pu": f"[composite:{char_id}:{outlook_id}] x"})


def _gate(db, pid: str, eid: str, monkeypatch):
    """`check_scene_images_ready` 를 그대로 부른다.

    `ensure_analysis_projection_current` 는 체크포인트 파일을 읽으러 가므로
    이 시험에서는 무력화한다 — 재려는 것은 참조/합성 판단이다.
    """
    import app.core.pipeline_gate as gate

    monkeypatch.setattr(gate, "ensure_analysis_projection_current",
                        lambda *a, **k: None)
    return gate.check_scene_images_ready(db, pid, eid)


def test_게이트_본체가_끝까지_돈다_합성까지(pg_session, world, monkeypatch):
    """★BLOCK-1·2 를 잡는 자리 — 합성 경로와 성공 반환까지 태운다."""
    pid, ep1, ep2, char, root = world
    rel = "refs/c01.png"
    _write_file(root, rel)
    _ref_asset(pg_session, pid, ep1, char, rel)

    outlook = _canon(pg_session, pid, "O01", "감색차장제복", "outlook")
    _outlook_pair(pg_session, pid, char, outlook, ep2)
    comp = "refs/c01_o01.png"
    _write_file(root, comp)
    _composite_asset(pg_session, pid, ep1, char, outlook, comp)
    _still(pg_session, pid, ep2)
    pg_session.commit()

    out = _gate(pg_session, pid, ep2, monkeypatch)
    assert out["entities"] == 1 and out["refs"] == 1, out
    assert out["combos"] == 1 and out["combo_done"] == 1, out
    assert out["scenes"] == 1, out


def test_게이트가_합성_누락을_그대로_잡는다(pg_session, world, monkeypatch):
    """양성 확인 — 합성 경로를 지나치게 만든 게 아니다."""
    from app.core.errors import AppError

    pid, ep1, ep2, char, root = world
    rel = "refs/c01.png"
    _write_file(root, rel)
    _ref_asset(pg_session, pid, ep1, char, rel)
    outlook = _canon(pg_session, pid, "O01", "감색차장제복", "outlook")
    _outlook_pair(pg_session, pid, char, outlook, ep2)     # 합성 이미지는 안 만든다
    _still(pg_session, pid, ep2)
    pg_session.commit()

    with pytest.raises(AppError) as e:
        _gate(pg_session, pid, ep2, monkeypatch)
    assert e.value.code == "gate.incomplete_composites"


def test_게이트가_참조_누락을_그대로_잡는다(pg_session, world, monkeypatch):
    from app.core.errors import AppError

    pid, ep1, ep2, char, root = world
    _still(pg_session, pid, ep2)
    pg_session.commit()

    with pytest.raises(AppError) as e:
        _gate(pg_session, pid, ep2, monkeypatch)
    assert e.value.code == "gate.incomplete_references"


def test_대상이_0이면_게이트와_화면이_둘_다_통과다(pg_session, monkeypatch,
                                                  tmp_path: Path):
    """★BLOCK-3 — 장소만 있는 에피소드.

    게이트는 「누락 0」이라 통과하는데 화면만 `> 0` 을 걸면 영영 미완성으로
    남는다. 준비됨(readiness) 계약이라 대상 0 은 **통과**다.
    """
    from app.core.pipeline_gate import get_pipeline_status

    monkeypatch.setattr("app.core.config.settings.projects_dir",
                        str(tmp_path / "projects"))
    monkeypatch.setattr("app.core.file_paths.PROJECT_ROOT", tmp_path)

    pid, eid = f"p-{uuid.uuid4()}", str(uuid.uuid4())
    _seed(pg_session, pid, eid)
    loc = _canon(pg_session, pid, "L01", "버스 안", "location")
    _link(pg_session, pid, eid, loc)
    _still(pg_session, pid, eid)
    pg_session.commit()

    out = _gate(pg_session, pid, eid, monkeypatch)
    assert out["entities"] == 0 and out["refs"] == 0, out

    st = get_pipeline_status(pg_session, pid, eid)["reference_images"]
    assert st["complete"] is True, st
    assert (st["ref_done"], st["entity_total"]) == (0, 0), st


def test_전부_저빈도_스킵이어도_게이트와_화면이_일치한다(pg_session, world,
                                                        monkeypatch):
    """char/prop 이 다 스킵 대상이면 대상 0 — 위와 같은 자리다."""
    from app.core.pipeline_gate import get_pipeline_status
    import app.core.pipeline_gate as gate

    pid, ep1, ep2, char, root = world
    _still(pg_session, pid, ep2)
    pg_session.commit()

    monkeypatch.setattr(gate, "ensure_analysis_projection_current",
                        lambda *a, **k: None)
    monkeypatch.setattr("app.core.low_freq_skip.load_low_freq_skip_ids",
                        lambda *a, **k: {char})

    out = gate.check_scene_images_ready(pg_session, pid, ep2)
    assert out["entities"] == 0, out
    st = get_pipeline_status(pg_session, pid, ep2)["reference_images"]
    assert st["complete"] is True and st["entity_total"] == 0, st


# ── ★단계 verify 도 같은 정본을 쓴다 ─────────────────────────────────
#
# 2026-08-29: `RefImageGenStep.verify_completion` 만 자산을 episode 범위로
# 물어서, 재사용 canon 을 **거짓 미완료**로 찍었다. 게이트는 4/4 통과이고
# 스틸컷 6장이 다 나온 에피소드였는데 단계는 `partial` 로 굳고 매 resume
# 마다 RERUN_SELF 를 불렀다. 사람(나)도 그 상태를 「옳은 동작」으로 오독했다.
#
# ★helper 가 아니라 `verify_completion()` 을 **그대로** 부른다.


def _verify(db, pid: str, eid: str):
    from app.core.steps.image_steps import RefImageGenStep

    return RefImageGenStep("ref_image_gen", pid, eid, db).verify_completion()


def test_verify_앞_에피소드_참조를_정본으로_인정한다(pg_session, world):
    """★이번 결함의 본체."""
    pid, ep1, ep2, char, root = world
    rel = "refs/c01.png"
    _write_file(root, rel)
    _ref_asset(pg_session, pid, ep1, char, rel)
    pg_session.commit()

    r = _verify(pg_session, pid, ep2)
    assert r.is_complete, r.metadata
    assert (r.metadata["expected"], r.metadata["found"]) == (1, 1), r.metadata


def test_verify_가_참조_없음을_그대로_잡는다(pg_session, world):
    pid, ep1, ep2, char, root = world
    r = _verify(pg_session, pid, ep2)
    assert not r.is_complete and r.metadata["found"] == 0, r.metadata


def test_verify_가_파일_없음을_잡는다(pg_session, world):
    pid, ep1, ep2, char, root = world
    _ref_asset(pg_session, pid, ep1, char, "refs/사라진.png")
    pg_session.commit()
    assert not _verify(pg_session, pid, ep2).is_complete


def test_verify_가_non_primary_를_안_친다(pg_session, world):
    pid, ep1, ep2, char, root = world
    rel = "refs/old.png"
    _write_file(root, rel)
    _ref_asset(pg_session, pid, ep1, char, rel, primary=0)
    pg_session.commit()
    assert not _verify(pg_session, pid, ep2).is_complete


def test_verify_가_남의_프로젝트_자산을_안_친다(pg_session, world):
    pid, ep1, ep2, char, root = world
    other_pid, other_ep = f"p-{uuid.uuid4()}", str(uuid.uuid4())
    _seed(pg_session, other_pid, other_ep)
    rel = "refs/남의것.png"
    _write_file(root, rel)
    _ref_asset(pg_session, other_pid, other_ep, char, rel)
    pg_session.commit()
    assert not _verify(pg_session, pid, ep2).is_complete


def test_verify_대상_0이면_clean(pg_session, monkeypatch, tmp_path: Path):
    """장소만 있는 에피소드 — 잴 것이 없으면 통과다."""
    monkeypatch.setattr("app.core.config.settings.projects_dir",
                        str(tmp_path / "projects"))
    monkeypatch.setattr("app.core.file_paths.PROJECT_ROOT", tmp_path)
    pid, eid = f"p-{uuid.uuid4()}", str(uuid.uuid4())
    _seed(pg_session, pid, eid)
    loc = _canon(pg_session, pid, "L01", "버스 안", "location")
    _link(pg_session, pid, eid, loc)
    pg_session.commit()

    r = _verify(pg_session, pid, eid)
    assert r.is_complete and r.metadata["expected"] == 0, r.metadata


def test_verify_저빈도_스킵은_대상에서_빠진다(pg_session, world, monkeypatch):
    pid, ep1, ep2, char, root = world
    monkeypatch.setattr("app.core.low_freq_skip.load_low_freq_skip_ids",
                        lambda *a, **k: {char})
    r = _verify(pg_session, pid, ep2)
    assert r.is_complete and r.metadata["expected"] == 0, r.metadata


def test_게이트와_verify_가_같은_수를_말한다(pg_session, world, monkeypatch):
    """★같은 물음에 두 자리가 다른 답을 내지 않는다."""
    from app.core.pipeline_gate import get_pipeline_status

    pid, ep1, ep2, char, root = world
    rel = "refs/c01.png"
    _write_file(root, rel)
    _ref_asset(pg_session, pid, ep1, char, rel)
    _still(pg_session, pid, ep2)
    pg_session.commit()

    gate = _gate(pg_session, pid, ep2, monkeypatch)
    st = get_pipeline_status(pg_session, pid, ep2)["reference_images"]
    v = _verify(pg_session, pid, ep2).metadata

    assert gate["entities"] == st["entity_total"] == v["expected"], (gate, st, v)
    assert gate["refs"] == st["ref_done"] == v["found"], (gate, st, v)