"""W21B-W8 outdoor_site_layout — deterministic 핵심 계약 테스트 (최소).

LLM layout/재작문 품질은 테스트 비대상 (canary + 육안 gate). 여기서는
조인/검증/merge/audit 의 데이터 계약만 잠근다.
"""
import json

from app.modules.pipeline.outdoor_site_layout_plan import (
    build_blocking_sketch_prompt,
    build_shot_spatial_summary,
    detect_site_seeds,
    merge_prompt_overrides,
    outdoor_loc_ids,
    render_pose_brief_text,
    revised_prompt_token_violations,
    validate_site_layout,
    zoom_member_shots,
)


def _staging(n_chars: int, moving: bool = False):
    constraints = []
    if moving:
        constraints.append({
            "target_kind": "character",
            "gesture_action": "moves_toward",
            "target_id": "C01",
        })
    return {
        "character_angles": [{"character": f"p{i}"} for i in range(n_chars)],
        "frame_spatial_contract": {"constraints": constraints},
    }


def test_detect_site_seeds_join_and_exclusions():
    """outdoor × 2인+ join / zoom 멤버 제외 / 1인 이동 진단 / 실내 제외."""
    selected_map = {15: {1, 5, 7, 9}, 3: {2}}
    ve_by_shot = {
        (15, 1): ["C01", "L12"],   # outdoor, 1인 이동 → 진단
        (15, 5): ["C09", "L12"],   # outdoor, 2인 → seed
        (15, 7): ["C06", "L12"],   # outdoor, 2인, zoom 멤버 → 제외
        (15, 9): ["C06", "L12", "L99"],  # outdoor 2곳 → 첫 번째 채택 + 진단
        (3, 2): ["C01", "L05"],    # 실내 → seed 아님 (진단도 없음)
    }
    staging_by_shot = {
        (15, 1): _staging(1, moving=True),
        (15, 5): _staging(2),
        (15, 7): _staging(2),
        (15, 9): _staging(3),
        (3, 2): _staging(2),
    }
    groups, skipped, seed_meta = detect_site_seeds(
        selected_map,
        ve_by_shot=ve_by_shot,
        staging_by_shot=staging_by_shot,
        outdoor_locs={"L12", "L99"},
        zoom_members={(15, 7)},
    )
    # legacy (single_shot_lane_enabled=False) — byte-identical.
    assert groups == {"L12": [(15, 5), (15, 9)]}
    reasons = {(s["shot"], s["reason"]) for s in skipped}
    assert ("S15sh1", "single_figure_moving") in reasons
    assert ("S15sh7", "zoom_continuity_member") in reasons
    assert ("S15sh9", "multiple_outdoor_locations") in reasons
    assert not any(s["shot"] == "S3sh2" for s in skipped)


def test_detect_site_seeds_broad_single_shot_lane():
    """4a fix: single_shot_lane_enabled → 단일 figure 복잡/구도샷도 seed +
    primary_location fallback(VE 에 location 없어도 scene primary 로 resolve)."""
    selected_map = {15: {1, 5}}
    ve_by_shot = {
        (15, 1): ["C01", "L11"],   # frame-visible outdoor location, 1인
        (15, 5): ["C02"],          # location 없음 → primary_location fallback
    }
    staging_by_shot = {
        (15, 1): _staging(1),      # 1인 (비이동)
        (15, 5): _staging(2),      # 2인
    }
    groups, skipped, seed_meta = detect_site_seeds(
        selected_map,
        ve_by_shot=ve_by_shot,
        staging_by_shot=staging_by_shot,
        outdoor_locs={"L11"},
        zoom_members=set(),
        primary_location_by_scene={15: "L11"},
        single_shot_lane_enabled=True,
    )
    # 두 샷 다 L11 seed (sh1=frame_visible 1인, sh5=primary_location fallback 2인).
    assert groups == {"L11": [(15, 1), (15, 5)]}
    assert seed_meta[(15, 1)]["location_source"] == "frame_visible"
    assert seed_meta[(15, 1)]["figure_count"] == 1
    assert seed_meta[(15, 1)]["why_candidate"] == "single_shot_outdoor_figure_present"
    assert seed_meta[(15, 5)]["location_source"] == "primary_location"
    # seed_meta 구조 계약(judge payload 로 전달되는 필드 — Codex 잠금).
    for _req in ("why_candidate", "location_source", "location_id", "figure_count",
                 "framing_scale", "character_angle_count", "frame_spatial_contract_present",
                 "available_structural_signal_names"):
        assert _req in seed_meta[(15, 1)], _req
    # _staging 은 character_angles 를 제공(framing_scale 은 None) — 구조 신호 목록에 반영.
    assert "character_angles" in seed_meta[(15, 1)]["available_structural_signal_names"]


def test_detect_site_seeds_broad_lane_no_figure_included():
    """broad lane 은 figure 0(구조/공간 establishing 샷)도 seed 에 포함한다.

    3-stage 필수화(2026-07-02): 인물 없는 outdoor 접근부/establishing 샷이 aerial
    공간 추론 없이 렌더되며 불가능한 카메라 프레이밍을 만들던 구조 누락 해소.
    attach 필요성은 judge 가 판정(seed 는 커버리지만)."""
    groups, skipped, seed_meta = detect_site_seeds(
        {15: {1}},
        ve_by_shot={(15, 1): ["L11"]},
        staging_by_shot={(15, 1): {"character_angles": []}},
        outdoor_locs={"L11"},
        zoom_members=set(),
        single_shot_lane_enabled=True,
    )
    assert groups == {"L11": [(15, 1)]}
    assert not any(s.get("reason") == "no_figure" for s in skipped)
    assert seed_meta[(15, 1)]["why_candidate"] == "single_shot_outdoor_no_figure"
    assert seed_meta[(15, 1)]["figure_count"] == 0


def test_detect_site_seeds_legacy_lane_no_figure_still_excluded():
    """legacy(single_shot_lane_enabled=False)는 figure<2 제외 그대로(byte-identical)."""
    groups, _skipped, _meta = detect_site_seeds(
        {15: {1}},
        ve_by_shot={(15, 1): ["L11"]},
        staging_by_shot={(15, 1): {"character_angles": []}},
        outdoor_locs={"L11"},
        zoom_members=set(),
        single_shot_lane_enabled=False,
    )
    assert groups == {}


def test_outdoor_and_zoom_member_joins_are_structured_only():
    """is_indoor structured field / ref_usage enum 조인만 — 텍스트 의미 판별 0."""
    groups = [
        {"members": [{"loc_id": "L01", "is_indoor": False},
                     {"loc_id": "L02", "is_indoor": True}]},
        {"members": [{"loc_id": "L03"}]},  # is_indoor 부재 → outdoor 아님
    ]
    assert outdoor_loc_ids(groups) == {"L01"}

    deps = [
        {"scene_index": 12, "shot_index": 12, "location_refs": [
            {"scene_index": 12, "shot_index": 7, "ref_usage": "zoom_in_detail"},
        ]},
        {"scene_index": 5, "shot_index": 3, "location_refs": [
            {"scene_index": 5, "shot_index": 1, "ref_usage": "exact_background"},
        ]},
    ]
    assert zoom_member_shots(deps) == {(12, 12), (12, 7)}


def test_validate_site_layout_shape_and_range_only():
    good = {
        "landmarks": [{"id": "lm1", "label": "road", "kind": "line",
                       "points": [[0, 50], [100, 60]]}],
        "figures": [{"figure_id": "f1", "label": "A", "entity_token": None,
                     "positions": [{"shot_index": 5, "pos": [40, 40],
                                    "moving_toward": [90, 55]}]}],
        "cameras": [{"shot_index": 5, "pos": [30, 30], "look_at": [60, 50]}],
    }
    assert validate_site_layout(good) == []

    bad = json.loads(json.dumps(good))
    bad["figures"][0]["positions"][0]["pos"] = [40, 140]  # 범위 밖
    bad["cameras"][0]["look_at"] = [60]  # shape 위반
    violations = validate_site_layout(bad)
    assert any("pos" in v for v in violations)
    assert any("look_at" in v for v in violations)


def test_spatial_summary_geography_is_deterministic():
    """R2 사용자 피드백 계약 — 요약이 프레임 기준 지형(경로 진행/이동 방향의
    landmark 기준 서술/안 가는 곳)을 좌표 산술로 출력한다. 카메라 상대 동사 금지."""
    layout = {
        "landmarks": [
            {"id": "rd", "label": "coastal road", "kind": "line",
             "points": [[20, 20], [95, 85]]},
            {"id": "sea", "label": "the sea", "kind": "area",
             "points": [[80, 5], [100, 5], [100, 35]]},
        ],
        "figures": [
            {"figure_id": "near", "label": "NearFig", "entity_token": "C06",
             "positions": [{"shot_index": 5, "pos": [35, 30], "moving_toward": None}]},
            {"figure_id": "far", "label": "FarFig", "entity_token": "C09",
             "positions": [{"shot_index": 5, "pos": [80, 70],
                            "moving_toward": [95, 85]}]},
        ],
        "cameras": [{"shot_index": 5, "pos": [30, 25], "look_at": [60, 50]}],
    }
    summary = build_shot_spatial_summary(layout, 5)
    assert summary is not None
    assert "coastal road (path): runs from the" in summary  # 경로 진행 서술
    near_line = next(l for l in summary.splitlines() if "NearFig" in l)
    far_line = next(l for l in summary.splitlines() if "FarFig" in l)
    assert "NEAREST" in near_line
    assert "smaller" in far_line
    # 이동 = 위치 진술 + landmark 기준 (카메라 상대 동사 금지)
    assert "already partway along their path" in far_line
    assert "follows the coastal road" in far_line
    assert "moving away from NearFig" in far_line
    assert "NOT heading into the sea" in far_line  # 관사 중복("the the") 금지
    assert "the the" not in summary
    assert "away from the camera" not in summary
    # 카메라 없는 shot → None (caller 가 skip 진단)
    assert build_shot_spatial_summary(layout, 99) is None


def test_path_classification_uses_aspect_not_just_extent():
    """경로성 판정 = 종횡비(길고 얇음)여야 한다 — 큰 blob(넓은 면적)은 extent 가
    커도 'at X' area, 길고 얇은 area 는 'path' (extent 만으로 승격하던 퇴화 수정).
    둘 다 kind='area' 라 kind 가 아니라 aspect 가 판별자임을 증명한다."""
    layout = {
        "landmarks": [
            # 길고 얇은 area (aspect≈16) → path 여야
            {"id": "strip", "label": "long strip", "kind": "area",
             "points": [[10, 50], [90, 50], [90, 55], [10, 55]]},
            # 넓은 정사각형 blob (extent≈42 ≥25 이지만 aspect≈2) → area 여야
            {"id": "blob", "label": "wide field", "kind": "area",
             "points": [[10, 10], [40, 10], [40, 40], [10, 40]]},
        ],
        "figures": [
            {"figure_id": "f", "label": "Fig", "entity_token": "C06",
             "positions": [{"shot_index": 1, "pos": [50, 52], "moving_toward": None}]},
        ],
        "cameras": [{"shot_index": 1, "pos": [50, 95], "look_at": [50, 30]}],
    }
    summary = build_shot_spatial_summary(layout, 1)
    assert summary is not None
    strip_line = next(l for l in summary.splitlines() if "long strip" in l)
    blob_line = next(l for l in summary.splitlines() if "wide field" in l)
    assert "(path): runs from the" in strip_line   # 길고 얇음 → path
    assert "(path)" not in blob_line               # 넓은 blob → path 아님
    assert "wide field: at the" in blob_line       # area 서술


