"""W21B-W7 — visual_continuity_anchor 핵심 계약 최소 테스트.

사용자 지시(2026-06-11): 결과물(LLM anchor 품질)은 TDD 비대상 — canary + 육안.
여기서는 deterministic 핵심 계약만 검증한다:
  1. C/D seed 조인 (zoom pair / prop 반복 + priority score / cap 진단)
  2. manifest 조인 무결성 validator
  3. flag OFF → applicability False (default 경로 영향 0)
"""
from __future__ import annotations

from app.modules.pipeline.visual_continuity_anchor_plan import (
    ANCHOR_TYPE_ZOOM,
    ANCHOR_TYPE_PROP,
    build_manifest,
    build_zoom_group,
    build_prop_group,
    detect_c_seeds,
    detect_d_seeds,
    select_seeds_with_cap,
    validate_anchor_manifest,
)

_SELECTED = {12: {7, 12}, 25: {4}, 29: {4}, 3: {1}}


def _deps():
    return [
        {  # 정상 C seed: S12 sh12(zoom) ← S12 sh7(source, selected)
            "scene_index": 12, "shot_index": 12,
            "location_refs": [
                {"scene_index": 12, "shot_index": 7, "ref_usage": "zoom_in_detail"},
            ],
        },
        {  # cross-scene zoom → skipped
            "scene_index": 25, "shot_index": 4,
            "location_refs": [
                {"scene_index": 12, "shot_index": 7, "ref_usage": "zoom_in_detail"},
            ],
        },
        {  # same-scene exact_background → diagnostics 후보만 (production 소비 0)
            "scene_index": 3, "shot_index": 1,
            "location_refs": [
                {"scene_index": 3, "shot_index": 2, "ref_usage": "exact_background"},
            ],
        },
        {  # zoom 샷이 selected 아님 → skipped
            "scene_index": 12, "shot_index": 9,
            "location_refs": [
                {"scene_index": 12, "shot_index": 7, "ref_usage": "zoom_in_detail"},
            ],
        },
    ]


def test_detect_c_seeds_zoom_pair_join():
    seeds, skipped, exact = detect_c_seeds(_deps(), _SELECTED)
    assert seeds == [{"zoom": [12, 12], "source": [12, 7]}]
    reasons = {s["reason"] for s in skipped}
    assert "cross_scene_zoom" in reasons
    assert "zoom_not_selected" in reasons
    assert exact == [{"zoom": [3, 1], "source": [3, 2]}]


def test_detect_d_seeds_priority_and_cap():
    ve = {
        (25, 4): ["C09", "P02"],
        (29, 4): ["P02"],
        (12, 7): ["C08", "P02"],
        (3, 1): ["P05"],  # 1샷 등장 → seed 아님
    }
    d_seeds = detect_d_seeds(
        ve, _SELECTED, {"P02", "P05"},
        framing_by_shot={(29, 4): "insert", (25, 4): "medium", (12, 7): "wide"},
        required_prop_refs_by_shot={(29, 4): {"P02"}},
    )
    assert len(d_seeds) == 1
    seed = d_seeds[0]
    assert seed["prop_short_id"] == "P02"
    assert seed["member_shots"] == [[12, 7], [25, 4], [29, 4]]
    # score = required(1)*10 + close/insert(1)*3 + members(3)*1
    assert seed["score"] == 16

    c_seeds = [{"zoom": [12, 12], "source": [12, 7]}]
    # D 예약 슬롯 (Codex W_A_STAGE_REVIEW): cap=1 이어도 D 가 먼저 슬롯 확보
    selected, cap_skipped = select_seeds_with_cap(c_seeds, d_seeds, cap=1)
    assert [t for t, _ in selected] == [ANCHOR_TYPE_PROP]
    assert cap_skipped == [
        {"seed": "zoom:S12sh12", "reason": "cap_after_priority_score"}]
    # cap 여유가 있으면 C + D 둘 다, 초과 D 는 score 기록과 함께 skipped
    selected2, skipped2 = select_seeds_with_cap(
        c_seeds, d_seeds + [{"prop_short_id": "P09", "member_shots": [[3, 1], [12, 7]],
                             "score": 2}], cap=2, d_reserved=1)
    assert [t for t, _ in selected2] == [ANCHOR_TYPE_ZOOM, ANCHOR_TYPE_PROP]
    assert skipped2 == [{"seed": "prop:P09", "reason": "cap_after_priority_score",
                         "score": 2}]


def test_manifest_join_integrity_validator():
    pose = build_zoom_group(
        {"zoom": [12, 12], "source": [12, 7]},
        {"locked_elements": [{"kind": "subject_pose", "description": "d",
                              "evidence_quote": "q", "confidence": "high"}],
         "wide_shot_contract": ["c"]},
        1,
    )
    prop = build_prop_group(
        {"prop_short_id": "P02", "member_shots": [[25, 4], [29, 4]]},
        {"printed_content": "pc", "physical_form": "pf",
         "scale_contract": "sc", "locked_elements": []},
        framing_by_shot={(29, 4): "insert"},
    )
    manifest = build_manifest([pose, prop], {})
    assert validate_anchor_manifest(manifest, _SELECTED, {"P02"}) == []
    # hint 키/role 계약
    assert manifest["groups"][0]["per_shot_consumption_hints"]["12:12"] == {
        "crop_from_source": True}
    assert manifest["groups"][1]["members"][1]["role"] == "close_insert"
    assert manifest["groups"][1]["per_shot_consumption_hints"]["25:4"] == {
        "keep_real_scale": True}

    # 무결성 위반: 미지의 prop short_id + evidence 빈 locked_element
    bad = build_manifest([
        {**prop, "prop_anchor": {**prop["prop_anchor"], "prop_short_id": "P99"},
         "locked_elements": [{"kind": "prop_state", "description": "d",
                              "evidence_quote": " ", "confidence": "low"}]},
    ], {})
    violations = validate_anchor_manifest(bad, _SELECTED, {"P02"})
    assert any("unknown prop_short_id" in v for v in violations)
    assert any("evidence_quote" in v for v in violations)


