"""space_set_bg focused tests — pure core 계약 + step opt-in/override 경로.

★주의(CLAUDE.md/메모리): LLM/VLM/T2I 출력 완성도는 TDD 로 못 올린다 — 여기서는
deterministic 한 것만 검증한다: 그룹핑/계획/마킹/프롬프트 조립/flag 게이트.
fixture 는 SAMPLE_FIXTURE_* generic (특정 시나리오 비의존)."""
from __future__ import annotations

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

import pytest

from app.modules.pipeline import space_set_bg as core


def _space(name, enclosure="enclosed", same=None, **kw):
    s = {
        "name": name, "enclosure": enclosure, "same_space_as": same,
        "construction_character": kw.get("construction"),
        "frame": {
            "boundaries": kw.get("boundaries") or ["wall"],
            "openings": kw.get("openings") or [{"name": f"{name} door", "kind": "door"}],
            "access_paths": kw.get("access") if "access" in kw else [{"via": "door", "description": "enter"}],
        },
        "addon": kw.get("addon") or [],
        "life_baseline": kw.get("life") or [],
    }
    return s


SAMPLE_FIXTURE_ANALYSIS: Dict[str, Any] = {
    "space_type": "SAMPLE_FIXTURE_unit",
    "genre_or_setting": "SAMPLE_FIXTURE_setting",
    "physical_realism": ["compact"],
    "spaces": [
        _space("hub room", addon=[{"role": "table_surface", "implementation": "small table"}],
               life=[{"item": "wall shelf", "reason": "daily use"}],
               construction="part of the original structure, plain finish"),
        _space("side room A"),
        _space("side room B", same="side room A"),
        _space("entry pocket", enclosure="threshold"),
        _space("open deck", enclosure="open_air", construction="added later, light materials"),
    ],
    "structure_kinds": [],
    "access_validation": [],
    "uncertain_structure_items": [],
    "excluded_items_with_reason": [],
}

SAMPLE_FIXTURE_BRIEFS = {
    "hub room": "wide view of hub room",
    "side room A": "wide view of side room A",
    "side room B": "wide view of side room B",
    "entry pocket": "narrow entry",
    "open deck": "open-air deck view",
}


# ──────────────────────── enclosure_groups ────────────────────────
def test_enclosure_groups_split_and_default():
    groups = core.enclosure_groups(SAMPLE_FIXTURE_ANALYSIS)
    assert list(groups.keys()) == ["indoor", "outdoor"]
    indoor_names = [s["name"] for s in groups["indoor"]]
    assert "entry pocket" in indoor_names          # threshold → indoor (connector)
    assert [s["name"] for s in groups["outdoor"]] == ["open deck"]
    # enclosure 미지정 → enclosed 취급
    g2 = core.enclosure_groups({"spaces": [{"name": "x"}]})
    assert list(g2.keys()) == ["indoor"]


# ──────────────────────── validate_fa ────────────────────────
def test_validate_fa_no_access_fail():
    a = {"spaces": [_space("sealed", openings=[], access=[])]}
    a["spaces"][0]["frame"]["openings"] = []
    a["spaces"][0]["frame"]["access_paths"] = []
    out = core.validate_fa(a)
    assert any("NO ACCESS" in f for f in out["_validation"]["fails"])


def test_validate_fa_cross_enclosure_same_space_fail():
    a = {"spaces": [
        _space("inside"),
        _space("outside view", enclosure="open_air", same="inside"),
    ]}
    out = core.validate_fa(a)
    assert any("CROSS-ENCLOSURE" in f for f in out["_validation"]["fails"])


def test_validate_fa_clean_pass():
    out = core.validate_fa(json.loads(json.dumps(SAMPLE_FIXTURE_ANALYSIS)))
    assert out["_validation"]["fails"] == []


# ──────────────────────── plan_space_views ────────────────────────
def test_plan_space_views_kinds_order_and_connectors():
    plans, connectors = core.plan_space_views(SAMPLE_FIXTURE_ANALYSIS, SAMPLE_FIXTURE_BRIEFS)
    by_room = {p["room"]: p for p in plans}
    # threshold = connector → plate 없음
    assert "entry pocket" not in by_room
    assert connectors and connectors[0]["room"] == "entry pocket"
    # kind 분기
    assert by_room["hub room"]["kind"] == "marked_indoor"
    assert by_room["open deck"]["kind"] == "outdoor_t2i"
    assert by_room["side room B"]["kind"] == "samespace_ref"
    assert by_room["side room B"]["ref_room"] == "side room A"
    # canonical(독립) 이 종속보다 먼저
    order = [p["room"] for p in plans]
    assert order.index("side room A") < order.index("side room B")
    # ★BG 한 장당 번호 1개 — num 은 plan 마다 고유
    nums = [p["num"] for p in plans]
    assert len(nums) == len(set(nums))
    # multi 그룹이면 indoor suffix 부여
    assert by_room["hub room"]["group_suffix"] == "_indoor"


def test_plan_space_views_single_group_no_suffix():
    a = {"spaces": [_space("only room")]}
    plans, _ = core.plan_space_views(a, {"only room": "v"})
    assert plans[0]["group_suffix"] == ""
    # ★단일 marked_indoor 그룹 = FP 무용(마킹할 다른 방 없음) → 단일 T2I 직행 (사용자 2026-06-10 fishing_boat 피드백)
    assert plans[0]["kind"] == "indoor_t2i"


