"""Group 2 #1 (visual_pipeline_contracts_plan) — scene_consistency status 회귀 가드.

scene_consistency 결과에 명시적 ``status`` 필드를 부여한다 (코드 후처리). LLM
출력에는 없는 코드 부여 필드:
- ``ok``: LLM 정상
- ``skipped_single_shot``: 1 샷만 선택 (clean)
- ``skipped_no_selection``: selected_shot_indices=[] (명시적 deselect, clean)
- ``blocked_no_selection``: shot_selection cp 자체 누락 (partial — silent skip 차단)
- ``failed_all_tiers``: 3-tier 모두 fail (partial)

verify_completion 이 status 기반으로 partial/clean 분류 + 옛 cp (status 없음)
backward-compat fallback 으로 ``analysis_summary`` prefix 매칭.

시나리오 의존성 0 — 모든 fixture 가 placeholder ID 만 사용.
"""
from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Dict, List
from unittest.mock import patch

import pytest


# ─────────────────────────────────────────────
# 공통 fixture: scene_consistency cp + 의존 cp 작성 helper
# ─────────────────────────────────────────────

def _write_cp(tmp_path: Path, project_id: str, episode_id: str, step_id: str, manifest: dict) -> None:
    cp_dir = tmp_path / project_id / "checkpoints" / "episodes" / episode_id / step_id
    cp_dir.mkdir(parents=True, exist_ok=True)
    (cp_dir / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")


def _make_step(tmp_path, monkeypatch, last_result=None, project_config=None):
    """SceneConsistencyStep instance — _execute 또는 verify_completion 모두 호출 가능."""
    from app.core.steps.scene_consistency_step import SceneConsistencyStep

    step = SceneConsistencyStep.__new__(SceneConsistencyStep)
    step.project_id = "P1"
    step.episode_id = "E1"
    step.step_id = "scene_consistency"
    step.project_config = project_config or {}
    step.opik_context = {}  # build_opik_metadata 가 dict 접근 — None 회피.
    if last_result is not None:
        step._last_execute_result = last_result

    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))
    return step


# ─────────────────────────────────────────────
# verify_completion: 신 status 기반 분류
# ─────────────────────────────────────────────