def test_flag_off_applicability_false(monkeypatch):
    from app.core import applicability as ap
    from app.core.config import settings

    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", False, raising=False)
    fn = ap.APPLICABILITY_VALIDATORS["if_visual_continuity_anchor_enabled"]
    assert fn(None) is False
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", True, raising=False)
    assert fn(None) is True


# ───────────────────── W-B (D 배선) 핵심 계약 ─────────────────────


def test_wb_prop_ref_overlay_and_label_helper():
    from app.core.steps.visual_continuity_anchor_step import (
        apply_keep_real_scale_to_prop_labels,
    )
    from app.modules.pipeline.visual_continuity_anchor_plan import (
        build_prop_ref_prompt_overlay,
    )

    anchor = {"printed_content": "PC", "physical_form": "PF",
              "scale_contract": "SC", "prop_short_id": "P02"}
    overlay = build_prop_ref_prompt_overlay(anchor, "CANON STYLE")
    # anchor 우선 + canon 보존(provenance) 계약
    assert overlay.index("PF") < overlay.index("PC") < overlay.index("SC")
    assert "CANON STYLE" in overlay and "the anchor wins" in overlay
    assert build_prop_ref_prompt_overlay(anchor, "") .count("CANON") == 0

    ctx = {"by_prop": {"P02": anchor},
           "keep_real_scale_by_shot": {(25, 4): {"P02"}}}
    refs = [("object P02 (required)", b"x"), ("char ref", b"y"),
            ("bg same room", b"z"), ("bg chain", b"w")]
    roles = ["prop_ref", "character_ref", "previous_shot_same_room",
             "background_chain_ref"]
    meta = [{"sid": "P02"}, {}, {}, {}]
    # hint 있는 still → prop_ref 라벨에 scale+physical_form+동일 실물 identity,
    # previous-shot bg 의 metadata["ignore"] 에 stale 사본 무시 지시
    # (P02 전수조사 fix — bg role 표기는 고정 문자열이라 ignore 채널 사용)
    out = apply_keep_real_scale_to_prop_labels(
        refs, roles, meta, scene_index=25, shot_index=4, anchor_ctx=ctx)
    assert out[0][0].startswith("object P02 (required) — SC PF ")
    assert "SAME single physical object" in out[0][0]
    assert out[1] == refs[1]
    assert out[2] == refs[2]  # bg 라벨 자체는 불변
    assert "separately-referenced printed/displayed object" in meta[2]["ignore"]
    # negative (Codex 의견 5): ignore 채널 미소비 role 은 mutation 금지
    assert out[3] == refs[3] and meta[3] == {}
    # hint 없는 still / 빈 ctx → byte-identical (+ metadata 비오염)
    meta2 = [{"sid": "P02"}, {}, {}, {}]
    assert apply_keep_real_scale_to_prop_labels(
        refs, roles, meta2, scene_index=12, shot_index=7, anchor_ctx=ctx) == refs
    assert apply_keep_real_scale_to_prop_labels(
        refs, roles, meta2, scene_index=25, shot_index=4, anchor_ctx={}) == refs
    assert meta2[2] == {}


def test_wb_loader_flag_off_returns_empty(monkeypatch, tmp_path):
    from app.core.config import settings
    from app.core.steps.visual_continuity_anchor_step import (
        load_printed_prop_anchor_context,
    )

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    cp_dir = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" / "visual_continuity_anchor"
    cp_dir.mkdir(parents=True)
    import json as _json
    cp_dir.joinpath("manifest.json").write_text(_json.dumps({
        "status": "completed", "schema_version": 1, "config_hash": "h",
        "data": {"groups": [{
            "group_id": "vca-p02-prop", "anchor_type": "printed_prop",
            "members": [{"scene_index": 25, "shot_index": 4, "role": "environment"}],
            "prop_anchor": {"prop_short_id": "P02", "printed_content": "PC",
                            "physical_form": "PF", "scale_contract": "SC"},
            "per_shot_consumption_hints": {"25:4": {"keep_real_scale": True}},
        }]},
    }))
    # flag OFF → 빈 dict (consumer 전원 no-op = byte-identical)
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", False, raising=False)
    assert load_printed_prop_anchor_context("p1", "e1") == {}
    # flag ON → 조인 맵 + stamp
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", True, raising=False)
    ctx = load_printed_prop_anchor_context("p1", "e1")
    assert ctx["by_prop"]["P02"]["scale_contract"] == "SC"
    assert ctx["members_by_shot"] == {(25, 4): ["P02"]}
    assert ctx["keep_real_scale_by_shot"] == {(25, 4): {"P02"}}
    assert ctx["stamp"]["group_count"] == 1 and ctx["stamp"]["digest"]


def test_prop_anchor_carries_printed_content_gate_overlay():
    """non-printed prop (carries_printed_content=False) → ref overlay 는 canon 만.

    표식(painted mark) 같은 prop 이 사진/문서 anchor 를 ref 에 덮어쓰지 못하게
    하는 generic 게이트 (06-19 conflation fix). 누락 필드는 backward-compat 로 active."""
    from app.modules.pipeline.visual_continuity_anchor_plan import (
        build_prop_ref_prompt_overlay,
    )
    active = {"prop_short_id": "P02", "carries_printed_content": True,
              "printed_content": "PC", "physical_form": "PF", "scale_contract": "SC"}
    overlay = build_prop_ref_prompt_overlay(active, "CANON")
    assert "PF" in overlay and "PC" in overlay and "the anchor wins" in overlay

    # 동일 anchor content 라도 carries_printed_content=False → canon 만 (anchor overlay 0)
    inactive = dict(active, prop_short_id="P03", carries_printed_content=False)
    out = build_prop_ref_prompt_overlay(inactive, "CANON RING")
    assert out == "CANON RING"
    assert "PF" not in out and "PC" not in out and "anchor wins" not in out

    # 누락 필드 (legacy) → active (byte-identical 보존)
    legacy = {"prop_short_id": "P02", "printed_content": "PC",
              "physical_form": "PF", "scale_contract": "SC"}
    assert "PF" in build_prop_ref_prompt_overlay(legacy, "CANON")