def test_layout_side_conflicts_join():
    """layout 좌/우 ↔ staged screen_zone 충돌 — entity_token↔target_id 조인."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        layout_side_conflicts,
        staged_sides_for_shot,
    )

    staging = {"frame_spatial_contract": {"constraints": [
        {"target_kind": "character", "target_id": "C09", "screen_zone": "middle_left"},
        {"target_kind": "background", "target_id": "", "screen_zone": "upper_right"},
    ]}}
    sides = staged_sides_for_shot(staging)
    assert sides == {"C09": "left"}

    layout = {
        "figures": [{"figure_id": "f", "label": "F", "entity_token": "C09",
                     "positions": [{"shot_index": 5, "pos": [50, 20],
                                    "moving_toward": None}]}],
        "cameras": [{"shot_index": 5, "pos": [10, 50], "look_at": [90, 50]}],
    }
    conflicts = layout_side_conflicts(layout, 5, sides)
    assert conflicts == [{"shot_index": 5, "entity_token": "C09",
                          "staged_side": "left", "layout_side": "right"}]
    # 일치하면 충돌 0
    layout["figures"][0]["positions"][0]["pos"] = [50, 80]
    assert layout_side_conflicts(layout, 5, sides) == []


def test_revised_prompt_token_violations_both_directions():
    original = "C06 stands near the shelter while C09 runs down the road L12"
    ok = "C06 stands near the shelter while C09, small in the distance, runs down the road L12"
    assert revised_prompt_token_violations(original, ok) == {}
    added = revised_prompt_token_violations(original, original + " and P02 lies there")
    assert added == {"new_tokens": ["P02"]}
    dropped = revised_prompt_token_violations(original, "C06 stands near the shelter on L12")
    assert dropped == {"missing_tokens": ["C09"]}


def test_merge_prompt_overrides_priority_zoom_over_site():
    """custom > zoom > site > original — 같은 (shot, variation) 은 zoom 승리,
    site 는 빈 곳만 채움. 둘 다 빈 맵이면 빈 맵 (no-op)."""
    assert merge_prompt_overrides({}, {}) == {}

    zoom = {(15, 5): {"group_id": "vca-s15-zoom-1",
                      "revised": {0: "zoom revised v0"},
                      "provenance": {"0": {"revised_prompt_hash": "zzz"}}}}
    site = {(15, 5): {"group_id": "osl-l12",
                      "revised": {0: "site revised v0", 1: "site revised v1"},
                      "provenance": {"0": {"revised_prompt_hash": "s0"},
                                     "1": {"revised_prompt_hash": "s1"}}},
            (15, 7): {"group_id": "osl-l12",
                      "revised": {0: "site only v0"},
                      "provenance": {"0": {"revised_prompt_hash": "s7"}}}}
    merged = merge_prompt_overrides(zoom, site)

    entry = merged[(15, 5)]
    assert entry["revised"] == {0: "zoom revised v0", 1: "site revised v1"}
    assert entry["prompt_source_by_index"] == {
        0: "zoom_continuity_anchor", 1: "outdoor_site_layout"}
    assert entry["prompt_source"] == "zoom_continuity_anchor"
    assert entry["provenance"]["0"]["revised_prompt_hash"] == "zzz"
    assert entry["group_id"] == "vca-s15-zoom-1"

    only_site = merged[(15, 7)]
    assert only_site["revised"] == {0: "site only v0"}
    assert only_site["prompt_source"] == "outdoor_site_layout"
    assert only_site["group_id"] == "osl-l12"


def test_loader_flag_off_and_on(monkeypatch, tmp_path):
    """flag OFF → {} (no-op) / ON + completed cp → (si, shi) int-key 맵."""
    from app.core.config import settings
    from app.core.steps.outdoor_site_layout_step import (
        load_outdoor_site_layout_context,
    )

    cp_dir = (tmp_path / "p1" / "checkpoints" / "episodes" / "e1"
              / "outdoor_site_layout")
    cp_dir.mkdir(parents=True)
    (cp_dir / "manifest.json").write_text(json.dumps({
        "status": "completed",
        "data": {"prompt_overrides": {"15:5": {
            "group_id": "osl-l12",
            "revised": {"0": "revised text"},
            "provenance": {"0": {"revised_prompt_hash": "abc"}},
        }}},
    }), encoding="utf-8")
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)

    monkeypatch.setattr(settings, "outdoor_site_layout_enabled", False, raising=False)
    assert load_outdoor_site_layout_context("p1", "e1") == {}

    monkeypatch.setattr(settings, "outdoor_site_layout_enabled", True, raising=False)
    ctx = load_outdoor_site_layout_context("p1", "e1")
    entry = ctx["prompt_overrides"][(15, 5)]
    assert entry["revised"] == {0: "revised text"}
    assert entry["prompt_source"] == "outdoor_site_layout"


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

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


def _write_cp(tmp_path, step_id, data):
    d = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" / step_id
    d.mkdir(parents=True, exist_ok=True)
    (d / "manifest.json").write_text(
        json.dumps({"status": "completed", "data": data}, ensure_ascii=False),
        encoding="utf-8",
    )


def test_layout_failure_is_non_blocking(monkeypatch, tmp_path):
    """Codex W8_CODE_REVIEW NARROW_1 계약 — 한 location 의 layout LLM 실패가
    failed_count 를 만들지 않고(status partial 차단 방지) 다른 valid group 의
    override 는 유지된다. 실패는 diagnostics 로만 surface."""
    from app.core.config import settings
    from app.core.steps.outdoor_site_layout_step import OutdoorSiteLayoutStep

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

    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "description": "d1"}]},
        {"scene_index": 2, "shots": [{"shot_index": 1, "description": "d2"}]},
    ]})
    _write_cp(tmp_path, "shot_selection", {"scenes": [
        {"scene_index": 1, "selected_shot_indices": [1]},
        {"scene_index": 2, "selected_shot_indices": [1]},
    ]})
    _write_cp(tmp_path, "background_classify", {"building_groups": [
        {"members": [{"loc_id": "L01", "is_indoor": False},
                     {"loc_id": "L02", "is_indoor": False}]},
    ]})
    _write_cp(tmp_path, "scene_save", {"segments": [
        {"scene_index": 1, "text": "t1"}, {"scene_index": 2, "text": "t2"},
    ]})
    _write_cp(tmp_path, "shot_director", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "visible_entity_ids": ["C01", "L01"]}]},
        {"scene_index": 2, "shots": [{"shot_index": 1, "visible_entity_ids": ["C02", "L02"]}]},
    ]})
    _write_cp(tmp_path, "shot_staging", {"shots": [
        {"scene_index": 1, "shot_index": 1,
         "character_angles": [{"character": "a"}, {"character": "b"}]},
        {"scene_index": 2, "shot_index": 1,
         "character_angles": [{"character": "c"}, {"character": "d"}]},
    ]})
    _write_cp(tmp_path, "scene_detail", {"scenes": [
        {"scene_index": 1, "_shot_index": 1,
         "t2i_variations": [{"t2i_prompt": "C01 stands in middle-center foreground"}]},
        {"scene_index": 2, "_shot_index": 1,
         "t2i_variations": [{"t2i_prompt": "C02 stands in middle-center foreground"}]},
    ]})

    step = OutdoorSiteLayoutStep.__new__(OutdoorSiteLayoutStep)
    step.step_id = "outdoor_site_layout"
    step.project_id = "p1"
    step.episode_id = "e1"
    step.project_config = {}
    step.build_opik_metadata = lambda **kw: {}

    good_layout = {
        "landmarks": [],
        "figures": [{"figure_id": "f1", "label": "fig", "entity_token": None,
                     "positions": [{"shot_index": 1, "pos": [10, 10],
                                    "moving_toward": None}]}],
        "cameras": [{"shot_index": 1, "pos": [5, 5], "look_at": [10, 10]}],
    }

    def layout_fn(scene_texts, member_shots, **kw):
        # L01 그룹(S1) 은 실패, L02 그룹(S2) 은 성공 — 그룹은 loc_id 정렬 순.
        if member_shots[0]["scene_index"] == 1:
            raise RuntimeError("llm boom")
        return good_layout

    def rewrite_fn(variations, summary, **kw):
        return {i: {"revised_prompt": p + " revised", "edited_spans": [],
                    "no_edit_reason": None} for i, p in variations}

    step.set_overrides_for_testing(layout=layout_fn, rewrite=rewrite_fn)
    out = step._execute()

    assert out["failed_count"] == 0  # 비차단 — StepRunner partial 금지
    assert out["applicable_count"] == 2
    assert out["completed_count"] == 1
    assert "2:1" in out["data"]["prompt_overrides"]  # valid group override 유지
    assert any(
        s.get("reason") == "llm_error"
        for s in out["data"]["diagnostics"]["skipped"]
    )


def test_side_conflict_shot_excluded_from_overrides(monkeypatch, tmp_path):
    """Codex R2~R4 리뷰 BLOCKING NARROW 1 계약 — retry 후에도 staged screen_zone
    충돌이 남은 shot 은 override 생성에서 제외(원본 prompt fallback)되고, 같은
    그룹의 충돌 없는 shot 의 override 는 유지된다. 비차단(failed_count=0)."""
    from app.core.config import settings
    from app.core.steps.outdoor_site_layout_step import OutdoorSiteLayoutStep

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

    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "description": "d1"},
                                     {"shot_index": 2, "description": "d2"}]},
    ]})
    _write_cp(tmp_path, "shot_selection", {"scenes": [
        {"scene_index": 1, "selected_shot_indices": [1, 2]},
    ]})
    _write_cp(tmp_path, "background_classify", {"building_groups": [
        {"members": [{"loc_id": "L01", "is_indoor": False}]},
    ]})
    _write_cp(tmp_path, "scene_save", {"segments": [
        {"scene_index": 1, "text": "t1"},
    ]})
    _write_cp(tmp_path, "shot_director", {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 1, "visible_entity_ids": ["C09", "L01"]},
            {"shot_index": 2, "visible_entity_ids": ["C09", "L01"]},
        ]},
    ]})
    _write_cp(tmp_path, "shot_staging", {"shots": [
        # sh1: C09 는 staged left — layout 이 right 에 놓으면 충돌.
        {"scene_index": 1, "shot_index": 1,
         "character_angles": [{"character": "a"}, {"character": "b"}],
         "frame_spatial_contract": {"constraints": [
             {"target_kind": "character", "target_id": "C09",
              "screen_zone": "middle_left"},
         ]}},
        # sh2: side 계약 없음 — 충돌 불가.
        {"scene_index": 1, "shot_index": 2,
         "character_angles": [{"character": "a"}, {"character": "b"}]},
    ]})
    _write_cp(tmp_path, "scene_detail", {"scenes": [
        {"scene_index": 1, "_shot_index": 1,
         "t2i_variations": [{"t2i_prompt": "C09 stands by the road"}]},
        {"scene_index": 1, "_shot_index": 2,
         "t2i_variations": [{"t2i_prompt": "C09 walks along the road"}]},
    ]})

    step = OutdoorSiteLayoutStep.__new__(OutdoorSiteLayoutStep)
    step.step_id = "outdoor_site_layout"
    step.project_id = "p1"
    step.episode_id = "e1"
    step.project_config = {}
    step.build_opik_metadata = lambda **kw: {}

    # 카메라가 +x 를 보는데 C09 가 시선축 아래(y 작음) = frame RIGHT — sh1 의
    # staged left 와 충돌. sh2 카메라/위치는 동일하지만 staged side 가 없다.
    conflicting_layout = {
        "landmarks": [],
        "figures": [{"figure_id": "f1", "label": "fig", "entity_token": "C09",
                     "positions": [
                         {"shot_index": 1, "pos": [50, 20], "moving_toward": None},
                         {"shot_index": 2, "pos": [50, 20], "moving_toward": None},
                     ]}],
        "cameras": [{"shot_index": 1, "pos": [10, 50], "look_at": [90, 50]},
                    {"shot_index": 2, "pos": [10, 50], "look_at": [90, 50]}],
    }
    calls = {"n": 0}

    def layout_fn(scene_texts, member_shots, **kw):
        calls["n"] += 1
        return conflicting_layout  # retry 도 동일 — 충돌 잔존

    def rewrite_fn(variations, summary, **kw):
        return {i: {"revised_prompt": p + " revised", "edited_spans": [],
                    "no_edit_reason": None} for i, p in variations}

    step.set_overrides_for_testing(layout=layout_fn, rewrite=rewrite_fn)
    out = step._execute()

    assert out["failed_count"] == 0
    assert calls["n"] == 2  # 초기 + constraint_feedback retry 1회
    overrides = out["data"]["prompt_overrides"]
    assert "1:1" not in overrides       # 충돌 잔존 shot — 원본 fallback
    assert "1:2" in overrides           # 충돌 없는 shot 은 유지
    diags = out["data"]["diagnostics"]
    assert diags["layout_side_conflicts"]
    assert any(
        s.get("reason") == "layout_side_conflict"
        for s in diags["spatial_summary_skipped"]
    )


# ───────────────────── composition guide (2026-06-13) ─────────────────────


def test_composition_guide_shot_selection():
    """기하 후보 = 2인+ AND 실제로 멀어지는(receding) figure 가 NEAREST 아님 —
    좌표 산술 계약. 제외: summary 없는 샷/1인/이동 figure 가 nearest/이동 figure 가
    카메라로 **다가오는(approaching)** 샷(lunge·근접 대치)."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        composition_guide_shot_keys,
        shot_hint_key,
    )

    layout = {
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C06",
             "positions": [
                 {"shot_index": 1, "pos": [20, 20], "moving_toward": None},
                 {"shot_index": 2, "pos": [20, 20], "moving_toward": [5, 5]},
                 {"shot_index": 3, "pos": [20, 20], "moving_toward": None},
                 {"shot_index": 5, "pos": [20, 20], "moving_toward": None},
             ]},
            {"figure_id": "b", "label": "B", "entity_token": "C09",
             "positions": [
                 # sh1: 이동 figure 가 카메라에서 더 멀다 = departing → 대상
                 {"shot_index": 1, "pos": [70, 70], "moving_toward": [95, 95]},
                 # sh2: 이동 figure(A)가 nearest → 제외 (R2 고스트 역조건)
                 {"shot_index": 2, "pos": [70, 70], "moving_toward": None},
                 # sh3: 아무도 이동 안 함 → 제외
                 {"shot_index": 3, "pos": [70, 70], "moving_toward": None},
                 # sh5: 깊은 figure 가 mover 지만 카메라 쪽으로 **다가옴**
                 # (moving_toward 가 현 위치보다 가까움) = departure 아님 → 제외
                 # (공격 lunge/근접 대치 오선별 제거 — S10sh5 패턴)
                 {"shot_index": 5, "pos": [70, 70], "moving_toward": [40, 40]},
             ]},
        ],
        "cameras": [
            {"shot_index": 1, "pos": [10, 10], "look_at": [50, 50]},
            {"shot_index": 2, "pos": [10, 10], "look_at": [50, 50]},
            {"shot_index": 3, "pos": [10, 10], "look_at": [50, 50]},
            {"shot_index": 4, "pos": [10, 10], "look_at": [50, 50]},
            {"shot_index": 5, "pos": [10, 10], "look_at": [50, 50]},
        ],
    }
    member_keys = [(15, 1), (15, 2), (15, 3), (15, 4), (15, 5)]
    summary_keys = {shot_hint_key(15, i) for i in (1, 2, 3, 5)}  # sh4 = summary 없음
    assert composition_guide_shot_keys(layout, member_keys, summary_keys) == [(15, 1)]
    # summary 가 없으면 (side-conflict skip 등) departing 이어도 제외
    assert composition_guide_shot_keys(layout, member_keys, set()) == []


