"""ShotValidatorStep v4 단위 테스트 (G4.6 RC-F + RC-G).

신규 동작 검증:
1. _build_entity_blocks_for_all_scenes — scene_director.present_entity_ids
   기반 entity map block 사전 빌드 (thread-safe)
2. user_prompt 에 [Entity map for this scene] block 포함
3. character_ids / characters merge — changed=false 여도 LLM 매핑 적용
4. failure path — validator_status='failed' + failed_carry_original 마킹

fixture 는 모두 generic — 시나리오 의존 어휘 0건. character name/description 은
universal "type-descriptor character" / "actor-action-target" 형식.
"""
from __future__ import annotations

import json
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

from app.core.errors import AppError
from app.core.steps.shot_validator_step import (
    ShotValidatorStep,
    _filter_character_ids,
    assert_no_failed_scenes,
)


# ──────────────────────────────────────────────────────────────────────
# Fixtures (generic — 시나리오 의존 어휘 0)
# ──────────────────────────────────────────────────────────────────────


@pytest.fixture
def project_episode(tmp_path, monkeypatch) -> tuple[str, str, Path]:
    pid = "p1"
    eid = "e1"
    ckpt_dir = tmp_path / pid / "checkpoints" / "episodes" / eid
    ckpt_dir.mkdir(parents=True, exist_ok=True)

    from app.core import config as _cfg
    monkeypatch.setattr(_cfg.settings, "projects_dir", str(tmp_path))
    return pid, eid, tmp_path


def _write_cp(base: Path, pid: str, eid: str, step: str, data: dict) -> None:
    target = base / pid / "checkpoints" / "episodes" / eid / step
    target.mkdir(parents=True, exist_ok=True)
    (target / "manifest.json").write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")


def _make_step(pid: str, eid: str) -> ShotValidatorStep:
    """최소 mock된 ShotValidatorStep — db 는 외부에서 set."""
    instance = ShotValidatorStep.__new__(ShotValidatorStep)
    instance.step_id = "shot_validator"
    instance.project_id = pid
    instance.episode_id = eid
    instance.db = MagicMock()
    instance.project_config = {}
    instance.manifest = {}
    instance.run_id = "r1"
    instance.opik_context = {}
    return instance


def _prompt_patches():
    return patch.multiple(
        "app.core.steps.shot_validator_step",
        load_prompt=MagicMock(return_value="SYSTEM PROMPT"),
        load_schema=MagicMock(return_value={"type": "object"}),
    )


def _mock_db_with_entities(rows: list[tuple[str, str, str]]) -> MagicMock:
    """rows = [(short_id, name, stable_traits_json)] → SQLAlchemy query mock."""
    db = MagicMock()
    row_objs = [SimpleNamespace(short_id=sid, name=nm, stable_traits=tr) for sid, nm, tr in rows]
    db.query.return_value.filter.return_value.all.return_value = row_objs
    return db


# ──────────────────────────────────────────────────────────────────────
# 1) _build_entity_blocks_for_all_scenes
# ──────────────────────────────────────────────────────────────────────


def test_entity_blocks_built_from_scene_director(project_episode):
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "scene_director", {
        "data": {"scenes": [
            {"scene_index": 1, "present_entity_ids": ["C01", "C02"]},
            {"scene_index": 2, "present_entity_ids": ["C01", "C02O01"]},  # composite → C02
        ]},
    })
    step = _make_step(pid, eid)
    step.db = _mock_db_with_entities([
        ("C01", "type-descriptor character A", json.dumps(["trait-1", "trait-2"])),
        ("C02", "type-descriptor character B", json.dumps(["trait-x"])),
    ])

    blocks = step._build_entity_blocks_for_all_scenes()

    assert 1 in blocks and 2 in blocks
    # scene 1: 두 entity 모두 등장
    assert "[Entity map for this scene]" in blocks[1]
    assert "C01: type-descriptor character A" in blocks[1]
    assert "trait-1, trait-2" in blocks[1]
    assert "C02: type-descriptor character B" in blocks[1]
    # scene 2: composite C02O01 → base C02 만 등장
    assert "C02: type-descriptor character B" in blocks[2]
    # block 끝에 빈 줄 (user_prompt prepend 호환)
    assert blocks[1].endswith("\n\n")


def test_entity_blocks_empty_when_no_director_cp(project_episode):
    pid, eid, _ = project_episode  # cp 미작성
    step = _make_step(pid, eid)
    blocks = step._build_entity_blocks_for_all_scenes()
    assert blocks == {}