def test_verify_clean_for_ok_and_skipped_status(tmp_path, monkeypatch):
    """ok + skipped_single_shot → clean."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok",
                 "fixed_elements": [{"id": "e"}], "status": "ok"},
                {"scene_index": 2, "analysis_summary": "단일 샷 — 교차 일관성 분석 불필요",
                 "fixed_elements": [], "status": "skipped_single_shot"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", cp_data)
    step = _make_step(tmp_path, monkeypatch)
    report = step.verify_completion()
    assert report.is_complete is True
    assert report.severity == "clean"
    assert report.metadata["failed_scenes"] == []
    assert report.metadata["blocked_scenes"] == []


def test_verify_partial_when_failed_all_tiers_status(tmp_path, monkeypatch):
    """status='failed_all_tiers' 가 있으면 partial."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok",
                 "fixed_elements": [], "status": "ok"},
                {"scene_index": 2, "analysis_summary": "분석 실패",
                 "fixed_elements": [], "status": "failed_all_tiers"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", cp_data)
    step = _make_step(tmp_path, monkeypatch)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "partial"
    assert report.metadata["failed_scenes"] == [2]
    assert report.metadata["blocked_scenes"] == []


def test_verify_partial_when_blocked_no_selection_status(tmp_path, monkeypatch):
    """status='blocked_no_selection' 가 있으면 partial — G2.1 핵심 회귀 가드."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok",
                 "fixed_elements": [{"id": "e"}], "status": "ok"},
                {"scene_index": 5, "analysis_summary": "분석 차단 — shot_selection 누락",
                 "fixed_elements": [], "status": "blocked_no_selection"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", cp_data)
    step = _make_step(tmp_path, monkeypatch)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "partial"
    assert report.metadata["failed_scenes"] == []
    assert report.metadata["blocked_scenes"] == [5]
    # 메시지에 사유 명시.
    assert any("shot_selection" in m for m in report.missing)


def test_verify_partial_with_both_failed_and_blocked(tmp_path, monkeypatch):
    """failed + blocked 동시 → 둘 다 metadata 에 노출."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok",
                 "fixed_elements": [], "status": "ok"},
                {"scene_index": 2, "analysis_summary": "분석 실패",
                 "fixed_elements": [], "status": "failed_all_tiers"},
                {"scene_index": 3, "analysis_summary": "분석 차단 — shot_selection 누락",
                 "fixed_elements": [], "status": "blocked_no_selection"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", cp_data)
    step = _make_step(tmp_path, monkeypatch)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "partial"
    assert report.metadata["failed_scenes"] == [2]
    assert report.metadata["blocked_scenes"] == [3]
    assert len(report.missing) == 2  # failed line + blocked line


def test_verify_metadata_keys_consistent_across_branches(tmp_path, monkeypatch):
    """missing/clean/partial 모든 분기에서 metadata key 동일 (consumer 보호)."""
    # missing (no cp, no last_result)
    step1 = _make_step(tmp_path, monkeypatch)
    r1 = step1.verify_completion()
    assert "total_scenes" in r1.metadata
    assert "failed_scenes" in r1.metadata
    assert "blocked_scenes" in r1.metadata

    # clean
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok",
                 "fixed_elements": [{"id": "e"}], "status": "ok"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", cp_data)
    step2 = _make_step(tmp_path, monkeypatch)
    r2 = step2.verify_completion()
    assert "blocked_scenes" in r2.metadata
    assert r2.metadata["blocked_scenes"] == []


# ─────────────────────────────────────────────
# verify_completion: backward-compat (status 없는 옛 cp)
# ─────────────────────────────────────────────

def test_verify_backward_compat_failure_prefix(tmp_path, monkeypatch):
    """옛 cp (status 필드 없음) — analysis_summary='분석 실패' prefix → failed."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok", "fixed_elements": []},
                {"scene_index": 2, "analysis_summary": "분석 실패", "fixed_elements": []},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", cp_data)
    step = _make_step(tmp_path, monkeypatch)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "partial"
    assert 2 in report.metadata["failed_scenes"]


def test_verify_backward_compat_blocked_prefix(tmp_path, monkeypatch):
    """옛 cp (status 필드 없음) — analysis_summary='분석 차단' prefix → blocked.

    G2.1 fix 후 새 cp 는 status='blocked_no_selection' 을 직접 부여하지만,
    G2.1 commit 이전 force-rerun 등으로 만든 옛 cp 는 prefix 만 갖는다 → fallback.
    """
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok", "fixed_elements": []},
                {"scene_index": 5, "analysis_summary": "분석 차단 — shot_selection 누락",
                 "fixed_elements": []},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", cp_data)
    step = _make_step(tmp_path, monkeypatch)
    report = step.verify_completion()
    assert report.is_complete is False
    assert report.severity == "partial"
    assert 5 in report.metadata["blocked_scenes"]


def test_verify_silent_fallback_caught_via_status(tmp_path, monkeypatch):
    """모두 failed_all_tiers → 옛 silent swallow 패턴 차단 (G1.3 회귀 가드 + G2.1 status)."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "분석 실패",
                 "fixed_elements": [], "status": "failed_all_tiers"},
                {"scene_index": 2, "analysis_summary": "분석 실패",
                 "fixed_elements": [], "status": "failed_all_tiers"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", cp_data)
    step = _make_step(tmp_path, monkeypatch)
    report = step.verify_completion()
    assert report.is_complete is False
    # 모두 fail 도 partial (clean 안 됨 — silent OK 차단).
    assert report.severity == "partial"


# ─────────────────────────────────────────────
# _execute: status 부여 (blocked / single_shot / ok / failed)
# ─────────────────────────────────────────────

def _setup_min_dependencies(tmp_path, *, scene_save_segs, shot_extract_scenes, shot_selection_scenes):
    """_execute 가 필요로 하는 최소 cp 조합 작성 — 생성 항목만 진실 매칭."""
    pid, eid = "P1", "E1"
    _write_cp(tmp_path, pid, eid, "scene_save", {"data": {"segments": scene_save_segs}})
    _write_cp(tmp_path, pid, eid, "shot_validator", {"data": {"scenes": shot_extract_scenes}})
    _write_cp(tmp_path, pid, eid, "shot_selection", {"data": {"scenes": shot_selection_scenes}})


def test_execute_assigns_blocked_status_when_shot_selection_missing(tmp_path, monkeypatch):
    """shot_selection cp 에 일부 씬만 있을 때, 누락 씬은 status='blocked_no_selection'."""
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[
            {"scene_index": 1, "text": "scene 1"},
            {"scene_index": 2, "text": "scene 2"},
        ],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s1.1"},
                {"shot_index": 2, "description": "s1.2"},
            ]},
            {"scene_index": 2, "shots": [
                {"shot_index": 1, "description": "s2.1"},
                {"shot_index": 2, "description": "s2.2"},
            ]},
        ],
        # scene 2 만 selection 에 있음 → scene 1 은 blocked.
        shot_selection_scenes=[
            {"scene_index": 2, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    # call_structured mock — scene 2 LLM 만 호출됨.
    def fake_call_structured(*args, **kwargs):
        si = kwargs.get("opik_metadata", {}).get("scene_index", 0)
        return {
            "scene_index": si,
            "analysis_summary": f"scene {si} ok",
            "fixed_elements": [{"element_id": "e", "element_type": "character_state",
                                "character_name": "n", "description": "d", "applies_to_shots": [1, 2]}],
        }

    with patch("app.core.steps.scene_consistency_step.call_structured", side_effect=fake_call_structured):
        result = step._execute(mode="force")

    scenes = result["data"]["scenes"]
    by_idx = {s["scene_index"]: s for s in scenes}
    assert by_idx[1]["status"] == "blocked_no_selection"
    assert by_idx[1]["fixed_elements"] == []
    assert "shot_selection" in by_idx[1]["analysis_summary"]
    assert by_idx[2]["status"] == "ok"
    assert result["blocked_count"] == 1
    # blocked 는 정직하게 failed_count 에 합산 (silent skip 차단).
    assert result["failed_count"] == 1


def test_execute_assigns_skipped_status_for_single_shot(tmp_path, monkeypatch):
    """선택 샷이 1개 → status='skipped_single_shot' (정상, clean)."""
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "scene 1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s1.1"},
                {"shot_index": 2, "description": "s1.2"},
            ]},
        ],
        # 1 개만 선택 → single_shot 분기.
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    with patch("app.core.steps.scene_consistency_step.call_structured") as mock_call:
        result = step._execute(mode="force")
        # single shot 은 LLM 호출 안 함.
        mock_call.assert_not_called()

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    assert by_idx[1]["status"] == "skipped_single_shot"
    assert by_idx[1]["fixed_elements"] == []