def test_plan_space_views_single_marked_indoor_becomes_t2i():
    """어선 구조 (enclosed 1 + open_air 1): 조타실=indoor_t2i(FP 미경유), 갑판=outdoor_t2i.
    FP 직역 위험(도트 그리드→벽지, 심볼→평면 장식) 차단 — 실 canary 실측 결함."""
    a = {"spaces": [
        _space("steering cabin"),
        _space("work deck", enclosure="open_air"),
    ]}
    plans, _ = core.plan_space_views(a, {"steering cabin": "v1", "work deck": "v2"})
    by_room = {p["room"]: p for p in plans}
    assert by_room["steering cabin"]["kind"] == "indoor_t2i"
    assert by_room["work deck"]["kind"] == "outdoor_t2i"
    # 멀티 marked_indoor 그룹은 FP 유지 (배치 관계 가치 있음 — SAMPLE_FIXTURE 기존 테스트가 커버)
    plans2, _ = core.plan_space_views(SAMPLE_FIXTURE_ANALYSIS, SAMPLE_FIXTURE_BRIEFS)
    assert {p["room"]: p["kind"] for p in plans2}["hub room"] == "marked_indoor"


def test_indoor_bg_user_life_and_construction():
    sp = _space("steering cabin", life=[{"item": "fixed captain seat", "reason": "operation"}],
                construction="weathered FRP panels, salt-aged")
    a = {"spaces": [sp], "physical_realism": ["cramped"]}
    u = core.build_indoor_bg_user(a, sp, "a worn Korean boat cabin", "wide cabin view")
    assert "fixed captain seat" in u            # life_baseline 주입 (FP 미경유라 텍스트로)
    assert "weathered FRP panels" in u          # construction 재질 전달
    assert "a worn Korean boat cabin" in u and "wide cabin view" in u


# ──────────────────────── 프롬프트 조립 ────────────────────────
def test_build_marked_view_prompt_life_injection():
    p = core.build_marked_view_prompt(3, "a place", "a view", ["wall shelf", "fixed lamp"])
    assert "number 3" in p
    assert "wall shelf; fixed lamp" in p
    assert "NOT drawn in the plan" in p
    p2 = core.build_marked_view_prompt(1, "a place", "a view", [])
    assert "NOT drawn in the plan" not in p2          # life 없으면 extra 미주입


def test_marked_view_prompt_blocks_plan_notation_literalism():
    """★FP 직역 차단 (사용자 2026-06-10 fishing_boat 피드백): 도면 그래픽 표기≠물리 객체,
    심볼=위치·종류 지시(모양 직역 금지), 재질·양식 SOT=place 설명."""
    p = core.build_marked_view_prompt(2, "a place", "a view", [])
    low = p.lower()
    assert "drawing notation" in low                  # 그리드/도트/라벨 = 표기일 뿐
    assert "wallpaper" in low                         # 벽지/패턴 직역 금지 명시
    assert "symbol" in low and "three-dimensional" in low   # 심볼→실물 3D 번역
    assert "only for layout" in low                   # 도면=배치 SOT 한정, 재질·양식=place


def test_build_place_group_user_adjacent_masses():
    groups = core.enclosure_groups(SAMPLE_FIXTURE_ANALYSIS)
    u_out = core.build_place_group_user(
        SAMPLE_FIXTURE_ANALYSIS, {"style_rules": "x"}, "outdoor", groups["outdoor"])
    assert "visible_adjacent_enclosed_masses" in u_out  # 반대그룹(실내) 매스 전달
    u_whole = core.build_place_group_user(
        SAMPLE_FIXTURE_ANALYSIS, {}, "whole", SAMPLE_FIXTURE_ANALYSIS["spaces"])
    assert "visible_adjacent_enclosed_masses" not in u_whole


def test_build_fp_user_frame_only_excludes_addon():
    groups = core.enclosure_groups(SAMPLE_FIXTURE_ANALYSIS)
    u = core.build_fp_user(SAMPLE_FIXTURE_ANALYSIS, groups["indoor"], "indoor", frame_only=True)
    assert "small table" not in u                     # frame only — addon 미포함
    u2 = core.build_fp_user(SAMPLE_FIXTURE_ANALYSIS, groups["indoor"], "indoor", frame_only=False)
    assert "small table" in u2
    # 반대그룹(옥외 open_air)은 enclosed 매스가 아니라 adjacent masses 없음
    assert "adjacent_enclosed_masses_outside_this_group" not in u
    u3 = core.build_fp_user(SAMPLE_FIXTURE_ANALYSIS, groups["outdoor"], "outdoor", frame_only=True)
    assert "adjacent_enclosed_masses_outside_this_group" in u3


def test_analyze_user_keeps_inputs_unabridged():
    blocks = "BLOCK" * 1000
    full = "FULL" * 5000
    u = core.build_analyze_user(blocks, full, [1, 2])
    assert blocks in u and full in u                  # ★무삭제 (CLAUDE.md)