def test_loader_excludes_non_printed_prop(monkeypatch, tmp_path):
    """load_printed_prop_anchor_context: carries_printed_content=False 그룹 제외,
    True/누락 그룹 유지 (06-19 conflation fix 소비 chokepoint 게이트)."""
    from app.core.config import settings
    from app.core.steps.visual_continuity_anchor_step import (
        load_printed_prop_anchor_context,
    )
    import json as _json

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", True, raising=False)
    cp_dir = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" / "visual_continuity_anchor"
    cp_dir.mkdir(parents=True)
    cp_dir.joinpath("manifest.json").write_text(_json.dumps({
        "status": "completed", "schema_version": 1, "config_hash": "h",
        "data": {"groups": [
            {  # printed photo prop — 유지
                "group_id": "vca-p02-prop", "anchor_type": "printed_prop",
                "members": [{"scene_index": 25, "shot_index": 4}],
                "prop_anchor": {"prop_short_id": "P02", "carries_printed_content": True,
                                "printed_content": "PC", "physical_form": "PF",
                                "scale_contract": "SC"},
            },
            {  # painted mark — carries_printed_content False → 제외
                "group_id": "vca-p03-prop", "anchor_type": "printed_prop",
                "members": [{"scene_index": 29, "shot_index": 7}],
                "prop_anchor": {"prop_short_id": "P03", "carries_printed_content": False,
                                "printed_content": "", "physical_form": "",
                                "scale_contract": ""},
            },
        ]},
    }))
    ctx = load_printed_prop_anchor_context("p1", "e1")
    assert "P02" in ctx["by_prop"]          # printed → 유지
    assert "P03" not in ctx["by_prop"]      # non-printed mark → 게이트 제외
    assert ctx["members_by_shot"] == {(25, 4): ["P02"]}


def test_build_prop_group_carries_field_flows_to_loader(monkeypatch, tmp_path):
    """REAL 경로 lock (Codex BLOCKING 1): provider result → build_prop_group →
    build_manifest → checkpoint → load_printed_prop_anchor_context. fixture 직접
    주입이 아니라 build_prop_group 이 carries_printed_content 를 실제로 보존해야
    loader 게이트가 fresh cp 에서 동작한다."""
    import json as _json
    from app.core.config import settings
    from app.core.steps.visual_continuity_anchor_step import (
        load_printed_prop_anchor_context,
    )
    from app.modules.pipeline.visual_continuity_anchor_plan import (
        build_manifest, build_prop_group,
    )

    g_false = build_prop_group(
        {"prop_short_id": "P03", "member_shots": [[29, 7]]},
        {"carries_printed_content": False, "applicability_evidence": "painted ritual mark",
         "printed_content": "", "physical_form": "", "scale_contract": "", "locked_elements": []})
    g_true = build_prop_group(
        {"prop_short_id": "P02", "member_shots": [[25, 8]]},
        {"carries_printed_content": True, "applicability_evidence": "photograph",
         "printed_content": "PC", "physical_form": "PF", "scale_contract": "SC", "locked_elements": []})
    g_legacy = build_prop_group(
        {"prop_short_id": "P07", "member_shots": [[23, 3]]},
        {"printed_content": "MAP", "physical_form": "PF", "scale_contract": "SC", "locked_elements": []})
    # build_prop_group 이 필드 보존 (BLOCKING 1 — 이게 빠지면 manifest 에 누락)
    assert g_false["prop_anchor"]["carries_printed_content"] is False
    assert g_false["prop_anchor"]["applicability_evidence"] == "painted ritual mark"
    assert g_true["prop_anchor"]["carries_printed_content"] is True
    assert g_legacy["prop_anchor"]["carries_printed_content"] is True  # 누락 → active default

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", True, raising=False)
    cp_dir = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" / "visual_continuity_anchor"
    cp_dir.mkdir(parents=True)
    cp = {"status": "completed", "config_hash": "h",
          "data": build_manifest([g_false, g_true, g_legacy], {})}
    cp_dir.joinpath("manifest.json").write_text(_json.dumps(cp))

    ctx = load_printed_prop_anchor_context("p1", "e1")
    assert "P03" not in ctx["by_prop"]   # false → 실제 경로로 제외
    assert "P02" in ctx["by_prop"]       # true → 유지
    assert "P07" in ctx["by_prop"]       # legacy 누락 → active 유지 (회귀 없음)