def _departing_step_fixture(monkeypatch, tmp_path, guide_enabled=True):
    """1 그룹(L01)·S1sh1 departing 샷 — composition guide step 테스트 공용."""
    from app.core.config import settings
    from app.core.steps.outdoor_site_layout_step import OutdoorSiteLayoutStep

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "outdoor_site_layout_enabled", True, raising=False)
    monkeypatch.setattr(
        settings, "outdoor_composition_guide_enabled", guide_enabled, raising=False)
    # legacy(OFF) 경로 전용 fixture — shared_model 은 반드시 OFF 로 pin 한다. .env(E2E)
    # 가 SHARED_MODEL_ENABLED=true 라 pin 안 하면 ON 경로를 타 dummy client 로 실패
    # (pre-existing 위생 부채; ON 경로는 _shared_model_step_fixture 가 별도 커버).
    monkeypatch.setattr(
        settings, "outdoor_composition_guide_shared_model_enabled", False,
        raising=False)

    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "description": "d1"}]},
    ]})
    _write_cp(tmp_path, "shot_selection", {"scenes": [
        {"scene_index": 1, "selected_shot_indices": [1]},
    ]})
    _write_cp(tmp_path, "background_classify", {"building_groups": [
        {"members": [{"loc_id": "L01", "is_indoor": False}]},
    ]})
    _write_cp(tmp_path, "scene_save", {"segments": [
        {"scene_index": 1, "text": "t1"},
    ]})
    _write_cp(tmp_path, "shot_director", {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 1, "visible_entity_ids": ["C06", "C09", "L01"]},
        ]},
    ]})
    _write_cp(tmp_path, "shot_staging", {"shots": [
        {"scene_index": 1, "shot_index": 1,
         "character_angles": [{"character": "a"}, {"character": "b"}]},
    ]})
    _write_cp(tmp_path, "scene_detail", {"scenes": [
        {"scene_index": 1, "_shot_index": 1,
         "t2i_variations": [{"t2i_prompt": "C06 watches C09 run down the road"}]},
    ]})

    step = OutdoorSiteLayoutStep.__new__(OutdoorSiteLayoutStep)
    step.step_id = "outdoor_site_layout"
    step.project_id = "p1"
    step.episode_id = "e1"
    step.project_config = {}
    step.build_opik_metadata = lambda **kw: {}

    departing_layout = {
        "landmarks": [],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C06",
             "positions": [{"shot_index": 1, "pos": [20, 20],
                            "moving_toward": None}]},
            {"figure_id": "b", "label": "B", "entity_token": "C09",
             "positions": [{"shot_index": 1, "pos": [70, 70],
                            "moving_toward": [95, 95]}]},
        ],
        "cameras": [{"shot_index": 1, "pos": [10, 10], "look_at": [50, 50]}],
    }

    def layout_fn(scene_texts, member_shots, **kw):
        return departing_layout

    def rewrite_fn(variations, summary, **kw):
        return {i: {"revised_prompt": p + " revised", "edited_spans": [],
                    "no_edit_reason": None} for i, p in variations}

    return step, layout_fn, rewrite_fn


def _admit_judge(scene_action, camera_direction, summary, *, model, **kw):
    """의미 게이트 통과(열린 외부 departure) stub — 기하 후보가 guide 까지 가게."""
    return {
        "is_open_exterior_departure": True, "confidence": "high",
        "reasoning": "open receding exterior", "evidence_quote": "recedes",
    }


def test_composition_guide_first_admitted_shot_is_sketch(monkeypatch, tmp_path):
    """option C — 그룹의 첫 admitted 샷(참조할 같은-그룹 완성 프레임 없음)은
    ``mode='sketch'`` — 마네킹 브리프+스케치를 실제 생성하고 path 를 기록한다.
    (단일 admitted = 첫 샷이므로 sketch.)"""
    step, layout_fn, rewrite_fn = _departing_step_fixture(monkeypatch, tmp_path)
    step._guide_openai_client = object()  # 실제 OpenAI client import/생성 회피
    calls = {"brief": 0, "sketch": 0}

    def brief_fn(scene_action, summary, *, model, **kw):
        calls["brief"] += 1
        return "COMPOSITION BRIEF"

    def sketch_fn(brief, *, openai_client, model, **kw):
        calls["sketch"] += 1
        return b"\x89PNG\r\n\x1a\nFAKE"

    step.set_overrides_for_testing(layout=layout_fn, rewrite=rewrite_fn,
                                   brief=brief_fn, sketch=sketch_fn,
                                   judge=_admit_judge)
    out = step._execute()

    assert out["failed_count"] == 0
    assert calls == {"brief": 1, "sketch": 1}  # 첫 샷은 스케치 생성
    guides = out["data"]["composition_guides"]
    assert "1:1" in guides
    assert guides["1:1"]["mode"] == "sketch"
    assert guides["1:1"]["path"].endswith("S1sh1.png")
    assert guides["1:1"]["admitted_order"] == 0
    assert guides["1:1"]["judge"]["confidence"] == "high"
    # 실제 스케치 파일이 checkpoint dir 에 기록됐는지
    sketch_path = step._checkpoint_dir() / guides["1:1"]["path"]
    assert sketch_path.exists()
    assert "1:1" in out["data"]["prompt_overrides"]  # text override 유지


def test_composition_continuity_chain_first_sketch_rest_anchor():
    """option C 순수 계약 — same scene admitted 샷: 첫=sketch, 이후=continuity_anchor
    +anchor_source(직전 admitted). 다른 scene 으로 체이닝 안 함."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        composition_continuity_chain,
    )
    # 입력 순서 무관 — scene/shot 정렬로 결정.
    chain = composition_continuity_chain([(15, 7), (15, 5), (16, 2)])
    by_key = {c["key"]: c for c in chain}
    assert by_key[(15, 5)]["mode"] == "sketch"
    assert by_key[(15, 5)]["anchor_source"] is None
    assert by_key[(15, 5)]["admitted_order"] == 0
    assert by_key[(15, 7)]["mode"] == "continuity_anchor"
    assert by_key[(15, 7)]["anchor_source"] == (15, 5)
    assert by_key[(15, 7)]["admitted_order"] == 1
    # scene 16 은 별도 그룹 → 첫 샷이므로 sketch (cross-scene 체이닝 없음)
    assert by_key[(16, 2)]["mode"] == "sketch"
    assert by_key[(16, 2)]["anchor_source"] is None


def test_composition_guide_second_shot_is_continuity_anchor(monkeypatch, tmp_path):
    """option C — 같은 scene 2 admitted 샷: sh1=sketch / sh2=continuity_anchor
    (anchor_source=sh1, forced_character_names = this ∩ source staged)."""
    from app.core.config import settings
    from app.core.steps.outdoor_site_layout_step import OutdoorSiteLayoutStep

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "outdoor_site_layout_enabled", True, raising=False)
    monkeypatch.setattr(
        settings, "outdoor_composition_guide_enabled", True, raising=False)
    # legacy(OFF) 경로 테스트 — shared_model OFF pin(.env E2E flag ON 대응, pre-existing).
    monkeypatch.setattr(
        settings, "outdoor_composition_guide_shared_model_enabled", False,
        raising=False)

    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "d1"},
            {"shot_index": 2, "description": "d2"}]},
    ]})
    _write_cp(tmp_path, "shot_selection", {"scenes": [
        {"scene_index": 1, "selected_shot_indices": [1, 2]},
    ]})
    _write_cp(tmp_path, "background_classify", {"building_groups": [
        {"members": [{"loc_id": "L01", "is_indoor": False}]},
    ]})
    _write_cp(tmp_path, "scene_save", {"segments": [{"scene_index": 1, "text": "t1"}]})
    _write_cp(tmp_path, "shot_director", {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 1, "visible_entity_ids": ["C06", "C09", "L01"]},
            {"shot_index": 2, "visible_entity_ids": ["C06", "C09", "L01"]}]},
    ]})
    # sh1: a,b staged / sh2: a,b staged — 교집합 {a,b}
    _write_cp(tmp_path, "shot_staging", {"shots": [
        {"scene_index": 1, "shot_index": 1,
         "character_angles": [{"character": "a"}, {"character": "b"}]},
        {"scene_index": 1, "shot_index": 2,
         "character_angles": [{"character": "a"}, {"character": "b"}]},
    ]})
    _write_cp(tmp_path, "scene_detail", {"scenes": [
        {"scene_index": 1, "_shot_index": 1,
         "t2i_variations": [{"t2i_prompt": "C06 watches C09 run down the road"}]},
        {"scene_index": 1, "_shot_index": 2,
         "t2i_variations": [{"t2i_prompt": "C06 watches C09 farther down the road"}]},
    ]})

    step = OutdoorSiteLayoutStep.__new__(OutdoorSiteLayoutStep)
    step.step_id = "outdoor_site_layout"
    step.project_id = "p1"
    step.episode_id = "e1"
    step.project_config = {}
    step.build_opik_metadata = lambda **kw: {}
    step._guide_openai_client = object()

    # 두 샷 모두 receding mover (C09 가 95,95 로 멀어짐) — 기하 후보.
    layout = {
        "landmarks": [],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C06",
             "positions": [{"shot_index": 1, "pos": [20, 20], "moving_toward": None},
                           {"shot_index": 2, "pos": [20, 20], "moving_toward": None}]},
            {"figure_id": "b", "label": "B", "entity_token": "C09",
             "positions": [{"shot_index": 1, "pos": [60, 60], "moving_toward": [95, 95]},
                           {"shot_index": 2, "pos": [70, 70], "moving_toward": [95, 95]}]},
        ],
        "cameras": [{"shot_index": 1, "pos": [10, 10], "look_at": [50, 50]},
                    {"shot_index": 2, "pos": [10, 10], "look_at": [50, 50]}],
    }

    def layout_fn(scene_texts, member_shots, **kw):
        return layout

    def rewrite_fn(variations, summary, **kw):
        return {i: {"revised_prompt": p + " revised", "edited_spans": [],
                    "no_edit_reason": None} for i, p in variations}

    def brief_fn(scene_action, summary, *, model, **kw):
        return "BRIEF"

    def sketch_fn(brief, *, openai_client, model, **kw):
        return b"\x89PNGFAKE"

    step.set_overrides_for_testing(layout=layout_fn, rewrite=rewrite_fn,
                                   brief=brief_fn, sketch=sketch_fn,
                                   judge=_admit_judge)
    out = step._execute()

    assert out["failed_count"] == 0
    guides = out["data"]["composition_guides"]
    assert guides["1:1"]["mode"] == "sketch"
    assert guides["1:2"]["mode"] == "continuity_anchor"
    assert guides["1:2"]["anchor_source"] == [1, 1]
    assert guides["1:2"]["admitted_order"] == 1
    # forced_character_names = this(sh2) ∩ source(sh1) staged = {a, b}
    assert sorted(guides["1:2"]["forced_character_names"]) == ["a", "b"]
    # 같은 그룹 group_id 공유
    assert guides["1:1"]["group_id"] == guides["1:2"]["group_id"]


# ───────── shared-model 카메라 가이드 (Phase II II-1, 2026-06-29) ─────────


def test_compute_camera_brief_coord_contract():
    """compute_camera_brief = 좌표 산술 계약 (LLM 0): near→far depth ordering +
    figure 상대 스케일 + in-motion. figure 는 featureless marker (이름/토큰 누출 0).
    카메라 좌표 부재 → None."""
    from app.modules.pipeline.outdoor_site_layout_plan import compute_camera_brief

    layout = {
        "landmarks": [
            {"id": "lm1", "label": "alpha structure", "kind": "area",
             "points": [[28, 48], [32, 48], [32, 52], [28, 52]]},   # 근 (depth ~20)
            {"id": "lm2", "label": "omega structure", "kind": "area",
             "points": [[83, 48], [87, 48], [87, 52], [83, 52]]},   # 원 (depth ~75)
        ],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C01",
             "positions": [{"shot_index": 1, "pos": [35, 55], "moving_toward": None}]},
            {"figure_id": "b", "label": "B", "entity_token": "C02",
             "positions": [{"shot_index": 1, "pos": [70, 52],
                            "moving_toward": [95, 52]}]},
        ],
        "cameras": [{"shot_index": 1, "pos": [10, 50], "look_at": [60, 50]}],
    }
    brief = compute_camera_brief(layout["cameras"][0], layout)
    assert brief is not None
    # near→far ordering (좌표 depth)
    assert brief.index("alpha structure") < brief.index("omega structure")
    assert "foreground" in brief and "background" in brief
    # figure = featureless marker — 이름/entity 토큰 누출 0
    assert "C01" not in brief and "C02" not in brief
    assert "a figure" in brief
    assert "in motion" in brief            # C02 receding mover
    # 카메라 좌표 부재 → None
    assert compute_camera_brief(
        {"shot_index": 9, "pos": None, "look_at": None}, layout) is None


def test_compute_camera_brief_enclosure_relation():
    """3-stage 필수화(2026-07-02): enclosure 위상 관계 = 순수 폴리곤 산술.

    figure 가 area landmark polygon 내부 + 카메라 밖 → 'INSIDE ... OUTSIDE' 관계
    서술(스케치가 인물을 개방 지면으로 옮기는 오류 방지). 카메라도 내부면 'inside'
    구절만. 어느 폴리곤에도 없으면 관계 서술 0 (기존 brief byte 구조 유지)."""
    from app.modules.pipeline.outdoor_site_layout_plan import compute_camera_brief

    layout = {
        "landmarks": [
            {"id": "room", "label": "walled room", "kind": "area",
             "points": [[40, 40], [60, 40], [60, 60], [40, 60]]},
        ],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C01",
             "positions": [{"shot_index": 1, "pos": [50, 50],
                            "moving_toward": None}]},   # room 내부
        ],
        "cameras": [{"shot_index": 1, "pos": [10, 50], "look_at": [50, 50]}],  # 밖
    }
    brief = compute_camera_brief(layout["cameras"][0], layout)
    assert brief is not None
    assert "INSIDE the mapped area 'walled room'" in brief
    assert "OUTSIDE" in brief and "boundary or opening" in brief

    # 카메라도 폴리곤 내부 → 'inside' 만 (through-boundary 관계 아님)
    layout_in = dict(layout)
    layout_in["cameras"] = [{"shot_index": 1, "pos": [45, 50], "look_at": [55, 50]}]
    brief_in = compute_camera_brief(layout_in["cameras"][0], layout_in)
    assert brief_in is not None
    assert "inside the mapped area 'walled room'" in brief_in
    assert "OUTSIDE" not in brief_in

    # figure 가 폴리곤 밖 → 관계 서술 없음
    layout_out = dict(layout)
    layout_out["figures"] = [
        {"figure_id": "a", "label": "A", "entity_token": "C01",
         "positions": [{"shot_index": 1, "pos": [30, 70], "moving_toward": None}]}]
    brief_out = compute_camera_brief(layout_out["cameras"][0], layout_out)
    assert brief_out is not None
    assert "mapped area" not in brief_out


def test_single_shot_candidates_broad_includes_no_figure():
    """3-stage 필수화(2026-07-02): broad lane 은 figure 0 layout 샷도 후보로 올린다
    (establishing/구조 샷 — 필요성 판정은 judge). broad=False 는 기존 제외 유지."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        single_shot_complexity_candidates, shot_hint_key,
    )
    layout = {
        "landmarks": [{"id": "rd", "label": "road", "kind": "line",
                       "points": [[10, 50], [90, 50]]}],
        "figures": [],
        "cameras": [{"shot_index": 4, "pos": [50, 10], "look_at": [50, 50]}]}
    mk = [(11, 4)]
    sk = {shot_hint_key(11, 4)}
    broad = single_shot_complexity_candidates(layout, mk, sk, broad=True)
    assert len(broad) == 1
    assert broad[0]["anchor_key"] == (11, 4)
    assert broad[0]["per_shot"][shot_hint_key(11, 4)]["entity_count"] == 0
    assert single_shot_complexity_candidates(layout, mk, sk, broad=False) == []