# ──────────────────────── Phase 2: shot→space 배정 조인 (순수) ────────────────────────
SAMPLE_FIXTURE_PLATES = {
    "hub room": {"status": "ok", "kind": "marked_indoor", "png": "bg_hub_room.png", "key": "hub_room", "num": 1},
    "side room A": {"status": "ok", "kind": "marked_indoor", "png": "bg_side_room_a.png", "key": "side_room_a", "num": 2},
    "open deck": {"status": "ok", "kind": "outdoor_t2i", "png": "bg_open_deck.png", "key": "open_deck", "num": 3},
    "broken room": {"status": "error", "error": "render failed"},
}


def test_build_shot_plate_map_join_ok():
    assignments = [
        {"scene": 1, "shot": 1, "space": "hub room", "basis": "table seen"},
        {"scene": 2, "shot": 3, "space": "open deck", "basis": "sky visible",
         "secondary_spaces": ["hub room"]},
    ]
    spm, diags = core.build_shot_plate_map(assignments, SAMPLE_FIXTURE_PLATES)
    assert set(spm.keys()) == {"1_1", "2_3"}
    e = spm["1_1"]
    assert e["space"] == "hub room" and e["plate_key"] == "hub_room"
    assert e["plate_png"] == "bg_hub_room.png" and e["plate_kind"] == "marked_indoor"
    assert e["shot_id"] == "S1_Shot1" and e["basis"] == "table seen"
    # 옥외 T2I plate 도 동일 contract (Codex ④ 합의)
    assert spm["2_3"]["plate_kind"] == "outdoor_t2i"
    # secondary 는 diagnostic 보존만 (multi-ref 금지 — Codex caveat)
    assert spm["2_3"]["secondary_spaces"] == ["hub room"]
    assert diags == []


def test_build_shot_plate_map_diagnostics_nonfatal():
    assignments = [
        {"scene": 1, "shot": 1, "space": None, "basis": "uncertain"},          # null → 미배정
        {"scene": 1, "shot": 2, "space": "no such space", "basis": "x"},       # unknown space
        {"scene": 1, "shot": 3, "space": "broken room", "basis": "x"},         # plate not-ok
        {"scene": None, "shot": 1, "space": "hub room", "basis": "x"},         # invalid index
        {"scene": 2, "shot": 1, "space": "hub room", "basis": "first"},
        {"scene": 2, "shot": 1, "space": "open deck", "basis": "dup"},         # 중복 → 첫 배정 보존
        {"scene": 3, "shot": 1, "space": "entry pocket", "basis": "x"},        # connector → plate 의도 생략
    ]
    spm, diags = core.build_shot_plate_map(
        assignments, SAMPLE_FIXTURE_PLATES, connector_names={"entry pocket"})
    assert set(spm.keys()) == {"2_1"}
    assert spm["2_1"]["space"] == "hub room"
    reasons = {d["reason"] for d in diags}
    # connector 배정은 LLM 발명(unknown_space)과 구분 — 운영 디버깅용 (실 canary 실측)
    assert reasons == {"unassigned", "unknown_space", "plate_not_ok", "invalid_index",
                       "duplicate_shot", "connector_no_plate"}


def test_build_shot_assign_user_keeps_inputs_unabridged():
    rows = [
        {"scene": 1, "shot": 1, "heading": "H1", "summary": "S" * 3000, "shot_desc": "D" * 2000},
        {"scene": 2, "shot": 4, "heading": "H2", "summary": "scene two", "shot_desc": "walks in"},
    ]
    u = core.build_shot_assign_user(SAMPLE_FIXTURE_ANALYSIS, rows)
    assert ("S" * 3000) in u and ("D" * 2000) in u    # ★무삭제 (CLAUDE.md)
    assert "scene 2" in u and "shot 4" in u           # 조인 키가 되는 인덱스 명시
    assert "hub room" in u and "open deck" in u       # 후보 공간 목록 전달


# ──────────────────────── Phase 3: plate_action derive (순수 조인 + 프롬프트) ────────────────────────
def test_shot_assign_sys_declares_plate_action():
    """배정 LLM 출력 계약에 plate_action enum + derive_instruction 이 선언돼야 한다."""
    s = core.SHOT_ASSIGN_SYS
    assert "plate_action" in s
    assert "reuse_base" in s and "derive_from_base" in s and "no_plate" in s
    assert "derive_instruction" in s
    # derive_instruction 은 카메라 이동만 — 인물/사건 금지가 프롬프트에 명시
    assert "카메라" in s


def test_build_shot_plate_map_default_reuse_base():
    """plate_action 미지정(구 LLM 출력 호환) → reuse_base 기본, derive 필드 없음."""
    assignments = [{"scene": 1, "shot": 1, "space": "hub room", "basis": "b"}]
    spm, diags = core.build_shot_plate_map(assignments, SAMPLE_FIXTURE_PLATES)
    e = spm["1_1"]
    assert e["plate_action"] == "reuse_base"
    assert "derive_instruction" not in e and "canonical_plate_key" not in e
    assert diags == []