def test_zoom_fill_prop_refs_anchor_contract(monkeypatch, tmp_path):
    """S12sh12 fix — crop/i2i-fill 의 object ref 라벨에 printed_prop anchor 계약.

    anchored member still → 라벨에 printed_content+physical_form+scale_contract
    +동일 실물 identity, anchored prop 이 cap 절단보다 우선. anchor 부재(다른
    still / flag OFF) → 기존 라벨 byte-identical.
    """
    import json as _json
    from app.core.config import settings
    from app.core.steps.visual_continuity_anchor_step import build_zoom_fill_prop_refs

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    cp_dir = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" / "visual_continuity_anchor"
    cp_dir.mkdir(parents=True)
    cp_dir.joinpath("manifest.json").write_text(_json.dumps({
        "status": "completed", "schema_version": 1, "config_hash": "h",
        "data": {"groups": [{
            "group_id": "vca-p02-prop", "anchor_type": "printed_prop",
            "members": [{"scene_index": 12, "shot_index": 12, "role": "environment"}],
            "prop_anchor": {"prop_short_id": "P02", "printed_content": "PC",
                            "physical_form": "PF", "scale_contract": "SC"},
            "per_shot_consumption_hints": {"12:12": {"keep_real_scale": True}},
        }]},
    }))
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", True, raising=False)

    # anchored P02 가 리스트 뒤에 있어도 cap=1 절단에서 살아남는다 (우선 정렬)
    ves = [
        {"id": "e-char", "short_id": "C08", "name": "char", "entity_type": "character"},
        {"id": "e-other", "short_id": "P09", "name": "other prop", "entity_type": "prop"},
        {"id": "e-p02", "short_id": "P02", "name": "photo", "entity_type": "prop"},
    ]
    ref_map = {"e-other": b"o", "e-p02": b"p"}
    out, has_contract = build_zoom_fill_prop_refs(
        "p1", "e1", ves, ref_map, scene_index=12, shot_index=12)
    assert has_contract is True
    assert len(out) == 1 and out[0][1] == b"p"
    label = out[0][0]
    assert label.startswith("Reference image 2 (object reference): photo — ")
    assert "PC" in label and "PF" in label and "SC" in label
    assert "SAME single physical object" in label

    # anchor 멤버가 아닌 still → 기존 라벨 그대로 (첫 prop, 계약 0)
    # + has_contract False = fill 프롬프트도 byte-identical (Codex narrow)
    out2, has_contract2 = build_zoom_fill_prop_refs(
        "p1", "e1", ves, ref_map, scene_index=25, shot_index=7)
    assert out2 == [("Reference image 2 (object reference): other prop", b"o")]
    assert has_contract2 is False

    # flag OFF → 동일하게 기존 라벨 byte-identical + 계약 없음
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", False, raising=False)
    assert build_zoom_fill_prop_refs(
        "p1", "e1", ves, ref_map, scene_index=12, shot_index=12) == (
        [("Reference image 2 (object reference): other prop", b"o")], False)


def test_zoom_fill_prompt_object_ref_clause():
    """S12sh12 fix — content-SOT 절은 anchor 계약 ref 가 있을 때만 붙는다.

    일반 prop ref 만 있는 still (has_object_content_sot_ref=False) 의 fill
    프롬프트는 기존과 byte-identical (Codex S12_FILL_PROP_REVIEW narrow).
    """
    from app.services.zoom_continuity_render_service import build_fill_prompt

    zt = {"locked_elements": [{"kind": "held_prop", "description": "LOCKED-FACT"}]}
    base = build_fill_prompt(zt, has_object_content_sot_ref=False)
    withref = build_fill_prompt(zt, has_object_content_sot_ref=True)
    assert "LOCKED-FACT" in base and "source of truth" not in base
    assert withref.startswith(base)
    assert "printed or displayed content" in withref
    assert "source of truth" in withref


# ───────────────────── W-C1 핵심 계약 ─────────────────────


def test_wc1_revised_prompt_token_audit():
    from app.modules.pipeline.visual_continuity_anchor_plan import (
        revised_prompt_new_tokens,
    )
    # 새 entity ID 토큰 도입 → 위반 집합 반환 (revised 폐기 대상)
    assert revised_prompt_new_tokens(
        "wide shot of C08 slumped", "wide shot of C08 holding P02 photo") == {"P02"}
    # 기존 토큰 유지/무토큰 text-only 계약 → 통과
    assert revised_prompt_new_tokens(
        "wide shot of C08", "wide shot of C08, a crumpled photo in her hand") == set()
    assert revised_prompt_new_tokens("no tokens", "still no tokens") == set()


def test_wc1_zoom_anchor_flag_off_applicability():
    from app.core import applicability as ap
    from app.core.config import settings
    fn = ap.APPLICABILITY_VALIDATORS["if_zoom_continuity_anchor_enabled"]
    import pytest  # noqa: F401
    orig = getattr(settings, "zoom_continuity_anchor_enabled", False)
    try:
        settings.zoom_continuity_anchor_enabled = False
        assert fn(None) is False
        settings.zoom_continuity_anchor_enabled = True
        assert fn(None) is True
    finally:
        settings.zoom_continuity_anchor_enabled = orig


# ───────────────────── W-C2 핵심 계약 ─────────────────────


def test_wc2_zoom_context_loader_and_flag_off(monkeypatch, tmp_path):
    import json as _json
    from app.core.config import settings
    from app.core.steps.zoom_continuity_anchor_step import load_zoom_continuity_context

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    cp_dir = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" / "zoom_continuity_anchor"
    cp_dir.mkdir(parents=True)
    cp_dir.joinpath("manifest.json").write_text(_json.dumps({
        "status": "completed", "schema_version": 1,
        "data": {"groups": [{
            "group_id": "vca-s12-zoom-1", "anchor_type": "zoom_continuity",
            "members": [
                {"scene_index": 12, "shot_index": 7, "role": "source_wide"},
                {"scene_index": 12, "shot_index": 12, "role": "zoom"},
            ],
            "locked_elements": [], "wide_shot_contract": ["c"],
            "revised_wide_t2i": {"0": "REVISED"},
            "revised_wide_prompt_provenance": {"0": {"revised_prompt_hash": "h"}},
        }]},
    }))
    monkeypatch.setattr(settings, "zoom_continuity_anchor_enabled", False, raising=False)
    assert load_zoom_continuity_context("p1", "e1") == {}  # flag OFF → no-op
    monkeypatch.setattr(settings, "zoom_continuity_anchor_enabled", True, raising=False)
    ctx = load_zoom_continuity_context("p1", "e1")
    assert ctx["source_overrides"][(12, 7)]["revised"] == {0: "REVISED"}
    assert ctx["zoom_targets"][(12, 12)]["source"] == (12, 7)