def test_execute_assigns_failed_status_when_all_tiers_fail(tmp_path, monkeypatch):
    """call_structured 가 모두 실패 → status='failed_all_tiers'."""
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "scene 1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s1.1"},
                {"shot_index": 2, "description": "s1.2"},
            ]},
        ],
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    def raise_always(*args, **kwargs):
        raise RuntimeError("3-tier fallback exhausted")

    with patch("app.core.steps.scene_consistency_step.call_structured", side_effect=raise_always):
        result = step._execute(mode="force")

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    assert by_idx[1]["status"] == "failed_all_tiers"
    assert by_idx[1]["analysis_summary"] == "분석 실패"
    assert result["failed_count"] >= 1


def test_execute_assigns_ok_status_for_normal_llm_result(tmp_path, monkeypatch):
    """정상 LLM 결과 → status='ok' 자동 부여 (LLM 출력엔 없음)."""
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "scene 1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s1.1"},
                {"shot_index": 2, "description": "s1.2"},
            ]},
        ],
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    # LLM 응답에는 status 가 없음 — 코드가 부여해야 함.
    fake_response = {
        "scene_index": 1,
        "analysis_summary": "common state",
        "fixed_elements": [{"element_id": "e", "element_type": "character_state",
                            "character_name": "n", "description": "d", "applies_to_shots": [1, 2]}],
    }
    with patch("app.core.steps.scene_consistency_step.call_structured", return_value=fake_response):
        result = step._execute(mode="force")

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    assert by_idx[1]["status"] == "ok"
    assert by_idx[1]["fixed_elements"]  # LLM 결과 보존