def test_build_shot_plate_map_derive_join():
    """derive_from_base 배정 = 결정론 조인만 — source/instruction/basis 기록,
    plate_png 는 조인 시점엔 canonical 그대로(step 이 derive 성공 후 교체),
    canonical plates 입력은 불변."""
    plates_before = json.loads(json.dumps(SAMPLE_FIXTURE_PLATES))
    assignments = [
        {"scene": 1, "shot": 5, "space": "hub room", "basis": "window detail",
         "plate_action": "derive_from_base",
         "derive_instruction": "camera now stands one meter in front of the window"},
        {"scene": 2, "shot": 1, "space": "open deck", "basis": "wide",
         "plate_action": "reuse_base"},
    ]
    spm, diags = core.build_shot_plate_map(assignments, SAMPLE_FIXTURE_PLATES)
    assert diags == []
    e = spm["1_5"]
    assert e["plate_action"] == "derive_from_base"
    assert e["derive_instruction"] == "camera now stands one meter in front of the window"
    assert e["canonical_plate_key"] == "hub_room"
    assert e["canonical_plate_png"] == "bg_hub_room.png"
    assert e["plate_png"] == "bg_hub_room.png"      # 조인 시점 = canonical (fallback 안전)
    assert e["basis"] == "window detail"
    assert spm["2_1"]["plate_action"] == "reuse_base"
    # ★canonical plates 불변 (조인은 읽기 전용)
    assert SAMPLE_FIXTURE_PLATES == plates_before


def test_build_shot_plate_map_derive_without_instruction_demotes():
    """얇은 gate: derive_from_base 인데 derive_instruction 누락 → reuse_base 강등 + 진단."""
    assignments = [{"scene": 1, "shot": 1, "space": "hub room", "basis": "b",
                    "plate_action": "derive_from_base"}]
    spm, diags = core.build_shot_plate_map(assignments, SAMPLE_FIXTURE_PLATES)
    e = spm["1_1"]
    assert e["plate_action"] == "reuse_base" and "derive_instruction" not in e
    assert any(d["reason"] == "derive_missing_instruction" for d in diags)


def test_build_shot_plate_map_no_plate_action():
    """no_plate = LLM 이 plate 참조 부적합 판정 → spm 제외 + 진단(기존 경로 유지)."""
    assignments = [{"scene": 1, "shot": 1, "space": "hub room", "basis": "extreme close-up",
                    "plate_action": "no_plate"}]
    spm, diags = core.build_shot_plate_map(assignments, SAMPLE_FIXTURE_PLATES)
    assert spm == {}
    assert diags == [{"reason": "plate_action_no_plate", "scene": 1, "shot": 1,
                      "space": "hub room", "basis": "extreme close-up"}]


def test_build_shot_plate_map_unknown_plate_action_demotes():
    """enum 밖 값 = LLM 발명 → reuse_base 강등 + 진단 (조인은 비치명 유지)."""
    assignments = [{"scene": 1, "shot": 1, "space": "hub room", "basis": "b",
                    "plate_action": "SAMPLE_FIXTURE_bogus"}]
    spm, diags = core.build_shot_plate_map(assignments, SAMPLE_FIXTURE_PLATES)
    assert spm["1_1"]["plate_action"] == "reuse_base"
    assert any(d["reason"] == "unknown_plate_action" for d in diags)


def test_build_derive_plate_prompt_contract():
    """derive i2i 프롬프트 = 배경 전용: 정체성 유지 + 카메라만 이동 + 인물/사건/스토리 소품 금지
    + 구조 발명 금지 + 입력 무삭제 (검증된 ~/tmp/geum_s10sh5_derive.py 패턴의 generic 화)."""
    instr = "camera now stands one meter in front of the window"
    p = core.build_derive_plate_prompt(instr)
    assert instr in p
    low = p.lower()
    assert "exactly identical" in low                 # 재질·구조·조명 정체성 유지
    assert "only the camera" in low                   # 카메라만 이동
    assert "no people" in low and "no story events" in low and "no story props" in low
    assert "do not invent" in low                     # 구조 발명 금지
    assert "do not remove" in low                     # 입력 무삭제
    assert "background plate" in low


# ──────────────────────── mark_fp (이미지 처리) ────────────────────────
def test_mark_fp_single_red_marker(tmp_path: Path):
    from PIL import Image
    src = tmp_path / "fp.png"
    Image.new("RGB", (200, 100), "white").save(src)
    out = tmp_path / "marked.png"
    core.mark_fp(str(src), str(out), 0.25, 0.5, 7)
    im = Image.open(out).convert("RGB")
    reds = [(x, y) for x in range(200) for y in range(100)
            if im.getpixel((x, y))[0] > 200 and im.getpixel((x, y))[1] < 80]
    assert reds, "빨간 마커가 그려져야 한다"
    xs = [p[0] for p in reds]
    assert max(xs) < 110, "마킹은 지정 위치(왼쪽 1/4) 주변에만 — 도면 전체에 흩지 않는다"


# ──────────────────────── step: flag 게이트 + override 전체 경로 ────────────────────────
def _mk_step(tmp_path, monkeypatch, enabled: bool):
    from app.core.config import settings
    from app.core.steps.space_set_bg_step import SpaceSetBgStep
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path), raising=False)
    monkeypatch.setattr(settings, "space_set_bg_enabled", enabled, raising=False)
    step = SpaceSetBgStep.__new__(SpaceSetBgStep)   # StepRunner.__init__ 우회 (gate/manifest 무관 단위검증)
    step.step_id = "space_set_bg"
    step.project_id = "p1"
    step.episode_id = "e1"
    step.db = MagicMock()
    step._cp_dir = tmp_path / "p1" / "checkpoints" / "episodes" / "e1" / "space_set_bg"
    return step


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({"data": data}, ensure_ascii=False), encoding="utf-8")