def test_entity_blocks_omit_scene_with_no_present_entity_ids(project_episode):
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "scene_director", {
        "data": {"scenes": [
            {"scene_index": 1, "present_entity_ids": ["C01"]},
            {"scene_index": 2, "present_entity_ids": []},  # 빈 list → 생략
            {"scene_index": 3},  # 필드 자체 없음 → 생략
        ]},
    })
    step = _make_step(pid, eid)
    step.db = _mock_db_with_entities([
        ("C01", "type-descriptor character A", "[]"),
    ])

    blocks = step._build_entity_blocks_for_all_scenes()

    assert 1 in blocks
    assert 2 not in blocks
    assert 3 not in blocks


def test_entity_blocks_omit_when_db_lookup_returns_empty(project_episode):
    """short_id 가 EntityCanon 에 없으면 (예: stale checkpoint) 해당 씬 entry 생략."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "scene_director", {
        "data": {"scenes": [{"scene_index": 1, "present_entity_ids": ["C99"]}]},
    })
    step = _make_step(pid, eid)
    step.db = _mock_db_with_entities([])  # DB 에 C99 없음

    blocks = step._build_entity_blocks_for_all_scenes()

    assert blocks == {}


def test_entity_blocks_uses_entity_merge_cp_as_primary_source(project_episode):
    """entity_merge cp 가 있으면 EntityCanon DB fallback 보다 우선."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "entity_merge", {
        "data": {"characters": [
            {"short_id": "C01", "name": "merge-character A", "stable_traits": ["m-trait-1"]},
            {"short_id": "C02", "name": "merge-character B"},
        ]},
    })
    _write_cp(base, pid, eid, "scene_director", {
        "data": {"scenes": [{"scene_index": 1, "present_entity_ids": ["C01", "C02"]}]},
    })
    step = _make_step(pid, eid)
    # DB 에 다른 데이터가 있더라도 entity_merge 가 우선 — db 호출되지 않아야 함
    step.db = _mock_db_with_entities([
        ("C01", "DB-NAME-SHOULD-NOT-APPEAR", "[]"),
    ])

    blocks = step._build_entity_blocks_for_all_scenes()

    assert "merge-character A" in blocks[1]
    assert "merge-character B" in blocks[1]
    assert "DB-NAME-SHOULD-NOT-APPEAR" not in blocks[1]


def test_entity_blocks_universal_when_no_director_cp_but_pool_present(project_episode):
    """scene_director 부재 + entity_merge 있음 → 모든 씬에 universal pool block."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": []},
            {"scene_index": 5, "shots": []},
        ]},
    })
    _write_cp(base, pid, eid, "entity_merge", {
        "data": {"characters": [
            {"short_id": "C01", "name": "character A"},
            {"short_id": "C02", "name": "character B"},
        ]},
    })
    # scene_director cp 미작성

    step = _make_step(pid, eid)
    blocks = step._build_entity_blocks_for_all_scenes()

    assert 1 in blocks and 5 in blocks
    # 두 씬 모두 동일 universal block (모든 character pool)
    assert blocks[1] == blocks[5]
    assert "C01: character A" in blocks[1]
    assert "C02: character B" in blocks[1]


def test_entity_blocks_filters_non_character_present_entities(project_episode):
    """scene_director.present_entity_ids 에 L##/P## 섞여도 C## 만 entity_block 에 포함."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "scene_director", {
        "data": {"scenes": [{"scene_index": 1, "present_entity_ids": [
            "C01", "L02", "P03", "C04O01",  # C04 base, L02/P03 제외
        ]}]},
    })
    step = _make_step(pid, eid)
    step.db = _mock_db_with_entities([
        ("C01", "character A", "[]"),
        ("C04", "character B", "[]"),
    ])

    blocks = step._build_entity_blocks_for_all_scenes()

    assert "C01: character A" in blocks[1]
    assert "C04: character B" in blocks[1]
    # L## / P## 는 character_ids 필드 대상 아님 — block 에 등장 X
    assert "L02" not in blocks[1]
    assert "P03" not in blocks[1]