def test_execute_blocked_count_in_return(tmp_path, monkeypatch):
    """blocked_count 가 반환 dict 에 노출 (UI/리포트 추적용)."""
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[
            {"scene_index": 1, "text": "1"},
            {"scene_index": 2, "text": "2"},
            {"scene_index": 3, "text": "3"},
        ],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
            {"scene_index": 2, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
            {"scene_index": 3, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
        ],
        # 1 만 선택, 2/3 누락 → blocked.
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    fake_resp = {
        "scene_index": 1, "analysis_summary": "ok",
        "fixed_elements": [{"element_id": "e", "element_type": "character_state",
                            "character_name": "n", "description": "d", "applies_to_shots": [1, 2]}],
    }
    with patch("app.core.steps.scene_consistency_step.call_structured", return_value=fake_resp):
        result = step._execute(mode="force")

    assert result["blocked_count"] == 2
    assert result["failed_count"] == 2  # blocked 가 합산됨
    # applicable 은 모든 씬 포함.
    assert result["applicable_count"] == 3


# ─────────────────────────────────────────────
# resume: existing_ok 가 status 필드 보존
# ─────────────────────────────────────────────

def test_resume_preserves_ok_in_existing_ok(tmp_path, monkeypatch):
    """resume 시 기존 cp 의 ok status 가 multi-shot 분기에서 그대로 보존됨."""
    # 첫 run 에서 만든 ok 결과 (status 포함).
    existing_cp = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok",
                 "fixed_elements": [{"id": "e"}], "status": "ok"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", existing_cp)
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
        ],
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    with patch("app.core.steps.scene_consistency_step.call_structured") as mock_call:
        result = step._execute(mode="resume")
        # multi-shot ok scene 은 existing_ok 재사용 — LLM 호출 X.
        mock_call.assert_not_called()

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    assert by_idx[1]["status"] == "ok"
    # G3.1: existing_ok reuse 직전 normalize 가 옛 fixture 4 evidence 필드 default
    # 주입 + confidence='legacy' 마킹 (Claude BLOCKING 1 fix). id 본 데이터는 보존.
    assert len(by_idx[1]["fixed_elements"]) == 1
    assert by_idx[1]["fixed_elements"][0]["id"] == "e"
    assert by_idx[1]["fixed_elements"][0]["confidence"] == "legacy"


# ─────────────────────────────────────────────
# Codex BLOCKING/IMPORTANT fix: existing_ok filter status 우선 + branch override
# ─────────────────────────────────────────────