def test_build_clean_birdseye_spec_no_labels_and_shot_filter():
    """birdseye spec = 모델 입력 — 인간 가독 라벨/이름 0 (텍스트 없음 by construction),
    색은 area index 로 generic 배정, shot_index 필터(단일 카메라뷰)."""
    import json as _json

    from app.modules.pipeline.outdoor_site_layout_plan import build_clean_birdseye_spec

    layout = {
        "landmarks": [
            {"id": "lm1", "label": "secret place name", "kind": "area",
             "points": [[10, 10], [20, 10], [20, 20], [10, 20]]},
            {"id": "lm2", "label": "named road", "kind": "line",
             "points": [[0, 50], [100, 50]]},
        ],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C01",
             "positions": [
                 {"shot_index": 1, "pos": [30, 30], "moving_toward": None},
                 {"shot_index": 2, "pos": [40, 40], "moving_toward": None}]},
        ],
        "cameras": [
            {"shot_index": 1, "pos": [5, 5], "look_at": [50, 50]},
            {"shot_index": 2, "pos": [9, 9], "look_at": [50, 50]}],
    }
    spec = build_clean_birdseye_spec(layout, shot_index=1)
    blob = _json.dumps(spec)
    # 모델 입력 spec 에 라벨/이름/엔티티 토큰 누출 0
    assert "secret place name" not in blob and "named road" not in blob
    assert "label" not in blob and "C01" not in blob
    # shot 필터: shot 1 카메라/figure 만
    assert [c["shot_index"] for c in spec["cameras"]] == [1]
    assert all(f["shot_index"] == 1 for f in spec["figures"])
    # landmark 는 고정 장소 → 항상 전부
    assert len(spec["landmarks"]) == 2
    poly = [lm for lm in spec["landmarks"] if lm["shape"] == "polygon"][0]
    line = [lm for lm in spec["landmarks"] if lm["shape"] == "line"][0]
    assert "fill" in poly and "edge" in poly       # area = generic 채움색
    assert "fill" not in line                       # line = stroke 만


def test_render_clean_birdseye_png_valid():
    """spec → PNG 렌더(PIL) — 유효 PNG bytes (모델 입력용, 텍스트 호출 없음)."""
    from app.modules.pipeline.outdoor_site_layout_plan import build_clean_birdseye_spec
    from app.modules.pipeline.outdoor_site_layout_provider import (
        _render_clean_birdseye_png,
    )

    layout = {
        "landmarks": [
            {"id": "lm1", "label": "x", "kind": "area",
             "points": [[10, 10], [30, 10], [30, 30], [10, 30]]},
            {"id": "lm2", "label": "y", "kind": "line", "points": [[0, 60], [100, 60]]},
        ],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C01",
             "positions": [{"shot_index": 1, "pos": [40, 40],
                            "moving_toward": [80, 40]}]}],
        "cameras": [{"shot_index": 1, "pos": [5, 50], "look_at": [60, 50]}],
    }
    png = _render_clean_birdseye_png(build_clean_birdseye_spec(layout, shot_index=1))
    assert png[:8] == b"\x89PNG\r\n\x1a\n"
    assert len(png) >= 1024


def test_shared_model_off_uses_old_sketch_producer(monkeypatch, tmp_path):
    """flag OFF (default) — shared-model producer 미호출, 기존 마네킹 브리프/스케치
    경로 그대로 (manifest 에 brief_hash/sketch_prompt_hash, producer_kind 없음;
    diagnostics 에 shared-model 필드 없음 = byte-identical)."""
    from app.core.config import settings

    step, layout_fn, rewrite_fn = _departing_step_fixture(monkeypatch, tmp_path)
    monkeypatch.setattr(
        settings, "outdoor_composition_guide_shared_model_enabled", False,
        raising=False)
    step._guide_openai_client = object()
    calls = {"brief": 0, "sketch": 0, "shared": 0}

    def brief_fn(*a, **k):
        calls["brief"] += 1
        return "BRIEF"

    def sketch_fn(*a, **k):
        calls["sketch"] += 1
        return b"\x89PNGFAKE"

    def shared_fn(*a, **k):
        calls["shared"] += 1
        raise AssertionError("shared-model producer must NOT run when flag OFF")

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn, brief=brief_fn, sketch=sketch_fn,
        judge=_admit_judge, shared_guide=shared_fn)
    out = step._execute()

    assert calls == {"brief": 1, "sketch": 1, "shared": 0}
    g = out["data"]["composition_guides"]["1:1"]
    assert g["mode"] == "sketch"
    assert "brief_hash" in g and "sketch_prompt_hash" in g
    assert "producer_kind" not in g
    assert "composition_guide_shared_model_enabled" not in out["data"]["diagnostics"]


# ───── shared-model v2 (라벨 마커 항공뷰 재배선, 2026-06-29) ─────


def _shared_model_step_fixture(monkeypatch, tmp_path):
    """2-shot same-scene cross-shot 그룹(L01, S1 sh1+sh2, DISTINCT 카메라, 2 figs +
    공통 landmarks) — shared-model v2 ON 경로 테스트 공용. (step, layout_fn, rewrite_fn)."""
    from app.core.config import settings
    from app.core.steps.outdoor_site_layout_step import OutdoorSiteLayoutStep

    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "outdoor_site_layout_enabled", True, raising=False)
    monkeypatch.setattr(
        settings, "outdoor_composition_guide_enabled", True, raising=False)
    monkeypatch.setattr(
        settings, "outdoor_composition_guide_shared_model_enabled", True,
        raising=False)

    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 1, "description": "d1"},
            {"shot_index": 2, "description": "d2"}]}]})
    _write_cp(tmp_path, "shot_selection", {"scenes": [
        {"scene_index": 1, "selected_shot_indices": [1, 2]}]})
    _write_cp(tmp_path, "background_classify", {"building_groups": [
        {"members": [{"loc_id": "L01", "is_indoor": False}]}]})
    _write_cp(tmp_path, "scene_save", {"segments": [{"scene_index": 1, "text": "t1"}]})
    _write_cp(tmp_path, "shot_director", {"scenes": [
        {"scene_index": 1, "shots": [
            {"shot_index": 1, "visible_entity_ids": ["C06", "C09", "L01"]},
            {"shot_index": 2, "visible_entity_ids": ["C06", "C09", "L01"]}]}]})
    _write_cp(tmp_path, "shot_staging", {"shots": [
        {"scene_index": 1, "shot_index": 1,
         "character_angles": [{"character": "a"}, {"character": "b"}]},
        {"scene_index": 1, "shot_index": 2,
         "character_angles": [{"character": "a"}, {"character": "b"}]}]})
    _write_cp(tmp_path, "scene_detail", {"scenes": [
        {"scene_index": 1, "_shot_index": 1,
         "t2i_variations": [{"t2i_prompt": "C06 and C09 by the road"}]},
        {"scene_index": 1, "_shot_index": 2,
         "t2i_variations": [{"t2i_prompt": "C06 and C09 farther down the road"}]}]})

    step = OutdoorSiteLayoutStep.__new__(OutdoorSiteLayoutStep)
    step.step_id = "outdoor_site_layout"
    step.project_id = "p1"
    step.episode_id = "e1"
    step.project_config = {}
    step.build_opik_metadata = lambda **kw: {}
    step._guide_openai_client = object()

    layout = {
        "landmarks": [
            {"id": "rd", "label": "road", "kind": "line",
             "points": [[10, 50], [90, 50]]},
            {"id": "sh", "label": "shelter", "kind": "area",
             "points": [[40, 40], [60, 40], [60, 60], [40, 60]]}],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C06",
             "positions": [{"shot_index": 1, "pos": [30, 30], "moving_toward": None},
                           {"shot_index": 2, "pos": [30, 30], "moving_toward": None}]},
            {"figure_id": "b", "label": "B", "entity_token": "C09",
             "positions": [{"shot_index": 1, "pos": [70, 60], "moving_toward": [90, 80]},
                           {"shot_index": 2, "pos": [70, 60], "moving_toward": [90, 80]}]}],
        # DISTINCT 카메라 (pos 다름) — cross-shot 후보 성립.
        "cameras": [{"shot_index": 1, "pos": [10, 10], "look_at": [50, 50]},
                    {"shot_index": 2, "pos": [90, 10], "look_at": [50, 50]}],
    }

    def layout_fn(scene_texts, member_shots, **kw):
        return layout

    def rewrite_fn(variations, summary, **kw):
        return {i: {"revised_prompt": p + " revised", "edited_spans": [],
                    "no_edit_reason": None} for i, p in variations}

    return step, layout_fn, rewrite_fn


def _route_admit(group_payload, *, model, **kw):
    """shared-model route judge 통과 stub — cross_shot_continuity + evidence non-empty."""
    return {
        "needs_shared_model_guide": True,
        "decision_type": "cross_shot_continuity",
        "confidence": "high",
        "evidence": [{"shot_key": "1:1", "source_field": "signals",
                      "quote": "same set, two angles"}],
        "reasons": ["multi_angle_same_set"],
        "guide_scope": "group", "risk_notes": "",
    }


def test_shared_model_candidate_cross_shot_only():
    """cross-shot candidate (순수) — single-shot/identical-camera 제외, 2-shot
    distinct-camera 같은 scene 만 후보. signals 에 이름/토큰 누출 0."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        shared_model_candidate_groups, shot_hint_key,
    )
    layout = {
        "landmarks": [{"id": "rd", "label": "road", "kind": "line",
                       "points": [[10, 50], [90, 50]]}],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C06", "positions": [
                {"shot_index": 1, "pos": [30, 30], "moving_toward": None},
                {"shot_index": 2, "pos": [30, 30], "moving_toward": None}]},
            {"figure_id": "b", "label": "B", "entity_token": "C09", "positions": [
                {"shot_index": 1, "pos": [70, 60], "moving_toward": [90, 80]},
                {"shot_index": 2, "pos": [70, 60], "moving_toward": [90, 80]}]}],
        "cameras": [{"shot_index": 1, "pos": [10, 10], "look_at": [50, 50]},
                    {"shot_index": 2, "pos": [90, 10], "look_at": [50, 50]}]}
    mk = [(1, 1), (1, 2)]
    sk = {shot_hint_key(s, h) for s, h in mk}
    cands = shared_model_candidate_groups(layout, mk, sk)
    assert len(cands) == 1
    assert cands[0]["anchor_key"] == (1, 1)
    assert cands[0]["cameras_non_identical"] is True
    # signals 에 entity 토큰 누출 0
    assert "C06" not in json.dumps(cands) and "C09" not in json.dumps(cands)
    # single-shot → 후보 아님 (cross-shot only, single 복잡도는 코드 확정 X)
    assert shared_model_candidate_groups(layout, [(1, 1)], {shot_hint_key(1, 1)}) == []
    # identical 카메라(non-identical geometry 아님) → 후보 아님
    lay2 = json.loads(json.dumps(layout))
    lay2["cameras"][1] = {"shot_index": 2, "pos": [10, 10], "look_at": [50, 50]}
    assert shared_model_candidate_groups(lay2, mk, sk) == []


def test_single_shot_complexity_candidate_multiple_figures():
    """★사용자 결정(2026-07-01): 단일 복잡샷도 후보. multiple figures → 단일 후보 생성,
    cross-shot 그룹 멤버는 exclude, 토큰 누출 0."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        single_shot_complexity_candidates, shot_hint_key,
    )
    layout = {
        "landmarks": [{"id": "rd", "label": "road", "kind": "line",
                       "points": [[10, 50], [90, 50]]}],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C06", "positions": [
                {"shot_index": 1, "pos": [30, 30], "moving_toward": None}]},
            {"figure_id": "b", "label": "B", "entity_token": "C09", "positions": [
                {"shot_index": 1, "pos": [70, 60], "moving_toward": None}]}],
        "cameras": [{"shot_index": 1, "pos": [10, 10], "look_at": [50, 50]}]}
    mk = [(1, 1)]
    sk = {shot_hint_key(1, 1)}
    cands = single_shot_complexity_candidates(layout, mk, sk)
    assert len(cands) == 1
    assert cands[0]["anchor_key"] == (1, 1)
    assert cands[0]["member_keys"] == [(1, 1)]
    assert cands[0]["is_single_shot"] is True
    assert "C06" not in json.dumps(cands) and "C09" not in json.dumps(cands)
    # exclude_keys (cross-shot 그룹에 이미 든 샷) → 제외
    assert single_shot_complexity_candidates(
        layout, mk, sk, exclude_keys={(1, 1)}) == []