def test_step_flag_off_not_applicable(tmp_path, monkeypatch):
    step = _mk_step(tmp_path, monkeypatch, enabled=False)
    # ★check_applicability override — OFF 면 StepRunner 가 not_applicable 로 마킹 (Codex 리뷰 #2)
    assert step.check_applicability() is False
    out = step._execute()
    assert out["applicable_count"] == 0 and out["data"] == {}


def test_step_full_path_with_overrides(tmp_path, monkeypatch):
    from PIL import Image
    step = _mk_step(tmp_path, monkeypatch, enabled=True)
    # 상류 checkpoint fixture (generic)
    _write_cp(tmp_path, "background_master_plan", {"plans": {
        "G1": {"status": "ok", "plan": {"backgrounds": [{"bg_id": "B1", "loc_id": "L1"}]}},
    }})
    _write_cp(tmp_path, "scene_save", {"segments": [
        {"scene_index": 1, "heading": "INT PLACE", "text": "scene one text"},
        {"scene_index": 2, "heading": "INT PLACE NIGHT", "text": "scene two text"},
    ]})
    # shot 에 location_id 없음 → scene_director primary_location fallback 경로 검증 (canary 실측 결함)
    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "description": "a shot"}]},
        {"scene_index": 2, "shots": [{"shot_index": 1, "description": "another shot"}]},
    ]})
    _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, "scene_director", {"scenes": [
        {"scene_index": 1, "primary_location": "L1"},
        {"scene_index": 2, "primary_location": "L1"},
    ]})
    _write_cp(tmp_path, "world_guide", {"guide": {"style_rules": "generic style"}})
    monkeypatch.setattr(step, "_load_fulltext", lambda: "full screenplay text")

    analysis = json.loads(json.dumps(SAMPLE_FIXTURE_ANALYSIS))

    def fake_text(*, system, user, max_tokens):
        if system is core.ANALYZE_FA_SYS:
            return json.dumps(analysis, ensure_ascii=False)
        if system is core.VIEW_BRIEF_SYS:
            return json.dumps(SAMPLE_FIXTURE_BRIEFS, ensure_ascii=False)
        if system is core.OUTDOOR_BG_T2I_SYS:
            return "outdoor t2i prompt"
        if system is core.SHOT_ASSIGN_SYS:
            return json.dumps({"assignments": [
                {"scene": 1, "shot": 1, "space": "hub room", "basis": "shot text"},
                {"scene": 2, "shot": 1, "space": "open deck", "basis": "shot text"},
            ]})
        return "a place description"               # PLACE_GROUP / FRAME_FP

    def fake_vision(*, system, user, image_paths, max_tokens):
        if system is core.SPACE_POS_SYS:
            return json.dumps({"positions": [
                {"name": "hub room", "cx": 0.3, "cy": 0.5},
                {"name": "side room A", "cx": 0.7, "cy": 0.5},
            ]})
        return json.dumps({"frame_preserved": True, "verdict": "ok"})

    def fake_img(*, prompt=None, out_path=None, base_image_path=None):
        Image.new("RGB", (160, 100), "white").save(out_path)

    step.set_overrides_for_testing(text=fake_text, vision=fake_vision, t2i=fake_img, i2i=fake_img)
    out = step._execute()
    assert out["applicable_count"] == 1 and out["failed_count"] == 0
    g = out["data"]["groups"]["G1"]
    plates = g["plates"]
    # threshold connector 는 plate 없음 / 나머지 4공간 plate
    assert "entry pocket" not in plates and len(plates) == 4
    assert plates["open deck"]["kind"] == "outdoor_t2i"
    assert plates["hub room"]["kind"] == "marked_indoor"
    assert plates["side room B"]["kind"] == "samespace_ref"
    assert all(v["status"] == "ok" for v in plates.values())
    adir = Path(g["assets_dir"])
    # 최종 2D FP 1장 + 공간당 단일 마킹 사본 (hub/side A 2장)
    assert (adir / "fp_final_2d_indoor.png").exists()
    assert len(list(adir.glob("fp_marked_*.png"))) == 2
    assert len(list(adir.glob("bg_*.png"))) == 4
    # frame_check 진단 기록
    assert g["frame_checks"]["indoor"]["frame_preserved"] is True
    # Phase 2: shot→space 배정이 manifest 에 deterministic 조인으로 고정 (Codex B 합의)
    spm = g["shot_plate_map"]
    assert spm["1_1"]["space"] == "hub room" and spm["1_1"]["plate_png"] == plates["hub room"]["png"]
    assert spm["2_1"]["space"] == "open deck" and spm["2_1"]["plate_kind"] == "outdoor_t2i"
    assert g["shot_assign_diagnostics"] == []
    # 프롬프트는 팩(파일)에서 온다 = 실행 입력 — 기록이 ★어느 판을 소비했는지★
    # 대변해야 한다. config_hash 는 달라진 것만 알려주고 무엇이었는지는 못 준다.
    pack = out["data"]["prompt_pack"]
    assert pack["module"] == core.PROMPT_PACK_MODULE
    assert set(pack["stems"]) == set(core.PACK_STEM_NAMES)
    assert all(v["version"] and v["sha256_16"] for v in pack["stems"].values())