def test_wc2_crop_expand_and_viability():
    from app.services.zoom_continuity_render_service import (
        assess_viability,
        expand_crop_16x9,
    )
    # 16:9 확장 + 클램프 (프레임 안)
    box = expand_crop_16x9((1600, 900), [0.4, 0.4, 0.6, 0.6])
    x0, y0, x1, y1 = box
    assert 0 <= x0 < x1 <= 1600 and 0 <= y0 < y1 <= 900
    assert abs((x1 - x0) / (y1 - y0) - 16 / 9) < 0.02
    # viability — 기록만(차단 아님)의 verdict 계약
    assert assess_viability({"found": False})["verdict"] == "bbox_not_found"
    assert assess_viability({"found": True, "bbox": [0.0, 0.0, 0.9, 0.9]})["verdict"] == "bbox_too_large_weak_zoom"
    assert assess_viability({"found": True, "bbox": [0.4, 0.4, 0.6, 0.6]})["verdict"] == "ok"


def test_wc2b_zoom_dep_edge_orders_source_before_zoom():
    """W-C2b 핵심 계약: zoom→source edge 가 위상 정렬에서 source 를 앞 batch 로."""
    from app.modules.entity_dependency import topological_sort_scenes

    # stills: [0]=sh12(zoom), [1]=sh7(source) — 의도적으로 zoom 이 앞 인덱스
    deps = {0: set(), 1: set()}
    batches_before = topological_sort_scenes(2, deps)
    assert batches_before == [[0, 1]]  # edge 없으면 같은 batch (병렬 race)
    deps[0].add(1)  # zoom(0) → source(1)
    batches_after = topological_sort_scenes(2, deps)
    assert batches_after == [[1], [0]]  # source 가 앞 batch = barrier 뒤에 zoom


# ───────────────────── S29 fix + Cinematography 축 핵심 계약 ─────────────────────


def test_s29_zoom_subject_char_token_join():
    """zoom 인물(C) 토큰 base(의상 O suffix 제거)가 source 프롬프트에 없으면 gap."""
    from app.modules.pipeline.visual_continuity_anchor_plan import (
        char_token_bases,
        zoom_subject_missing_chars,
    )

    assert char_token_bases("C08O06 near [L19: room] with P02") == {"C08"}
    assert char_token_bases("no tokens at all") == set()
    # S29 실측 형상: 무인물 wide ← 인물 클로즈업 zoom
    assert zoom_subject_missing_chars(
        ["C24O15 close-up beside C09"], "wide view of [L19: wheelhouse] walls",
    ) == {"C09", "C24"}
    # 의상 suffix 가 달라도 같은 인물 base 면 충족
    assert zoom_subject_missing_chars(["C08 slack hand"], "wide with C08O06") == set()
    # prop-only zoom (인물 토큰 0) 은 어떤 source 와도 무모순
    assert zoom_subject_missing_chars(["P02 photograph insert"], "empty room") == set()


def test_camera_relation_gates_crop_and_zoom_override(monkeypatch, tmp_path):
    """Cinematography 축: punch-in 만 crop(zoom_targets), different 는
    revised_zoom 이 prompt override 로 — camera_relation 부재(legacy)는 punch-in."""
    import json as _json
    from app.core.config import settings
    from app.core.steps.zoom_continuity_anchor_step import load_zoom_continuity_context
    from app.modules.pipeline.visual_continuity_anchor_plan import build_zoom_group

    # build_zoom_group: enum → crop hint
    g_diff = build_zoom_group(
        {"source": [7, 2], "zoom": [7, 5]},
        {"camera_relation": "different_camera_same_moment",
         "camera_relation_evidence": "frontal vs profile"},
        1,
    )
    assert g_diff["camera_relation"] == "different_camera_same_moment"
    assert g_diff["per_shot_consumption_hints"]["7:5"]["crop_from_source"] is False
    g_legacy = build_zoom_group({"source": [7, 2], "zoom": [7, 5]}, {}, 1)
    assert g_legacy["per_shot_consumption_hints"]["7:5"]["crop_from_source"] is True

    # loader: different → zoom_targets 제외 + revised_zoom 은 override 맵으로
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "zoom_continuity_anchor_enabled", True, raising=False)
    cp_dir = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" / "zoom_continuity_anchor"
    cp_dir.mkdir(parents=True)
    cp_dir.joinpath("manifest.json").write_text(_json.dumps({
        "status": "completed", "schema_version": 1,
        "data": {"groups": [
            {
                "group_id": "vca-s7-zoom-1", "anchor_type": "zoom_continuity",
                "camera_relation": "different_camera_same_moment",
                "members": [
                    {"scene_index": 7, "shot_index": 2, "role": "source_wide"},
                    {"scene_index": 7, "shot_index": 5, "role": "zoom"},
                ],
                "revised_wide_t2i": {"0": "WIDE-REV"},
                "revised_zoom_t2i": {"0": "ZOOM-REV"},
                "revised_zoom_prompt_provenance": {"0": {"revised_prompt_hash": "zh"}},
            },
            {
                "group_id": "vca-s12-zoom-2", "anchor_type": "zoom_continuity",
                "camera_relation": "same_axis_punch_in",
                "members": [
                    {"scene_index": 12, "shot_index": 7, "role": "source_wide"},
                    {"scene_index": 12, "shot_index": 12, "role": "zoom"},
                ],
                "revised_wide_t2i": {"0": "W2"},
            },
        ]},
    }))
    ctx = load_zoom_continuity_context("p1", "e1")
    assert (7, 5) not in ctx["zoom_targets"]          # different → crop 안 함
    assert ctx["source_overrides"][(7, 5)]["revised"] == {0: "ZOOM-REV"}
    assert ctx["source_overrides"][(7, 2)]["revised"] == {0: "WIDE-REV"}
    assert ctx["zoom_targets"][(12, 12)]["source"] == (12, 7)  # punch-in 은 crop 유지


# ─────────────────────────── P8 immobilized_subject ───────────────────────────
# Codex 합의 acceptance: (1) pure seed detector, (2) manifest→loader→detail 주입.
# 데이터는 분석-run 구조 재현(시신 = 한 인물이 같은 씬 2샷 immobilized)이되 작품
# 토큰 0 — 이름/short_id 는 generic placeholder.