def test_entity_blocks_warns_when_pool_fully_empty(project_episode, caplog):
    """entity_merge 부재 + EntityCanon 비어있음 → WARNING + 빈 dict (RC-F 부분 비활성)."""
    import logging
    caplog.set_level(logging.WARNING)
    pid, eid, _ = project_episode  # cp 미작성

    step = _make_step(pid, eid)
    step.db = _mock_db_with_entities([])  # DB 에도 character 없음

    blocks = step._build_entity_blocks_for_all_scenes()

    assert blocks == {}
    assert any(
        "RC-F entity mapping inactive" in rec.message
        for rec in caplog.records
    )


# ──────────────────────────────────────────────────────────────────────
# 2) user_prompt 에 entity_block 포함
# ──────────────────────────────────────────────────────────────────────


def test_user_prompt_includes_entity_block_when_director_cp_present(project_episode):
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "actor at venue", "based_on_beat": 1, "characters": []},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})
    _write_cp(base, pid, eid, "scene_director", {
        "data": {"scenes": [{"scene_index": 1, "present_entity_ids": ["C01"]}]},
    })

    step = _make_step(pid, eid)
    step.db = _mock_db_with_entities([
        ("C01", "type-descriptor character A", "[]"),
    ])

    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.return_value = {
            "scene_index": 1,
            "shots": [{"shot_index": 1, "changed": False, "revised_description": "", "reason": ""}],
        }
        step._execute()

    user_prompt = llm.call_args.kwargs["user_prompt"]
    assert "[Entity map for this scene]" in user_prompt
    assert "C01: type-descriptor character A" in user_prompt
    assert "character_ids 는 entity map 의 short_id 사용" in user_prompt


def test_user_prompt_includes_universal_pool_block_when_only_entity_merge_cp(project_episode):
    """entity_merge cp 있고 scene_director cp 부재 → user_prompt 에 universal pool block.

    I3 — universal-pool branch 의 LLM call path 통합 검증 (prompt 안 entity_block 도달).
    """
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "actor at venue", "based_on_beat": 1, "characters": []},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})
    _write_cp(base, pid, eid, "entity_merge", {
        "data": {"characters": [
            {"short_id": "C01", "name": "type-descriptor character A"},
            {"short_id": "C02", "name": "type-descriptor character B"},
        ]},
    })
    # scene_director cp 미작성 → universal pool branch

    step = _make_step(pid, eid)

    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.return_value = {
            "scene_index": 1,
            "shots": [{"shot_index": 1, "changed": False, "revised_description": "", "reason": ""}],
        }
        step._execute()

    user_prompt = llm.call_args.kwargs["user_prompt"]
    assert "[Entity map for this scene]" in user_prompt
    # 모든 character pool 이 prompt 에 포함 (universal block)
    assert "C01: type-descriptor character A" in user_prompt
    assert "C02: type-descriptor character B" in user_prompt


def test_user_prompt_omits_entity_block_when_no_director_cp(project_episode):
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "actor at venue", "based_on_beat": 1, "characters": []},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})
    # scene_director cp 미작성

    step = _make_step(pid, eid)

    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.return_value = {
            "scene_index": 1,
            "shots": [{"shot_index": 1, "changed": False, "revised_description": "", "reason": ""}],
        }
        step._execute()

    user_prompt = llm.call_args.kwargs["user_prompt"]
    assert "[Entity map for this scene]" not in user_prompt


# ──────────────────────────────────────────────────────────────────────
# 3) character_ids / characters merge (changed=false 여도)
# ──────────────────────────────────────────────────────────────────────


def test_validator_merges_character_ids_when_changed_false(project_episode):
    """LLM 이 changed=false + character_ids/characters 매핑만 채운 경우 — merge 적용."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "actor A acts on actor B", "based_on_beat": 1, "characters": []},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})

    step = _make_step(pid, eid)
    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.return_value = {
            "scene_index": 1,
            "shots": [{
                "shot_index": 1,
                "changed": False,
                "revised_description": "",
                "reason": "",
                "character_ids": ["C01", "C02"],
                "characters": ["type-descriptor character A", "type-descriptor character B"],
            }],
        }
        out = step._execute()

    shot = out["data"]["scenes"][0]["shots"][0]
    assert shot["description"] == "actor A acts on actor B"  # 원본 유지
    assert shot["character_ids"] == ["C01", "C02"]
    assert shot["characters"] == ["type-descriptor character A", "type-descriptor character B"]
    assert "original_description" not in shot  # changed=false → 백업 없음


def test_validator_merges_character_ids_when_changed_true(project_episode):
    """changed=true + character_ids 동시 — description 재작성과 매핑이 함께 적용."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "actor A acts on actor B then turns", "based_on_beat": 1, "characters": []},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})

    step = _make_step(pid, eid)
    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.return_value = {
            "scene_index": 1,
            "shots": [{
                "shot_index": 1,
                "changed": True,
                "revised_description": "actor A acts on actor B mid-action",
                "reason": "시간 연결어 'then' 제거",
                "character_ids": ["C01", "C02"],
                "characters": ["type-descriptor character A", "type-descriptor character B"],
            }],
        }
        out = step._execute()

    shot = out["data"]["scenes"][0]["shots"][0]
    assert shot["description"] == "actor A acts on actor B mid-action"
    assert shot["original_description"] == "actor A acts on actor B then turns"
    assert shot["character_ids"] == ["C01", "C02"]
    assert shot["characters"] == ["type-descriptor character A", "type-descriptor character B"]