def test_step_shot_assign_failure_nonfatal(tmp_path, monkeypatch):
    """배정 LLM 호출 실패는 plate 산출을 죽이지 않는다 (thick gate 금지 — Codex 합의)."""
    from PIL import Image
    step = _mk_step(tmp_path, monkeypatch, enabled=True)
    _write_cp(tmp_path, "background_master_plan", {"plans": {
        "G1": {"status": "ok", "plan": {"backgrounds": [{"bg_id": "B1", "loc_id": "L1"}]}},
    }})
    _write_cp(tmp_path, "scene_save", {"segments": [
        {"scene_index": 1, "heading": "INT PLACE", "text": "t"},
        {"scene_index": 2, "heading": "INT PLACE 2", "text": "t2"},
    ]})
    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "description": "d", "location_id": "L1"}]},
        {"scene_index": 2, "shots": [{"shot_index": 1, "description": "d2", "location_id": "L1"}]},
    ]})
    _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, "world_guide", {"guide": {}})
    monkeypatch.setattr(step, "_load_fulltext", lambda: "x")

    analysis = json.loads(json.dumps(SAMPLE_FIXTURE_ANALYSIS))

    def fake_text(*, system, user, max_tokens):
        if system is core.ANALYZE_FA_SYS:
            return json.dumps(analysis, ensure_ascii=False)
        if system is core.VIEW_BRIEF_SYS:
            return json.dumps(SAMPLE_FIXTURE_BRIEFS, ensure_ascii=False)
        if system is core.OUTDOOR_BG_T2I_SYS:
            return "outdoor t2i prompt"
        if system is core.SHOT_ASSIGN_SYS:
            raise RuntimeError("assign provider down")
        return "a place description"

    def fake_vision(*, system, user, image_paths, max_tokens):
        if system is core.SPACE_POS_SYS:
            return json.dumps({"positions": [
                {"name": "hub room", "cx": 0.3, "cy": 0.5},
                {"name": "side room A", "cx": 0.7, "cy": 0.5},
            ]})
        return json.dumps({"frame_preserved": True, "verdict": "ok"})

    def fake_img(*, prompt=None, out_path=None, base_image_path=None):
        Image.new("RGB", (160, 100), "white").save(out_path)

    step.set_overrides_for_testing(text=fake_text, vision=fake_vision, t2i=fake_img, i2i=fake_img)
    out = step._execute()
    assert out["failed_count"] == 0
    g = out["data"]["groups"]["G1"]
    assert g["status"] == "ok" and len(g["plates"]) == 4   # plate 산출 무사
    assert g["shot_plate_map"] == {}
    assert any(d["reason"] == "assign_call_failed" for d in g["shot_assign_diagnostics"])


def test_step_single_indoor_space_skips_fp(tmp_path, monkeypatch):
    """단일 marked_indoor 그룹 = FP 미경유 (frame T2I/addon i2i/마킹 0) + 단일 T2I plate.
    어선 조타실 canary 실측 결함(FP 도트 그리드→벽지 직역) 차단."""
    from PIL import Image
    step = _mk_step(tmp_path, monkeypatch, enabled=True)
    _write_cp(tmp_path, "background_master_plan", {"plans": {
        "G1": {"status": "ok", "plan": {"backgrounds": [{"bg_id": "B1", "loc_id": "L1"}]}},
    }})
    _write_cp(tmp_path, "scene_save", {"segments": [
        {"scene_index": 1, "heading": "INT CABIN", "text": "t"},
        {"scene_index": 2, "heading": "EXT DECK", "text": "t2"},
    ]})
    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "description": "inside", "location_id": "L1"}]},
        {"scene_index": 2, "shots": [{"shot_index": 1, "description": "on deck", "location_id": "L1"}]},
    ]})
    _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, "world_guide", {"guide": {}})
    monkeypatch.setattr(step, "_load_fulltext", lambda: "x")

    analysis = {
        "space_type": "SAMPLE_FIXTURE_vessel", "genre_or_setting": "SAMPLE_FIXTURE",
        "physical_realism": ["cramped"],
        "spaces": [_space("steering cabin"), _space("work deck", enclosure="open_air")],
        "structure_kinds": [], "access_validation": [],
        "uncertain_structure_items": [], "excluded_items_with_reason": [],
    }
    briefs = {"steering cabin": "wide cabin", "work deck": "wide deck"}

    def fake_text(*, system, user, max_tokens):
        if system is core.ANALYZE_FA_SYS:
            return json.dumps(analysis, ensure_ascii=False)
        if system is core.VIEW_BRIEF_SYS:
            return json.dumps(briefs, ensure_ascii=False)
        if system is core.FRAME_FP_SYS:
            raise AssertionError("단일 marked_indoor 그룹은 frame FP 프롬프트를 만들면 안 된다")
        if system is core.SHOT_ASSIGN_SYS:
            return json.dumps({"assignments": [
                {"scene": 1, "shot": 1, "space": "steering cabin", "basis": "b"},
                {"scene": 2, "shot": 1, "space": "work deck", "basis": "b"},
            ]})
        return "a t2i prompt or description"          # PLACE_GROUP / INDOOR_BG / OUTDOOR_BG

    def fake_vision(*, system, user, image_paths, max_tokens):
        raise AssertionError("FP 미경유면 VLM(frame_check/positions) 호출이 없어야 한다")

    calls = {"t2i": 0, "i2i": 0}

    def fake_t2i(*, prompt=None, out_path=None):
        calls["t2i"] += 1
        Image.new("RGB", (160, 100), "white").save(out_path)

    def fake_i2i(*, base_image_path=None, prompt=None, out_path=None):
        calls["i2i"] += 1
        Image.new("RGB", (160, 100), "white").save(out_path)

    step.set_overrides_for_testing(text=fake_text, vision=fake_vision, t2i=fake_t2i, i2i=fake_i2i)
    out = step._execute()
    assert out["failed_count"] == 0 and out["applicable_count"] == 1
    g = out["data"]["groups"]["G1"]
    plates = g["plates"]
    assert plates["steering cabin"]["kind"] == "indoor_t2i"
    assert plates["work deck"]["kind"] == "outdoor_t2i"
    adir = Path(g["assets_dir"])
    assert not list(adir.glob("fp_*.png"))            # FP/마킹 산출물 0
    assert calls["i2i"] == 0 and calls["t2i"] == 2    # 공간당 단일 T2I 만
    assert g["frame_checks"] == {}
    # shot_plate_map 도 indoor_t2i plate 로 정상 조인
    assert g["shot_plate_map"]["1_1"]["plate_kind"] == "indoor_t2i"