def test_single_shot_complexity_candidate_simple_shot_excluded():
    """단일 figure + depth 단층 + 무모션 = 단순 → 후보 아님(LLM 비용 절약 pre-filter)."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        single_shot_complexity_candidates, shot_hint_key,
    )
    layout = {
        "landmarks": [{"id": "rd", "label": "road", "kind": "line",
                       "points": [[10, 50], [90, 50]]}],
        "figures": [
            {"figure_id": "a", "label": "A", "entity_token": "C06", "positions": [
                {"shot_index": 1, "pos": [50, 50], "moving_toward": None}]}],
        "cameras": [{"shot_index": 1, "pos": [50, 10], "look_at": [50, 50]}]}
    mk = [(1, 1)]
    sk = {shot_hint_key(1, 1)}
    cands = single_shot_complexity_candidates(layout, mk, sk)
    # 1 figure, 단일 depth, 무모션 → 복잡 신호 없음 → 후보 아님
    assert cands == []


def test_evaluate_shared_model_judge_opens_single_shot():
    """★사용자 결정(2026-07-01): _evaluate 가 single_shot_complexity 도 attach 허용
    (실내와 동일 lane). no_guide/저신뢰/근거없음은 여전히 deny."""
    from app.core.steps.outdoor_site_layout_step import (
        _evaluate_shared_model_guide_judge,
    )
    ev = [{"shot_key": "1_1", "source_field": "signals", "quote": "two figures"}]
    # single_shot_complexity + 근거 + medium/high → attach
    admit, reason = _evaluate_shared_model_guide_judge({
        "needs_shared_model_guide": True, "decision_type": "single_shot_complexity",
        "confidence": "high", "evidence": ev})
    assert admit is True and reason is None
    # cross_shot_continuity 도 그대로 attach
    admit2, _ = _evaluate_shared_model_guide_judge({
        "needs_shared_model_guide": True, "decision_type": "cross_shot_continuity",
        "confidence": "medium", "evidence": ev})
    assert admit2 is True
    # no_guide → deny
    admit3, r3 = _evaluate_shared_model_guide_judge({
        "needs_shared_model_guide": True, "decision_type": "no_guide",
        "confidence": "high", "evidence": ev})
    assert admit3 is False and r3 == "judge_no_guide_decision"
    # single_shot + 근거 없음 → deny
    admit4, r4 = _evaluate_shared_model_guide_judge({
        "needs_shared_model_guide": True, "decision_type": "single_shot_complexity",
        "confidence": "high", "evidence": []})
    assert admit4 is False and r4 == "judge_missing_evidence"


def test_shared_model_judge_schema_has_camera_or_framing_risk():
    """4d B' (2026-07-01): judge schema reasons enum 에 camera_or_framing_risk 추가 —
    극단 카메라/framing directive 단독으로도 single_shot guide 정당화(코드 파싱 X,
    judge 계층만 의미판정)."""
    from app.modules.pipeline.outdoor_site_layout_provider import (
        SHARED_MODEL_GUIDE_JUDGE_SCHEMA,
    )
    reasons_enum = (SHARED_MODEL_GUIDE_JUDGE_SCHEMA["properties"]["reasons"]
                    ["items"]["enum"])
    assert "camera_or_framing_risk" in reasons_enum
    # 기존 reason 도 보존(회귀 방지)
    assert "occlusion_or_framing_risk" in reasons_enum


def test_evaluate_admits_camera_or_framing_risk_single_shot():
    """B' attach gate: camera_or_framing_risk 근거의 single_shot verdict 도
    _evaluate 가 attach 허용(needs+decision+confidence+grounded evidence 충족)."""
    from app.core.steps.outdoor_site_layout_step import (
        _evaluate_shared_model_guide_judge,
    )
    ev = [{"shot_key": "19:5", "source_field": "camera_direction",
           "quote": "the camera looks steeply down"}]
    admit, reason = _evaluate_shared_model_guide_judge({
        "needs_shared_model_guide": True, "decision_type": "single_shot_complexity",
        "confidence": "high", "reasons": ["camera_or_framing_risk"], "evidence": ev})
    assert admit is True and reason is None


def test_single_shot_evidence_target_guard():
    """★Codex NARROW 1: single-shot judge evidence 가 후보 anchor 샷을 가리킬 때만 통과.
    엉뚱한 shot_key evidence 면 False(wrong single-shot guide 차단)."""
    from app.core.steps.outdoor_site_layout_step import (
        _single_shot_target_in_evidence,
    )
    ev = [{"shot_key": "13:3", "source_field": "signals", "quote": "two figures"}]
    assert _single_shot_target_in_evidence(ev, "13:3") is True
    # 다른 shot_key → 차단
    assert _single_shot_target_in_evidence(ev, "13:4") is False
    # 빈/비list → 차단
    assert _single_shot_target_in_evidence([], "13:3") is False
    assert _single_shot_target_in_evidence(None, "13:3") is False
    assert _single_shot_target_in_evidence(
        [{"shot_key": "   ", "source_field": "x", "quote": "y"}], "13:3") is False


def _usable_pose(*a, **k):
    """P1 테스트용 fake pose brief — usable figure 1개(자세 확정 high conf).
    실제 generate_pose_brief 는 LLM 호출이라 결정론 테스트는 override 주입."""
    return {"figures": [{
        "slot": "the one leaving", "body_posture": "walking",
        "limb_action": "carrying a held object", "head_body_orientation": "away from camera",
        "contact_or_support": "ground", "interaction_target_role": "none",
        "evidence_quote": "walks off", "confidence": "high"}]}


def test_shared_model_on_uses_v2_producer_not_old(monkeypatch, tmp_path):
    """flag ON — anchor 샷은 shared-model v2(base→sketch) producer 로 생성. old
    candidate/G1·G2 judge/brief/sketch **절대 미호출**. manifest 에 producer_kind=
    shared_model_aerial_v2 + base/blocking/sketch hash + blocking_stage + base_cache_key.
    2번째 샷=continuity_anchor(producer 미호출). diagnostics 단계별 + image call count."""
    step, layout_fn, rewrite_fn = _shared_model_step_fixture(monkeypatch, tmp_path)
    calls = {"base": 0, "guide": 0, "old_brief": 0, "old_sketch": 0, "old_judge": 0}

    def base_fn(layout, *, openai_client, model, **k):
        calls["base"] += 1
        return b"\x89PNG\r\n\x1a\nBASE", {
            "base_prompt": "bp", "base_prompt_hash": "bph", "base_png_hash": "bnh",
            "base_prompt_version": "v2"}

    def guide_fn(layout, shot_index, *, base_png, use_blocking, openai_client,
                 model, **k):
        calls["guide"] += 1
        assert shot_index == 1                  # anchor = 첫 샷
        assert base_png == b"\x89PNG\r\n\x1a\nBASE"   # 캐싱된 base 전달
        assert use_blocking is True             # 다수 figure → 블로킹
        return b"\x89PNG\r\n\x1a\nSKETCH", {
            "camera_brief_hash": "cbh", "blocking_prompt_hash": "blkph",
            "blocking_png_hash": "blknh", "sketch_prompt_hash": "skph",
            "camera_view_sketch_hash": "cvsh", "helper_version": "hv"}

    def old_brief(*a, **k):
        calls["old_brief"] += 1
        raise AssertionError("ON path must NOT call old brief")

    def old_sketch(*a, **k):
        calls["old_sketch"] += 1
        raise AssertionError("ON path must NOT call old sketch")

    def old_judge(*a, **k):
        calls["old_judge"] += 1
        raise AssertionError("ON path must NOT call old G1/G2 judge")

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn, brief=old_brief, sketch=old_sketch,
        judge=old_judge, shared_base=base_fn, shared_guide=guide_fn,
        shared_judge=_route_admit, pose_brief=_usable_pose)
    out = step._execute()

    assert out["failed_count"] == 0
    assert calls == {"base": 1, "guide": 1, "old_brief": 0, "old_sketch": 0,
                     "old_judge": 0}
    g = out["data"]["composition_guides"]["1:1"]
    assert g["mode"] == "sketch"
    assert g["producer_kind"] == "shared_model_aerial_v2"
    assert g["guide_generation_status"] == "ok"
    assert g["base_hash"] == "bnh"
    assert g["base_prompt_hash"] == "bph"
    assert g["blocking_stage"] == "used"
    assert g["blocking_hash"] == "blknh"
    assert g["camera_view_sketch_hash"] == "cvsh"
    assert g["helper_version"] == "hv"
    assert g["base_cache_key"]
    assert g["sketch_prompt_hash"] == "skph"   # v2 블로킹-스케치 프롬프트 해시
    assert "brief_hash" not in g               # old 마네킹 brief 필드 없음(v2 미사용)
    assert g["path"].endswith("S1sh1.png")
    assert (step._checkpoint_dir() / g["path"]).exists()
    # 2번째 샷 = continuity_anchor (producer 미호출 — anchor 1회만)
    g2 = out["data"]["composition_guides"]["1:2"]
    assert g2["mode"] == "continuity_anchor" and g2["anchor_source"] == [1, 1]
    assert "producer_kind" not in g2
    assert sorted(g2["forced_character_names"]) == ["a", "b"]
    diag = out["data"]["diagnostics"]
    assert diag["composition_guide_shared_model_enabled"] is True
    assert diag["shared_model_candidate_signals"]
    assert diag["shared_model_judge_decisions"][0]["admit"] is True
    assert diag["shared_model_attach_decisions"][0]["attached"] is True
    # base 1 + sketch(C) 1 + blocking(B) 1 = 3
    assert diag["shared_model_image_calls"]["actual_call_count"] == 3


def test_shared_model_v2_failure_no_guide_no_old_fallback(monkeypatch, tmp_path):
    """flag ON + shared-model 생성 실패 = no-guide + diagnostic, old sketch fallback
    절대 금지 (비차단). anchor 실패 → 그룹 전체 가이드 0."""
    step, layout_fn, rewrite_fn = _shared_model_step_fixture(monkeypatch, tmp_path)
    calls = {"old_sketch": 0}

    def old_sketch(*a, **k):
        calls["old_sketch"] += 1
        return b"OLD"

    def base_fn(*a, **k):
        return b"BASE", {"base_png_hash": "bnh", "base_prompt_hash": "bph"}

    def guide_fail(*a, **k):
        raise RuntimeError("shared-model boom")

    def old_judge(*a, **k):
        raise AssertionError("ON path must NOT call old G1/G2 judge")

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn, brief=lambda *a, **k: "B",
        sketch=old_sketch, judge=old_judge, shared_base=base_fn,
        shared_guide=guide_fail, shared_judge=_route_admit, pose_brief=_usable_pose)
    out = step._execute()

    assert out["failed_count"] == 0                        # 비차단
    assert out["data"]["composition_guides"] == {}         # 가이드 0
    assert calls["old_sketch"] == 0                        # old fallback 0
    failed = out["data"]["diagnostics"]["composition_guide_failed"]
    assert any(f.get("producer_kind") == "shared_model_aerial_v2" for f in failed)
    attach = out["data"]["diagnostics"]["shared_model_attach_decisions"][0]
    assert attach["attached"] is False


def test_shared_model_pose_fail_closed_skips_guide(monkeypatch, tmp_path):
    """P1 fail-closed — pose 추출 usable figure 0(전부 저신뢰/unknown)이면 이 그룹 가이드
    미부착(generic standing 마네킹 재발 방지). base/guide producer 미호출, 진단 기록."""
    step, layout_fn, rewrite_fn = _shared_model_step_fixture(monkeypatch, tmp_path)
    calls = {"base": 0, "guide": 0}

    def base_fn(*a, **k):
        calls["base"] += 1
        return b"BASE", {"base_png_hash": "bnh", "base_prompt_hash": "bph"}

    def guide_fn(*a, **k):
        calls["guide"] += 1
        return b"G", {}

    def unusable_pose(*a, **k):
        return {"figures": [{
            "slot": "x", "body_posture": "unknown", "limb_action": "none",
            "head_body_orientation": "", "contact_or_support": "none",
            "interaction_target_role": "none", "evidence_quote": "",
            "confidence": "low"}]}

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn, brief=lambda *a, **k: "B",
        sketch=lambda *a, **k: b"S", judge=lambda *a, **k: {},
        shared_base=base_fn, shared_guide=guide_fn, shared_judge=_route_admit,
        pose_brief=unusable_pose)
    out = step._execute()

    assert out["failed_count"] == 0
    assert out["data"]["composition_guides"] == {}       # fail-closed → attach 0
    assert calls == {"base": 0, "guide": 0}              # pose 게이트가 base 前 차단
    diag = out["data"]["diagnostics"]
    assert diag["shared_model_attach_decisions"][0]["attached"] is False
    assert any(p["shot"] == "S1sh1"
               for p in diag["shared_model_pose_fail_closed"])


def test_shared_model_judge_default_deny(monkeypatch, tmp_path):
    """route judge default-deny — needs+evidence 비움 / low conf / no_guide 면 attach
    안 함(candidate-only diagnostic), v2 producer 미호출.

    ★사용자 결정(2026-07-01): single_shot_complexity 단독은 더 이상 deny 아님(실내와 동일
    lane 개방). 따라서 deny 목록에서 제외 — 별도 admit 테스트로 커버."""
    deny_verdicts = [
        {"needs_shared_model_guide": True, "decision_type": "cross_shot_continuity",
         "confidence": "high", "evidence": [], "reasons": [], "guide_scope": "group",
         "risk_notes": ""},                                          # evidence empty
        {"needs_shared_model_guide": True, "decision_type": "cross_shot_continuity",
         "confidence": "high", "reasons": [], "guide_scope": "group", "risk_notes": "",
         "evidence": [{"shot_key": "1:1", "source_field": "signals",
                       "quote": "   "}]},                            # quote 공백만 → deny
        {"needs_shared_model_guide": True, "decision_type": "cross_shot_continuity",
         "confidence": "high", "reasons": [], "guide_scope": "group", "risk_notes": "",
         "evidence": [{"shot_key": "", "source_field": "",
                       "quote": "x"}]},                              # shot/source 빈값 → deny
        {"needs_shared_model_guide": True, "decision_type": "cross_shot_continuity",
         "confidence": "low", "reasons": [], "guide_scope": "group", "risk_notes": "",
         "evidence": [{"shot_key": "1:1", "source_field": "x", "quote": "y"}]},  # low conf
        {"needs_shared_model_guide": False, "decision_type": "no_guide",
         "confidence": "high", "evidence": [], "reasons": [], "guide_scope": "none",
         "risk_notes": ""},                                          # no need
    ]
    for verdict in deny_verdicts:
        step, layout_fn, rewrite_fn = _shared_model_step_fixture(monkeypatch, tmp_path)
        calls = {"base": 0, "guide": 0}

        def base_fn(*a, **k):
            calls["base"] += 1
            return b"B", {}

        def guide_fn(*a, **k):
            calls["guide"] += 1
            return b"G", {}

        step.set_overrides_for_testing(
            layout=layout_fn, rewrite=rewrite_fn, brief=lambda *a, **k: "B",
            sketch=lambda *a, **k: b"S", judge=lambda *a, **k: {},
            shared_base=base_fn, shared_guide=guide_fn,
            shared_judge=lambda gp, *, model, **k: verdict)
        out = step._execute()

        assert out["failed_count"] == 0
        assert out["data"]["composition_guides"] == {}    # attach 0
        assert calls == {"base": 0, "guide": 0}            # producer 미호출
        jd = out["data"]["diagnostics"]["shared_model_judge_decisions"][0]
        assert jd["admit"] is False


def test_shared_model_judge_disabled_candidate_only(monkeypatch, tmp_path):
    """judge enable 내부옵션 OFF — judge 미호출 + candidate-only diagnostic, attach 0."""
    from app.core.config import settings

    step, layout_fn, rewrite_fn = _shared_model_step_fixture(monkeypatch, tmp_path)
    monkeypatch.setattr(
        settings, "outdoor_shared_model_guide_judge_enabled", False, raising=False)
    calls = {"base": 0, "guide": 0, "judge": 0}

    def judge_fn(*a, **k):
        calls["judge"] += 1
        raise AssertionError("judge disabled — must NOT call route judge")

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn, brief=lambda *a, **k: "B",
        sketch=lambda *a, **k: b"S",
        shared_base=lambda *a, **k: (calls.__setitem__("base", calls["base"] + 1)
                                     or (b"B", {})),
        shared_guide=lambda *a, **k: (calls.__setitem__("guide", calls["guide"] + 1)
                                      or (b"G", {})),
        shared_judge=judge_fn)
    out = step._execute()

    assert out["failed_count"] == 0
    assert out["data"]["composition_guides"] == {}
    assert calls == {"base": 0, "guide": 0, "judge": 0}
    diag = out["data"]["diagnostics"]
    assert diag["shared_model_candidate_signals"]            # 후보는 기록
    assert diag["shared_model_judge_decisions"][0]["decision"] == (
        "judge_disabled_diagnostic_only")


def test_composition_guide_judge_default_deny(monkeypatch, tmp_path):
    """의미 게이트 default-deny — judge 가 열린 외부 departure 아님/저신뢰/판정
    실패면 guide 미생성 + failed_count=0 + candidate_skipped 진단(reason), sketch
    미호출. (구조물 프레이밍 샷 = S10/S28 오선별 제거의 step 계약)."""
    step, layout_fn, rewrite_fn = _departing_step_fixture(monkeypatch, tmp_path)

    def brief_fn(scene_action, summary, *, model, **kw):
        raise AssertionError("brief/sketch must not run when judge denies")

    def sketch_fn(brief, *, openai_client, model, **kw):
        raise AssertionError("brief/sketch must not run when judge denies")

    def deny_judge(scene_action, camera_direction, summary, *, model, **kw):
        return {
            "is_open_exterior_departure": False, "confidence": "high",
            "reasoning": "framed by a structural opening",
            "evidence_quote": "the window surface sits flat across the frame",
        }

    step.set_overrides_for_testing(layout=layout_fn, rewrite=rewrite_fn,
                                   brief=brief_fn, sketch=sketch_fn,
                                   judge=deny_judge)
    out = step._execute()

    assert out["failed_count"] == 0
    assert out["data"]["composition_guides"] == {}
    skipped = out["data"]["diagnostics"]["composition_guide_candidate_skipped"]
    assert skipped and skipped[0]["reason"] == "judge_not_open_departure"
    # text override(별개 경로)는 유지 = guide 부재가 기존 경로를 막지 않는다
    assert "1:1" in out["data"]["prompt_overrides"]


def test_composition_guide_judge_failure_is_non_blocking(monkeypatch, tmp_path):
    """judge 호출 자체가 예외 → skip(judge_failed) + 비차단(brief/sketch 미호출)."""
    step, layout_fn, rewrite_fn = _departing_step_fixture(monkeypatch, tmp_path)

    def boom_judge(scene_action, camera_direction, summary, *, model, **kw):
        raise RuntimeError("judge boom")

    def sketch_fn(brief, *, openai_client, model, **kw):
        raise AssertionError("sketch must not run when judge fails")

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn,
        brief=lambda *a, **k: "b", sketch=sketch_fn, judge=boom_judge)
    out = step._execute()

    assert out["failed_count"] == 0
    assert out["data"]["composition_guides"] == {}
    skipped = out["data"]["diagnostics"]["composition_guide_candidate_skipped"]
    assert skipped and skipped[0]["reason"] == "judge_failed"


def test_composition_guide_marker_load_attach(monkeypatch, tmp_path):
    """option C sketch 경로 — 첫 admitted 샷은 sketch marker(path) 기록 → loader 조인
    (절대경로 보강) → consumer attach 가 same-room bg 라벨 완화 + composition_guide
    스케치 ref 를 append. guide 없는 샷은 4-list 불변(byte-identical)."""
    import json as _json

    from app.core.steps.outdoor_site_layout_step import (
        attach_composition_guide_ref,
        load_composition_guide_context,
    )

    step, layout_fn, rewrite_fn = _departing_step_fixture(monkeypatch, tmp_path)
    step._guide_openai_client = object()
    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn,
        brief=lambda *a, **k: "BRIEF", sketch=lambda *a, **k: b"PNGFAKE",
        judge=_admit_judge)
    out = step._execute()

    guides = out["data"]["composition_guides"]
    assert "1:1" in guides
    assert guides["1:1"]["mode"] == "sketch"      # 첫 admitted = 스케치
    assert guides["1:1"]["path"].endswith("S1sh1.png")
    assert guides["1:1"]["judge"]["confidence"] == "high"

    # loader — cp manifest 조인 + path 절대경로 보강
    cp_dir = (tmp_path / "p1" / "checkpoints" / "episodes" / "e1"
              / "outdoor_site_layout")
    cp_dir.mkdir(parents=True, exist_ok=True)
    (cp_dir / "manifest.json").write_text(_json.dumps({
        "status": "completed", "data": out["data"],
    }), encoding="utf-8")
    ctx = load_composition_guide_context("p1", "e1")
    assert (1, 1) in ctx
    from pathlib import Path as _P
    assert _P(ctx[(1, 1)]["path"]).is_absolute()

    # flag OFF 면 빈 dict (byte-identical 소비)
    from app.core.config import settings
    monkeypatch.setattr(
        settings, "outdoor_composition_guide_enabled", False, raising=False)
    assert load_composition_guide_context("p1", "e1") == {}
    monkeypatch.setattr(
        settings, "outdoor_composition_guide_enabled", True, raising=False)

    # attach (sketch mode) — same-room bg 완화 + composition_guide ref append
    labeled_refs = [
        ("previous shot at same location (SAME ROOM) — use this background as-is.",
         b"prev-frame-bytes"),
    ]
    ref_roles = ["previous_shot_same_room"]
    ref_role_metadata = [{"keep_elements": [], "ignore": "",
                          "ref_usage": "exact_background"}]
    attached_meta = [("background_prev_shot", "L01")]
    attached = attach_composition_guide_ref(
        labeled_refs, ref_roles, ref_role_metadata, attached_meta,
        scene_index=1, shot_index=1, guide_ctx=ctx,
    )
    assert attached is True
    # composition_guide 스케치 ref 가 마지막에 append, same-room 은 완화 stamp
    assert ref_roles == ["previous_shot_same_room", "composition_guide"]
    assert attached_meta[-1] == ("composition_guide", "1:1")
    assert labeled_refs[-1][1] == b"PNGFAKE"
    assert ref_role_metadata[0]["composition_relaxed"] is True
    assert labeled_refs[0][1] == b"prev-frame-bytes"  # bg bytes 보존

    # guide 없는 샷 → no-op (4-list 불변)
    before = (list(labeled_refs), list(ref_roles), list(attached_meta))
    assert attach_composition_guide_ref(
        labeled_refs, ref_roles, ref_role_metadata, attached_meta,
        scene_index=2, shot_index=9, guide_ctx=ctx,
    ) is False
    assert (labeled_refs, ref_roles, attached_meta) == before


def test_composition_continuity_anchor_role_continuity_also_upgraded():
    """option C — role previous_shot_continuity 도 승격 대상 (same_room 외). 승격 시
    bytes 는 anchor_source 의 현재-run v2(resolver)로 명시 교체된다."""
    from app.core.steps.outdoor_site_layout_step import (
        COMPOSITION_CONTINUITY_ANCHOR_LABEL, attach_composition_guide_ref,
    )
    labeled_refs = [("env ref", b"stale-frame")]
    ref_roles = ["previous_shot_continuity"]
    ref_role_metadata = [{"ref_usage": "atmosphere_reference"}]
    attached_meta = [("background_prev_shot", "L01")]
    ok = attach_composition_guide_ref(
        labeled_refs, ref_roles, ref_role_metadata, attached_meta,
        scene_index=1, shot_index=2,
        guide_ctx={(1, 2): {"mode": "continuity_anchor",
                            "anchor_source": [1, 1], "location_id": "L01"}},
        source_bytes_resolver=lambda k: b"source-v2" if k == (1, 1) else None,
    )
    assert ok is True
    # bytes 가 stale 가 아닌 source v2 로 명시 교체
    assert labeled_refs[0] == (COMPOSITION_CONTINUITY_ANCHOR_LABEL, b"source-v2")
    assert ref_role_metadata[0]["composition_continuity_anchor"] is True
    assert ref_role_metadata[0]["anchor_source"] == [1, 1]
    assert ref_roles == ["previous_shot_continuity"]


def test_composition_continuity_anchor_noop_when_source_bytes_absent():
    """option C 핵심 — anchor_source 의 현재-run bytes 가 없으면 stale fallback 없이
    no-op (False, 4-list 불변). '현재-run v2 source' 계약 — stale DB primary 금지."""
    from app.core.steps.outdoor_site_layout_step import attach_composition_guide_ref
    labeled_refs = [("prev frame", b"stale")]
    ref_roles = ["previous_shot_same_room"]
    ref_role_metadata = [{"ref_usage": "exact_background"}]
    attached_meta = [("background_prev_shot", "L01")]
    before = (list(labeled_refs), list(ref_roles),
              [dict(m) for m in ref_role_metadata], list(attached_meta))
    ok = attach_composition_guide_ref(
        labeled_refs, ref_roles, ref_role_metadata, attached_meta,
        scene_index=1, shot_index=2,
        guide_ctx={(1, 2): {"mode": "continuity_anchor", "anchor_source": [1, 1]}},
        source_bytes_resolver=lambda k: None,  # source 아직 미생성
    )
    assert ok is False
    assert labeled_refs == before[0]          # stale bytes 그대로 (승격 안 함)
    assert ref_roles == before[1]
    assert ref_role_metadata == before[2]
    assert attached_meta == before[3]


def test_composition_continuity_anchor_skips_zoomed_role():
    """Codex 가드 — previous_shot_same_frame_zoomed(zoom crop)는 건드리지 않음
    (zoom_continuity 와 충돌 방지). upgrade/relax/append 모두 안 함."""
    from app.core.steps.outdoor_site_layout_step import attach_composition_guide_ref
    labeled_refs = [("zoom frame", b"zframe")]
    ref_roles = ["previous_shot_same_frame_zoomed"]
    ref_role_metadata = [{"ref_usage": "zoom_in_detail"}]
    attached_meta = [("background_prev_shot", "L01")]
    before = (list(labeled_refs), list(ref_roles),
              [dict(m) for m in ref_role_metadata], list(attached_meta))
    # anchor_source + source bytes 가 있어도 zoom crop ref 만 있으면 no-op
    # (zoom_continuity 충돌 회피 — append 도 안 함).
    ok = attach_composition_guide_ref(
        labeled_refs, ref_roles, ref_role_metadata, attached_meta,
        scene_index=1, shot_index=2,
        guide_ctx={(1, 2): {"mode": "continuity_anchor", "anchor_source": [1, 1]}},
        source_bytes_resolver=lambda k: b"source-v2",
    )
    assert ok is False
    assert labeled_refs == before[0]
    assert ref_roles == before[1]
    assert "composition_continuity_anchor" not in ref_role_metadata[0]
    assert attached_meta == before[3]


def test_composition_continuity_anchor_appends_when_no_prev_ref():
    """option C — 승격할 prev-shot same-place ref 가 없으면(그리고 zoom crop ref 도
    없으면) anchor_source 의 현재-run v2 를 background_prev_shot 로 append (Codex:
    '없으면 삽입'). character ref 만 있던 4-list 에 continuity ref 1장 추가."""
    from app.core.steps.outdoor_site_layout_step import (
        COMPOSITION_CONTINUITY_ANCHOR_LABEL, attach_composition_guide_ref,
    )
    labeled_refs = [("character ref", b"char")]  # prev-shot ref 없음
    ref_roles = ["character_ref"]
    ref_role_metadata = [{}]
    attached_meta = [("character", "C09")]
    ok = attach_composition_guide_ref(
        labeled_refs, ref_roles, ref_role_metadata, attached_meta,
        scene_index=1, shot_index=2,
        guide_ctx={(1, 2): {"mode": "continuity_anchor",
                            "anchor_source": [1, 1], "location_id": "L01"}},
        source_bytes_resolver=lambda k: b"source-v2",
    )
    assert ok is True
    assert labeled_refs[-1] == (COMPOSITION_CONTINUITY_ANCHOR_LABEL, b"source-v2")
    assert ref_roles[-1] == "previous_shot_same_room"
    assert attached_meta[-1] == ("background_prev_shot", "L01")
    assert ref_role_metadata[-1]["composition_continuity_anchor"] is True
    # 기존 character ref 는 보존
    assert labeled_refs[0] == ("character ref", b"char")


def test_composition_continuity_anchor_prompt_instructions():
    """prompt_service — composition_continuity_anchor 메타가 연속성 지시를 발화:
    장소·인물 배치 연속성 + '이 샷의 카메라/POV/crop 은 본문 SOT(prompt wins)'.
    same_room / continuity 양 role 모두. (sh7 의 고유 프레이밍 보존 계약)."""
    from app.services.prompt_service import (
        make_labeled_ref_payload, resolve_ref_roles,
    )
    for role in ("previous_shot_same_room", "previous_shot_continuity"):
        payload = make_labeled_ref_payload(
            labeled_refs=[("prev frame", b"png")],
            ref_roles=[role],
            ref_role_metadata=[{"composition_continuity_anchor": True}],
            attached_meta=[("background_prev_shot", "L01")],
        )
        res = resolve_ref_roles(payload)
        joined = " ".join(res.ref_roles + res.ref_instructions).lower()
        assert "continuity" in joined
        assert "prompt text wins" in joined          # framing 은 본문 SOT
        # '연속성 예외' 절 — composition 복사 금지의 명시 예외
        assert "one exception to the 'do not copy compositions'" in joined


def test_composition_continuity_anchor_preserves_background_contract():
    """재배선 v2 ref_contract 불변 (Codex 가드) — 연속성 anchor 승격은
    attached_meta=background_prev_shot 를 그대로 유지하고 role/길이를 바꾸지 않으므로,
    prev_shot lineage 가 required background 를 계속 충족한다 (composition_guide 라는
    별도 attached_meta 를 만들지 않는다 = 계약 검증 불변)."""
    from app.core.ref_contract_validator import validate_attached_refs
    from app.core.steps.outdoor_site_layout_step import attach_composition_guide_ref

    labeled_refs = [("prev frame", b"png")]
    ref_roles = ["previous_shot_same_room"]
    ref_role_metadata = [{"ref_usage": "exact_background"}]
    attached_meta = [("background_prev_shot", "L01")]
    ok = attach_composition_guide_ref(
        labeled_refs, ref_roles, ref_role_metadata, attached_meta,
        scene_index=1, shot_index=2,
        guide_ctx={(1, 2): {"mode": "continuity_anchor",
                            "anchor_source": [1, 1], "location_id": "L01"}},
        source_bytes_resolver=lambda k: b"source-v2",
    )
    assert ok is True
    # composition_guide attached_meta 는 생기지 않는다 (role 신규 0 — 기존 슬롯 교체)
    assert attached_meta == [("background_prev_shot", "L01")]
    assert all(m[0] != "composition_guide" for m in attached_meta)
    # prev_shot lineage 가 required background(L01) 를 충족 — validator 통과
    rpc = {"asset_requirements": {"required_refs": {"background": ["L01"]}}}
    validate_attached_refs(
        rpc, labeled_refs, attached_meta, "prompt text",
        False, chain_bg_lookup=lambda bg_id: "L01",
        reference_phrase_kinds=[],
    )


# ── P1(2026-07-01) shared-model v2 Stage C 포즈 복원 — 결정론 계약 ──
# LLM pose 추출 품질은 canary/육안 gate; 여기서는 렌더/degrade/계층 계약만 잠근다.


def test_render_pose_brief_text_filters_and_formats():
    """usable figure(자세 확정 + confidence medium/high)만 텍스트로 렌더한다."""
    pose = {"figures": [
        {"slot": "the seated one", "body_posture": "crouching",
         "limb_action": "covering the face", "head_body_orientation": "away from camera",
         "contact_or_support": "ground", "interaction_target_role": "none",
         "evidence_quote": "crouches down", "confidence": "high"},
        # low confidence → 제외
        {"slot": "the other", "body_posture": "standing", "limb_action": "none",
         "head_body_orientation": "", "contact_or_support": "none",
         "interaction_target_role": "none", "evidence_quote": "", "confidence": "low"},
        # unknown posture → 제외
        {"slot": "ghost", "body_posture": "unknown", "limb_action": "none",
         "head_body_orientation": "", "contact_or_support": "none",
         "interaction_target_role": "none", "evidence_quote": "", "confidence": "high"},
    ]}
    out = render_pose_brief_text(pose)
    assert out is not None
    lines = out.splitlines()
    assert len(lines) == 1  # usable 1개만
    line = lines[0]
    assert "the seated one is crouching" in line
    assert "covering the face" in line
    assert "resting on/against ground" in line
    assert "facing away from camera" in line
    # 배제된 figure 는 흔적 없음
    assert "the other" not in out and "ghost" not in out


def test_render_pose_brief_text_fail_closed_returns_none():
    """usable figure 0(빈/전부 저신뢰/unknown) → None (step fail-closed)."""
    assert render_pose_brief_text(None) is None
    assert render_pose_brief_text({"figures": []}) is None
    assert render_pose_brief_text({"figures": [
        {"slot": "x", "body_posture": "unknown", "limb_action": "none",
         "head_body_orientation": "", "contact_or_support": "none",
         "interaction_target_role": "none", "evidence_quote": "", "confidence": "low"},
    ]}) is None


def test_build_blocking_sketch_prompt_pose_vs_featureless():
    """pose_brief 있으면 posed mannequin 경로, 없으면 기존 featureless(byte-identical)."""
    cam = "The eye-level camera view for shot 5."
    # pose 없음 → 기존 프롬프트(포즈 강제어 없음, featureless placeholder)
    plain = build_blocking_sketch_prompt(cam)
    assert "featureless mannequin/placeholder" in plain
    assert "BODY POSE AND ACTION" not in plain
    # pose 있음 → posed mannequin + 계층 + identity NOT SOT
    posed = build_blocking_sketch_prompt(cam, "- the seated one is crouching.")
    assert "BODY POSE AND ACTION" in posed
    assert "the seated one is crouching" in posed
    assert "posed EXACTLY in the posture and action" in posed
    assert "LOOMIS METHOD" in posed
    assert "does NOT define" in posed and "identity" in posed
    # camera brief 는 두 경로 모두에 포함(framing SOT)
    assert cam in plain and cam in posed
    # 빈 pose 는 featureless 로 degrade
    assert build_blocking_sketch_prompt(cam, "   ") == plain


def test_shared_model_optical_risk_shot_excluded(monkeypatch, tmp_path):
    """optical/reflective 구도 게이트(2026-07-02): judge 가 optical_risk_shot_keys 로
    지목한 샷은 그룹 admit 이어도 sketch/anchor 엔트리를 만들지 않는다(무-가이드
    baseline). 나머지 멤버는 정상 attach. 코드 텍스트 파싱 0 — judge 필드만."""
    from app.modules.pipeline.outdoor_site_layout_plan import shot_hint_key

    step, layout_fn, rewrite_fn = _shared_model_step_fixture(monkeypatch, tmp_path)

    def base_fn(layout, *, openai_client, model, **k):
        return b"\x89PNG\r\n\x1a\nBASE", {
            "base_prompt": "bp", "base_prompt_hash": "bph", "base_png_hash": "bnh",
            "base_prompt_version": "v3"}

    def guide_fn(layout, shot_index, *, base_png, use_blocking, openai_client,
                 model, **k):
        # 3-stage 필수화: blocking 은 항상 사용
        assert use_blocking is True
        return b"\x89PNG\r\n\x1a\nSKETCH", {
            "camera_brief_hash": "cbh", "blocking_prompt_hash": "blkph",
            "blocking_png_hash": "blknh", "sketch_prompt_hash": "skph",
            "camera_view_sketch_hash": "cvsh", "helper_version": "hv"}

    def judge_optical_sh2(group_payload, *, model, **kw):
        return {
            "needs_shared_model_guide": True,
            "decision_type": "cross_shot_continuity",
            "confidence": "high",
            "evidence": [{"shot_key": "1:1", "source_field": "signals",
                          "quote": "same set, two angles"}],
            "reasons": ["multi_angle_same_set"],
            "guide_scope": "group",
            "optical_risk_shot_keys": [shot_hint_key(1, 2)],
            "risk_notes": "shot 2 frames the subject as a reflection",
        }

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn,
        shared_base=base_fn, shared_guide=guide_fn,
        shared_judge=judge_optical_sh2, pose_brief=_usable_pose)
    out = step._execute()

    cg = out["data"]["composition_guides"]
    assert "1:1" in cg and cg["1:1"]["mode"] == "sketch"
    assert cg["1:1"]["blocking_stage"] == "used"
    assert cg["1:1"]["blocking_stage_reason"] == "mandatory_3stage"
    # optical 지목 샷은 어떤 모드의 엔트리도 없다
    assert "1:2" not in cg
    diag = out["data"]["diagnostics"]
    assert any(
        d.get("shot") == "S1sh2"
        for d in diag.get("shared_model_optical_risk_excluded", []))
    assert any(
        s.get("reason") == "optical_composition_risk" and s.get("shot") == "S1sh2"
        for s in diag.get("composition_guide_candidate_skipped", []))


def test_shared_model_optical_first_shot_rechains_anchor(monkeypatch, tmp_path):
    """★Codex 리뷰 NARROW(2026-07-02): 첫 admitted 샷이 optical risk 면 체인을
    eligible 멤버로 재구성 — 두 번째 샷이 sketch anchor 가 되고(anchor_source 없음),
    skip 샷을 가리키는 continuity_anchor 모순이 없어야 한다."""
    from app.modules.pipeline.outdoor_site_layout_plan import shot_hint_key

    step, layout_fn, rewrite_fn = _shared_model_step_fixture(monkeypatch, tmp_path)

    def base_fn(layout, *, openai_client, model, **k):
        return b"\x89PNG\r\n\x1a\nBASE", {
            "base_prompt": "bp", "base_prompt_hash": "bph", "base_png_hash": "bnh",
            "base_prompt_version": "v3"}

    def guide_fn(layout, shot_index, *, base_png, use_blocking, openai_client,
                 model, **k):
        assert use_blocking is True
        assert shot_index == 2          # 재구성된 anchor = 두 번째 샷
        return b"\x89PNG\r\n\x1a\nSKETCH", {
            "camera_brief_hash": "cbh", "blocking_prompt_hash": "blkph",
            "blocking_png_hash": "blknh", "sketch_prompt_hash": "skph",
            "camera_view_sketch_hash": "cvsh", "helper_version": "hv"}

    def judge_optical_sh1(group_payload, *, model, **kw):
        return {
            "needs_shared_model_guide": True,
            "decision_type": "cross_shot_continuity",
            "confidence": "high",
            "evidence": [{"shot_key": "1:2", "source_field": "signals",
                          "quote": "same set, two angles"}],
            "reasons": ["multi_angle_same_set"],
            "guide_scope": "group",
            "optical_risk_shot_keys": [shot_hint_key(1, 1)],
            "risk_notes": "shot 1 frames the subject as a reflection",
        }

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn,
        shared_base=base_fn, shared_guide=guide_fn,
        shared_judge=judge_optical_sh1, pose_brief=_usable_pose)
    out = step._execute()

    cg = out["data"]["composition_guides"]
    assert "1:1" not in cg
    assert cg["1:2"]["mode"] == "sketch"           # 두 번째 샷이 anchor 로 승격
    assert "anchor_source" not in cg["1:2"]
    diag = out["data"]["diagnostics"]
    assert any(d.get("shot") == "S1sh1"
               for d in diag.get("shared_model_optical_risk_excluded", []))


def test_shared_model_all_members_optical_no_guide(monkeypatch, tmp_path):
    """★Codex 리뷰 NARROW(2026-07-02): 전 멤버 optical risk → no-guide
    (attached=false, reason=all_members_optical_risk), 이미지 콜 0."""
    from app.modules.pipeline.outdoor_site_layout_plan import shot_hint_key

    step, layout_fn, rewrite_fn = _shared_model_step_fixture(monkeypatch, tmp_path)
    calls = {"base": 0, "guide": 0}

    def base_fn(layout, *, openai_client, model, **k):
        calls["base"] += 1
        return b"B", {"base_prompt": "bp", "base_prompt_hash": "bph",
                      "base_png_hash": "bnh", "base_prompt_version": "v3"}

    def guide_fn(layout, shot_index, **k):
        calls["guide"] += 1
        return b"S", {}

    def judge_optical_all(group_payload, *, model, **kw):
        return {
            "needs_shared_model_guide": True,
            "decision_type": "cross_shot_continuity",
            "confidence": "high",
            "evidence": [{"shot_key": "1:1", "source_field": "signals",
                          "quote": "reflections"}],
            "reasons": ["multi_angle_same_set"],
            "guide_scope": "group",
            "optical_risk_shot_keys": [shot_hint_key(1, 1), shot_hint_key(1, 2)],
            "risk_notes": "both shots are reflections",
        }

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn,
        shared_base=base_fn, shared_guide=guide_fn,
        shared_judge=judge_optical_all, pose_brief=_usable_pose)
    out = step._execute()

    cg = out["data"]["composition_guides"]
    assert "1:1" not in cg and "1:2" not in cg
    assert calls == {"base": 0, "guide": 0}
    diag = out["data"]["diagnostics"]
    ad = [a for a in diag.get("shared_model_attach_decisions", [])
          if a.get("fail_stage") == "all_members_optical_risk"]
    assert ad and ad[0]["attached"] is False
    assert any(s.get("reason") == "all_members_optical_risk"
               for s in diag.get("composition_guide_candidate_skipped", []))


def test_shared_model_staging_zero_figures_bypasses_pose(monkeypatch, tmp_path):
    """figure 0 바이패스 기준 = staging character_angles(텍스트 SOT).

    2차 canary 실측: layout LLM 이 staging 에 없는 figure 를 발명(entity_count 1)
    → layout 기준 바이패스는 pose 추출을 강제하고 텍스트 근거가 없어 항상
    fail-closed → establishing 샷 전멸. staging 0 이면 pose 추출을 걸지 않고
    featureless 스케치로 진행(pose_brief_present=False)해야 한다."""
    from app.core.config import settings

    step, layout_fn, rewrite_fn = _shared_model_step_fixture(monkeypatch, tmp_path)
    # staging 을 figure 0 으로 재작성(텍스트 SOT 상 인물 없는 establishing) —
    # broad single-shot lane 으로 seed 되도록 flag ON.
    monkeypatch.setattr(
        settings, "outdoor_single_shot_seed_lane_enabled", True, raising=False)
    _write_cp(tmp_path, "shot_staging", {"shots": [
        {"scene_index": 1, "shot_index": 1, "character_angles": []},
        {"scene_index": 1, "shot_index": 2, "character_angles": []}]})

    pose_calls = {"n": 0}

    def pose_fn(*a, **k):
        pose_calls["n"] += 1
        raise AssertionError("staging figure 0 — pose 추출이 호출되면 안 된다")

    def base_fn(layout, *, openai_client, model, **k):
        return b"\x89PNG\r\n\x1a\nBASE", {
            "base_prompt": "bp", "base_prompt_hash": "bph", "base_png_hash": "bnh",
            "base_prompt_version": "v3"}

    def guide_fn(layout, shot_index, *, base_png, use_blocking, openai_client,
                 model, pose_brief=None, **k):
        assert use_blocking is True
        assert pose_brief is None      # featureless 경로
        return b"\x89PNG\r\n\x1a\nSKETCH", {
            "camera_brief_hash": "cbh", "blocking_prompt_hash": "blkph",
            "blocking_png_hash": "blknh", "sketch_prompt_hash": "skph",
            "camera_view_sketch_hash": "cvsh", "helper_version": "hv"}

    step.set_overrides_for_testing(
        layout=layout_fn, rewrite=rewrite_fn,
        shared_base=base_fn, shared_guide=guide_fn,
        shared_judge=_route_admit, pose_brief=pose_fn)
    out = step._execute()

    cg = out["data"]["composition_guides"]
    assert cg["1:1"]["mode"] == "sketch"
    assert cg["1:1"]["pose_brief_present"] is False
    assert pose_calls["n"] == 0
    diag = out["data"]["diagnostics"]
    # staging 0 은 fail-closed 대상이 아님
    assert not any(
        d.get("shot") == "S1sh1"
        for d in diag.get("shared_model_pose_fail_closed", []))


# ───────── W-A/W-B (2026-07-03) — faces_toward 방향성 + 프레이밍 게이트 ─────────


def test_validate_site_layout_faces_toward_shape():
    """W-A: faces_toward 는 존재 시에만 shape 검증(point or null) — 부재는 구 layout
    하위호환(위반 아님), 무효 shape 만 violation."""
    from app.modules.pipeline.outdoor_site_layout_plan import validate_site_layout

    base = {
        "figures": [{"figure_id": "a", "label": "A", "entity_token": None,
                     "positions": [{"shot_index": 1, "pos": [50, 50],
                                    "moving_toward": None}]}],
        "cameras": [{"shot_index": 1, "pos": [10, 50], "look_at": [50, 50]}],
    }
    # null / 유효 point / 필드 부재 → 전부 통과
    ok = dict(base)
    ok["landmarks"] = [
        {"id": "l1", "label": "shelter", "kind": "area", "faces_toward": None,
         "points": [[40, 40], [60, 40], [60, 60]]},
        {"id": "l2", "label": "road", "kind": "line", "faces_toward": [50, 90],
         "points": [[0, 80], [100, 80]]},
        {"id": "l3", "label": "legacy", "kind": "point", "points": [[10, 10]]},
    ]
    assert validate_site_layout(ok) == []
    # 무효 shape → violation
    bad = dict(base)
    bad["landmarks"] = [
        {"id": "l1", "label": "shelter", "kind": "area", "faces_toward": "north",
         "points": [[40, 40], [60, 40], [60, 60]]},
    ]
    assert any("faces_toward" in v for v in validate_site_layout(bad))


def test_compute_camera_brief_faces_toward_four_directions():
    """W-A: faces_toward open_vec(=ft−centroid) 카메라 투영 → 지배 성분 4방향 서술.
    부재 landmark 는 방향구 생략(구 layout 하위호환). 순수 좌표 산술."""
    from app.modules.pipeline.outdoor_site_layout_plan import compute_camera_brief

    def _layout(ft):
        return {
            "landmarks": [
                {"id": "s", "label": "shelter box", "kind": "area",
                 "points": [[40, 45], [50, 45], [50, 55], [40, 55]],
                 "faces_toward": ft},
            ],
            "figures": [],
            "cameras": [{"shot_index": 1, "pos": [10, 50], "look_at": [60, 50]}],
        }

    # forward=+x, right=(0,-1): ft 가 카메라쪽(-x) → toward the camera
    b = compute_camera_brief(_layout([20, 50])["cameras"][0], _layout([20, 50]))
    assert "its open/front side faces toward the camera" in b
    b = compute_camera_brief(_layout([80, 50])["cameras"][0], _layout([80, 50]))
    assert "faces away from the camera" in b
    b = compute_camera_brief(_layout([45, 20])["cameras"][0], _layout([45, 20]))
    assert "faces toward the right of frame" in b
    b = compute_camera_brief(_layout([45, 80])["cameras"][0], _layout([45, 80]))
    assert "faces toward the left of frame" in b
    # faces_toward 부재/None → 방향구 없음
    lo = _layout(None)
    b = compute_camera_brief(lo["cameras"][0], lo)
    assert "open/front side" not in b


def test_sketch_systems_orientation_scale_invention_guards():
    """W-A: 스케치 SYSTEM 2종(featureless/posed)에 방향 준수·스케일 앵커·발명 억제
    강문 3종 포함 (S15sh1 방향 반전 / S11sh4 miniature / S15' 이중 벤치 재발 방지)."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        build_blocking_sketch_prompt,
    )

    for prompt in (
        build_blocking_sketch_prompt("BRIEF"),                       # featureless
        build_blocking_sketch_prompt("BRIEF", pose_brief="POSE"),    # posed
    ):
        assert "never mirror or flip" in prompt                      # 방향 준수
        assert "doorway reads as adult height" in prompt             # 스케일 앵커
        assert "miniature" in prompt
        assert "do NOT add" in prompt                                # 발명 억제
        assert "second copy of a listed structure" in prompt