def test_validator_skips_merge_when_llm_omits_character_fields(project_episode):
    """LLM 이 character_ids/characters 필드 자체를 안 보내면 — 원본 characters 보존, character_ids 추가 X."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "actor at venue", "based_on_beat": 1, "characters": ["legacy-name"]},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})

    step = _make_step(pid, eid)
    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.return_value = {
            "scene_index": 1,
            "shots": [{"shot_index": 1, "changed": False, "revised_description": "", "reason": ""}],
        }
        out = step._execute()

    shot = out["data"]["scenes"][0]["shots"][0]
    assert shot["characters"] == ["legacy-name"]  # 원본 보존
    assert "character_ids" not in shot  # 추가 X


def test_validator_strict_drops_malformed_character_ids(project_episode, caplog):
    """strict validation — '42' / 'C-A' / None / '' 모두 drop, 유효한 C## 만 보존.

    pool 부재 (entity_merge + EntityCanon 비어있음) 시 regex 만 검증 — pool
    candidate set 검증은 skip.
    """
    import logging
    caplog.set_level(logging.WARNING)
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "actor at venue", "based_on_beat": 1, "characters": []},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})

    step = _make_step(pid, eid)
    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.return_value = {
            "scene_index": 1,
            "shots": [{
                "shot_index": 1,
                "changed": False,
                "revised_description": "",
                "reason": "",
                "character_ids": ["C01", 42, None, "", "C-A", "C02O01"],  # 다양한 invalid
                "characters": ["A", None, "B"],
            }],
        }
        out = step._execute()

    shot = out["data"]["scenes"][0]["shots"][0]
    # 42 / None / "" / "C-A" 모두 drop. C02O01 은 base C02 로 collapse → 보존.
    assert shot["character_ids"] == ["C01", "C02"]
    # characters 는 None 만 drop (str 보존)
    assert shot["characters"] == ["A", "B"]
    # 1+ warning 로그 발생 (drop 안내)
    assert any("dropping malformed" in r.message for r in caplog.records)


def test_validator_strict_drops_out_of_pool_character_ids(project_episode, caplog):
    """character_pool 이 있으면 pool 안 candidate set 검증 — 외부 ID drop."""
    import logging
    caplog.set_level(logging.WARNING)
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "actor at venue", "based_on_beat": 1, "characters": []},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})
    # entity_merge cp 작성 — pool 안에 C01 만
    _write_cp(base, pid, eid, "entity_merge", {
        "data": {"characters": [{"short_id": "C01", "name": "A"}]},
    })

    step = _make_step(pid, eid)
    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.return_value = {
            "scene_index": 1,
            "shots": [{
                "shot_index": 1,
                "changed": False,
                "revised_description": "",
                "reason": "",
                "character_ids": ["C01", "C99"],  # C99 는 pool 밖 → drop
                "characters": ["A"],
            }],
        }
        out = step._execute()

    shot = out["data"]["scenes"][0]["shots"][0]
    assert shot["character_ids"] == ["C01"]
    assert any("out-of-pool" in r.message or "malformed" in r.message for r in caplog.records)


def test_validator_strict_dedupes_character_ids(project_episode):
    """character_ids 중복 entry — 첫 등장 순서 보존하며 dedupe."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "actor at venue", "based_on_beat": 1, "characters": []},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})

    step = _make_step(pid, eid)
    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.return_value = {
            "scene_index": 1,
            "shots": [{
                "shot_index": 1,
                "changed": False,
                "revised_description": "",
                "reason": "",
                "character_ids": ["C02", "C01", "C02", "C01O03"],
                "characters": [],
            }],
        }
        out = step._execute()

    shot = out["data"]["scenes"][0]["shots"][0]
    # C02, C01 첫 등장 순서. 중복 C02 + C01O03 (base C01 dedupe) drop.
    assert shot["character_ids"] == ["C02", "C01"]