def _derive_fixture_step(tmp_path, monkeypatch):
    """derive 경로 공용 fixture — hub room derive 1샷 + open deck reuse 1샷."""
    step = _mk_step(tmp_path, monkeypatch, enabled=True)
    _write_cp(tmp_path, "background_master_plan", {"plans": {
        "G1": {"status": "ok", "plan": {"backgrounds": [{"bg_id": "B1", "loc_id": "L1"}]}},
    }})
    _write_cp(tmp_path, "scene_save", {"segments": [
        {"scene_index": 1, "heading": "INT PLACE", "text": "scene one text"},
        {"scene_index": 2, "heading": "EXT PLACE", "text": "scene two text"},
    ]})
    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 5, "description": "very close to the window", "location_id": "L1"}]},
        {"scene_index": 2, "shots": [{"shot_index": 1, "description": "wide on deck", "location_id": "L1"}]},
    ]})
    _write_cp(tmp_path, "shot_selection", {"scenes": [
        {"scene_index": 1, "selected_shot_indices": [5]},
        {"scene_index": 2, "selected_shot_indices": [1]},
    ]})
    _write_cp(tmp_path, "world_guide", {"guide": {}})
    monkeypatch.setattr(step, "_load_fulltext", lambda: "x")

    analysis = json.loads(json.dumps(SAMPLE_FIXTURE_ANALYSIS))

    def fake_text(*, system, user, max_tokens):
        if system is core.ANALYZE_FA_SYS:
            return json.dumps(analysis, ensure_ascii=False)
        if system is core.VIEW_BRIEF_SYS:
            return json.dumps(SAMPLE_FIXTURE_BRIEFS, ensure_ascii=False)
        if system is core.OUTDOOR_BG_T2I_SYS:
            return "outdoor t2i prompt"
        if system is core.SHOT_ASSIGN_SYS:
            return json.dumps({"assignments": [
                {"scene": 1, "shot": 5, "space": "hub room", "basis": "window detail",
                 "plate_action": "derive_from_base",
                 "derive_instruction": "camera now stands one meter in front of the window"},
                {"scene": 2, "shot": 1, "space": "open deck", "basis": "wide",
                 "plate_action": "reuse_base"},
            ]})
        return "a place description"

    def fake_vision(*, system, user, image_paths, max_tokens):
        if system is core.SPACE_POS_SYS:
            return json.dumps({"positions": [
                {"name": "hub room", "cx": 0.3, "cy": 0.5},
                {"name": "side room A", "cx": 0.7, "cy": 0.5},
            ]})
        return json.dumps({"frame_preserved": True, "verdict": "ok"})

    return step, fake_text, fake_vision


def test_step_derive_path_generates_derived_plate(tmp_path, monkeypatch):
    """derive_from_base 샷 = canonical plate 를 base 로 배경 전용 i2i →
    spm.plate_png 가 derived png 로 교체(+canonical_plate_key 보존),
    derived_plates additive 저장. reuse_base 샷은 canonical 그대로 ★불변★."""
    from PIL import Image
    step, fake_text, fake_vision = _derive_fixture_step(tmp_path, monkeypatch)

    derive_calls = []

    def fake_img(*, prompt=None, out_path=None, base_image_path=None):
        if base_image_path is not None and "ONLY the CAMERA moves" in (prompt or ""):
            derive_calls.append({"base": base_image_path, "out": out_path, "prompt": prompt})
        Image.new("RGB", (160, 100), "white").save(out_path)

    step.set_overrides_for_testing(text=fake_text, vision=fake_vision, t2i=fake_img, i2i=fake_img)
    out = step._execute()
    assert out["failed_count"] == 0
    g = out["data"]["groups"]["G1"]
    plates = g["plates"]
    spm = g["shot_plate_map"]

    # derive 샷: derived png 로 교체 + canonical 출처 보존
    e = spm["1_5"]
    assert e["plate_action"] == "derive_from_base"
    assert e["plate_png"] == "bg_derived_1_5.png"
    assert e["canonical_plate_key"] == plates["hub room"]["key"]
    assert e["canonical_plate_png"] == plates["hub room"]["png"]
    adir = Path(g["assets_dir"])
    assert (adir / "bg_derived_1_5.png").exists()
    # derive i2i 는 canonical plate 를 base 로, instruction 이 프롬프트에 주입
    assert len(derive_calls) == 1
    assert derive_calls[0]["base"] == str(adir / plates["hub room"]["png"])
    assert "one meter in front of the window" in derive_calls[0]["prompt"]
    # derived_plates additive 저장
    dp = g["derived_plates"]["1_5"]
    assert dp["status"] == "ok" and dp["png"] == "bg_derived_1_5.png"
    assert dp["canonical_plate_key"] == plates["hub room"]["key"]
    # ★reuse_base 샷 불변 (acceptance 기준)
    assert spm["2_1"]["plate_action"] == "reuse_base"
    assert spm["2_1"]["plate_png"] == plates["open deck"]["png"]
    assert "2_1" not in g["derived_plates"]
    # ★canonical plates 자체 불변 — derive 는 additive
    assert plates["hub room"]["png"] == "bg_hub_room.png"
    assert g["shot_assign_diagnostics"] == []