from app.modules.pipeline.visual_continuity_anchor_plan import (  # noqa: E402
    ANCHOR_TYPE_IMMOBILIZED,
    build_immobilized_subject_group,
    detect_immobilized_subject_seeds,
    select_immobilized_seeds_with_cap,
)


def _staging(state_by_char):
    """{(si,shi): [(name, state, body_pose)]} → staging_by_shot dict."""
    out = {}
    for key, cas in state_by_char.items():
        out[key] = {
            "character_angles": [
                {"character": n, "subject_state": s, "body_pose": bp}
                for (n, s, bp) in cas
            ]
        }
    return out


def test_immobilized_seed_basic_two_shots_same_scene():
    # 같은 씬(7)에서 한 인물(Name-A=C04)이 2 selected 샷에 dead → seed.
    staging = _staging({
        (7, 8): [("Name-A", "dead", "seated head fallen"), ("Name-B", "alive", "standing")],
        (7, 13): [("Name-A", "dead", "stiff hand clenched")],
    })
    ve = {(7, 8): ["C04", "C09", "L04"], (7, 13): ["C04", "P09"]}
    selected = {7: {8, 13}}
    name_to_sids = {"Name-A": {"C04"}, "Name-B": {"C09"}}
    framing = {(7, 8): "medium", (7, 13): "insert"}

    seeds, skipped = detect_immobilized_subject_seeds(
        staging, ve, selected, name_to_sids, framing_by_shot=framing)
    assert len(seeds) == 1
    s = seeds[0]
    assert s["character_short_id"] == "C04"
    assert s["scene_index"] == 7
    assert s["subject_state"] == "dead"
    assert s["member_shots"] == [[7, 8], [7, 13]]
    # insert 1개 → close_count*3 + len*2 = 3 + 2 = 5
    assert s["score"] == 5
    assert skipped == []


def test_immobilized_seed_gates_ve_name_state_mixed():
    staging = _staging({
        # alive only → 제외
        (1, 1): [("Solo", "alive", "x")],
        # immobilized 1샷뿐 → <2 제외 (다른 씬)
        (2, 1): [("Lonely", "unconscious", "x")],
        # name 미상 → unknown_subject_name skip
        (3, 1): [("Ghost", "dead", "x")],
        (3, 2): [("Ghost", "dead", "x")],
        # name ambiguous(2 sids) → ambiguous skip
        (4, 1): [("Twin", "dead", "x")],
        (4, 2): [("Twin", "dead", "x")],
        # VE 미포함 → subject_not_in_visible_entities skip (sh2 에 C50 없음)
        (5, 1): [("Body", "dead", "x")],
        (5, 2): [("Body", "dead", "x")],
        # mixed states (dead + unconscious) → mixed_immobilized_states skip
        (6, 1): [("Mixed", "dead", "x")],
        (6, 2): [("Mixed", "unconscious", "x")],
    })
    ve = {
        (1, 1): ["C01"], (2, 1): ["C02"],
        (3, 1): ["C03"], (3, 2): ["C03"],
        (4, 1): ["C04", "C44"], (4, 2): ["C04", "C44"],
        (5, 1): ["C50"], (5, 2): ["C99"],   # sh2 에 C50 없음
        (6, 1): ["C06"], (6, 2): ["C06"],
    }
    selected = {1: {1}, 2: {1}, 3: {1, 2}, 4: {1, 2}, 5: {1, 2}, 6: {1, 2}}
    name_to_sids = {
        "Solo": {"C01"}, "Lonely": {"C02"},
        "Twin": {"C04", "C44"},                 # ambiguous
        "Body": {"C50"}, "Mixed": {"C06"},
        # "Ghost" 미등록 → unknown
    }
    seeds, skipped = detect_immobilized_subject_seeds(
        staging, ve, selected, name_to_sids)
    assert seeds == []  # 전부 gate 에 걸림
    reasons = {sk["reason"] for sk in skipped}
    assert "unknown_subject_name" in reasons
    assert "ambiguous_subject_name" in reasons
    assert "subject_not_in_visible_entities" in reasons
    assert "mixed_immobilized_states" in reasons


def test_immobilized_cap_is_independent_and_diagnoses_overflow():
    seeds = [
        {"character_short_id": f"C{i:02d}", "scene_index": i,
         "subject_state": "dead", "member_shots": [[i, 1], [i, 2]], "score": 10 - i}
        for i in range(5)
    ]
    selected, skipped = select_immobilized_seeds_with_cap(seeds, 2)
    assert len(selected) == 2
    assert len(skipped) == 3
    assert all(sk["reason"] == "immobilized_cap" for sk in skipped)
    assert all("score" in sk for sk in skipped)


def test_build_immobilized_group_shape_and_focus_filter():
    seed = {
        "character_short_id": "C04", "scene_index": 7, "subject_state": "dead",
        "member_shots": [[7, 8], [7, 13]], "score": 5,
    }
    llm_result = {
        "shared_state_contract": "The body sits slumped, head fallen forward.",
        "locked_elements": [
            {"kind": "subject_pose", "description": "seated, head fallen",
             "evidence_quote": "쓰러진 채 앉아", "confidence": "high"},
        ],
        "per_shot_visible_focus": [
            {"shot_index": 8, "visible_focus": "full seated body"},
            {"shot_index": 13, "visible_focus": "only the clenched hand"},
            {"shot_index": 99, "visible_focus": "ignored — not a member"},
        ],
    }
    g = build_immobilized_subject_group(
        seed, llm_result, framing_by_shot={(7, 8): "medium", (7, 13): "insert"},
        related_zoom_group_ids=["vca-s7-zoom-5"])
    assert g["anchor_type"] == ANCHOR_TYPE_IMMOBILIZED
    assert g["group_id"] == "vca-s7-c04-immobilized"
    assert g["subject_anchor"]["character_short_id"] == "C04"
    assert g["subject_anchor"]["subject_state"] == "dead"
    assert g["subject_anchor"]["per_shot_visible_focus"] == {
        "8": "full seated body", "13": "only the clenched hand"}  # 99 필터됨
    assert g["related_zoom_group_ids"] == ["vca-s7-zoom-5"]
    # role: insert → close_insert, medium → environment
    roles = {(m["scene_index"], m["shot_index"]): m["role"] for m in g["members"]}
    assert roles[(7, 13)] == "close_insert"
    assert roles[(7, 8)] == "environment"
    assert g["per_shot_consumption_hints"]["7:8"]["enforce_state_continuity"] is True


