"""같은 인물이 화마다 **다른 옷**일 때, 2화가 1화 옷을 안 본다 (2026-09-04).

## 무엇이 결함이었나 — Codex BLOCK B

`character_outlook` 에 화 범위가 없어서(2026-09-04 alembic 014 이전),
production 조회 열 곳이 **프로젝트 전체**를 읽었다:

    image_steps · reference_phase2 · reference_phase3 · scene_reference ·
    still_recipe · reference_pipeline_orchestrator · pipeline_gate ·
    asset_readiness

그래서 같은 인물 `C01` 이 1화에 `O01`, 2화에 `O02` 를 입으면 **2화가 두 벌 다**
생성·검증·프롬프트 재료로 봤고, **한 화의 `O00`(배정 없음)이 다른 화 판단까지
오염**시켰다.

## 여기서 재는 것 — 실제 소비자의 **끝점**

2화가 보는 것에 `O02` 만 있고 `O01` 은 **0**이어야 한다:

1. `episode_outlook_rows` 정본 자체
2. 참조 준비도(`asset_readiness`) 의 합성 기대 목록
3. 이미지 게이트(`pipeline_gate`) 의 합성 대상
4. 보류(`shelved`) 인물은 어느 자리에도 안 들어온다

★legacy(`episode_id IS NULL`) 는 **exact 가 하나도 없을 때만** 되돌아간다.
그때도 이 화의 active 링크 **양 끝**에 걸린 행만 — 무조건 합치면 014 가 남긴
NULL 행 때문에 과거 옷이 모든 화로 다시 들어온다 (Codex 재지적).

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

import uuid

import pytest
from sqlalchemy import text as sql_text

pytestmark = pytest.mark.pg


def _seed(session, pid: str, ep1: str, ep2: str) -> None:
    uid = f"epscope-{uuid.uuid4()}"
    session.execute(sql_text(
        "INSERT INTO user_account (id, username, display_name, password_hash, "
        "role, is_active, created_at, updated_at) VALUES "
        "(:u, :un, 't', 'x', 'creator', 1, 'x', 'x')"), {"u": uid, "un": uid})
    session.execute(sql_text(
        "INSERT INTO project_registry (id, name, created_by, created_at, "
        "updated_at) VALUES (:p, 'epscope', :u, 'x', 'x')"),
        {"p": pid, "u": uid})
    for eid, num in ((ep1, 1), (ep2, 2)):
        session.execute(sql_text(
            "INSERT INTO episode (id, project_id, title, episode_number, "
            "source_filename, source_path, status, created_at, updated_at) "
            "VALUES (:e, :p, :t, :n, 'x', 'x', 'analyzed', 'x', 'x')"),
            {"e": eid, "p": pid, "t": f"{num}화", "n": num})
    session.commit()


def _canon(session, pid: str, sid: str, name: str, etype: str) -> str:
    cid = str(uuid.uuid4())
    session.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 "
        "(:c, :p, :s, :n, :e, '', '[]', '{}', 't', 'active', 'x', 'x')"),
        {"c": cid, "p": pid, "s": sid, "n": name, "e": etype})
    return cid


def _link(session, pid: str, eid: str, cid: str, presence: str = "active") -> None:
    session.execute(sql_text(
        "INSERT INTO entity_episode_link (id, canon_id, project_id, episode_id, "
        "presence_status) VALUES (:l, :c, :p, :e, :s)"),
        {"l": str(uuid.uuid4()), "c": cid, "p": pid, "e": eid, "s": presence})


def _pair(session, pid: str, char_id: str, ol_id: str, eid) -> None:
    session.execute(sql_text(
        "INSERT INTO character_outlook (id, project_id, episode_id, "
        "character_id, outlook_id, created_at) VALUES (:i, :p, :e, :c, :o, 'x')"),
        {"i": str(uuid.uuid4()), "p": pid, "e": eid, "c": char_id, "o": ol_id})


@pytest.fixture
def scene(pg_session, tmp_path, monkeypatch):
    """같은 인물 C01 이 1화엔 O01, 2화엔 O02. 2화에 보류된 소품 하나."""
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    pid = f"p-{uuid.uuid4()}"
    ep1, ep2 = str(uuid.uuid4()), str(uuid.uuid4())
    _seed(pg_session, pid, ep1, ep2)

    ids = {
        "C01": _canon(pg_session, pid, "C01", "정순", "character"),
        "O01": _canon(pg_session, pid, "O01", "앞치마", "outlook"),
        "O02": _canon(pg_session, pid, "O02", "외투", "outlook"),
        "P09": _canon(pg_session, pid, "P09", "보류소품", "prop"),
    }
    for eid in (ep1, ep2):
        _link(pg_session, pid, eid, ids["C01"])
    _link(pg_session, pid, ep1, ids["O01"])
    _link(pg_session, pid, ep2, ids["O02"])
    # ★2화에서 보류된 소품 — 어느 자리에도 안 들어와야 한다.
    _link(pg_session, pid, ep2, ids["P09"], presence="shelved")

    _pair(pg_session, pid, ids["C01"], ids["O01"], ep1)
    _pair(pg_session, pid, ids["C01"], ids["O02"], ep2)
    pg_session.commit()
    return pid, ep1, ep2, ids


def test_정본이_이_화의_배정만_돌려준다(pg_session, scene):
    from app.core.entity_identity import episode_outlook_pairs

    pid, ep1, ep2, ids = scene
    p1 = episode_outlook_pairs(pg_session, pid, ep1)
    p2 = episode_outlook_pairs(pg_session, pid, ep2)

    assert p1 == [(ids["C01"], ids["O01"])], f"1화가 2화 옷을 봤다: {p1}"
    assert p2 == [(ids["C01"], ids["O02"])], f"2화가 1화 옷을 봤다: {p2}"


def test_exact_가_있으면_legacy_는_안_섞인다(pg_session, scene):
    """★★★뒤집은 단정 (Codex BLOCK 2026-09-04 재지적).

    앞 판은 exact 와 **모든 NULL 행을 OR 로** 합쳤고, 이 시험이 바로 그
    합집합을 「정상」으로 잠가서 반례를 못 보고 있었다. 그런데 alembic 014 는
    기존 행을 전부 NULL 로 남긴다 — 그러면 다중 화 legacy 프로젝트에서
    **과거 옷이 모든 화로 다시 들어온다.** 고치려던 것이 돌아오는 셈이다.
    """
    from app.core.entity_identity import episode_outlook_pairs

    pid, ep1, ep2, ids = scene
    legacy_ol = _canon(pg_session, pid, "O07", "옛외투", "outlook")
    _pair(pg_session, pid, ids["C01"], legacy_ol, None)
    pg_session.commit()

    got = {o for _c, o in episode_outlook_pairs(pg_session, pid, ep2)}
    assert got == {ids["O02"]}, (
        f"exact 배정이 있는데 legacy 가 섞였다: {got}")


def test_exact_가_하나도_없을_때만_legacy_로_되돌아간다(pg_session, scene):
    """★그때도 **이 화의 active 링크 양 끝**에 걸린 행만. 모호하면 안 넓힌다."""
    from app.core.entity_identity import episode_outlook_pairs

    pid, ep1, ep2, ids = scene
    # 이 화의 exact 행을 지운다 — 014 이전 프로젝트의 모양.
    pg_session.execute(sql_text(
        "DELETE FROM character_outlook WHERE project_id = :p AND episode_id = :e"),
        {"p": pid, "e": ep2})
    # 양 끝이 이 화에 걸린 legacy 하나 + 한쪽만 걸린 legacy 하나.
    _pair(pg_session, pid, ids["C01"], ids["O02"], None)   # 둘 다 2화 링크 있음
    _pair(pg_session, pid, ids["C01"], ids["O01"], None)   # O01 은 1화만
    pg_session.commit()

    got = {o for _c, o in episode_outlook_pairs(pg_session, pid, ep2)}
    assert got == {ids["O02"]}, (
        f"양 끝이 이 화에 안 걸린 legacy 까지 들어왔다: {got}")


def test_참조_준비도가_이_화_합성만_기대한다(pg_session, scene):
    """`asset_readiness` 끝점 — 종전에는 두 벌을 다 기대했다."""
    from app.core.asset_readiness import compute_episode_asset_readiness

    pid, _ep1, ep2, ids = scene
    card = compute_episode_asset_readiness(pg_session, pid, ep2)
    combos = [x for x in card.expected_ids if x.startswith("composite:")]

    assert any(ids["O02"] in c for c in combos), f"2화 옷이 기대에 없다: {combos}"
    assert not any(ids["O01"] in c for c in combos), (
        f"1화 옷을 2화가 기대한다: {combos}")


def test_보류된_요소는_참조_대상이_아니다(pg_session, scene):
    """게이트와 준비도가 **같은 물음**을 쓰는지도 같이 본다."""
    from app.core.asset_readiness import compute_episode_asset_readiness
    from app.core.pipeline_gate import episode_reference_entities

    pid, _ep1, ep2, ids = scene
    gate = {e.short_id for e in episode_reference_entities(pg_session, pid, ep2)}
    assert "P09" not in gate, f"보류 소품이 참조 대상에 들어왔다: {gate}"

    card = compute_episode_asset_readiness(pg_session, pid, ep2)
    assert not any(ids["P09"] in x for x in card.expected_ids), (
        "보류 소품을 준비도가 기대한다")


def test_한_화의_O00_이_다른_화_판단을_오염시키지_않는다(pg_session, scene):
    """`O00` 은 「배정 없음」 표식이라 화마다 뜻이 다르다."""
    from app.core.entity_identity import (NULL_OUTLOOK_SHORT_ID,
                                          episode_outlook_pairs)

    pid, ep1, ep2, ids = scene
    o00 = _canon(pg_session, pid, NULL_OUTLOOK_SHORT_ID, "Null Outlook", "outlook")
    # 1화에서만 이 인물이 O00 이다.
    _pair(pg_session, pid, ids["C01"], o00, ep1)
    pg_session.commit()

    p2 = {o for _c, o in episode_outlook_pairs(pg_session, pid, ep2)}
    assert o00 not in p2, "1화의 O00 배정이 2화로 샜다"
    p1 = {o for _c, o in episode_outlook_pairs(pg_session, pid, ep1)}
    assert o00 in p1, "1화 자신의 O00 이 사라졌다"


def test_미래_화_착용자가_앞_화_앵커에_안_섞인다(pg_session, scene):
    """★2화가 생긴 뒤 1화를 force 하면 2화 착용자가 1화 앵커에 섞였다.

    그 잘못된 값이 새 스냅샷에 그대로 얼어붙는다 (Codex BLOCK 2026-09-04).
    """
    from app.modules.pipeline.episode_carry import build_prior_roster

    pid, ep1, ep2, ids = scene
    # 같은 아웃룩을 두 화가 서로 다른 인물에게 배정한다.
    c2 = _canon(pg_session, pid, "C02", "태욱", "character")
    _link(pg_session, pid, ep2, c2)
    _pair(pg_session, pid, c2, ids["O01"], ep2)
    pg_session.commit()

    # ★`force` 판이다 — 이 화 자신을 보는 **유일한 자리**(2026-09-04).
    #  평상시(normal)에는 1화가 앞 화가 없어 명부 자체가 비고, 이 결함은
    #  force 로 다시 뜰 때만 드러난다.
    lines, allowed = build_prior_roster(pg_session, pid, "outlook", ep1,
                                        include_self=True)
    assert allowed == ["O01"], allowed
    line = lines[0]
    assert "C01" in line, f"1화 착용자가 앵커에 없다: {line}"
    assert "C02" not in line, f"2화 착용자가 1화 앵커에 섞였다: {line}"


def test_앞_화_착용자는_다음_화_앵커에_남는다(pg_session, scene):
    """★★★미래를 막다가 **과거까지 지우면** 안 된다 (Codex 3차).

    2화를 **처음** 추출할 때는 2화 아웃룩 배정이 아직 없다. 그때 앵커를
    「이 화의 배정」으로만 보면 1화의 `O01` 이 명부에 나오면서도 착용자가
    비어, 같은 이름의 옷을 가르는 핵심 정보가 사라진다.
    """
    from app.modules.pipeline.episode_carry import build_prior_roster

    pid, ep1, ep2, ids = scene
    # 2화의 아웃룩 배정을 지운다 — 「처음 추출하는 시점」의 모양.
    pg_session.execute(sql_text(
        "DELETE FROM character_outlook WHERE project_id = :p AND episode_id = :e"),
        {"p": pid, "e": ep2})
    pg_session.commit()

    lines, allowed = build_prior_roster(pg_session, pid, "outlook", ep2)
    assert "O01" in allowed, f"1화 아웃룩이 2화 명부에 없다: {allowed}"
    o01 = [x for x in lines if x.startswith("- O01")][0]
    assert "C01" in o01, f"1화 착용자 앵커가 사라졌다: {o01}"


def test_그래도_미래_화_착용자는_안_들어온다(pg_session, scene):
    """앞을 살리면서 뒤를 막는가 — 둘 다여야 한다."""
    from app.modules.pipeline.episode_carry import build_prior_roster

    pid, ep1, ep2, ids = scene
    ep3 = str(uuid.uuid4())
    pg_session.execute(sql_text(
        "INSERT INTO episode (id, project_id, title, episode_number, "
        "source_filename, source_path, status, created_at, updated_at) VALUES "
        "(:e, :p, '3화', 3, 'x', 'x', 'analyzed', 'x', 'x')"),
        {"e": ep3, "p": pid})
    c2 = _canon(pg_session, pid, "C02", "태욱", "character")
    _link(pg_session, pid, ep3, c2)
    _pair(pg_session, pid, c2, ids["O01"], ep3)
    pg_session.execute(sql_text(
        "DELETE FROM character_outlook WHERE project_id = :p AND episode_id = :e"),
        {"p": pid, "e": ep2})
    pg_session.commit()

    lines, _allowed = build_prior_roster(pg_session, pid, "outlook", ep2)
    o01 = [x for x in lines if x.startswith("- O01")][0]
    assert "C01" in o01, f"앞 화 착용자가 사라졌다: {o01}"
    assert "C02" not in o01, f"미래 화(3화) 착용자가 섞였다: {o01}"


def test_legacy_혼합_프로젝트에서_화별로_따진다(pg_session, scene):
    """★★★범위 전체로 한 번 접으면 **앞 화 legacy 앵커를 통째로 잃는다**.

    014 뒤 2화만 sync 돼 exact 가 생기고 1화는 아직 NULL 인 흔한 중간 상태
    (Codex BLOCK 4차). 3화 명부에 1화(legacy)와 2화(exact) 것이 **둘 다**
    남아야 한다.
    """
    from app.modules.pipeline.episode_carry import build_prior_roster

    pid, ep1, ep2, ids = scene
    ep3 = str(uuid.uuid4())
    pg_session.execute(sql_text(
        "INSERT INTO episode (id, project_id, title, episode_number, "
        "source_filename, source_path, status, created_at, updated_at) VALUES "
        "(:e, :p, '3화', 3, 'x', 'x', 'analyzed', 'x', 'x')"), {"e": ep3, "p": pid})
    # 1화: legacy(NULL) 배정으로 되돌린다 — 014 이전에 쓰인 모양.
    pg_session.execute(sql_text(
        "DELETE FROM character_outlook WHERE project_id = :p AND episode_id = :e"),
        {"p": pid, "e": ep1})
    _pair(pg_session, pid, ids["C01"], ids["O01"], None)
    # 2화: exact 배정(C02 → O02). 1화 것과 다른 인물이어야 섞임이 드러난다.
    c2 = _canon(pg_session, pid, "C02", "태욱", "character")
    _link(pg_session, pid, ep2, c2)
    pg_session.execute(sql_text(
        "DELETE FROM character_outlook WHERE project_id = :p AND episode_id = :e"),
        {"p": pid, "e": ep2})
    _pair(pg_session, pid, c2, ids["O02"], ep2)
    pg_session.commit()

    lines, allowed = build_prior_roster(pg_session, pid, "outlook", ep3)
    assert set(allowed) == {"O01", "O02"}, allowed
    o01 = [x for x in lines if x.startswith("- O01")][0]
    o02 = [x for x in lines if x.startswith("- O02")][0]
    assert "C01" in o01, f"1화 legacy 앵커를 잃었다: {o01}"
    assert "C02" in o02, f"2화 exact 앵커가 없다: {o02}"


def test_양_끝이_같은_화에서_active_여야_한다(pg_session, scene):
    """★화들을 합쳐 보면 C01 은 1화에만, O02 는 2화에만 active 인데도 통과한다.

    「같은 화에서 함께 active 였다」는 근거가 없다 (Codex BLOCK 4차).
    """
    from app.modules.pipeline.episode_carry import build_prior_roster

    pid, ep1, ep2, ids = scene
    ep3 = str(uuid.uuid4())
    pg_session.execute(sql_text(
        "INSERT INTO episode (id, project_id, title, episode_number, "
        "source_filename, source_path, status, created_at, updated_at) VALUES "
        "(:e, :p, '3화', 3, 'x', 'x', 'analyzed', 'x', 'x')"), {"e": ep3, "p": pid})
    # 모든 exact 배정을 지우고 legacy 하나만 남긴다.
    pg_session.execute(sql_text(
        "DELETE FROM character_outlook WHERE project_id = :p"), {"p": pid})
    # C01 은 1화에만, O02 는 2화에만 active — 같은 화에 함께 있은 적이 없다.
    pg_session.execute(sql_text(
        "DELETE FROM entity_episode_link WHERE project_id = :p AND canon_id = :c "
        "AND episode_id = :e"), {"p": pid, "c": ids["C01"], "e": ep2})
    _pair(pg_session, pid, ids["C01"], ids["O02"], None)
    pg_session.commit()

    lines, _allowed = build_prior_roster(pg_session, pid, "outlook", ep3)
    o02 = [x for x in lines if x.startswith("- O02")]
    assert o02 and "C01" not in o02[0], (
        f"같은 화에 함께 active 였던 적 없는 legacy 가 앵커로 들어왔다: {o02}")