def test_resume_retries_failed_all_tiers_status(tmp_path, monkeypatch):
    """status='failed_all_tiers' scene 은 resume 에서 retry 대상 (existing_ok 진입 X)."""
    # 첫 run 에서 만든 failed 결과 — analysis_summary text drift 시나리오까지 포함.
    existing_cp = {
        "data": {
            "scenes": [
                # 정확히 prefix "분석 실패" 시작 — 옛 cp 와 신 cp 양쪽 catch.
                {"scene_index": 1, "analysis_summary": "분석 실패",
                 "fixed_elements": [], "status": "failed_all_tiers"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", existing_cp)
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
        ],
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    fake_resp = {"scene_index": 1, "analysis_summary": "ok",
                 "fixed_elements": [{"element_id": "e", "element_type": "character_state",
                                     "character_name": "n", "description": "d",
                                     "applies_to_shots": [1, 2]}]}
    with patch("app.core.steps.scene_consistency_step.call_structured", return_value=fake_resp) as mock_call:
        result = step._execute(mode="resume")
        # failed scene 은 existing_ok 진입 X → LLM 재시도.
        mock_call.assert_called_once()

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    assert by_idx[1]["status"] == "ok"


def test_resume_blocked_status_overrides_prior_ok(tmp_path, monkeypatch):
    """Codex IMPORTANT: prior cp 가 ok 더라도 현재 input 이 blocked 면 fresh blocked emit.

    silent fallback 차단 — stale ok 가 현재 blocked 진실을 덮으면 안 됨.
    """
    # 첫 run 에서 정상 처리된 prior ok 결과 (scene 1 ok + scene 2 ok).
    existing_cp = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "stale ok",
                 "fixed_elements": [{"id": "stale"}], "status": "ok"},
                {"scene_index": 2, "analysis_summary": "stale ok 2",
                 "fixed_elements": [{"id": "stale2"}], "status": "ok"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", existing_cp)
    # 현재 shot_selection 에 scene 2 만 있음 → scene 1 은 blocked (cp 누락).
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[
            {"scene_index": 1, "text": "1"},
            {"scene_index": 2, "text": "2"},
        ],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
            {"scene_index": 2, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
        ],
        # scene 2 만 selection 에 있음, scene 1 누락.
        shot_selection_scenes=[
            {"scene_index": 2, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    with patch("app.core.steps.scene_consistency_step.call_structured") as mock_call:
        # scene 2 는 existing_ok 재사용 — LLM 호출 X.
        result = step._execute(mode="resume")
        mock_call.assert_not_called()

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    # scene 1: prior ok 무시, fresh blocked emit.
    assert by_idx[1]["status"] == "blocked_no_selection"
    assert by_idx[1]["fixed_elements"] == []  # stale [{"id":"stale"}] 사라짐.
    # scene 2: prior ok 보존 (existing_ok 재사용 — multi shot 분기).
    assert by_idx[2]["status"] == "ok"
    # G3.1: existing_ok reuse normalize 가 옛 fixture 4 evidence 필드 default
    # 주입 + confidence='legacy' 마킹. id 본 데이터는 보존.
    assert len(by_idx[2]["fixed_elements"]) == 1
    assert by_idx[2]["fixed_elements"][0]["id"] == "stale2"
    assert by_idx[2]["fixed_elements"][0]["confidence"] == "legacy"
    assert result["blocked_count"] == 1


def test_resume_single_shot_overrides_prior_ok(tmp_path, monkeypatch):
    """현재 input 이 single shot 이면 prior multi-shot ok 를 덮음 — 진실 우선."""
    existing_cp = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "prior multi ok",
                 "fixed_elements": [{"id": "stale"}], "status": "ok"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", existing_cp)
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
        ],
        # 현재는 1 샷만 선택 → single_shot 분기.
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    result = step._execute(mode="resume")
    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    # prior ok 무시, fresh skipped_single_shot.
    assert by_idx[1]["status"] == "skipped_single_shot"


def test_resume_status_none_old_cp_preserved_when_summary_clean(tmp_path, monkeypatch):
    """Claude I3: status 없는 옛 cp 의 정상 결과 (summary='ok') 는 existing_ok 보존."""
    # status 필드 없는 옛 G1.3 시대 cp.
    existing_cp = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok",
                 "fixed_elements": [{"id": "preserved"}]},  # status 키 없음
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", existing_cp)
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
        ],
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    with patch("app.core.steps.scene_consistency_step.call_structured") as mock_call:
        result = step._execute(mode="resume")
        # 옛 cp 라도 status=None + summary=ok 는 existing_ok 진입 → LLM 안 호출.
        mock_call.assert_not_called()

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    # 옛 dict 그대로 보존 — status key 강제 추가 안 함.
    # G3.1: existing_ok reuse normalize 가 옛 fixture 4 evidence 필드 default
    # 주입 + confidence='legacy' 마킹. id 본 데이터는 보존.
    assert len(by_idx[1]["fixed_elements"]) == 1
    assert by_idx[1]["fixed_elements"][0]["id"] == "preserved"
    assert by_idx[1]["fixed_elements"][0]["confidence"] == "legacy"
    assert "status" not in by_idx[1]  # 옛 dict 그대로 (status 는 normalize 안 건드림)


def test_resume_status_none_old_cp_with_failure_prefix_retried(tmp_path, monkeypatch):
    """Claude I3: status 없는 옛 cp 의 '분석 실패' prefix 도 retry 대상."""
    existing_cp = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "분석 실패",
                 "fixed_elements": []},  # 옛 cp — status 없음.
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", existing_cp)
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
        ],
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    fake_resp = {"scene_index": 1, "analysis_summary": "now ok",
                 "fixed_elements": [{"element_id": "e", "element_type": "character_state",
                                     "character_name": "n", "description": "d",
                                     "applies_to_shots": [1, 2]}]}
    with patch("app.core.steps.scene_consistency_step.call_structured", return_value=fake_resp) as mock_call:
        result = step._execute(mode="resume")
        # 옛 prefix 매칭 → existing_ok 진입 X → retry.
        mock_call.assert_called_once()

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    assert by_idx[1]["status"] == "ok"