# ──────────────────────────────────────────────────────────────────────
# 4) failure path — validator_status='failed' + failed_carry_original
# ──────────────────────────────────────────────────────────────────────


def test_validator_marks_failed_scene_with_metadata(project_episode):
    """LLM 3번 다 실패 → validator_status='failed' + reason + 각 shot 마킹."""
    pid, eid, base = project_episode
    original = {"scene_index": 7, "shots": [
        {"shot_index": 1, "description": "shot 1 raw", "based_on_beat": 1, "characters": []},
        {"shot_index": 2, "description": "shot 2 raw", "based_on_beat": 2, "characters": ["legacy"]},
    ]}
    _write_cp(base, pid, eid, "shot_extract", {"data": {"scenes": [original]}})
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 7, "text": ""}]}})

    step = _make_step(pid, eid)
    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.side_effect = RuntimeError("upstream model error")
        out = step._execute()

    sc = out["data"]["scenes"][0]
    assert sc["scene_index"] == 7
    assert sc["validator_status"] == "failed"
    assert "upstream model error" in sc["validator_failure_reason"]
    # description 원본 보존
    assert sc["shots"][0]["description"] == "shot 1 raw"
    assert sc["shots"][1]["description"] == "shot 2 raw"
    # 각 shot 마킹
    for shot in sc["shots"]:
        assert shot["validator_status"] == "failed_carry_original"
    # 다른 필드 보존
    assert sc["shots"][0]["based_on_beat"] == 1
    assert sc["shots"][1]["characters"] == ["legacy"]


def test_validator_truncates_failure_reason_to_200_chars(project_episode):
    """LLM exception message 가 길어도 validator_failure_reason 은 200 자 cap."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [{"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
        ]}]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": [{"scene_index": 1, "text": ""}]}})

    step = _make_step(pid, eid)
    long_msg = "x" * 500
    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        llm.side_effect = RuntimeError(long_msg)
        out = step._execute()

    sc = out["data"]["scenes"][0]
    assert len(sc["validator_failure_reason"]) == 200


def test_validator_preserves_partial_success_alongside_failure(project_episode):
    """한 씬은 성공, 다른 씬은 실패 → 성공 씬은 마킹 없음, 실패 씬만 마킹."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_extract", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "ok scene", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "shots": [
                {"shot_index": 1, "description": "fail scene", "based_on_beat": 1, "characters": []},
            ]},
        ]},
    })
    _write_cp(base, pid, eid, "scene_save", {"data": {"segments": []}})

    step = _make_step(pid, eid)
    with _prompt_patches(), patch("app.core.steps.shot_validator_step.call_structured") as llm:
        import re as _re
        def _side(**kw):
            m = _re.match(r"shot_validator_s(\d+)_", kw["schema_name"])
            si = int(m.group(1))
            if si == 2:
                raise RuntimeError("blocked")
            return {
                "scene_index": si,
                "shots": [{"shot_index": 1, "changed": False, "revised_description": "", "reason": ""}],
            }
        llm.side_effect = _side
        out = step._execute()

    scenes_by_idx = {sc["scene_index"]: sc for sc in out["data"]["scenes"]}
    # scene 1 — 성공, 마킹 없음
    assert "validator_status" not in scenes_by_idx[1]
    # scene 2 — 실패 마킹
    assert scenes_by_idx[2]["validator_status"] == "failed"
    assert scenes_by_idx[2]["shots"][0]["validator_status"] == "failed_carry_original"


# ──────────────────────────────────────────────────────────────────────
# 5) assert_no_failed_scenes helper (downstream fail-fast guard, RO-15)
# ──────────────────────────────────────────────────────────────────────


# ──────────────────────────────────────────────────────────────────────
# 5a) _filter_character_ids — unit-level strict validation
# ──────────────────────────────────────────────────────────────────────


def test_filter_character_ids_drops_none_and_empty():
    accepted, rejected = _filter_character_ids(
        ["C01", None, "", "C02"], pool_keys=set(),
    )
    assert accepted == ["C01", "C02"]
    assert None in rejected and "" in rejected