def test_step_derive_failure_falls_back_to_canonical(tmp_path, monkeypatch):
    """derive i2i 실패 = non-blocking — spm 은 canonical plate 로 fallback,
    derive_failed diagnostic 크게 남기고 그룹 status 는 ok 유지."""
    from PIL import Image
    step, fake_text, fake_vision = _derive_fixture_step(tmp_path, monkeypatch)

    def fake_img(*, prompt=None, out_path=None, base_image_path=None):
        if base_image_path is not None and "ONLY the CAMERA moves" in (prompt or ""):
            raise RuntimeError("image provider rejected")
        Image.new("RGB", (160, 100), "white").save(out_path)

    step.set_overrides_for_testing(text=fake_text, vision=fake_vision, t2i=fake_img, i2i=fake_img)
    out = step._execute()
    assert out["failed_count"] == 0
    g = out["data"]["groups"]["G1"]
    assert g["status"] == "ok"
    e = g["shot_plate_map"]["1_5"]
    # canonical fallback — plate_png 는 canonical 그대로, 실패 마크
    assert e["plate_png"] == g["plates"]["hub room"]["png"]
    assert e["derive_failed"] is True
    dp = g["derived_plates"]["1_5"]
    assert dp["status"] == "error" and "image provider rejected" in dp["error"]
    diag = [d for d in g["shot_assign_diagnostics"] if d["reason"] == "derive_failed"]
    assert len(diag) == 1
    assert diag[0]["scene"] == 1 and diag[0]["shot"] == 5 and diag[0]["space"] == "hub room"


def test_step_group_error_isolated(tmp_path, monkeypatch):
    """한 그룹 실패가 step 전체를 죽이지 않는다 (failed_count 로 격리)."""
    step = _mk_step(tmp_path, monkeypatch, enabled=True)
    _write_cp(tmp_path, "background_master_plan", {"plans": {
        "G1": {"status": "ok", "plan": {"backgrounds": [{"bg_id": "B1", "loc_id": "L1"}]}},
    }})
    _write_cp(tmp_path, "scene_save", {"segments": [
        {"scene_index": 1, "heading": "INT PLACE", "text": "t"},
        {"scene_index": 2, "heading": "INT PLACE 2", "text": "t2"},
    ]})
    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "description": "d", "location_id": "L1"}]},
        {"scene_index": 2, "shots": [{"shot_index": 1, "description": "d2", "location_id": "L1"}]},
    ]})
    _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, "world_guide", {"guide": {}})
    monkeypatch.setattr(step, "_load_fulltext", lambda: "x")

    def boom(**kw):
        raise RuntimeError("provider down")

    step.set_overrides_for_testing(text=boom, vision=boom, t2i=boom, i2i=boom)
    out = step._execute()
    assert out["applicable_count"] == 1 and out["failed_count"] == 1
    assert out["data"]["groups"]["G1"]["status"] == "error"


def test_step_low_frequency_group_skipped(tmp_path, monkeypatch):
    """★1번 정도 나타나는 배경은 기준 BG 생략 (사용자 2026-06-10 규칙)."""
    step = _mk_step(tmp_path, monkeypatch, enabled=True)
    _write_cp(tmp_path, "background_master_plan", {"plans": {
        "G1": {"status": "ok", "plan": {"backgrounds": [{"bg_id": "B1", "loc_id": "L1"}]}},
    }})
    _write_cp(tmp_path, "scene_save", {"segments": [
        {"scene_index": 1, "heading": "INT PLACE", "text": "only once"},
    ]})
    _write_cp(tmp_path, "shot_validator", {"scenes": [
        {"scene_index": 1, "shots": [{"shot_index": 1, "description": "d", "location_id": "L1"}]},
    ]})
    _write_cp(tmp_path, "shot_selection", {"scenes": [{"scene_index": 1, "selected_shot_indices": [1]}]})
    _write_cp(tmp_path, "world_guide", {"guide": {}})
    monkeypatch.setattr(step, "_load_fulltext", lambda: "x")

    def no_call(**kw):
        raise AssertionError("low-frequency group must not call any provider")

    step.set_overrides_for_testing(text=no_call, vision=no_call, t2i=no_call, i2i=no_call)
    out = step._execute()
    assert out["applicable_count"] == 0
    assert out["data"]["groups"]["G1"]["status"] == "skipped_low_frequency"