def test_attach_composition_guide_framing_gate():
    """W-B: mode='sketch' + framing close/insert → attach skip(False, 4-list 불변).
    wide/None 은 기존 동작. continuity_anchor 는 framing 무관(별도 계약)."""
    from app.core.steps.outdoor_site_layout_step import attach_composition_guide_ref

    def _mk():
        return ([("prev", b"x")], ["previous_shot_same_room"],
                [{"ref_usage": "exact_background"}], [("background_prev_shot", "L01")])

    sketch_ctx = {(1, 2): {"mode": "sketch", "path": "/nonexistent/sketch.png"}}
    for framing in ("close", "insert"):
        lr, rr, rm, am = _mk()
        before = (list(lr), list(rr), [dict(m) for m in rm], list(am))
        ok = attach_composition_guide_ref(
            lr, rr, rm, am, scene_index=1, shot_index=2,
            guide_ctx=sketch_ctx, framing=framing)
        assert ok is False
        assert (lr, rr, rm, am) == before

    # wide → 게이트 통과 (path 부재로 False 지만 게이트 이후 분기 도달 확인은
    # framing=None 과 동일 결과 — 게이트가 wide 를 막지 않는다는 계약만 고정)
    lr, rr, rm, am = _mk()
    assert attach_composition_guide_ref(
        lr, rr, rm, am, scene_index=1, shot_index=2,
        guide_ctx=sketch_ctx, framing="wide") is False  # path 부재 사유(게이트 아님)

    # continuity_anchor 는 close 여도 승격 유지
    lr = [("env ref", b"stale")]
    rr = ["previous_shot_same_room"]
    rm = [{"ref_usage": "exact_background"}]
    am = [("background_prev_shot", "L01")]
    ok = attach_composition_guide_ref(
        lr, rr, rm, am, scene_index=1, shot_index=3,
        guide_ctx={(1, 3): {"mode": "continuity_anchor", "anchor_source": [1, 1]}},
        source_bytes_resolver=lambda k: b"source-v2", framing="close")
    assert ok is True
    assert lr[0][1] == b"source-v2"