def test_resume_status_none_old_cp_with_blocked_prefix_retried(tmp_path, monkeypatch):
    """Claude I3: status 없는 옛 cp 의 '분석 차단' prefix 도 retry 대상."""
    existing_cp = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "분석 차단 — shot_selection 누락",
                 "fixed_elements": []},  # 옛 cp — status 없음.
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", existing_cp)
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
        ],
        # 현재는 selection 채워져있음 → 정상 처리되어야.
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": [1, 2]},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    fake_resp = {"scene_index": 1, "analysis_summary": "now ok",
                 "fixed_elements": [{"element_id": "e", "element_type": "character_state",
                                     "character_name": "n", "description": "d",
                                     "applies_to_shots": [1, 2]}]}
    with patch("app.core.steps.scene_consistency_step.call_structured", return_value=fake_resp) as mock_call:
        result = step._execute(mode="resume")
        mock_call.assert_called_once()

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    assert by_idx[1]["status"] == "ok"


# ─────────────────────────────────────────────
# Codex MINOR fix: selected_shot_indices=[] (명시적 deselect)
# ─────────────────────────────────────────────

def test_execute_skipped_no_selection_for_explicit_empty(tmp_path, monkeypatch):
    """selected_shot_indices=[] (명시적 deselect) → status='skipped_no_selection' (clean)."""
    _setup_min_dependencies(
        tmp_path,
        scene_save_segs=[{"scene_index": 1, "text": "1"}],
        shot_extract_scenes=[
            {"scene_index": 1, "shots": [
                {"shot_index": 1, "description": "s"}, {"shot_index": 2, "description": "s"},
            ]},
        ],
        # 명시적 빈 선택 — blocked (cp 누락) 와 구분.
        shot_selection_scenes=[
            {"scene_index": 1, "selected_shot_indices": []},
        ],
    )
    step = _make_step(tmp_path, monkeypatch)

    with patch("app.core.steps.scene_consistency_step.call_structured") as mock_call:
        result = step._execute(mode="force")
        mock_call.assert_not_called()

    by_idx = {s["scene_index"]: s for s in result["data"]["scenes"]}
    assert by_idx[1]["status"] == "skipped_no_selection"
    assert by_idx[1]["fixed_elements"] == []
    # blocked 와 다른 분기 — failed_count 에 안 합산 (의도된 deselect 는 정상).
    assert result["blocked_count"] == 0
    assert result["failed_count"] == 0


def test_verify_clean_for_skipped_no_selection_status(tmp_path, monkeypatch):
    """status='skipped_no_selection' → clean (사용자/LLM 의도)."""
    cp_data = {
        "data": {
            "scenes": [
                {"scene_index": 1, "analysis_summary": "ok",
                 "fixed_elements": [{"id": "e"}], "status": "ok"},
                {"scene_index": 2, "analysis_summary": "스킵 — selected_shot_indices=[]",
                 "fixed_elements": [], "status": "skipped_no_selection"},
            ]
        }
    }
    _write_cp(tmp_path, "P1", "E1", "scene_consistency", cp_data)
    step = _make_step(tmp_path, monkeypatch)
    report = step.verify_completion()
    assert report.is_complete is True
    assert report.severity == "clean"


# ─────────────────────────────────────────────
# Codex BLOCKING fix: manifest allow_partial_downstream=False
# ─────────────────────────────────────────────

def test_manifest_blocks_partial_cascade():
    """scene_consistency manifest 가 partial 시 downstream cascade 차단."""
    from app.core.step_manifest import STEP_MANIFEST
    meta = STEP_MANIFEST["scene_consistency"]
    assert meta.get("allow_partial_downstream") is False, (
        "scene_consistency partial 시 scene_detail/shot_dependency_t2i 가 fixed_elements"
        " 를 빈 array 로 silent 처리 회귀 — fail-fast 정책 위반"
    )