def test_filter_character_ids_drops_non_C_pattern():
    accepted, rejected = _filter_character_ids(
        ["C01", "L02", "P03", "42", "None", "C-A"], pool_keys=set(),
    )
    assert accepted == ["C01"]
    assert set(rejected) == {"L02", "P03", "42", "None", "C-A"}


def test_filter_character_ids_collapses_composite_to_base():
    accepted, _ = _filter_character_ids(
        ["C01O02", "C03"], pool_keys=set(),
    )
    assert accepted == ["C01", "C03"]


def test_filter_character_ids_rejects_partial_trailing_O():
    """'C1O' / 'C01O' / 'CO01' 같은 partial composite 는 base 추출 X → drop.

    Claude iter 2 I1 — composite split 가 'O' 위치만 보고 collapse 하면 'C1O' 가
    잘못 'C1' 통과. anchored full-pattern regex 로 strict 검증.
    """
    accepted, rejected = _filter_character_ids(
        ["C1O", "C01O", "CO01", "C1OO2", "C01"],
        pool_keys=set(),
    )
    assert accepted == ["C01"]
    assert set(rejected) == {"C1O", "C01O", "CO01", "C1OO2"}


def test_filter_character_ids_documents_digit_count_acceptance():
    """현재 정책 (G4.6 Phase 3): ^C\\d+$ — base ID 의 digit count 강제 X.

    Codex iter 2 B2 — 'C0' / 'C0001' 같은 패딩 anomaly 는 currently accepted.
    canonical short-id (C##) 강제 narrowing 은 별도 Phase 5+ 결정 (project_short_id_design).
    이 test 는 documented behavior 가드 — 변경 시 명시 결정 필요.
    """
    accepted, rejected = _filter_character_ids(
        ["C0", "C01", "C100", "C0001"], pool_keys=set(),
    )
    assert "C0" in accepted
    assert "C01" in accepted
    assert "C100" in accepted
    assert "C0001" in accepted
    assert rejected == []


def test_filter_character_ids_rejects_whitespace_and_special_chars():
    """공백/특수문자 prefix-suffix entry — regex anchored 라 모두 drop."""
    accepted, rejected = _filter_character_ids(
        [" C01", "C01 ", "C01\n", "*C01", "C01;", "C-01", "C_01"],
        pool_keys=set(),
    )
    assert accepted == []
    assert len(rejected) == 7


def test_filter_character_ids_pool_constraint_when_pool_present():
    accepted, rejected = _filter_character_ids(
        ["C01", "C99"], pool_keys={"C01", "C02"},
    )
    assert accepted == ["C01"]
    assert "C99" in rejected


def test_filter_character_ids_pool_skipped_when_pool_empty():
    """pool 부재 (첫 run RC-F 부분 비활성) — pool 검증 skip, regex 만."""
    accepted, _ = _filter_character_ids(
        ["C01", "C99"], pool_keys=set(),
    )
    assert accepted == ["C01", "C99"]


def test_filter_character_ids_dedupes_preserve_order():
    accepted, _ = _filter_character_ids(
        ["C02", "C01", "C02", "C01O03"], pool_keys=set(),
    )
    assert accepted == ["C02", "C01"]


def test_filter_character_ids_coerces_int_to_string_via_base_check():
    """LLM 이 숫자 (int) 보낸 경우 — str() 후 regex 검증, '42' 는 fail → drop."""
    accepted, rejected = _filter_character_ids(
        [42, "C01"], pool_keys=set(),
    )
    assert accepted == ["C01"]
    assert 42 in rejected


def test_assert_no_failed_scenes_passes_when_cp_is_none():
    assert_no_failed_scenes(None, {}, consumer_step="test")


def test_assert_no_failed_scenes_passes_when_no_failed_marker():
    cp = {"data": {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "description": "ok"}]},
        {"scene_index": 2, "shots": [{"shot_index": 1, "description": "ok"}]},
    ]}}
    assert_no_failed_scenes(cp, {}, consumer_step="test")


def test_assert_no_failed_scenes_raises_app_error_when_any_scene_failed():
    cp = {"data": {"scenes": [
        {"scene_index": 1, "shots": []},
        {"scene_index": 7, "validator_status": "failed", "shots": []},
        {"scene_index": 3, "validator_status": "failed", "shots": []},
    ]}}
    with pytest.raises(AppError) as exc_info:
        assert_no_failed_scenes(cp, {}, consumer_step="downstream_step")
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert exc_info.value.status_code == 400
    assert "downstream_step" in exc_info.value.message
    # sorted indices in the message
    assert "[3, 7]" in exc_info.value.message