def test_immobilized_manifest_to_loader(monkeypatch, tmp_path):
    import json as _json
    from app.core.config import settings
    from app.core.steps.visual_continuity_anchor_step import (
        load_immobilized_subject_anchor_context,
        load_printed_prop_anchor_context,
    )

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", True, raising=False)
    monkeypatch.setattr(
        settings, "immobilized_subject_continuity_enabled", True, raising=False)

    cp_dir = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" / "visual_continuity_anchor"
    cp_dir.mkdir(parents=True)
    manifest = {
        "status": "completed", "schema_version": 1, "config_hash": "h",
        "data": {"groups": [
            {  # printed_prop (loader 분리 검증용 — immobilized loader 가 무시해야)
                "group_id": "vca-p09-prop", "anchor_type": "printed_prop",
                "members": [{"scene_index": 7, "shot_index": 13, "role": "close_insert"}],
                "prop_anchor": {"prop_short_id": "P09", "carries_printed_content": True,
                                "printed_content": "x", "physical_form": "y",
                                "scale_contract": "z"},
                "per_shot_consumption_hints": {"7:13": {"keep_real_scale": True}},
            },
            {
                "group_id": "vca-s7-c04-immobilized",
                "anchor_type": ANCHOR_TYPE_IMMOBILIZED,
                "members": [
                    {"scene_index": 7, "shot_index": 8, "role": "environment"},
                    {"scene_index": 7, "shot_index": 13, "role": "close_insert"},
                ],
                "locked_elements": [],
                "subject_anchor": {
                    "character_short_id": "C04", "subject_state": "dead",
                    "shared_state_contract": "The body sits slumped.",
                    "per_shot_visible_focus": {"8": "full body", "13": "hand"},
                },
                "related_zoom_group_ids": ["vca-s7-zoom-5"],
                "per_shot_consumption_hints": {
                    "7:8": {"enforce_state_continuity": True},
                    "7:13": {"enforce_state_continuity": True},
                },
            },
        ]},
    }
    cp_dir.joinpath("manifest.json").write_text(_json.dumps(manifest))

    imm = load_immobilized_subject_anchor_context("p1", "e1")
    assert set(imm["anchors_by_group"]) == {"vca-s7-c04-immobilized"}
    assert imm["members_by_shot"][(7, 8)] == ["vca-s7-c04-immobilized"]
    assert imm["members_by_shot"][(7, 13)] == ["vca-s7-c04-immobilized"]
    sa = imm["anchors_by_group"]["vca-s7-c04-immobilized"]
    assert sa["character_short_id"] == "C04"
    assert sa["shared_state_contract"] == "The body sits slumped."
    assert sa["group_id"] == "vca-s7-c04-immobilized"

    # loader 분리: printed_prop loader 는 immobilized 그룹을 by_prop 에 넣지 않는다.
    prop = load_printed_prop_anchor_context("p1", "e1")
    assert set(prop["by_prop"]) == {"P09"}

    # immobilized flag OFF → 빈 dict (default no-op).
    monkeypatch.setattr(
        settings, "immobilized_subject_continuity_enabled", False, raising=False)
    assert load_immobilized_subject_anchor_context("p1", "e1") == {}


def test_immobilized_loader_same_char_two_scenes_no_contract_bleed(monkeypatch, tmp_path):
    """BLOCKING1 회귀: 같은 C04 가 S7·S12 두 씬에서 각각 immobilized group 을 가져도
    group_id 키라 한 씬 계약이 다른 씬 샷에 새지 않는다."""
    import json as _json
    from app.core.config import settings
    from app.core.steps.visual_continuity_anchor_step import (
        load_immobilized_subject_anchor_context,
    )

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", True, raising=False)
    monkeypatch.setattr(
        settings, "immobilized_subject_continuity_enabled", True, raising=False)

    def _imm_group(gid, si, contract):
        return {
            "group_id": gid, "anchor_type": ANCHOR_TYPE_IMMOBILIZED,
            "members": [
                {"scene_index": si, "shot_index": 1, "role": "environment"},
                {"scene_index": si, "shot_index": 2, "role": "close_insert"},
            ],
            "locked_elements": [],
            "subject_anchor": {
                "character_short_id": "C04", "subject_state": "dead",
                "shared_state_contract": contract, "per_shot_visible_focus": {},
            },
            "per_shot_consumption_hints": {},
        }

    cp_dir = tmp_path / "p2" / "checkpoints" / "episodes" / "e2" / "visual_continuity_anchor"
    cp_dir.mkdir(parents=True)
    cp_dir.joinpath("manifest.json").write_text(_json.dumps({
        "status": "completed", "schema_version": 1, "config_hash": "h",
        "data": {"groups": [
            _imm_group("vca-s7-c04-immobilized", 7, "S7: lying on the floor."),
            _imm_group("vca-s12-c04-immobilized", 12, "S12: seated behind curtain."),
        ]},
    }))

    imm = load_immobilized_subject_anchor_context("p2", "e2")
    # 두 그룹 모두 보존 (덮이지 않음).
    assert set(imm["anchors_by_group"]) == {
        "vca-s7-c04-immobilized", "vca-s12-c04-immobilized"}
    # 각 샷은 자기 씬 group_id 만 가리킨다 — 계약 누출 0.
    assert imm["members_by_shot"][(7, 1)] == ["vca-s7-c04-immobilized"]
    assert imm["members_by_shot"][(12, 1)] == ["vca-s12-c04-immobilized"]
    assert (imm["anchors_by_group"]["vca-s7-c04-immobilized"]["shared_state_contract"]
            == "S7: lying on the floor.")
    assert (imm["anchors_by_group"]["vca-s12-c04-immobilized"]["shared_state_contract"]
            == "S12: seated behind curtain.")