def test_shot_spatial_summary_degenerate_zero_distance_no_crash():
    """E2E11 실측: 인물 좌표가 카메라와 동일(distance 0)한 저작에서
    상대크기 문장 나눗셈이 ZeroDivisionError — 보강 문장 생략 fail-safe."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        build_shot_spatial_summary,
    )

    layout = {
        "cameras": [{"shot_index": 1, "pos": [0.0, 0.0],
                     "look_at": [1.0, 0.0]}],
        "figures": [
            {"figure_id": "F1", "label": "runner",
             "positions": [{"shot_index": 1, "pos": [0.0, 0.0]}]},
            {"figure_id": "F2", "label": "watcher",
             "positions": [{"shot_index": 1, "pos": [0.0, 0.0]}]},
        ],
        "landmarks": [],
    }
    out = build_shot_spatial_summary(layout, 1)
    assert out is not None and "FIGURES:" in out
    assert "0.00x" not in out  # 왜곡 문장 미출력


def test_shot_spatial_summary_zero_nearest_positive_second_no_false_ratio():
    """Codex HIGH-3: nearest=0(카메라 좌표 인물)+second=10 — 임의 1m 기준
    비율("10.0x/0.10x")이 거짓 SOT 로 나가면 안 된다. 상대크기 문장 생략."""
    from app.modules.pipeline.outdoor_site_layout_plan import (
        build_shot_spatial_summary,
    )

    layout = {
        "cameras": [{"shot_index": 1, "pos": [0.0, 0.0],
                     "look_at": [1.0, 0.0]}],
        "figures": [
            {"figure_id": "F1", "label": "near",
             "positions": [{"shot_index": 1, "pos": [0.0, 0.0]}]},
            {"figure_id": "F2", "label": "far",
             "positions": [{"shot_index": 1, "pos": [10.0, 0.0]}]},
        ],
        "landmarks": [],
    }
    out = build_shot_spatial_summary(layout, 1)
    assert out is not None
    assert "as far from the camera" not in out  # 비율 문장 전체 생략
    assert "x the nearest figure" not in out