def test_assert_no_failed_scenes_respects_explicit_operator_override(caplog):
    import logging
    caplog.set_level(logging.WARNING)
    cp = {"data": {"scenes": [
        {"scene_index": 7, "validator_status": "failed", "shots": []},
    ]}}
    assert_no_failed_scenes(cp, {"allow_failed_validator": True}, consumer_step="test")
    assert any(
        "allow_failed_validator=True" in rec.message
        and "explicit operator override" in rec.message
        for rec in caplog.records
    )


def test_assert_no_failed_scenes_rejects_truthy_string_override(caplog):
    """Codex iter 3 I1 — assert_no_failed_scenes 도 strict ``is True`` 적용.

    'true' / '1' / 1 같은 truthy 표현은 명시 bool override X → AppError raise.
    """
    cp = {"data": {"scenes": [
        {"scene_index": 7, "validator_status": "failed", "shots": []},
    ]}}
    for non_strict in ["true", "True", "1", 1, "yes"]:
        with pytest.raises(AppError) as exc_info:
            assert_no_failed_scenes(
                cp, {"allow_failed_validator": non_strict}, consumer_step="test",
            )
        assert exc_info.value.code == "step.upstream_validator_failed", (
            f"non-strict value {non_strict!r} should not unlock override"
        )


def test_assert_no_failed_scenes_handles_none_project_config():
    cp = {"data": {"scenes": [{"scene_index": 1}]}}
    assert_no_failed_scenes(cp, None, consumer_step="test")


# ──────────────────────────────────────────────────────────────────────
# 6) Integration — 3 downstream consumers actually invoke the guard
# ──────────────────────────────────────────────────────────────────────


def _make_consumer_step(step_cls, pid: str, eid: str):
    instance = step_cls.__new__(step_cls)
    instance.step_id = step_cls.__name__.replace("Step", "").lower()
    instance.project_id = pid
    instance.episode_id = eid
    instance.db = MagicMock()
    instance.project_config = {}
    instance.manifest = {}
    instance.run_id = "r1"
    instance.opik_context = {}
    return instance


def test_shot_selection_step_blocks_on_failed_validator(project_episode):
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    from app.core.steps.shot_selection_step import ShotSelectionStep
    step = _make_consumer_step(ShotSelectionStep, pid, eid)

    with pytest.raises(AppError) as exc_info:
        step._execute()
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert "shot_selection" in exc_info.value.message


def test_shot_dependency_step_blocks_on_failed_validator(project_episode):
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    from app.core.steps.shot_dependency_step import ShotDependencyStep
    step = _make_consumer_step(ShotDependencyStep, pid, eid)

    with pytest.raises(AppError) as exc_info:
        step._execute()
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert "shot_dependency" in exc_info.value.message


def test_shot_director_step_blocks_on_failed_validator(project_episode):
    pid, eid, base = project_episode
    # scene_director 도 필요 (shot_director 가 먼저 검사)
    _write_cp(base, pid, eid, "scene_director", {
        "data": {"scenes": [{"scene_index": 1, "present_entity_ids": []}]},
    })
    _write_cp(base, pid, eid, "scene_save", {
        "data": {"segments": [{"scene_index": 1, "text": ""}]},
    })
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    from app.core.steps.shot_director_step import ShotDirectorStep
    step = _make_consumer_step(ShotDirectorStep, pid, eid)

    with pytest.raises(AppError) as exc_info:
        step._execute()
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert "shot_director" in exc_info.value.message


def test_shot_dependency_t2i_step_blocks_on_failed_validator(project_episode):
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    from app.core.steps.shot_dependency_t2i_step import ShotDependencyT2iStep
    step = _make_consumer_step(ShotDependencyT2iStep, pid, eid)

    with pytest.raises(AppError) as exc_info:
        step._execute()
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert "shot_dependency_t2i" in exc_info.value.message