def test_immobilized_loader_drops_empty_contract_group(monkeypatch, tmp_path):
    """MINOR1 회귀: shared_state_contract 가 빈 그룹은 loader 가 제외 (무의미 주입 차단)."""
    import json as _json
    from app.core.config import settings
    from app.core.steps.visual_continuity_anchor_step import (
        load_immobilized_subject_anchor_context,
    )
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "visual_continuity_anchor_enabled", True, raising=False)
    monkeypatch.setattr(
        settings, "immobilized_subject_continuity_enabled", True, raising=False)
    cp_dir = tmp_path / "p3" / "checkpoints" / "episodes" / "e3" / "visual_continuity_anchor"
    cp_dir.mkdir(parents=True)
    cp_dir.joinpath("manifest.json").write_text(_json.dumps({
        "status": "completed", "schema_version": 1,
        "data": {"groups": [{
            "group_id": "vca-s1-c01-immobilized", "anchor_type": ANCHOR_TYPE_IMMOBILIZED,
            "members": [
                {"scene_index": 1, "shot_index": 1, "role": "environment"},
                {"scene_index": 1, "shot_index": 2, "role": "close_insert"},
            ],
            "locked_elements": [],
            "subject_anchor": {
                "character_short_id": "C01", "subject_state": "dead",
                "shared_state_contract": "   ", "per_shot_visible_focus": {},
            },
            "per_shot_consumption_hints": {},
        }]},
    }))
    assert load_immobilized_subject_anchor_context("p3", "e3") == {}


# ─────────────────── identity-variant family (2026-07-02) ───────────────────

from app.modules.pipeline.visual_continuity_anchor_plan import (  # noqa: E402
    build_identity_family_by_sid,
)


def test_identity_family_pairs_chain_and_type_filter():
    relations = [
        {"entity_type": "character", "base_short_id": "C05", "variant_short_id": "C16"},
        # 체인: C16 의 또 다른 변형 → 3인 family 로 union
        {"entity_type": "character", "base_short_id": "C16", "variant_short_id": "C21"},
        # prop 관계는 결합 금지 (character 한정)
        {"entity_type": "prop", "base_short_id": "P01", "variant_short_id": "P02"},
        # malformed → 무시
        {"entity_type": "character", "base_short_id": "", "variant_short_id": "C30"},
        {"entity_type": "character", "base_short_id": "C31", "variant_short_id": "C31"},
        "not-a-dict",
    ]
    fam = build_identity_family_by_sid(relations)
    assert fam["C05"] == {"C05", "C16", "C21"}
    assert fam["C16"] == fam["C05"] == fam["C21"]
    assert "P01" not in fam and "P02" not in fam
    assert "C30" not in fam and "C31" not in fam


def test_identity_family_empty_inputs():
    assert build_identity_family_by_sid([]) == {}
    assert build_identity_family_by_sid(None) == {}


def test_immobilized_seed_matches_ve_via_identity_variant():
    """subject sid 는 VE 에 없지만 variant EntityCanon 이 VE 에 실재 → 멤버 인정.

    (S12 시신 실측 재현: staging=base C05, VE=variant C16.)
    """
    staging = _staging({
        (12, 10): [("Body", "dead", "seated slumped")],
        (12, 16): [("Body", "dead", "hand clutching")],
    })
    ve = {(12, 10): ["C16", "L05"], (12, 16): ["C16", "P02"]}
    selected = {12: {10, 16}}
    name_to_sids = {"Body": {"C05"}}
    fam = {"C05": {"C05", "C16"}, "C16": {"C05", "C16"}}

    seeds, skipped = detect_immobilized_subject_seeds(
        staging, ve, selected, name_to_sids, identity_family_by_sid=fam)
    assert len(seeds) == 1
    s = seeds[0]
    assert s["character_short_id"] == "C05"
    assert s["member_shots"] == [[12, 10], [12, 16]]
    # 진단 채널에 매칭 감사 기록
    matched = [d for d in skipped
               if d.get("reason") == "subject_matched_via_identity_variant"]
    assert len(matched) == 2
    assert matched[0]["ve_member"] == "C16"
    assert set(matched[0]["identity_family"]) == {"C05", "C16"}


def test_immobilized_seed_family_absent_keeps_skip():
    """family 미제공/무관계 → 기존 subject_not_in_visible_entities skip 유지."""
    staging = _staging({
        (12, 10): [("Body", "dead", "x")],
        (12, 16): [("Body", "dead", "x")],
    })
    ve = {(12, 10): ["C16"], (12, 16): ["C16"]}
    selected = {12: {10, 16}}
    name_to_sids = {"Body": {"C05"}}

    seeds, skipped = detect_immobilized_subject_seeds(
        staging, ve, selected, name_to_sids)
    assert seeds == []
    reasons = {d["reason"] for d in skipped}
    assert reasons == {"subject_not_in_visible_entities"}

    # 무관 family (다른 인물끼리) → 여전히 skip + family 진단 포함
    seeds2, skipped2 = detect_immobilized_subject_seeds(
        staging, ve, selected, name_to_sids,
        identity_family_by_sid={"C05": {"C05", "C09"}})
    assert seeds2 == []
    assert all(d["reason"] == "subject_not_in_visible_entities" for d in skipped2)
    assert skipped2[0]["identity_family"] == ["C05", "C09"]