def test_scene_context_loader_blocks_on_failed_validator(project_episode):
    """SceneContextLoader 의 _load_shot_scenes_map 이 failed scene 검출 시 block."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    from app.core.steps.scene_context_loader import SceneContextLoader
    runner = MagicMock()
    runner.project_id = pid
    runner.episode_id = eid
    runner.project_config = {}

    def _load_cp_helper(step_id):
        cp_path = (
            base / pid / "checkpoints" / "episodes" / eid / step_id / "manifest.json"
        )
        if cp_path.exists():
            return json.loads(cp_path.read_text(encoding="utf-8"))
        return None

    runner._load_prev_checkpoint = _load_cp_helper
    loader = SceneContextLoader(runner)

    with pytest.raises(AppError) as exc_info:
        loader._load_shot_scenes_map({})
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert "scene_context_loader" in exc_info.value.message


def test_entity_all_character_step_blocks_on_failed_validator(project_episode):
    """Codex iter 3 I2 — entity_steps.EntityAllCharacterStep 가 failed scene 차단."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    _write_cp(base, pid, eid, "visual_world_rules", {"data": {}})
    from app.core.steps.entity_steps import EntityAllCharacterStep
    step = _make_consumer_step(EntityAllCharacterStep, pid, eid)

    with pytest.raises(AppError) as exc_info:
        step._execute()
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert "entity_all_character" in exc_info.value.message


def test_scene_camera_flow_step_blocks_on_failed_validator(project_episode):
    """Codex iter 3 I2 — SceneCameraFlowStep 가 failed scene 차단 (description grouping)."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "scene_save", {
        "data": {"segments": [{"scene_index": 1, "text": "x"}]},
    })
    _write_cp(base, pid, eid, "shot_selection", {
        "data": {"scenes": [{"scene_index": 1, "selected_shot_indices": [1]}]},
    })
    _write_cp(base, pid, eid, "scene_director", {
        "data": {"scenes": [{"scene_index": 1, "primary_location": "loc"}]},
    })
    _write_cp(base, pid, eid, "entity_merge", {"data": {"characters": []}})
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    from app.core.steps.scene_camera_flow_step import SceneCameraFlowStep
    step = _make_consumer_step(SceneCameraFlowStep, pid, eid)

    with pytest.raises(AppError) as exc_info:
        step._execute()
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert "scene_camera_flow" in exc_info.value.message


def test_shot_staging_step_blocks_on_failed_validator(project_episode):
    """Codex iter 3 I2 — ShotStagingStep 가 failed scene 차단 (shot_extract_data 통째 전달)."""
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_selection", {
        "data": {"scenes": [{"scene_index": 1, "selected_shot_indices": [1]}]},
    })
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    from app.core.steps.shot_staging_step import ShotStagingStep
    step = _make_consumer_step(ShotStagingStep, pid, eid)

    with pytest.raises(AppError) as exc_info:
        step._execute()
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert "shot_staging" in exc_info.value.message


def test_background_master_plan_step_blocks_on_failed_validator(project_episode, monkeypatch):
    """Codex iter 3 I2 — BackgroundMasterPlanStep 가 failed scene 차단 (description 추출).

    background_mode=on 환경 강제 (default off skip 회피).
    """
    from app.core import config as _cfg
    monkeypatch.setattr(_cfg.settings, "background_mode", "on")

    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "background_classify", {"data": {"building_groups": []}})
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [{"shot_index": 1, "description": "raw"}]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    from app.core.steps.background_master_plan_step import BackgroundMasterPlanStep
    step = _make_consumer_step(BackgroundMasterPlanStep, pid, eid)

    with pytest.raises(AppError) as exc_info:
        step._execute()
    assert exc_info.value.code == "step.upstream_validator_failed"
    assert "background_master_plan" in exc_info.value.message


def test_shot_selection_step_respects_explicit_operator_override(project_episode):
    pid, eid, base = project_episode
    _write_cp(base, pid, eid, "shot_validator", {
        "data": {"scenes": [
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "raw", "based_on_beat": 1, "characters": []},
            ]},
            {"scene_index": 2, "validator_status": "failed", "shots": [
                {"shot_index": 1, "description": "raw", "validator_status": "failed_carry_original"},
            ]},
        ]},
    })
    from app.core.steps.shot_selection_step import ShotSelectionStep
    step = _make_consumer_step(ShotSelectionStep, pid, eid)
    step.project_config = {"allow_failed_validator": True}

    # explicit operator override — downstream may still surface other errors (e.g. LLM not mocked).
    # We only assert that it does NOT raise step.upstream_validator_failed.
    try:
        step._execute()
    except AppError as e:
        assert e.code != "step.upstream_validator_failed", (
            f"Expected operator override path but got upstream_validator_failed: {e.message}"
        )
    except Exception:
        # 다른 downstream 에러는 무관 — guard 가 explicit override 인식했음만 검증.
        pass